refactor: 凭证下沉账号层——平台不再持有API凭证；账号支持登录凭证+API配置，SDK测试/同步改为账号维度，同步产出自动挂账号；含存量数据自动迁移

app/database.py

@@ -98,6 +98,18 @@ def _migrate_assets_db() -> None:
                 conn.execute(sa.text("ALTER TABLE ai_accounts ADD COLUMN api_key_encrypted VARCHAR"))
             # 迁移：将明文 api_key 加密后存入 api_key_encrypted，并清空原字段
             _migrate_plaintext_api_keys(conn)
+        if insp.has_table("accounts"):
+            cols = {c["name"] for c in insp.get_columns("accounts")}
+            if "login_user" not in cols:
+                conn.execute(sa.text("ALTER TABLE accounts ADD COLUMN login_user VARCHAR"))
+            if "login_password_encrypted" not in cols:
+                conn.execute(sa.text("ALTER TABLE accounts ADD COLUMN login_password_encrypted VARCHAR"))
+            if "api_config_encrypted" not in cols:
+                conn.execute(sa.text("ALTER TABLE accounts ADD COLUMN api_config_encrypted VARCHAR"))
+            if "last_synced_at" not in cols:
+                conn.execute(sa.text("ALTER TABLE accounts ADD COLUMN last_synced_at DATETIME"))
+            # 凭证下沉：平台 api_config 迁到默认账号；AI 资产自动挂靠账号
+            _backfill_account_credentials(conn)
 
 
 def _migrate_indexes() -> None:
@@ -151,6 +163,84 @@ def _backfill_provider_services(conn) -> None:
         )
 
 
+def _backfill_account_credentials(conn) -> None:
+    """凭证下沉迁移（幂等，仅在首次加列后产生效果）：
+
+    1. 平台凭证下沉：api_config_encrypted 非空的 Provider → 确保存在
+       「{slug}-默认」账号并拷入凭证（不删平台原值，保留可回滚）。
+    2. AI 资产挂靠：account 为空的 ai_agent 资产 → 按 ai_accounts.provider
+       找/建账号并关联；账号无 API 配置时把该资产的 api_key 写入账号配置。
+    """
+    import json
+    from datetime import datetime, timezone
+
+    import sqlalchemy as sa
+
+    from app.core.crypto import decrypt, encrypt
+
+    def _ensure_account(name: str, platform: str) -> int:
+        row = conn.execute(sa.text("SELECT id FROM accounts WHERE name = :name"), {"name": name}).first()
+        if row:
+            return row[0]
+        conn.execute(
+            sa.text("INSERT INTO accounts (name, platform, created_at) VALUES (:name, :platform, :ts)"),
+            {"name": name, "platform": platform, "ts": datetime.now(timezone.utc).replace(tzinfo=None)},
+        )
+        return conn.execute(sa.text("SELECT id FROM accounts WHERE name = :name"), {"name": name}).first()[0]
+
+    # 1. 平台凭证下沉到默认账号
+    for pid, slug, cfg in conn.execute(
+        sa.text(
+            "SELECT id, slug, api_config_encrypted FROM providers "
+            "WHERE api_config_encrypted IS NOT NULL AND api_config_encrypted != ''"
+        )
+    ).fetchall():
+        acc_name = f"{slug}-默认"
+        _ensure_account(acc_name, slug)
+        conn.execute(
+            sa.text(
+                "UPDATE accounts SET api_config_encrypted = :cfg "
+                "WHERE name = :name AND (api_config_encrypted IS NULL OR api_config_encrypted = '')"
+            ),
+            {"cfg": cfg, "name": acc_name},
+        )
+
+    # 2. 无账号的 AI 资产按 provider 挂靠
+    rows = conn.execute(
+        sa.text(
+            "SELECT a.id, COALESCE(ai.provider, '') FROM assets a "
+            "JOIN ai_accounts ai ON ai.asset_id = a.id "
+            "WHERE a.asset_type = 'ai_agent' AND (a.account IS NULL OR a.account = '')"
+        )
+    ).fetchall()
+    for asset_id, provider in rows:
+        if not provider:
+            continue
+        _ensure_account(provider, provider)
+        conn.execute(
+            sa.text("UPDATE assets SET account = :acc WHERE id = :id"),
+            {"acc": provider, "id": asset_id},
+        )
+        # 账号尚无 API 配置时，把该资产的 api_key 写入账号配置（解密后重组 JSON 再加密）
+        key_row = conn.execute(
+            sa.text("SELECT api_key_encrypted FROM ai_accounts WHERE asset_id = :id"),
+            {"id": asset_id},
+        ).first()
+        if key_row and key_row[0]:
+            api_key = decrypt(key_row[0])
+            if api_key:
+                cfg_json = json.dumps({"api_key": api_key})
+                cfg_enc = encrypt(cfg_json)
+                if cfg_enc:
+                    conn.execute(
+                        sa.text(
+                            "UPDATE accounts SET api_config_encrypted = :cfg "
+                            "WHERE name = :name AND (api_config_encrypted IS NULL OR api_config_encrypted = '')"
+                        ),
+                        {"cfg": cfg_enc, "name": provider},
+                    )
+
+
 def get_session() -> Generator[Session, None, None]:
     """资产库会话（默认）
 

app/models/asset.py

@@ -202,6 +202,8 @@ class Account(SQLModel, table=True):
     """平台账号字典：统一管理各平台下的账号（多账号场景区分归属）
 
     资产的 Asset.account 字段按名称引用本表，重命名时由服务层同步更新引用。
+    凭证层：登录用户名/密码（网站登录）+ API 配置 JSON（SDK 同步用），
+    平台本身不再持有凭证（Provider.api_config_encrypted 已弃用，仅留历史值）。
     """
 
     __tablename__ = "accounts"
@@ -212,4 +214,14 @@ class Account(SQLModel, table=True):
         default=None, index=True, description="所属平台（slug 或名称）"
     )
     remark: Optional[str] = Field(default=None, description="备注")
+    login_user: Optional[str] = Field(default=None, description="登录用户名/邮箱")
+    login_password_encrypted: Optional[str] = Field(
+        default=None, description="加密的登录密码（Fernet）"
+    )
+    api_config_encrypted: Optional[str] = Field(
+        default=None, description="加密的 API 配置 JSON（access_key/secret/api_key 等，供 SDK 同步）"
+    )
+    last_synced_at: Optional[datetime] = Field(
+        default=None, description="最近一次 SDK 同步时间"
+    )
     created_at: datetime = Field(default_factory=utcnow, description="创建时间")

app/routers/accounts.py

@@ -2,6 +2,7 @@
 
 统一入口 /api/accounts，写操作（POST/PUT/DELETE）受 API Key 保护。
 资产的 Asset.account 按名称引用账号；重命名账号时服务层会同步更新引用资产。
+凭证（登录密码/API 配置）存在账号上；账号维度的 SDK 测试/同步见 test/sync 端点。
 """
 
 from typing import List
@@ -12,7 +13,7 @@ from sqlmodel import Session
 from app.core.security import require_api_key
 from app.database import get_session
 from app.schemas.account import AccountCreate, AccountRead, AccountUpdate
-from app.services import account_service
+from app.services import account_service, sync_service
 
 router = APIRouter(prefix="/api/accounts", tags=["accounts"])
 
@@ -52,3 +53,21 @@ def update_account(
 )
 def delete_account(account_id: int, session: Session = Depends(get_session)) -> dict:
     return account_service.delete_account(session, account_id)
+
+
+@router.post(
+    "/{account_id}/test",
+    summary="测试账号凭证有效性（按账号所属平台的 SDK）",
+    dependencies=[Depends(require_api_key)],
+)
+def test_account(account_id: int, session: Session = Depends(get_session)) -> dict:
+    return sync_service.test_account(session, account_id)
+
+
+@router.post(
+    "/{account_id}/sync",
+    summary="同步账号资产到本地库（产出自动挂到该账号名下）",
+    dependencies=[Depends(require_api_key)],
+)
+def sync_account(account_id: int, session: Session = Depends(get_session)) -> dict:
+    return sync_service.sync_account(session, account_id)

app/schemas/account.py

@@ -1,4 +1,9 @@
-"""平台账号 Schema"""
+"""平台账号 Schema
+
+凭证语义约定：
+- login_password / api_config：Create 时传入即加密存储；Update 时 None 表示不修改。
+- Read 永不返回凭证明文/密文，只给布尔标记（has_login_password / has_api_config）。
+"""
 
 from datetime import datetime
 from typing import Optional
@@ -10,19 +15,27 @@ class AccountBase(SQLModel):
     name: str
     platform: Optional[str] = None
     remark: Optional[str] = None
+    login_user: Optional[str] = None
 
 
 class AccountCreate(AccountBase):
-    pass
+    login_password: Optional[str] = None  # 登录密码（明文传入，加密存储）
+    api_config: Optional[str] = None  # API 配置 JSON 字符串（加密存储，供 SDK 同步）
 
 
 class AccountUpdate(SQLModel):
     name: Optional[str] = None
     platform: Optional[str] = None
     remark: Optional[str] = None
+    login_user: Optional[str] = None
+    login_password: Optional[str] = None  # None = 不修改
+    api_config: Optional[str] = None  # None = 不修改
 
 
 class AccountRead(AccountBase):
     id: int
     created_at: datetime
     asset_count: int = 0  # 引用该账号的资产数（列表接口填充）
+    has_login_password: bool = False
+    has_api_config: bool = False
+    last_synced_at: Optional[datetime] = None

app/services/account_service.py

@@ -2,6 +2,7 @@
 
 账号与资产的关系：Asset.account 按名称引用账号（字符串，兼容历史自由文本）。
 重命名账号时同步更新所有引用资产，保证两边一致。
+凭证层：登录密码与 API 配置加密存储，Read 仅返回布尔标记。
 """
 
 from typing import Dict, List
@@ -10,6 +11,7 @@ from fastapi import HTTPException, status
 from sqlalchemy import func
 from sqlmodel import Session, select
 
+from app.core import crypto
 from app.models.asset import Account, Asset
 from app.schemas.account import AccountCreate, AccountRead, AccountUpdate
 
@@ -27,6 +29,8 @@ def _asset_counts(session: Session) -> Dict[str, int]:
 def _to_read(account: Account, counts: Dict[str, int]) -> AccountRead:
     read = AccountRead.model_validate(account)
     read.asset_count = counts.get(account.name, 0)
+    read.has_login_password = bool(account.login_password_encrypted)
+    read.has_api_config = bool(account.api_config_encrypted)
     return read
 
 
@@ -58,7 +62,9 @@ def create_account(session: Session, data: AccountCreate) -> AccountRead:
     if not name:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="账号名称不能为空")
     _check_name_taken(session, name)
-    account = Account(name=name, platform=data.platform, remark=data.remark)
+    account = Account(name=name, platform=data.platform, remark=data.remark, login_user=data.login_user)
+    account.login_password_encrypted = crypto.encrypt(data.login_password)
+    account.api_config_encrypted = crypto.encrypt(data.api_config)
     session.add(account)
     session.commit()
     session.refresh(account)
@@ -83,6 +89,13 @@ def update_account(session: Session, account_id: int, data: AccountUpdate) -> Ac
         account.platform = data.platform or None
     if data.remark is not None:
         account.remark = data.remark or None
+    if data.login_user is not None:
+        account.login_user = data.login_user or None
+    # 凭证：None = 不修改；空串 = 清除；非空 = 重新加密存储
+    if data.login_password is not None:
+        account.login_password_encrypted = crypto.encrypt(data.login_password)
+    if data.api_config is not None:
+        account.api_config_encrypted = crypto.encrypt(data.api_config)
     session.add(account)
     session.commit()
     session.refresh(account)

app/services/sync_service.py

@@ -3,6 +3,10 @@
 将适配器返回的标准化资产（NormalizedVPS / NormalizedDomain / AccountInfo）
 写入或更新到资产库。以 (provider_id, external_id) 作为去重键，
 已存在则更新状态/详情，不存在则新建资产。
+
+凭证层级：API 配置存在账号（Account.api_config_encrypted）上，
+同步按账号维度进行（test_account/sync_account）；同步产出的资产
+自动挂到该账号名下（Asset.account）。平台不再持有凭证。
 """
 
 import json
@@ -20,6 +24,7 @@ from app.core import crypto
 from app.core.timeutils import utcnow
 from app.models.asset import (
     AIAccount,
+    Account,
     Asset,
     AssetStatus,
     AssetType,
@@ -32,7 +37,7 @@ _VALID_STATUS = {s.value for s in AssetStatus}
 
 
 def _load_config(provider: Provider) -> dict:
-    """解密平台的 API 配置 JSON"""
+    """（已弃用）解密平台级 API 配置，仅为兼容历史数据保留"""
     plain = crypto.decrypt(provider.api_config_encrypted)
     if not plain:
         return {}
@@ -42,6 +47,30 @@ def _load_config(provider: Provider) -> dict:
         return {}
 
 
+def _load_account_config(account: Account) -> dict:
+    """解密账号的 API 配置 JSON"""
+    plain = crypto.decrypt(account.api_config_encrypted)
+    if not plain:
+        return {}
+    try:
+        return json.loads(plain)
+    except (json.JSONDecodeError, TypeError):
+        return {}
+
+
+def _find_provider_by_platform(session: Session, platform: str) -> Provider:
+    """按账号的 platform（slug 或名称）定位平台"""
+    provider = session.exec(select(Provider).where(Provider.slug == platform)).first()
+    if not provider:
+        provider = session.exec(select(Provider).where(Provider.name == platform)).first()
+    if not provider:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"账号所属平台不存在：{platform}（请先在账号编辑中选择有效平台）",
+        )
+    return provider
+
+
 def _get_provider(session: Session, provider_id: int) -> Provider:
     provider = session.get(Provider, provider_id)
     if not provider:
@@ -49,7 +78,7 @@ def _get_provider(session: Session, provider_id: int) -> Provider:
     return provider
 
 
-def _build_adapter(provider: Provider) -> BaseAdapter:
+def _build_adapter(provider: Provider, config: dict) -> BaseAdapter:
     if not provider.sdk_type:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST, detail="该平台未配置 SDK 类型（sdk_type）"
@@ -59,7 +88,27 @@ def _build_adapter(provider: Provider) -> BaseAdapter:
             status_code=status.HTTP_400_BAD_REQUEST,
             detail=f"暂不支持的 SDK 类型：{provider.sdk_type}",
         )
-    return registry.get_adapter(provider.sdk_type, _load_config(provider))
+    return registry.get_adapter(provider.sdk_type, config)
+
+
+def _get_account(session: Session, account_id: int) -> Account:
+    account = session.get(Account, account_id)
+    if not account:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="账号不存在")
+    return account
+
+
+def _build_adapter_for_account(session: Session, account: Account):
+    """按账号构建适配器：平台定 sdk_type，账号提供凭证"""
+    if not account.platform:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="该账号未指定所属平台，无法确定 SDK 类型",
+        )
+    provider = _find_provider_by_platform(session, account.platform)
+    config = _load_account_config(account)
+    adapter = _build_adapter(provider, config)
+    return provider, adapter
 
 
 def _norm_status(raw: Optional[str]) -> AssetStatus:
@@ -72,15 +121,37 @@ def _missing_config(adapter: BaseAdapter) -> list:
 
 
 def test_provider(session: Session, provider_id: int) -> dict:
-    """测试平台连接 / 凭证有效性"""
+    """（兼容入口）测试平台连接：凭证已下沉到账号，自动找该平台第一个配了凭证的账号"""
     provider = _get_provider(session, provider_id)
-    adapter = _build_adapter(provider)
-    base = {"capabilities": adapter.capabilities(), "sdk_type": provider.sdk_type}
+    account = _first_account_with_config(session, provider)
+    if not account:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="凭证已改为在账号上配置：请先在该平台的账号管理中新建账号并填写 API 配置",
+        )
+    return test_account(session, account.id)
+
+
+def _first_account_with_config(session: Session, provider: Provider):
+    """找该平台下第一个配置了 API 凭证的账号（兼容旧入口）"""
+    return session.exec(
+        select(Account)
+        .where(Account.platform == provider.slug)
+        .where(Account.api_config_encrypted.is_not(None))  # type: ignore[union-attr]
+        .order_by(Account.id.asc())
+    ).first()
+
+
+def test_account(session: Session, account_id: int) -> dict:
+    """测试账号凭证有效性（按账号 platform 定 SDK，凭证取自账号）"""
+    account = _get_account(session, account_id)
+    provider, adapter = _build_adapter_for_account(session, account)
+    base = {"capabilities": adapter.capabilities(), "sdk_type": provider.sdk_type, "account": account.name}
     missing = _missing_config(adapter)
     if missing:
         return {
             "ok": False,
-            "message": f"缺少凭证配置：{', '.join(missing)}（请在平台编辑里填写 API 配置）",
+            "message": f"缺少凭证配置：{', '.join(missing)}（请在账号编辑里填写 API 配置）",
             **base,
         }
     result = adapter.test_connection()
@@ -98,7 +169,7 @@ def _find_asset(session: Session, provider_id: int, external_id: str, asset_type
     ).first()
 
 
-def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter) -> dict:
+def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter, account_name: Optional[str] = None) -> dict:
     created = updated = 0
     for vps in adapter.list_vps():
         existing = _find_asset(session, provider.id, vps.external_id, AssetType.VPS)
@@ -108,6 +179,8 @@ def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter) -> dic
             if vps.monthly_cost is not None:
                 existing.cost = vps.monthly_cost
                 existing.currency = vps.currency
+            if account_name:
+                existing.account = account_name
             session.add(existing)
             detail = session.exec(
                 select(VPSDetail).where(VPSDetail.asset_id == existing.id)
@@ -128,6 +201,7 @@ def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter) -> dic
                 provider=provider.slug,
                 provider_id=provider.id,
                 external_id=vps.external_id,
+                account=account_name,
                 status=_norm_status(vps.status),
                 cost=vps.monthly_cost or 0,
                 currency=vps.currency,
@@ -150,7 +224,7 @@ def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter) -> dic
     return {"created": created, "updated": updated}
 
 
-def _sync_domains(session: Session, provider: Provider, adapter: BaseAdapter) -> dict:
+def _sync_domains(session: Session, provider: Provider, adapter: BaseAdapter, account_name: Optional[str] = None) -> dict:
     created = updated = 0
     for dom in adapter.list_domains():
         existing = _find_asset(session, provider.id, dom.external_id, AssetType.DOMAIN)
@@ -158,6 +232,8 @@ def _sync_domains(session: Session, provider: Provider, adapter: BaseAdapter) ->
             existing.status = _norm_status(dom.status)
             if dom.expiry_date:
                 existing.expiry_date = dom.expiry_date
+            if account_name:
+                existing.account = account_name
             session.add(existing)
             detail = session.exec(
                 select(DomainDetail).where(DomainDetail.asset_id == existing.id)
@@ -174,6 +250,7 @@ def _sync_domains(session: Session, provider: Provider, adapter: BaseAdapter) ->
                 provider=provider.slug,
                 provider_id=provider.id,
                 external_id=dom.external_id,
+                account=account_name,
                 status=_norm_status(dom.status),
                 expiry_date=dom.expiry_date,
             )
@@ -192,39 +269,51 @@ def _sync_domains(session: Session, provider: Provider, adapter: BaseAdapter) ->
 
 
 def sync_provider(session: Session, provider_id: int) -> dict:
-    """同步平台资产到本地库"""
+    """（兼容入口）同步平台资产：凭证已下沉到账号，自动找该平台第一个配了凭证的账号"""
     provider = _get_provider(session, provider_id)
-    adapter = _build_adapter(provider)
+    account = _first_account_with_config(session, provider)
+    if not account:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="凭证已改为在账号上配置：请先在该平台的账号管理中新建账号并填写 API 配置",
+        )
+    return sync_account(session, account.id)
+
+
+def sync_account(session: Session, account_id: int) -> dict:
+    """同步账号资产到本地库（凭证取自账号，同步产出自动挂到该账号名下）"""
+    account = _get_account(session, account_id)
+    provider, adapter = _build_adapter_for_account(session, account)
     missing = _missing_config(adapter)
     if missing:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"缺少凭证配置：{', '.join(missing)}（请先在平台编辑里填写 API 配置）",
+            detail=f"缺少凭证配置：{', '.join(missing)}（请先在账号编辑里填写 API 配置）",
         )
     caps = adapter.capabilities()
-    result = {"provider": provider.slug, "sdk_type": provider.sdk_type}
+    result = {"provider": provider.slug, "sdk_type": provider.sdk_type, "account": account.name}
 
     if caps["list_vps"]:
         try:
-            result["vps"] = _sync_vps(session, provider, adapter)
+            result["vps"] = _sync_vps(session, provider, adapter, account.name)
         except Exception as e:  # noqa: BLE001
             result["vps_error"] = str(e)
     if caps["list_domains"]:
         try:
-            result["domains"] = _sync_domains(session, provider, adapter)
+            result["domains"] = _sync_domains(session, provider, adapter, account.name)
         except Exception as e:  # noqa: BLE001
             result["domains_error"] = str(e)
     if caps["get_account"]:
         try:
-            result["account"] = adapter.get_account().to_dict()
+            result["account_info"] = adapter.get_account().to_dict()
         except Exception as e:  # noqa: BLE001
-            result["account_error"] = str(e)
+            result["account_info_error"] = str(e)
 
-    provider.last_synced_at = utcnow()
-    session.add(provider)
+    account.last_synced_at = utcnow()
+    session.add(account)
     session.commit()
-    result["last_synced_at"] = provider.last_synced_at.isoformat()
-    logger.info("同步平台 provider=%s result=%s", provider.slug, result)
+    result["last_synced_at"] = account.last_synced_at.isoformat()
+    logger.info("同步账号 account=%s provider=%s result=%s", account.name, provider.slug, result)
     return result