feat: 数据导出/导入（资产迁移/备份，支撑部署到其他服务器）

app/routers/assets.py

@@ -5,7 +5,7 @@
 
 from typing import List, Optional
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Body, Depends, Query
 from sqlmodel import Session
 
 from app.core.security import require_api_key
@@ -32,6 +32,22 @@ def list_assets(
     )
 
 
+@router.get("/export", summary="导出所有资产（数据迁移/备份）")
+def export_assets(session: Session = Depends(get_session)) -> dict:
+    return asset_service.export_assets(session)
+
+
+@router.post(
+    "/import",
+    summary="导入资产（从导出数据，按 name+类型+服务商去重）",
+    dependencies=[Depends(require_api_key)],
+)
+def import_assets_endpoint(
+    data: dict = Body(...), session: Session = Depends(get_session)
+) -> dict:
+    return asset_service.import_assets(session, data.get("assets", []))
+
+
 @router.get("/{asset_id}", response_model=AssetRead, summary="资产详情")
 def get_asset(asset_id: int, session: Session = Depends(get_session)):
     return asset_service.get_asset(session, asset_id)

app/services/asset_service.py

@@ -3,7 +3,7 @@
 统一处理 Asset 主表与其一对一详情表（VPSDetail/DomainDetail/AIAccount）的联动。
 """
 
-from datetime import date
+from datetime import date, datetime
 from typing import List, Optional
 
 from fastapi import HTTPException, status
@@ -235,6 +235,47 @@ def delete_asset(session: Session, asset_id: int) -> None:
     session.commit()
 
 
+def export_assets(session: Session) -> dict:
+    """导出所有资产（含 detail）为 JSON，用于数据迁移/备份"""
+    assets = session.exec(select(Asset)).all()
+    result = []
+    for asset in assets:
+        read = _to_read(session, asset, _get_detail(session, asset))
+        result.append(read.model_dump(mode="json"))
+    return {
+        "count": len(result),
+        "exported_at": datetime.utcnow().isoformat(),
+        "assets": result,
+    }
+
+
+def import_assets(session: Session, assets_data: list) -> dict:
+    """从导出数据导入资产（按 name+asset_type+provider 去重：存在则更新，不存在则创建）"""
+    created = 0
+    updated = 0
+    errors = []
+    skip_fields = {"id", "created_at", "updated_at", "days_to_expiry", "provider_name"}
+    for item in assets_data:
+        try:
+            payload = {k: v for k, v in item.items() if k not in skip_fields}
+            existing = session.exec(
+                select(Asset).where(
+                    Asset.name == payload.get("name"),
+                    Asset.asset_type == payload.get("asset_type"),
+                    Asset.provider == payload.get("provider"),
+                )
+            ).first()
+            if existing:
+                update_asset(session, existing.id, AssetUpdate(**payload))
+                updated += 1
+            else:
+                create_asset(session, AssetCreate(**payload))
+                created += 1
+        except Exception as e:  # noqa: BLE001
+            errors.append(f"{item.get('name', '?')}: {e}")
+    return {"created": created, "updated": updated, "errors": errors}
+
+
 def list_assets(
     session: Session,
     asset_type: Optional[AssetType] = None,