gouki/vps-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(account): 支持跨平台同名账号——(platform,name)联合唯一+Asset.account_id外键重构

Browse Source
This commit is contained in:
gouki
2026-08-11 07:27:01 +00:00
parent bc1b804931
commit b4705331c0
9 changed files with 209 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/database.py
+108 -11
View File
@@ -84,6 +84,10 @@ def _migrate_assets_db() -> None:
conn.execute(sa.text("ALTER TABLE assets ADD COLUMN renew_url VARCHAR"))
if "cancel_url" not in cols:
conn.execute(sa.text("ALTER TABLE assets ADD COLUMN cancel_url VARCHAR"))
if "account_id" not in cols:
conn.execute(sa.text("ALTER TABLE assets ADD COLUMN account_id INTEGER"))
# 资产账号引用从 name 字符串迁移到 account_id 外键（幂等）
_backfill_asset_account_id(conn)
if insp.has_table("providers"):
cols = {c["name"] for c in insp.get_columns("providers")}
if "last_synced_at" not in cols:
@@ -110,14 +114,24 @@ def _migrate_assets_db() -> None:
conn.execute(sa.text("ALTER TABLE accounts ADD COLUMN last_synced_at DATETIME"))
# 凭证下沉：平台 api_config 迁到默认账号；AI 资产自动挂靠账号
_backfill_account_credentials(conn)
# 合并存量 (platform,name) 重复账号，为建唯一索引做准备
_dedupe_accounts(conn)
def _migrate_indexes() -> None:
"""为已有数据库补充复合索引（create_all 不会为已存在的表补索引，IF NOT EXISTS 幂等）"""
import sqlalchemy as sa
# 旧的全局 name 唯一索引（早期 Account.name unique=True 产物）会阻止跨平台同名，
# 与新的 (platform,name) 联合唯一冲突，需先删除
with assets_engine.begin() as conn:
conn.execute(sa.text("DROP INDEX IF EXISTS ix_accounts_name"))
stmts = [
(assets_engine, "CREATE INDEX IF NOT EXISTS ix_assets_provider_ext_type ON assets (provider_id, external_id, asset_type)"),
(assets_engine, "CREATE INDEX IF NOT EXISTS ix_assets_account_id ON assets (account_id)"),
# (platform, name) 联合唯一：同邮箱可跨平台复用，同平台内不重名
(assets_engine, "CREATE UNIQUE INDEX IF NOT EXISTS uq_accounts_platform_name ON accounts (platform, name)"),
(metrics_engine, "CREATE INDEX IF NOT EXISTS ix_metric_points_asset_ts ON metric_points (asset_id, ts)"),
(metrics_engine, "CREATE INDEX IF NOT EXISTS ix_security_checks_asset_ts ON security_checks (asset_id, ts)"),
]
@@ -163,6 +177,80 @@ def _backfill_provider_services(conn) -> None:
)
def _dedupe_accounts(conn) -> None:
"""建 (platform,name) 唯一索引前的防护：合并存量重复账号。
platform 统一按 COALESCE(platform,'') 规范化比较（避免 NULL 绕过）。
重复时保留 id 最小者，将其余行的资产引用(account_id/account)与凭证并入后删除。
当前数据量小，预期无重复；此函数仅在发现重复时产生写操作。
"""
import sqlalchemy as sa
rows = conn.execute(
sa.text(
"SELECT COALESCE(platform,'') AS p, name, COUNT(*) AS c, MIN(id) AS keep_id "
"FROM accounts GROUP BY p, name HAVING c > 1"
)
).fetchall()
for p, name, _c, keep_id in rows:
dups = conn.execute(
sa.text(
"SELECT id FROM accounts WHERE COALESCE(platform,'') = :p AND name = :n AND id != :keep"
),
{"p": p, "n": name, "keep": keep_id},
).fetchall()
for (dup_id,) in dups:
# 资产引用并入保留行
conn.execute(
sa.text("UPDATE assets SET account_id = :keep WHERE account_id = :dup"),
{"keep": keep_id, "dup": dup_id},
)
# 凭证：保留行为空时才从重复行拷贝
conn.execute(
sa.text(
"UPDATE accounts SET "
"api_config_encrypted = COALESCE(api_config_encrypted, (SELECT api_config_encrypted FROM accounts WHERE id = :dup)), "
"login_password_encrypted = COALESCE(login_password_encrypted, (SELECT login_password_encrypted FROM accounts WHERE id = :dup)), "
"login_user = COALESCE(login_user, (SELECT login_user FROM accounts WHERE id = :dup)) "
"WHERE id = :keep"
),
{"keep": keep_id, "dup": dup_id},
)
conn.execute(sa.text("DELETE FROM accounts WHERE id = :dup"), {"dup": dup_id})
def _backfill_asset_account_id(conn) -> None:
"""资产账号引用迁移（幂等）：把 Asset.account 的 name 字符串引用回填为 account_id 外键。
匹配规则：按 accounts.name 匹配；同名多个时优先 platform 与 asset.provider 一致者。
匹配不到（历史自由文本/未登记账号）则跳过并保留 account 字符串，供人工处理。
"""
import sqlalchemy as sa
rows = conn.execute(
sa.text(
"SELECT id, account, provider FROM assets "
"WHERE account_id IS NULL AND account IS NOT NULL AND account != ''"
)
).fetchall()
if not rows:
return
for asset_id, acc_name, provider in rows:
# 同名账号可能多个：优先 platform 与资产 provider 一致的
cand = conn.execute(
sa.text(
"SELECT id, COALESCE(platform,'') FROM accounts WHERE name = :n "
"ORDER BY (COALESCE(platform,'') = :prov) DESC, id ASC"
),
{"n": acc_name, "prov": provider or ""},
).fetchall()
if cand:
conn.execute(
sa.text("UPDATE assets SET account_id = :aid WHERE id = :id"),
{"aid": cand[0][0], "id": asset_id},
)
def _backfill_account_credentials(conn) -> None:
"""凭证下沉迁移（幂等，仅在首次加列后产生效果）：
@@ -179,14 +267,21 @@ def _backfill_account_credentials(conn) -> None:
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()
# 按 (platform, name) 查，避免同名跨平台账号混淆
row = conn.execute(
sa.text("SELECT id FROM accounts WHERE name = :name AND COALESCE(platform,'') = :platform"),
{"name": name, "platform": platform},
).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]
return conn.execute(
sa.text("SELECT id FROM accounts WHERE name = :name AND COALESCE(platform,'') = :platform"),
{"name": name, "platform": platform},
).first()[0]
# 1. 平台凭证下沉到默认账号
for pid, slug, cfg in conn.execute(
@@ -200,26 +295,27 @@ def _backfill_account_credentials(conn) -> None:
conn.execute(
sa.text(
"UPDATE accounts SET api_config_encrypted = :cfg "
"WHERE name = :name AND (api_config_encrypted IS NULL OR api_config_encrypted = '')"
"WHERE name = :name AND COALESCE(platform,'') = :platform "
"AND (api_config_encrypted IS NULL OR api_config_encrypted = '')"
),
{"cfg": cfg, "name": acc_name},
{"cfg": cfg, "name": acc_name, "platform": slug},
)
# 2. 无账号的 AI 资产按 provider 挂靠
# 2. 无账号的 AI 资产按 provider 挂靠（account_id 外键 + account 字符串兼容）
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 = '')"
"WHERE a.asset_type = 'ai_agent' AND a.account_id IS NULL AND (a.account IS NULL OR a.account = '')"
)
).fetchall()
for asset_id, provider in rows:
if not provider:
continue
_ensure_account(provider, provider)
acc_id = _ensure_account(provider, provider)
conn.execute(
sa.text("UPDATE assets SET account = :acc WHERE id = :id"),
{"acc": provider, "id": asset_id},
sa.text("UPDATE assets SET account = :acc, account_id = :aid WHERE id = :id"),
{"acc": provider, "aid": acc_id, "id": asset_id},
)
# 账号尚无 API 配置时，把该资产的 api_key 写入账号配置（解密后重组 JSON 再加密）
key_row = conn.execute(
@@ -235,9 +331,10 @@ def _backfill_account_credentials(conn) -> None:
conn.execute(
sa.text(
"UPDATE accounts SET api_config_encrypted = :cfg "
"WHERE name = :name AND (api_config_encrypted IS NULL OR api_config_encrypted = '')"
"WHERE name = :name AND COALESCE(platform,'') = :platform "
"AND (api_config_encrypted IS NULL OR api_config_encrypted = '')"
),
{"cfg": cfg_enc, "name": provider},
{"cfg": cfg_enc, "name": provider, "platform": provider},
)