feat: resolve account platforms prioritizing structural snapshot values and correction on existing accounts

2026-07-02 22:58:15 +08:00
parent 1cb13e878e
commit fd2755d539
4 changed files with 359 additions and 13 deletions
@@ -462,6 +462,17 @@ def import_groups_from_upstream(
    )


+SUPPORTED_PLATFORMS = {"openai", "anthropic", "gemini", "antigravity"}
+
+
+def _resolve_platform(text: str, group_snapshot: dict | None, fallback: str = "openai") -> str:
+    if group_snapshot and isinstance(group_snapshot, dict):
+        platform = group_snapshot.get("platform")
+        if platform and str(platform).lower() in SUPPORTED_PLATFORMS:
+            return str(platform).lower()
+    return _detect_platform(text, fallback)
+
+
 def _detect_platform(text: str, fallback: str = "openai") -> str:
    """根据 Key 名或分组名关键词判断平台类型。"""
    lower = text.lower()
@@ -496,6 +507,12 @@ def sync_imported_upstream_keys(
        .all()
    )
    items: list[ImportAccountItem] = []
+    # 预载最新快照映射，用于结构化 platform 解析
+    upstream_ids = {row.upstream_id for row in rows}
+    latest_snapshots = {}
+    for uid in upstream_ids:
+        latest_snapshots[uid] = latest_rate_map(db, uid)
+
    with _client(website) as c:
        remote_accounts = None
        try:
@@ -513,7 +530,9 @@ def sync_imported_upstream_keys(
            remote_ids = None

        for row in rows:
-            platform = _detect_platform(f"{row.group_name} {row.group_id} {row.key_name}", "openai")
+            snap_groups = latest_snapshots.get(row.upstream_id) or {}
+            group_snap = snap_groups.get(row.group_id) if isinstance(snap_groups, dict) else None
+            platform = _resolve_platform(f"{row.group_name} {row.group_id} {row.key_name}", group_snap, "openai")
            if not row.imported_account_id:
                continue
            old_account_id = row.imported_account_id
@@ -645,12 +664,17 @@ def import_upstream_keys_as_accounts(
        )
        for kid in missing_ids
    ]
-    # 预载上游以方便获取名称和 base_url
+    # 预载上游以方便获取名称 and base_url
    from app.models.upstream import Upstream as _Up
    upstream_ids = {row.upstream_id for row in rows}
    upstreams_db = db.query(_Up).filter(_Up.id.in_(upstream_ids)).all()
    upstreams_map = {u.id: u for u in upstreams_db}

+    # 预载最新快照映射，用于结构化 platform 解析
+    latest_snapshots: dict[int, dict[str, Any]] = {}
+    for uid in upstream_ids:
+        latest_snapshots[uid] = latest_rate_map(db, uid)
+
    # 按倍率自动分配优先级
    rate_priority_map: dict[tuple[str, int, str], int] = {}
    if body.auto_priority_by_rate:
@@ -706,8 +730,11 @@ def import_upstream_keys_as_accounts(
                continue
            # 先确定平台（失败项也需要记录）
            if body.platform_mode == "auto":
-                platform = _detect_platform(
+                snap_groups = latest_snapshots.get(row.upstream_id) or {}
+                group_snap = snap_groups.get(row.group_id) if isinstance(snap_groups, dict) else None
+                platform = _resolve_platform(
                    f"{row.group_name} {row.group_id} {row.key_name}",
+                    group_snap,
                    body.default_platform,
                )
            else:
@@ -1644,6 +1671,18 @@ def organize_website_groups(
                if acc_id:
                    remote_account_map[str(acc_id)] = acc

+        # 预载所有相关的上游快照映射
+        all_upstream_ids = set()
+        for b in bindings:
+            for src in binding_sources(b):
+                _uid = src.get("upstream_id")
+                if _uid:
+                    all_upstream_ids.add(int(_uid))
+
+        latest_snapshots: dict[int, dict[str, Any]] = {}
+        for _uid in all_upstream_ids:
+            latest_snapshots[_uid] = latest_rate_map(db, _uid)
+
        # 对每个绑定关系遍历其对应的上游 Key 进行整理
        for b in bindings:
            target_group_id = b.target_group_id
@@ -1659,14 +1698,10 @@ def organize_website_groups(
                _u = db.query(Upstream).filter(Upstream.id == uid).first()
                upstream_name = _u.name if _u else f"#{uid}"

+                snapshot_groups = latest_snapshots.get(int(uid)) or {}
                # 来源名称优先取 src.group_name，再 fallback 快照，再 fallback ID
                source_group_name = src.get("group_name") or ""
                if not source_group_name:
-                    snapshot_groups = {}
-                    try:
-                        snapshot_groups = latest_rate_map(db, uid)
-                    except Exception:
-                        pass
                    if gid in snapshot_groups:
                        g_data = snapshot_groups[gid]
                        if isinstance(g_data, dict):
@@ -1725,8 +1760,10 @@ def organize_website_groups(
                        continue

                    # 检测平台类型
-                    platform = _detect_platform(
+                    group_snap = snapshot_groups.get(row.group_id) if isinstance(snapshot_groups, dict) else None
+                    platform = _resolve_platform(
                        f"{row.group_name} {row.group_id} {row.key_name}",
+                        group_snap,
                        "openai",
                    )

@@ -1771,11 +1808,31 @@ def organize_website_groups(
                                old_set = {str(g) for g in current_group_ids}
                                new_set = {str(g) for g in new_group_ids}

+                                remote_platform = remote_acc.get("platform")
+                                if remote_platform is not None:
+                                    platform_mismatch = (str(remote_platform).lower() != str(platform).lower())
+                                else:
+                                    platform_mismatch = (str(platform).lower() != "openai")
+
+                                update_payload = {}
                                if old_set != new_set:
-                                    c.update_account(old_account_id, {"group_ids": new_group_ids})
-                                    # 顺便更新下 remote_account_map 中的数据，防止同个账号后续在其他绑定中重复处理
-                                    remote_acc["group_ids"] = new_group_ids
-                                    msg = "已存在，已迁移到目标分组"
+                                    update_payload["group_ids"] = new_group_ids
+                                if platform_mismatch:
+                                    update_payload["platform"] = platform
+
+                                if update_payload:
+                                    c.update_account(old_account_id, update_payload)
+                                    if "group_ids" in update_payload:
+                                        remote_acc["group_ids"] = new_group_ids
+                                    if "platform" in update_payload:
+                                        remote_acc["platform"] = platform
+
+                                    msg_parts = []
+                                    if old_set != new_set:
+                                        msg_parts.append("已迁移到目标分组")
+                                    if platform_mismatch:
+                                        msg_parts.append(f"平台从 {remote_platform} 修正为 {platform}")
+                                    msg = "已存在，" + "且".join(msg_parts)
                                else:
                                    msg = "已存在且已对齐，已跳过"

@@ -0,0 +1,36 @@
+import os
+import sys
+import json
+
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+from app.database import SessionLocal
+from app.models.website import Website
+from app.routers.websites import _client
+
+db = SessionLocal()
+try:
+    website = db.query(Website).filter(Website.id == 2).first()
+    if not website:
+        print("Error: website wangwang888 (id=2) not found!")
+        sys.exit(1)
+        
+    print(f"Loaded website: {website.name}, URL: {website.base_url}")
+    
+    with _client(website) as c:
+        accounts = c.list_accounts()
+        acc = next((a for a in accounts if str(c.extract_id(a)) == "935"), None)
+        if not acc:
+            print("Error: Account 935 not found on remote site!")
+            sys.exit(1)
+            
+        print(f"Found account 935: {acc.get('name')}, current platform: {acc.get('platform')}")
+        
+        res = c.update_account("935", {"platform": "anthropic"})
+        print(f"Update response: {res}")
+        
+        accounts_new = c.list_accounts()
+        acc_new = next((a for a in accounts_new if str(c.extract_id(a)) == "935"), None)
+        print(f"Verified account 935: {acc_new.get('name')}, new platform: {acc_new.get('platform')}")
+finally:
+    db.close()
@@ -479,3 +479,102 @@ def test_organize_groups_force_alignment(db_session, monkeypatch):
    # 既然已经一致，不会触发 update_account
    assert len(updated_accounts) == 0
    assert "已存在且已对齐 2" in response_aligned.message
+
+
+def test_organize_groups_corrects_mismatched_platform_instead_of_recreating(db_session, monkeypatch):
+    """测试一键整理遇到已存在账号平台错配时，只调用 update_account() 修正平台，不创建新账号。"""
+    from app.models.snapshot import UpstreamRateSnapshot
+
+    w = Website(
+        name="W1",
+        site_type="sub2api",
+        base_url="http://w1",
+        enabled=True,
+        auth_config_json="{}",
+        timeout_seconds=30
+    )
+    u1 = Upstream(name="U1", base_url="http://u1")
+    db_session.add_all([w, u1])
+    db_session.commit()
+    db_session.refresh(w)
+    db_session.refresh(u1)
+
+    # 绑定关系：目标分组 TG1 ↔ G1
+    b1 = WebsiteGroupBinding(
+        website_id=w.id,
+        target_group_id="TG1",
+        target_group_name="TG1-Group",
+        source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G1"}]),
+        enabled=True
+    )
+    db_session.add(b1)
+
+    # G1: 已导入，导入的目标分组为 TG1，账号为 ACC-G1
+    k1 = UpstreamGeneratedKey(
+        upstream_id=u1.id,
+        group_id="G1",
+        group_name="G1-Group",
+        key_name="Key-G1",
+        key_value="sk-g1-secret",
+        status="imported",
+        imported_website_id=w.id,
+        imported_account_id="ACC-G1",
+        imported_target_group_id="TG1",
+    )
+    db_session.add(k1)
+
+    # 写入快照指定平台为 anthropic
+    snapshot = UpstreamRateSnapshot(
+        upstream_id=u1.id,
+        snapshot_json=json.dumps({
+            "groups": {
+                "G1": {
+                    "group_name": "G1-Group",
+                    "rate": 0.1,
+                    "platform": "anthropic"
+                }
+            }
+        })
+    )
+    db_session.add(snapshot)
+    db_session.commit()
+
+    updated_accounts = []
+    create_called = False
+
+    class MockClient:
+        def __init__(self, **kwargs): pass
+        def __enter__(self): return self
+        def __exit__(self, *a): pass
+        def get_groups(self, *a, **kw):
+            return [{"id": "TG1", "name": "TG1-Group"}]
+        def list_accounts(self):
+            return [
+                {
+                    "id": "ACC-G1",
+                    "name": "SmartUp-G1",
+                    "group_ids": ["TG1"],
+                    "platform": "openai",
+                }
+            ]
+        def extract_id(self, val):
+            return val.get("id") if isinstance(val, dict) else str(val)
+        def create_account(self, body):
+            nonlocal create_called
+            create_called = True
+            return {"id": "NEW-ACC", "name": body["name"], "group_ids": body["group_ids"]}
+        def update_account(self, account_id, body):
+            updated_accounts.append((account_id, body))
+            return {"id": account_id, "platform": body.get("platform")}
+
+    monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient)
+    monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: [])
+
+    response = organize_website_groups(wid=w.id, db=db_session)
+
+    assert response.success is True
+    assert not create_called
+    assert len(updated_accounts) == 1
+    assert updated_accounts[0][0] == "ACC-G1"
+    assert updated_accounts[0][1]["platform"] == "anthropic"
+    assert "平台从 openai 修正为 anthropic" in response.items[0].message
@@ -410,3 +410,157 @@ def test_import_upstream_key_with_custom_concurrency_and_priority(monkeypatch, d
    assert body["concurrency"] == 20
    assert body["priority"] == 5
    assert body["credentials"]["base_url"] == "http://packy.local"
+
+
+def test_import_platform_from_snapshot(monkeypatch, db_session):
+    """验证快照 platform=anthropic 且名称不含 Claude 时，创建账号平台必须为 anthropic"""
+    from app.models.snapshot import UpstreamRateSnapshot
+    website, generated = seed_account_import_rows(db_session)
+    generated.key_name = "MyCustomKey" # 不含 Claude
+
+    # 写入快照指定平台为 anthropic
+    snapshot = UpstreamRateSnapshot(
+        upstream_id=generated.upstream_id,
+        snapshot_json=json.dumps({
+            "groups": {
+                "vip": {
+                    "group_name": "VIP",
+                    "rate": 0.1,
+                    "platform": "anthropic"
+                }
+            }
+        })
+    )
+    db_session.add(snapshot)
+    db_session.commit()
+
+    account_bodies = []
+    class FakeWebsiteClient:
+        def __init__(self, **kwargs):
+            pass
+        def __enter__(self): return self
+        def __exit__(self, *args): return False
+        def create_account(self, body, endpoint="/accounts"):
+            account_bodies.append(body)
+            return {"id": 101, "name": body["name"]}
+        def account_exists(self, account_id): return True
+        @staticmethod
+        def extract_id(data): return "101"
+
+    monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeWebsiteClient)
+
+    websites_router.import_upstream_keys_as_accounts(
+        website.id,
+        ImportAccountsRequest(
+            upstream_key_ids=[generated.id],
+            target_group_map={"vip": "7"},
+            default_platform="openai",
+            platform_mode="auto",
+        ),
+        db_session,
+        object(),
+    )
+    assert len(account_bodies) == 1
+    assert account_bodies[0]["platform"] == "anthropic"
+
+
+def test_import_platform_fallback_to_name(monkeypatch, db_session):
+    """验证快照无 platform 但名称含 Claude 时，仍兜底为 anthropic"""
+    from app.models.snapshot import UpstreamRateSnapshot
+    website, generated = seed_account_import_rows(db_session)
+    generated.key_name = "Claude-Key"
+
+    # 写入快照但不含 platform 字段
+    snapshot = UpstreamRateSnapshot(
+        upstream_id=generated.upstream_id,
+        snapshot_json=json.dumps({
+            "groups": {
+                "vip": {
+                    "group_name": "VIP",
+                    "rate": 0.1,
+                }
+            }
+        })
+    )
+    db_session.add(snapshot)
+    db_session.commit()
+
+    account_bodies = []
+    class FakeWebsiteClient:
+        def __init__(self, **kwargs):
+            pass
+        def __enter__(self): return self
+        def __exit__(self, *args): return False
+        def create_account(self, body, endpoint="/accounts"):
+            account_bodies.append(body)
+            return {"id": 101, "name": body["name"]}
+        def account_exists(self, account_id): return True
+        @staticmethod
+        def extract_id(data): return "101"
+
+    monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeWebsiteClient)
+
+    websites_router.import_upstream_keys_as_accounts(
+        website.id,
+        ImportAccountsRequest(
+            upstream_key_ids=[generated.id],
+            target_group_map={"vip": "7"},
+            default_platform="openai",
+            platform_mode="auto",
+        ),
+        db_session,
+        object(),
+    )
+    assert len(account_bodies) == 1
+    assert account_bodies[0]["platform"] == "anthropic"
+
+
+def test_import_platform_manual_override(monkeypatch, db_session):
+    """手动模式下不使用快照平台，仍按用户指定平台"""
+    from app.models.snapshot import UpstreamRateSnapshot
+    website, generated = seed_account_import_rows(db_session)
+
+    # 快照虽然指定为 anthropic
+    snapshot = UpstreamRateSnapshot(
+        upstream_id=generated.upstream_id,
+        snapshot_json=json.dumps({
+            "groups": {
+                "vip": {
+                    "group_name": "VIP",
+                    "rate": 0.1,
+                    "platform": "anthropic"
+                }
+            }
+        })
+    )
+    db_session.add(snapshot)
+    db_session.commit()
+
+    account_bodies = []
+    class FakeWebsiteClient:
+        def __init__(self, **kwargs):
+            pass
+        def __enter__(self): return self
+        def __exit__(self, *args): return False
+        def create_account(self, body, endpoint="/accounts"):
+            account_bodies.append(body)
+            return {"id": 101, "name": body["name"]}
+        def account_exists(self, account_id): return True
+        @staticmethod
+        def extract_id(data): return "101"
+
+    monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeWebsiteClient)
+
+    websites_router.import_upstream_keys_as_accounts(
+        website.id,
+        ImportAccountsRequest(
+            upstream_key_ids=[generated.id],
+            target_group_map={"vip": "7"},
+            default_platform="gemini",
+            platform_mode="manual", # 手动模式
+        ),
+        db_session,
+        object(),
+    )
+    assert len(account_bodies) == 1
+    assert account_bodies[0]["platform"] == "gemini"