feat: 资产 CRUD API + 统计接口 + Vue3 管理前端（可选 API Key 鉴权）
This commit is contained in:
gouki
@@ -0,0 +1,246 @@
|
||||
"""资产业务逻辑：CRUD + 统计聚合
|
||||
|
||||
统一处理 Asset 主表与其一对一详情表（VPSDetail/DomainDetail/AIAccount）的联动。
|
||||
"""
|
||||
|
||||
from datetime import date
|
||||
from typing import List, Optional
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.models.asset import (
|
||||
AIAccount,
|
||||
Asset,
|
||||
AssetStatus,
|
||||
AssetType,
|
||||
DomainDetail,
|
||||
VPSDetail,
|
||||
)
|
||||
from app.schemas.asset import (
|
||||
AIAccountRead,
|
||||
AssetCreate,
|
||||
AssetRead,
|
||||
AssetUpdate,
|
||||
DomainDetailRead,
|
||||
VPSDetailRead,
|
||||
)
|
||||
|
||||
# 资产类型 -> (AssetCreate/Update 中的字段名, 详情表模型)
|
||||
DETAIL_MAP = {
|
||||
AssetType.VPS: ("vps_detail", VPSDetail),
|
||||
AssetType.DOMAIN: ("domain_detail", DomainDetail),
|
||||
AssetType.AI_AGENT: ("ai_detail", AIAccount),
|
||||
}
|
||||
|
||||
# 允许排序的字段白名单
|
||||
SORTABLE_FIELDS = {"expiry_date", "name", "created_at", "cost", "updated_at"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 辅助函数
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _days_to_expiry(expiry_date: Optional[date]) -> Optional[int]:
|
||||
"""计算距到期天数（已过期为负数），无到期日返回 None"""
|
||||
if not expiry_date:
|
||||
return None
|
||||
return (expiry_date - date.today()).days
|
||||
|
||||
|
||||
def _get_detail(session: Session, asset: Asset):
|
||||
"""根据资产类型读取对应详情记录"""
|
||||
item = DETAIL_MAP.get(asset.asset_type)
|
||||
if not item:
|
||||
return None
|
||||
_, model = item
|
||||
return session.exec(select(model).where(model.asset_id == asset.id)).first()
|
||||
|
||||
|
||||
def _to_read(asset: Asset, detail) -> AssetRead:
|
||||
"""组装 AssetRead 输出（主表 + 详情 + 计算字段）"""
|
||||
read = AssetRead.model_validate(asset)
|
||||
read.days_to_expiry = _days_to_expiry(asset.expiry_date)
|
||||
if isinstance(detail, VPSDetail):
|
||||
read.vps_detail = VPSDetailRead.model_validate(detail)
|
||||
elif isinstance(detail, DomainDetail):
|
||||
read.domain_detail = DomainDetailRead.model_validate(detail)
|
||||
elif isinstance(detail, AIAccount):
|
||||
read.ai_detail = AIAccountRead.model_validate(detail)
|
||||
return read
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CRUD
|
||||
# --------------------------------------------------------------------------- #
|
||||
def create_asset(session: Session, data: AssetCreate) -> AssetRead:
|
||||
"""创建资产及其详情"""
|
||||
item = DETAIL_MAP.get(data.asset_type)
|
||||
detail_in = None
|
||||
model = None
|
||||
if item:
|
||||
field, model = item
|
||||
detail_in = getattr(data, field)
|
||||
if detail_in is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"asset_type={data.asset_type.value} 需要提供 {field}",
|
||||
)
|
||||
|
||||
asset_data = data.model_dump(
|
||||
exclude={"vps_detail", "domain_detail", "ai_detail"}
|
||||
)
|
||||
asset = Asset(**asset_data)
|
||||
session.add(asset)
|
||||
session.commit()
|
||||
session.refresh(asset)
|
||||
|
||||
detail = None
|
||||
if model is not None and detail_in is not None:
|
||||
detail = model(asset_id=asset.id, **detail_in.model_dump())
|
||||
session.add(detail)
|
||||
session.commit()
|
||||
session.refresh(detail)
|
||||
|
||||
return _to_read(asset, detail)
|
||||
|
||||
|
||||
def get_asset(session: Session, asset_id: int) -> AssetRead:
|
||||
"""读取单个资产（含详情）"""
|
||||
asset = session.get(Asset, asset_id)
|
||||
if not asset:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="资产不存在")
|
||||
return _to_read(asset, _get_detail(session, asset))
|
||||
|
||||
|
||||
def update_asset(session: Session, asset_id: int, data: AssetUpdate) -> AssetRead:
|
||||
"""更新资产主表及详情（仅更新传入字段）"""
|
||||
asset = session.get(Asset, asset_id)
|
||||
if not asset:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="资产不存在")
|
||||
|
||||
main_fields = data.model_dump(
|
||||
exclude_unset=True, exclude={"vps_detail", "domain_detail", "ai_detail"}
|
||||
)
|
||||
for key, value in main_fields.items():
|
||||
setattr(asset, key, value)
|
||||
session.add(asset)
|
||||
|
||||
# 详情表：以更新后的 asset_type 为准
|
||||
detail = None
|
||||
item = DETAIL_MAP.get(asset.asset_type)
|
||||
if item:
|
||||
field, model = item
|
||||
detail_in = getattr(data, field)
|
||||
existing = session.exec(
|
||||
select(model).where(model.asset_id == asset.id)
|
||||
).first()
|
||||
if detail_in is not None:
|
||||
if existing:
|
||||
for key, value in detail_in.model_dump().items():
|
||||
setattr(existing, key, value)
|
||||
session.add(existing)
|
||||
detail = existing
|
||||
else:
|
||||
detail = model(asset_id=asset.id, **detail_in.model_dump())
|
||||
session.add(detail)
|
||||
else:
|
||||
detail = existing
|
||||
|
||||
session.commit()
|
||||
session.refresh(asset)
|
||||
if detail is not None:
|
||||
session.refresh(detail)
|
||||
return _to_read(asset, detail)
|
||||
|
||||
|
||||
def delete_asset(session: Session, asset_id: int) -> None:
|
||||
"""删除资产及其详情"""
|
||||
asset = session.get(Asset, asset_id)
|
||||
if not asset:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="资产不存在")
|
||||
detail = _get_detail(session, asset)
|
||||
if detail is not None:
|
||||
session.delete(detail)
|
||||
session.delete(asset)
|
||||
session.commit()
|
||||
|
||||
|
||||
def list_assets(
|
||||
session: Session,
|
||||
asset_type: Optional[AssetType] = None,
|
||||
asset_status: Optional[AssetStatus] = None,
|
||||
is_archived: Optional[bool] = None,
|
||||
q: Optional[str] = None,
|
||||
sort: str = "expiry_date",
|
||||
order: str = "asc",
|
||||
) -> List[AssetRead]:
|
||||
"""资产列表（筛选 + 搜索 + 排序）"""
|
||||
stmt = select(Asset)
|
||||
if asset_type is not None:
|
||||
stmt = stmt.where(Asset.asset_type == asset_type)
|
||||
if asset_status is not None:
|
||||
stmt = stmt.where(Asset.status == asset_status)
|
||||
if is_archived is not None:
|
||||
stmt = stmt.where(Asset.is_archived == is_archived)
|
||||
if q:
|
||||
pattern = f"%{q}%"
|
||||
stmt = stmt.where(Asset.name.like(pattern) | Asset.provider.like(pattern))
|
||||
|
||||
sort_col = getattr(Asset, sort if sort in SORTABLE_FIELDS else "expiry_date")
|
||||
stmt = stmt.order_by(sort_col.desc() if order == "desc" else sort_col.asc())
|
||||
|
||||
assets = session.exec(stmt).all()
|
||||
return [_to_read(a, _get_detail(session, a)) for a in assets]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 统计
|
||||
# --------------------------------------------------------------------------- #
|
||||
def get_overview(session: Session) -> dict:
|
||||
"""资产总览：数量分布、异常数、到期预警、支出合计"""
|
||||
assets = session.exec(select(Asset)).all()
|
||||
today = date.today()
|
||||
|
||||
by_type: dict = {}
|
||||
by_status: dict = {}
|
||||
abnormal = 0
|
||||
expiring_30 = 0
|
||||
year_cost = 0.0
|
||||
month_cost = 0.0
|
||||
|
||||
for a in assets:
|
||||
by_type[a.asset_type.value] = by_type.get(a.asset_type.value, 0) + 1
|
||||
by_status[a.status.value] = by_status.get(a.status.value, 0) + 1
|
||||
if a.status in (AssetStatus.STOPPED, AssetStatus.EXPIRED):
|
||||
abnormal += 1
|
||||
if a.expiry_date:
|
||||
days = (a.expiry_date - today).days
|
||||
if 0 <= days <= 30:
|
||||
expiring_30 += 1
|
||||
if a.expiry_date.year == today.year:
|
||||
year_cost += a.cost
|
||||
if a.expiry_date.month == today.month:
|
||||
month_cost += a.cost
|
||||
|
||||
return {
|
||||
"total": len(assets),
|
||||
"by_type": by_type,
|
||||
"by_status": by_status,
|
||||
"abnormal_count": abnormal,
|
||||
"expiring_30": expiring_30,
|
||||
"year_cost": round(year_cost, 2),
|
||||
"month_cost": round(month_cost, 2),
|
||||
}
|
||||
|
||||
|
||||
def get_expiring(session: Session, days: int = 30) -> List[AssetRead]:
|
||||
"""N 天内到期资产列表（按剩余天数升序）"""
|
||||
today = date.today()
|
||||
assets = session.exec(select(Asset).where(Asset.expiry_date.is_not(None))).all()
|
||||
result = []
|
||||
for a in assets:
|
||||
delta = (a.expiry_date - today).days
|
||||
if 0 <= delta <= days:
|
||||
result.append(_to_read(a, _get_detail(session, a)))
|
||||
result.sort(key=lambda x: x.days_to_expiry if x.days_to_expiry is not None else 10**9)
|
||||
return result
|
||||
Reference in New Issue
Block a user