diff --git a/backend/app/database.py b/backend/app/database.py index 8b830e4..5002fc9 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -47,6 +47,8 @@ def init_db(): _migrate_custom_pages() _migrate_upstreams() _migrate_upstream_generated_keys() + _migrate_website_group_bindings_platform() + _migrate_upstream_key_account_links() # ── 已有数据库幂等索引迁移 ───────────────────────────────── @@ -197,3 +199,43 @@ def _migrate_upstream_generated_keys(): except Exception: logger = __import__("logging").getLogger(__name__) logger.warning("could not create unique indexes on upstream_generated_keys (non-fatal)") + + +def _migrate_upstream_key_account_links(): + """Create the multi-platform account mapping table and indexes.""" + inspector = inspect(engine) + if "upstream_key_account_links" not in inspector.get_table_names(): + with engine.begin() as conn: + conn.execute(text(""" + CREATE TABLE IF NOT EXISTS upstream_key_account_links ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + upstream_key_id INTEGER NOT NULL REFERENCES upstream_generated_keys(id) ON DELETE CASCADE, + website_id INTEGER NOT NULL REFERENCES websites(id) ON DELETE CASCADE, + target_group VARCHAR(255) NOT NULL DEFAULT 'default', + platform VARCHAR(64) NOT NULL, + remote_account_id VARCHAR(255) NOT NULL, + status VARCHAR(32) NOT NULL DEFAULT 'active', + created_at DATETIME, + updated_at DATETIME + ) + """)) + + with engine.begin() as conn: + conn.execute(text(""" + CREATE UNIQUE INDEX IF NOT EXISTS uq_key_website_platform + ON upstream_key_account_links(upstream_key_id, website_id, platform) + """)) + conn.execute(text(""" + CREATE UNIQUE INDEX IF NOT EXISTS uq_website_remote_account + ON upstream_key_account_links(website_id, remote_account_id) + """)) + +def _migrate_website_group_bindings_platform(): + """Add platform column to website_group_bindings if missing.""" + inspector = inspect(engine) + if "website_group_bindings" not in inspector.get_table_names(): + return + columns = {col["name"] for col in inspector.get_columns("website_group_bindings")} + if "platform" not in columns: + with engine.begin() as conn: + conn.execute(text("ALTER TABLE website_group_bindings ADD COLUMN platform VARCHAR(64)")) diff --git a/backend/app/models/upstream_key.py b/backend/app/models/upstream_key.py index 6e47cb9..50a2d49 100644 --- a/backend/app/models/upstream_key.py +++ b/backend/app/models/upstream_key.py @@ -2,7 +2,7 @@ from datetime import datetime, timezone from typing import Optional from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint -from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base @@ -30,7 +30,30 @@ class UpstreamGeneratedKey(Base): created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc)) updated_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) + account_links = relationship("UpstreamKeyAccountLink", back_populates="upstream_key", cascade="all, delete-orphan") + __table_args__ = ( UniqueConstraint("upstream_id", "group_id", "key_name", name="uq_upstream_group_key"), Index("ix_key_upstream_name", "upstream_id", "key_name"), ) + + +class UpstreamKeyAccountLink(Base): + __tablename__ = "upstream_key_account_links" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + upstream_key_id: Mapped[int] = mapped_column(Integer, ForeignKey("upstream_generated_keys.id", ondelete="CASCADE"), index=True, nullable=False) + website_id: Mapped[int] = mapped_column(Integer, ForeignKey("websites.id", ondelete="CASCADE"), index=True, nullable=False) + target_group: Mapped[str] = mapped_column(String(255), nullable=False) + platform: Mapped[str] = mapped_column(String(64), nullable=False) + remote_account_id: Mapped[str] = mapped_column(String(255), nullable=False) + status: Mapped[str] = mapped_column(String(32), default="active", nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc)) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) + + upstream_key = relationship("UpstreamGeneratedKey", back_populates="account_links") + + __table_args__ = ( + UniqueConstraint("upstream_key_id", "website_id", "platform", name="uq_key_website_platform"), + UniqueConstraint("website_id", "remote_account_id", name="uq_website_remote_account"), + ) diff --git a/backend/app/models/website.py b/backend/app/models/website.py index fc5fe56..e70215f 100644 --- a/backend/app/models/website.py +++ b/backend/app/models/website.py @@ -42,6 +42,7 @@ class WebsiteGroupBinding(Base): percent: Mapped[str] = mapped_column(String(32), default="0") algorithm: Mapped[str] = mapped_column(String(64), default="priority_weighted_plus_percent") enabled: Mapped[bool] = mapped_column(Boolean, default=True) + platform: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc)) updated_at: Mapped[datetime] = mapped_column( DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc) diff --git a/backend/app/routers/upstreams.py b/backend/app/routers/upstreams.py index cf7eabc..55dce37 100644 --- a/backend/app/routers/upstreams.py +++ b/backend/app/routers/upstreams.py @@ -15,7 +15,7 @@ from sqlalchemy.orm import Session from app.database import SessionLocal, get_db from app.models.admin_user import AdminUser from app.models.upstream import Upstream -from app.models.upstream_key import UpstreamGeneratedKey +from app.models.upstream_key import UpstreamGeneratedKey, UpstreamKeyAccountLink from app.models.upstream_recharge_event import UpstreamRechargeEvent from app.models.snapshot import UpstreamRateSnapshot from app.schemas.upstream import ( @@ -25,6 +25,7 @@ from app.schemas.upstream import ( UpstreamCreate, UpstreamUpdate, UpstreamResponse, SnapshotResponse, TestResult, UpstreamBatchActionItem, UpstreamBatchActionSummary, UpstreamBatchActionResponse, UpstreamRechargeCreate, UpstreamRechargeUpdate, UpstreamRechargeResponse, + UpstreamKeyAccountLinkResponse, ) from app.services.upstream_client import UpstreamClient, UpstreamError, _PendingKeyError, build_snapshot, build_new_api_token_name, mask_secret, _extract_key_value from app.services.auth_config import MASK, mask_auth_config, normalize_auth_config @@ -49,7 +50,13 @@ def _group_name(group: dict, gid: str) -> str: return str(group.get("name") or group.get("group_name") or gid) -def _key_response(row: UpstreamGeneratedKey, include_value: bool = False) -> GeneratedUpstreamKeyResponse: +def _key_response(row: UpstreamGeneratedKey, include_value: bool = False, db: Session | None = None) -> GeneratedUpstreamKeyResponse: + sess = db or Session.object_session(row) + imported_accounts = [] + if sess and row.id is not None: + links = sess.query(UpstreamKeyAccountLink).filter(UpstreamKeyAccountLink.upstream_key_id == row.id).all() + imported_accounts = [UpstreamKeyAccountLinkResponse.model_validate(l) for l in links] + return GeneratedUpstreamKeyResponse( id=row.id, upstream_id=row.upstream_id, @@ -67,6 +74,7 @@ def _key_response(row: UpstreamGeneratedKey, include_value: bool = False) -> Gen created_at=row.created_at, updated_at=row.updated_at, has_key_value=bool(row.key_value), + imported_accounts=imported_accounts, ) diff --git a/backend/app/routers/websites.py b/backend/app/routers/websites.py index ec5afc4..ef849c9 100644 --- a/backend/app/routers/websites.py +++ b/backend/app/routers/websites.py @@ -16,7 +16,7 @@ from sqlalchemy.orm import Session from app.database import get_db from app.models.snapshot import UpstreamRateSnapshot from app.models.upstream import Upstream -from app.models.upstream_key import UpstreamGeneratedKey +from app.models.upstream_key import UpstreamGeneratedKey, UpstreamKeyAccountLink from app.models.website import Website, WebsiteGroupBinding, WebsiteSyncLog from app.schemas.website import ( BindingCreate, @@ -63,6 +63,10 @@ from app.services.website_sync import ( sync_account_priorities_for_upstream, latest_rate_map, build_imported_account_name, + sync_key_legacy_fields, + get_or_backfill_account_links, + get_account_mapping_pairs, + deactivate_account_mapping, ) from app.utils.auth import get_current_user from app.utils.number import fixed_decimal_number @@ -248,6 +252,7 @@ def _binding_response(db: Session, row: WebsiteGroupBinding) -> BindingResponse: percent=float(row.percent or 0), algorithm=row.algorithm, enabled=row.enabled, + platform=row.platform, created_at=row.created_at, updated_at=row.updated_at, ) @@ -539,9 +544,10 @@ def import_groups_from_upstream( create_body = { "name": target_name, "description": group.get("description") or f"Imported from {upstream.name} / {source_name}", - "platform": group.get("platform") or "openai", "rate_multiplier": fixed_decimal_number(_source_group_rate(group), 2), } + if group.get("platform"): + create_body["platform"] = group["platform"] if group.get("rpm_limit") is not None: create_body["rpm_limit"] = group.get("rpm_limit") try: @@ -593,8 +599,10 @@ def import_groups_from_upstream( ) -def _normalize_platform(platform: str) -> str: +def _normalize_platform(platform: str | None) -> str: """统一平台值：小写、xai → grok。""" + if not platform: + return "" p = platform.lower().strip() if p == "xai": return "grok" @@ -604,28 +612,22 @@ def _normalize_platform(platform: str) -> str: SUPPORTED_PLATFORMS = {"openai", "anthropic", "gemini", "grok", "antigravity"} -def _resolve_platform(text: str, group_snapshot: dict | None, fallback: str = "openai") -> str: - if group_snapshot and isinstance(group_snapshot, dict): - raw = group_snapshot.get("platform") - if raw: - platform = _normalize_platform(str(raw)) - if platform in SUPPORTED_PLATFORMS: - return platform - return _detect_platform(text, fallback) +def _target_group_platform(group: dict[str, Any] | None) -> str | None: + platform = _normalize_platform(group.get("platform") if group else None) + return platform if platform in SUPPORTED_PLATFORMS else None -def _detect_platform(text: str, fallback: str = "openai") -> str: - """根据 Key 名或分组名关键词判断平台类型。""" - lower = text.lower() - if "claude" in lower or "anthropic" in lower: - return "anthropic" - if "gemini" in lower: - return "gemini" - if "grok" in lower or "xai" in lower: - return "grok" - if "antigravity" in lower: - return "antigravity" - return _normalize_platform(fallback) +def _fetch_target_group(website: Website, target_group_id: str) -> tuple[dict[str, Any], str | None]: + try: + with _client(website) as client: + groups = client.get_groups(website.groups_endpoint) or [] + except Exception as exc: + raise HTTPException(502, f"读取目标网站分组失败: {exc}") from exc + target = next((group for group in groups if str(group.get("id")) == str(target_group_id)), None) + if target is None: + raise HTTPException(400, "目标分组不存在") + platform = _target_group_platform(target) + return target, platform @router.post("/api/websites/{wid}/accounts/sync-imported-upstream-keys", response_model=ImportAccountsResponse) @@ -640,21 +642,7 @@ def sync_imported_upstream_keys( if not website: raise HTTPException(404, "website not found") - rows = ( - db.query(UpstreamGeneratedKey) - .filter( - UpstreamGeneratedKey.upstream_id == body.upstream_id, - UpstreamGeneratedKey.imported_website_id == wid, - UpstreamGeneratedKey.imported_account_id.isnot(None), - ) - .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 @@ -665,20 +653,34 @@ def sync_imported_upstream_keys( if remote_accounts is not None: remote_ids = set() + remote_map = {} for acc in remote_accounts: acc_id = _extract_id(acc) if acc_id: remote_ids.add(str(acc_id)) + remote_map[str(acc_id)] = acc else: remote_ids = None + remote_map = {} - for row in rows: - 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 + target_groups: dict[str, dict[str, Any]] = {} + try: + for group in c.get_groups(website.groups_endpoint) or []: + if group.get("id") is not None: + target_groups[str(group["id"])] = group + except Exception as exc: + logger.warning("failed to fetch groups during import status sync for website %s: %s", wid, exc) + get_or_backfill_account_links(db, wid, remote_map, target_groups) + + rows = [ + pair + for pair in get_account_mapping_pairs(db, wid, include_orphaned=True) + if pair[0].upstream_id == body.upstream_id + ] + + for row, link in rows: + old_account_id = str(link.remote_account_id) + platform = link.platform if remote_ids is None: items.append(ImportAccountItem( @@ -688,16 +690,13 @@ def sync_imported_upstream_keys( message="无法校验目标账号存在性（拉取账号列表失败）", )) elif str(old_account_id) not in remote_ids: - row.imported_website_id = None - row.imported_account_id = None - row.imported_at = None - row.status = "created" items.append(ImportAccountItem( upstream_key_id=row.id, source_group_id=row.group_id, source_group_name=row.group_name, account_id=old_account_id, platform=platform, status="stale_cleared", message="目标账号已删除，已清除导入标记", )) + deactivate_account_mapping(db, row, link) else: items.append(ImportAccountItem( upstream_key_id=row.id, source_group_id=row.group_id, @@ -801,7 +800,7 @@ def import_upstream_keys_as_accounts( upstream_key_id=kid, source_group_id="", source_group_name="", - platform=body.default_platform, + platform="", status="failed", message="key not found", ) @@ -813,11 +812,6 @@ def import_upstream_keys_as_accounts( 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: @@ -834,12 +828,15 @@ def import_upstream_keys_as_accounts( with _client(website) as c: target_group_names: dict[str, str] = {} + target_groups: dict[str, dict[str, Any]] = {} try: grps = c.get_groups(endpoint=website.groups_endpoint) or [] for g in grps: g_id = g.get("id") if g_id is not None: - target_group_names[str(g_id)] = g.get("name") or str(g_id) + group_id = str(g_id) + target_group_names[group_id] = g.get("name") or group_id + target_groups[group_id] = g except Exception as e: logger.warning("failed to fetch groups for website %s at import: %s", wid, e) @@ -854,34 +851,40 @@ def import_upstream_keys_as_accounts( for acc in remote_accounts_list: aid = c.extract_id(acc) if aid: - remote_accounts_map[aid] = acc + remote_accounts_map[str(aid)] = acc except Exception as e: logger.warning("failed to fetch remote accounts at import: %s", e) + get_or_backfill_account_links(db, wid, remote_accounts_map, target_groups) + for row in rows: - # 跳过远端已不存在或导入失败的 Key（对账后标记为 orphaned / import_failed） - if row.status in ("orphaned", "failed", "import_failed"): + target_group_id = body.target_group_map.get(row.group_id) + target_group = target_groups.get(str(target_group_id)) if target_group_id else None + platform = _target_group_platform(target_group) + if not target_group_id or target_group is None or platform is None: items.append(ImportAccountItem( upstream_key_id=row.id, source_group_id=row.group_id, source_group_name=row.group_name, - target_group_id=body.target_group_map.get(row.group_id), - platform=body.default_platform, + target_group_id=target_group_id, + platform=platform or "", + status="failed", + message="必须选择存在且包含合法 platform 的目标分组", + )) + continue + + # 跳过远端已不存在或导入失败的 Key（对账后标记为 orphaned / import_failed） + if row.status in ("orphaned", "failed"): + items.append(ImportAccountItem( + upstream_key_id=row.id, + source_group_id=row.group_id, + source_group_name=row.group_name, + target_group_id=target_group_id, + platform=platform, status="failed", message="远端 Key 已不存在，请重新生成", )) continue - # 先确定平台（失败项也需要记录） - if body.platform_mode == "auto": - 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: - platform = _normalize_platform(body.default_platform) upstream = upstreams_map.get(row.upstream_id) upstream_name = upstream.name if upstream else "Unknown" @@ -893,87 +896,114 @@ def import_upstream_keys_as_accounts( group_name=row.group_name, group_id=row.group_id, key_id=row.id, + platform=platform, ) - # 幂等校验：已导入过则检查远端账号是否仍存在 - if row.imported_website_id == wid and row.imported_account_id: - old_account_id = row.imported_account_id - if has_list_accounts: - if remote_accounts_list is None: - # 接口获取失败，保守判定为校验失败 - exists = None - else: - exists = old_account_id in remote_accounts_map - else: - # 如果 client 没有 list_accounts，则 fallback 回原先的 account_exists 检测以兼容旧 Mock 测试 - try: - exists = c.account_exists(row.imported_account_id) - except Exception: - exists = None - - if exists is True: - # 顺手回填 imported_target_group_id（老数据升级后可通过重导自动补齐） - new_tgid = body.target_group_map.get(row.group_id) or None - if new_tgid: - tg_name = target_group_names.get(str(new_tgid)) - if row.imported_target_group_id != new_tgid or row.imported_target_group_name != tg_name: - row.imported_target_group_id = new_tgid - row.imported_target_group_name = tg_name - db.commit() - - remote_name = "" - if old_account_id in remote_accounts_map: - remote_name = remote_accounts_map[old_account_id].get("name") or "" + link = db.query(UpstreamKeyAccountLink).filter( + UpstreamKeyAccountLink.upstream_key_id == row.id, + UpstreamKeyAccountLink.website_id == wid, + UpstreamKeyAccountLink.platform == platform, + ).first() + if link is None and row.imported_website_id == wid and row.imported_account_id: + legacy_account = remote_accounts_map.get(str(row.imported_account_id)) + legacy_platform = _normalize_platform(legacy_account.get("platform")) if legacy_account else "" + if legacy_account is not None and legacy_platform == platform: + link = UpstreamKeyAccountLink( + upstream_key_id=row.id, + website_id=wid, + target_group=str(target_group_id), + platform=platform, + remote_account_id=str(row.imported_account_id), + status="active", + ) + db.add(link) + db.commit() + sync_key_legacy_fields(db, row.id) + has_any_link = db.query(UpstreamKeyAccountLink.id).filter( + UpstreamKeyAccountLink.upstream_key_id == row.id, + ).first() + if ( + link is None + and has_any_link is None + and row.imported_website_id == wid + and row.imported_account_id + and remote_accounts_list is None + ): + items.append(ImportAccountItem( + upstream_key_id=row.id, + source_group_id=row.group_id, + source_group_name=row.group_name, + target_group_id=target_group_id, + account_id=str(row.imported_account_id), + platform=platform, + upstream_base_url=upstream_base_url, + status="check_failed", + message="无法校验旧账号状态，已保守跳过", + )) + continue + if link and link.status == "active": + old_account_id = str(link.remote_account_id) + if remote_accounts_list is None: items.append(ImportAccountItem( upstream_key_id=row.id, source_group_id=row.group_id, source_group_name=row.group_name, - target_group_id=body.target_group_map.get(row.group_id), + target_group_id=target_group_id, account_id=old_account_id, - account_name=remote_name or expected_new_name, platform=platform, upstream_base_url=upstream_base_url, - status="exists", - message="已导入过，已跳过", - )) - continue - elif exists is False: - # 远端已删除，清空标记后继续创建 - row.imported_website_id = None - row.imported_account_id = None - row.imported_at = None - row.status = "created" - # 继续往下走（不 continue） - else: - # 校验失败，保守跳过 - items.append(ImportAccountItem( - upstream_key_id=row.id, - source_group_id=row.group_id, - source_group_name=row.group_name, - target_group_id=body.target_group_map.get(row.group_id), - account_id=old_account_id, - platform=platform, status="check_failed", message="无法校验目标账号状态，已保守跳过", )) continue + remote_account = remote_accounts_map.get(old_account_id) + remote_platform = _normalize_platform(remote_account.get("platform")) if remote_account else "" + if remote_account is not None and remote_platform == platform: + current_group_ids = remote_account.get("group_ids") or [] + if not isinstance(current_group_ids, list): + current_group_ids = [current_group_ids] + target_value = _numeric_group_id(target_group_id) + target_value = target_value if target_value is not None else target_group_id + new_group_ids = [g for g in current_group_ids if str(g) != str(link.target_group)] + if str(target_group_id) not in {str(g) for g in new_group_ids}: + new_group_ids.append(target_value) + if {str(g) for g in current_group_ids} != {str(g) for g in new_group_ids}: + c.update_account(old_account_id, {"group_ids": new_group_ids}) + link.target_group = str(target_group_id) + db.commit() + sync_key_legacy_fields(db, row.id) + items.append(ImportAccountItem( + upstream_key_id=row.id, + source_group_id=row.group_id, + source_group_name=row.group_name, + target_group_id=target_group_id, + account_id=old_account_id, + account_name=str(remote_account.get("name") or expected_new_name), + platform=platform, + upstream_base_url=upstream_base_url, + status="exists", + message="已存在并已对齐目标分组", + raw=remote_account, + )) + continue + link.status = "stale" + db.commit() + sync_key_legacy_fields(db, row.id) if not row.key_value: items.append(ImportAccountItem( upstream_key_id=row.id, source_group_id=row.group_id, source_group_name=row.group_name, + target_group_id=target_group_id, platform=platform, - status="failed", + status="check_failed", message="该 Key 无明文值，无法导入（远端已存在 Key 不会保留明文，请重新创建或手动填入）", )) continue - target_group_id = body.target_group_map.get(row.group_id) - group_ids = [] numeric_target = _numeric_group_id(target_group_id) - if numeric_target is not None: - group_ids.append(numeric_target) + group_ids = [numeric_target if numeric_target is not None else target_group_id] account_name = expected_new_name account_body = { "name": account_name, @@ -992,14 +1022,25 @@ def import_upstream_keys_as_accounts( try: created = c.create_account(account_body) account_id = c.extract_id(created) - row.imported_website_id = wid - row.imported_account_id = account_id or None - row.imported_at = datetime.now(timezone.utc) - row.imported_target_group_id = target_group_id or None - row.imported_target_group_name = target_group_names.get(str(target_group_id)) if target_group_id else None - row.status = "imported" + if not account_id: + raise WebsiteError("创建账号成功但响应缺少账号 ID") + if link: + link.target_group = str(target_group_id) + link.remote_account_id = str(account_id) + link.status = "active" + else: + db.add(UpstreamKeyAccountLink( + upstream_key_id=row.id, + website_id=wid, + target_group=str(target_group_id), + platform=platform, + remote_account_id=str(account_id), + status="active", + )) row.error = None db.commit() + sync_key_legacy_fields(db, row.id) + remote_accounts_map[str(account_id)] = created items.append(ImportAccountItem( upstream_key_id=row.id, source_group_id=row.group_id, @@ -1015,8 +1056,13 @@ def import_upstream_keys_as_accounts( )) except Exception as exc: logger.exception("import upstream key as account failed website=%s key=%s", wid, row.id) - row.status = "import_failed" row.error = str(exc) + has_active_link = db.query(UpstreamKeyAccountLink.id).filter( + UpstreamKeyAccountLink.upstream_key_id == row.id, + UpstreamKeyAccountLink.status == "active", + ).first() + if not has_active_link: + row.status = "import_failed" db.commit() items.append(ImportAccountItem( upstream_key_id=row.id, @@ -1060,20 +1106,31 @@ def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClie for acc in remote_accounts: aid = c.extract_id(acc) if aid: - remote_accounts_map[aid] = acc + remote_accounts_map[str(aid)] = acc + + target_groups: dict[str, dict[str, Any]] = {} + try: + for group in c.get_groups(website.groups_endpoint) or []: + if group.get("id") is not None: + target_groups[str(group["id"])] = group + except Exception as exc: + logger.warning("failed to fetch groups during cleanup for website %s: %s", website.id, exc) + get_or_backfill_account_links(db, website.id, remote_accounts_map, target_groups) candidates: dict[str, dict[str, Any]] = {} # 2. Category: missing_remote_account # Local keys that still reference an imported_account_id, # but that account no longer appears in the remote account list. - imported_keys = db.query(UpstreamGeneratedKey).filter( - UpstreamGeneratedKey.imported_website_id == website.id, - UpstreamGeneratedKey.imported_account_id.isnot(None), - ).all() + all_pairs = get_account_mapping_pairs(db, website.id, include_orphaned=True) + local_pair_by_account = {str(link.remote_account_id): (key, link) for key, link in all_pairs} + imported_pairs = [ + (key, link) for key, link in all_pairs + if key.status != "orphaned" and link.status == "active" + ] - for key in imported_keys: - acc_id = key.imported_account_id + for key, link in imported_pairs: + acc_id = str(link.remote_account_id) if acc_id in candidates: continue if acc_id in remote_accounts_map: @@ -1089,18 +1146,18 @@ def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClie "reason": "目标账号已被手动删除", "cleanup_action": "clear_local_marker", "db_key": key, + "db_link": link, } # 3. Category: local_orphan_account # Local keys with status=orphaned that still carry an imported_account_id. - orphaned_keys = db.query(UpstreamGeneratedKey).filter( - UpstreamGeneratedKey.status == "orphaned", - UpstreamGeneratedKey.imported_website_id == website.id, - UpstreamGeneratedKey.imported_account_id.isnot(None), - ).all() + orphaned_pairs = [ + (key, link) for key, link in all_pairs + if key.status == "orphaned" and link.status == "active" + ] - for key in orphaned_keys: - acc_id = key.imported_account_id + for key, link in orphaned_pairs: + acc_id = str(link.remote_account_id) remote_acc = remote_accounts_map.get(acc_id) acc_name = remote_acc.get("name") if remote_acc else key.key_name upstream = db.query(Upstream).filter(Upstream.id == key.upstream_id).first() @@ -1113,6 +1170,7 @@ def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClie candidates[acc_id]["upstream_key_id"] = key.id candidates[acc_id]["upstream_name"] = upstream_name candidates[acc_id]["source_group_name"] = key.group_name + candidates[acc_id]["db_link"] = link continue if remote_acc: @@ -1125,6 +1183,7 @@ def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClie "reason": "本地上游 Key 已归档", "cleanup_action": "delete_remote_and_clear_marker", "db_key": key, + "db_link": link, } else: candidates[acc_id] = { @@ -1136,6 +1195,7 @@ def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClie "reason": "本地上游 Key 已归档，且目标账号已被手动删除", "cleanup_action": "clear_local_marker", "db_key": key, + "db_link": link, } # 4. Category: remote_orphan_account @@ -1166,10 +1226,12 @@ def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClie if not key: is_orphan = True reason = "本地找不到对应 Key" - elif key.imported_website_id != website.id or key.imported_account_id != aid: + local_pair = local_pair_by_account.get(aid) + link = local_pair[1] if local_pair else None + if key and (local_pair is None or local_pair[0].id != key.id): is_orphan = True reason = "本地 Key 与目标账号关联不一致" - elif key.status == 'orphaned': + elif key and key.status == 'orphaned': is_orphan = True reason = "本地上游 Key 已归档" @@ -1190,6 +1252,7 @@ def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClie "reason": reason, "cleanup_action": "delete_remote_account", "db_key": key, + "db_link": link, } return candidates @@ -1215,8 +1278,23 @@ def _test_candidate_account(c: Sub2ApiWebsiteClient, account_id: str) -> tuple[s return "test_unknown", False, f"测试时发生异常: {e}" -def _clear_local_import_marker(key: UpstreamGeneratedKey, db: Session) -> None: - """清空单条 Key 上的 SmartUp 导入标记字段。""" +def _clear_local_import_marker( + key: UpstreamGeneratedKey, + db: Session, + link: UpstreamKeyAccountLink | None = None, +) -> None: + """Deactivate one account mapping and refresh the compatibility marker.""" + if link is not None: + if isinstance(link, UpstreamKeyAccountLink): + deactivate_account_mapping(db, key, link) + else: + key.imported_website_id = None + key.imported_account_id = None + key.imported_at = None + key.imported_target_group_id = None + key.imported_target_group_name = None + db.add(key) + return key.imported_website_id = None key.imported_account_id = None key.imported_at = None @@ -1300,7 +1378,7 @@ def execute_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _= if action == "clear_local_marker": key = cand.get("db_key") if key: - _clear_local_import_marker(key, db) + _clear_local_import_marker(key, db, cand.get("db_link")) action_status = "cleared" msg = "已清除本地导入标记" else: @@ -1320,7 +1398,7 @@ def execute_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _= if action == "delete_remote_and_clear_marker": key = cand.get("db_key") if key: - _clear_local_import_marker(key, db) + _clear_local_import_marker(key, db, cand.get("db_link")) msg += "，已清除本地导入标记" else: action_status = "failed" @@ -1405,16 +1483,22 @@ def set_website_accounts_concurrency( for acc in remote_accounts: aid = c.extract_id(acc) if aid: - remote_map[aid] = acc + remote_map[str(aid)] = acc - keys = db.query(UpstreamGeneratedKey).filter( - UpstreamGeneratedKey.imported_website_id == wid, - UpstreamGeneratedKey.imported_account_id.isnot(None) - ).all() + target_groups: dict[str, dict[str, Any]] = {} + try: + for group in c.get_groups(website.groups_endpoint) or []: + if group.get("id") is not None: + target_groups[str(group["id"])] = group + except Exception as exc: + logger.warning("failed to fetch groups while setting concurrency for website %s: %s", wid, exc) + get_or_backfill_account_links(db, wid, remote_map, target_groups) + + account_pairs = get_account_mapping_pairs(db, wid, include_orphaned=False) candidates = {} - for key in keys: - aid = key.imported_account_id + for key, link in account_pairs: + aid = str(link.remote_account_id) if aid not in candidates: candidates[aid] = { "account_id": aid, @@ -1595,17 +1679,22 @@ def _sync_upstream_models_generator(wid: int, db: Session): for acc in remote_accounts: aid = c.extract_id(acc) if aid: - remote_map[aid] = acc + remote_map[str(aid)] = acc - # 1. 查找候选账号：本地 UpstreamGeneratedKey.imported_website_id == wid 且 imported_account_id 非空 - keys = db.query(UpstreamGeneratedKey).filter( - UpstreamGeneratedKey.imported_website_id == wid, - UpstreamGeneratedKey.imported_account_id.isnot(None) - ).all() + target_groups: dict[str, dict[str, Any]] = {} + try: + for group in c.get_groups(website.groups_endpoint) or []: + if group.get("id") is not None: + target_groups[str(group["id"])] = group + except Exception as exc: + logger.warning("failed to fetch groups while syncing models for website %s: %s", wid, exc) + get_or_backfill_account_links(db, wid, remote_map, target_groups) + + account_pairs = get_account_mapping_pairs(db, wid, include_orphaned=False) candidates = {} - for key in keys: - aid = key.imported_account_id + for key, link in account_pairs: + aid = str(link.remote_account_id) if aid not in candidates: candidates[aid] = { "account_id": aid, @@ -1937,6 +2026,11 @@ def _organize_website_groups_generator(wid: int, db: Session): ) for row in hist_group_ids: smartup_managed_group_ids.add(str(row[0])) + linked_group_ids = db.query(UpstreamKeyAccountLink.target_group).filter( + UpstreamKeyAccountLink.website_id == wid, + UpstreamKeyAccountLink.status == "active", + ).distinct().all() + smartup_managed_group_ids.update(str(row[0]) for row in linked_group_ids if row[0]) items: list[OrganizeGroupsItem] = [] @@ -1989,13 +2083,16 @@ def _organize_website_groups_generator(wid: int, db: Session): with _client(website) as c: # 获取远端分组名称列表和账号列表 target_group_names: dict[str, str] = {} + target_group_map: dict[str, dict[str, Any]] = {} remote_accounts: list[dict[str, Any]] = [] try: grps = c.get_groups(endpoint=website.groups_endpoint) or [] for g in grps: g_id = g.get("id") if g_id is not None: - target_group_names[str(g_id)] = g.get("name") or str(g_id) + group_id = str(g_id) + target_group_names[group_id] = g.get("name") or group_id + target_group_map[group_id] = g except Exception as e: logger.warning("failed to fetch groups for website %s at organize: %s", wid, e) @@ -2019,6 +2116,9 @@ def _organize_website_groups_generator(wid: int, db: Session): if acc_id: remote_account_map[str(acc_id)] = acc + # 回填遗留 Key 到 UpstreamKeyAccountLink 映射 + get_or_backfill_account_links(db, wid, remote_account_map, target_group_map) + # 预载所有相关的上游快照映射 all_upstream_ids = set() for b in bindings: @@ -2035,6 +2135,36 @@ def _organize_website_groups_generator(wid: int, db: Session): for b in bindings: target_group_id = b.target_group_id target_group_name = target_group_names.get(str(target_group_id)) or b.target_group_name or str(target_group_id) + target_group = target_group_map.get(str(target_group_id)) + binding_platform = _target_group_platform(target_group) + if binding_platform is None: + for src in binding_sources(b): + uid = src.get("upstream_id") + gid = src.get("group_id") + if not uid or not gid: + continue + upstream = db.query(Upstream).filter(Upstream.id == uid).first() + sname = src.get("group_name") or gid + _item = OrganizeGroupsItem( + target_group_id=str(target_group_id), + target_group_name=target_group_name, + upstream_name=upstream.name if upstream else f"#{uid}", + source_group_id=str(gid), + source_group_name=str(sname), + key_name="", + status="failed", + message="目标分组不存在或缺少合法 platform，无法确定账号平台", + ) + items.append(_item) + yield "item", _item.model_dump() + if time.monotonic() - last_event_at >= 10: + yield "heartbeat", {} + last_event_at = time.monotonic() + continue + if b.platform != binding_platform or b.target_group_name != target_group_name: + b.platform = binding_platform + b.target_group_name = target_group_name + db.commit() sources = binding_sources(b) for src in sources: @@ -2113,18 +2243,33 @@ def _organize_website_groups_generator(wid: int, db: Session): last_event_at = time.monotonic() continue - # 检测平台类型 - 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", - ) - - # 1. 幂等校验：已导入过则检查远端账号是否仍存在 - if row.imported_website_id == wid and row.imported_account_id: - old_account_id = row.imported_account_id + # 平台唯一来源：目标分组绑定的 platform 字段 + platform = binding_platform + # 1. 幂等校验：仅按 UpstreamKeyAccountLink (key_id, website_id, platform) 查找，不复用 legacy 字段 + link = db.query(UpstreamKeyAccountLink).filter( + UpstreamKeyAccountLink.upstream_key_id == row.id, + UpstreamKeyAccountLink.website_id == wid, + UpstreamKeyAccountLink.platform == platform, + UpstreamKeyAccountLink.status == "active", + ).first() + if link is None and row.imported_website_id == wid and row.imported_account_id: + legacy_account = remote_account_map.get(str(row.imported_account_id)) + legacy_platform = _normalize_platform(legacy_account.get("platform")) if legacy_account else "" + if legacy_account is not None and legacy_platform == platform: + link = UpstreamKeyAccountLink( + upstream_key_id=row.id, + website_id=wid, + target_group=str(target_group_id), + platform=platform, + remote_account_id=str(row.imported_account_id), + status="active", + ) + db.add(link) + db.commit() + sync_key_legacy_fields(db, row.id) + old_account_id = link.remote_account_id if link else None + if old_account_id: if remote_account_ids is None: # 账号列表获取失败，无法校验状态，保守跳过 _item = OrganizeGroupsItem( @@ -2148,7 +2293,17 @@ def _organize_website_groups_generator(wid: int, db: Session): remote_acc = remote_account_map.get(str(old_account_id)) if remote_acc is not None: - # 目标账号存在：检测并对齐目标分组，移除其余已启用的 SmartUp 托管分组，保留非 SmartUp 托管分组 + remote_platform = _normalize_platform(remote_acc.get("platform")) + if remote_platform != platform: + link.status = "stale" + db.commit() + sync_key_legacy_fields(db, row.id) + old_account_id = None + is_recreated = False + remote_acc = None + + if remote_acc is not None: + # 目标账号存在：仅对齐目标分组的 group_ids，绝不修改远端账号 platform try: current_group_ids = remote_acc.get("group_ids") or [] if not isinstance(current_group_ids, list): @@ -2157,61 +2312,23 @@ def _organize_website_groups_generator(wid: int, db: Session): numeric_target = _numeric_group_id(target_group_id) target_val = numeric_target if numeric_target is not None else target_group_id - # 过滤出非 SmartUp 托管的 group_ids new_group_ids = [g for g in current_group_ids if str(g) not in smartup_managed_group_ids] - # 追加当前目标分组 new_group_ids.append(target_val) 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: - 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: - # 主动核对防线：防止子站忽略 platform 修改而发生静默失败 - fresh_accs = c.list_accounts() - if fresh_accs is None: - raise WebsiteError("复查失败：拉取远端账号列表返回空，无法核对平台修改是否生效") - fresh_acc = next((a for a in fresh_accs if str(c.extract_id(a)) == str(old_account_id)), None) - if not fresh_acc: - raise WebsiteError(f"复查失败：在远端账号列表中未找到要更新的账号 {old_account_id}") - - remote_account_map[str(old_account_id)] = fresh_acc - new_platform = fresh_acc.get("platform") - if str(new_platform).lower() != str(platform).lower(): - raise WebsiteError( - f"远端账号不支持修改平台属性，尝试将平台从 {remote_platform} 修改为 {platform} 失败" - ) - 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) + c.update_account(old_account_id, {"group_ids": new_group_ids}) + remote_acc["group_ids"] = new_group_ids + msg = "已存在，已迁移到目标分组" else: msg = "已存在且已对齐，已跳过" - # 补齐本地 DB 的目标分组标记 - if row.imported_target_group_id != target_group_id: - row.imported_target_group_id = target_group_id - row.imported_target_group_name = target_group_names.get(str(target_group_id)) - db.commit() + # 更新链接的 target_group（同平台换组迁移） + link.target_group = str(target_group_id) + db.commit() + sync_key_legacy_fields(db, row.id) items.append(_item := OrganizeGroupsItem( target_group_id=str(target_group_id), @@ -2248,14 +2365,13 @@ def _organize_website_groups_generator(wid: int, db: Session): yield "heartbeat", {} last_event_at = time.monotonic() continue - else: - # 远端已删除，清理标记后进行重建 - row.imported_website_id = None - row.imported_account_id = None - row.imported_at = None - row.status = "created" + elif old_account_id: + link.status = "stale" db.commit() + sync_key_legacy_fields(db, row.id) is_recreated = True + elif not old_account_id: + is_recreated = False else: is_recreated = False @@ -2291,6 +2407,7 @@ def _organize_website_groups_generator(wid: int, db: Session): group_name=row.group_name, group_id=row.group_id, key_id=row.id, + platform=platform, ) priority_val = rate_priority_map.get((str(target_group_id), row.upstream_id, row.group_id), 1) account_body = { @@ -2310,12 +2427,30 @@ def _organize_website_groups_generator(wid: int, db: Session): try: created = c.create_account(account_body) account_id = c.extract_id(created) - row.imported_website_id = wid - row.imported_account_id = account_id or None - row.imported_at = datetime.now(timezone.utc) - row.imported_target_group_id = target_group_id or None - row.imported_target_group_name = target_group_names.get(str(target_group_id)) if target_group_id else None - row.status = "imported" + + # 更新或创建 UpstreamKeyAccountLink 映射（upsert） + if account_id: + existing_link = db.query(UpstreamKeyAccountLink).filter( + UpstreamKeyAccountLink.upstream_key_id == row.id, + UpstreamKeyAccountLink.website_id == wid, + UpstreamKeyAccountLink.platform == platform, + ).first() + if existing_link: + existing_link.remote_account_id = str(account_id) + existing_link.target_group = str(target_group_id) + existing_link.status = "active" + else: + new_link = UpstreamKeyAccountLink( + upstream_key_id=row.id, + website_id=wid, + target_group=str(target_group_id), + platform=platform, + remote_account_id=str(account_id), + status="active", + ) + db.add(new_link) + db.commit() + sync_key_legacy_fields(db, row.id) row.error = None db.commit() @@ -2545,14 +2680,16 @@ def create_binding(body: BindingCreate, db: Session = Depends(get_db), _=Depends if body.algorithm not in ALGORITHMS: raise HTTPException(400, "不支持的算法") _ensure_unique_target(db, body.website_id, body.target_group_id) + target_group, platform = _fetch_target_group(website, body.target_group_id) row = WebsiteGroupBinding( website_id=body.website_id, target_group_id=body.target_group_id, - target_group_name=body.target_group_name, + target_group_name=str(target_group.get("name") or body.target_group_name or body.target_group_id), source_groups_json=json.dumps([item.model_dump() for item in body.source_groups], ensure_ascii=False), percent=str(body.percent), algorithm=body.algorithm, enabled=body.enabled, + platform=platform, ) db.add(row) db.commit() @@ -2577,12 +2714,21 @@ def update_binding(bid: int, body: BindingUpdate, db: Session = Depends(get_db), next_website_id = int(data.get("website_id", row.website_id)) next_target_group_id = str(data.get("target_group_id", row.target_group_id)) _ensure_unique_target(db, next_website_id, next_target_group_id, exclude_id=row.id) + website = db.query(Website).filter(Website.id == next_website_id).first() + target_changed = next_website_id != row.website_id or next_target_group_id != row.target_group_id + if not target_changed and not row.enabled and data.get("enabled") is not True: + target_group = {"name": row.target_group_name} + platform = row.platform + else: + target_group, platform = _fetch_target_group(website, next_target_group_id) if "source_groups" in data: row.source_groups_json = json.dumps(data.pop("source_groups"), ensure_ascii=False) if "percent" in data: row.percent = str(data.pop("percent")) for key, value in data.items(): setattr(row, key, value) + row.target_group_name = str(target_group.get("name") or row.target_group_name or next_target_group_id) + row.platform = platform row.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(row) diff --git a/backend/app/schemas/upstream.py b/backend/app/schemas/upstream.py index 0063a83..9ad57bc 100644 --- a/backend/app/schemas/upstream.py +++ b/backend/app/schemas/upstream.py @@ -134,6 +134,20 @@ class GenerateKeysByGroupsRequest(BaseModel): endpoint: str = "/keys" +class UpstreamKeyAccountLinkResponse(BaseModel): + id: int + upstream_key_id: int + website_id: int + target_group: str + platform: str + remote_account_id: str + status: str + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None + + model_config = {"from_attributes": True} + + class GeneratedUpstreamKeyResponse(BaseModel): id: Optional[int] = None upstream_id: int @@ -151,6 +165,7 @@ class GeneratedUpstreamKeyResponse(BaseModel): created_at: Optional[datetime] = None updated_at: Optional[datetime] = None has_key_value: bool = False + imported_accounts: list[UpstreamKeyAccountLinkResponse] = Field(default_factory=list) class GenerateKeysByGroupsResponse(BaseModel): diff --git a/backend/app/schemas/website.py b/backend/app/schemas/website.py index bc702fb..1b00636 100644 --- a/backend/app/schemas/website.py +++ b/backend/app/schemas/website.py @@ -65,6 +65,7 @@ class WebsiteGroupResponse(BaseModel): name: str rate_multiplier: Optional[str] = None description: Optional[str] = None + platform: Optional[str] = None raw: dict[str, Any] = {} @@ -115,6 +116,7 @@ class BindingResponse(BaseModel): percent: float algorithm: str enabled: bool + platform: Optional[str] = None created_at: datetime updated_at: datetime diff --git a/backend/app/services/website_client.py b/backend/app/services/website_client.py index 5cc27bc..2c311a4 100644 --- a/backend/app/services/website_client.py +++ b/backend/app/services/website_client.py @@ -134,13 +134,17 @@ def normalize_groups(value: Any) -> list[dict[str, Any]]: name = item.get("name") or item.get("group_name") or str(gid) rate = item.get("rate_multiplier") or item.get("rateMultiplier") or item.get("ratio") desc = item.get("description") or item.get("desc") or item.get("remark") - groups.append({ + platform = item.get("platform") + group = { "id": str(gid), "name": str(name), "rate_multiplier": fixed_decimal_string(rate, 2) if rate is not None else None, "description": str(desc) if desc is not None else None, "raw": item, - }) + } + if platform is not None: + group["platform"] = str(platform) + groups.append(group) return groups diff --git a/backend/app/services/website_sync.py b/backend/app/services/website_sync.py index 4452333..b837dd0 100644 --- a/backend/app/services/website_sync.py +++ b/backend/app/services/website_sync.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import logging +from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_UP from datetime import datetime, timezone from typing import Any @@ -11,7 +12,7 @@ from sqlalchemy.orm import Session from app.database import SessionLocal from app.models.snapshot import UpstreamRateSnapshot from app.models.upstream import Upstream -from app.models.upstream_key import UpstreamGeneratedKey +from app.models.upstream_key import UpstreamGeneratedKey, UpstreamKeyAccountLink from app.models.website import Website, WebsiteGroupBinding, WebsiteSyncLog from app.services.auth_config import normalize_auth_config from app.services.website_client import Sub2ApiWebsiteClient, WebsiteError, _extract_id, calculate_target_rate, parse_positive_decimal @@ -22,10 +23,219 @@ from app.utils.number import fixed_decimal_string logger = logging.getLogger(__name__) -def build_imported_account_name(upstream_name: str, group_name: str | None, group_id: str, key_id: int) -> str: - """构建导入的账号名称：{上游网站名}-{上游分组名或分组ID}-{KeyID}""" +@dataclass +class LegacyAccountLink: + """Read-only compatibility view used only when a key has no mapping rows.""" + + upstream_key_id: int + website_id: int + target_group: str + remote_account_id: str + platform: str = "" + status: str = "active" + + +def get_account_mapping_pairs( + db: Session, + website_id: int, + *, + include_orphaned: bool = True, +) -> list[tuple[UpstreamGeneratedKey, UpstreamKeyAccountLink | LegacyAccountLink]]: + query = db.query(UpstreamGeneratedKey, UpstreamKeyAccountLink).join( + UpstreamKeyAccountLink, UpstreamKeyAccountLink.upstream_key_id == UpstreamGeneratedKey.id, + ).filter( + UpstreamKeyAccountLink.website_id == website_id, + UpstreamKeyAccountLink.status == "active", + ) + if not include_orphaned: + query = query.filter(UpstreamGeneratedKey.status != "orphaned") + pairs = list(query.all()) + mapped_key_ids = {row.id for row, _ in pairs} + + legacy_filters = [ + UpstreamGeneratedKey.imported_website_id == website_id, + UpstreamGeneratedKey.imported_account_id.isnot(None), + ] + if mapped_key_ids: + legacy_filters.append(~UpstreamGeneratedKey.id.in_(mapped_key_ids)) + legacy_query = db.query(UpstreamGeneratedKey).filter(*legacy_filters) + if not include_orphaned: + legacy_query = legacy_query.filter(UpstreamGeneratedKey.status != "orphaned") + for row in legacy_query.all(): + pairs.append((row, LegacyAccountLink( + upstream_key_id=row.id, + website_id=website_id, + target_group=str(row.imported_target_group_id or row.group_id), + remote_account_id=str(row.imported_account_id), + ))) + return pairs + + +def deactivate_account_mapping( + db: Session, + key: UpstreamGeneratedKey, + link: UpstreamKeyAccountLink | LegacyAccountLink, +) -> None: + if isinstance(link, UpstreamKeyAccountLink): + link.status = "stale" + db.commit() + sync_key_legacy_fields(db, key.id) + return + key.imported_website_id = None + key.imported_account_id = None + key.imported_at = None + key.imported_target_group_id = None + key.imported_target_group_name = None + if key.status == "imported": + key.status = "created" + db.commit() + + +def _normalize_platform(platform: str | None) -> str: + if not platform: + return "" + p = str(platform).lower().strip() + if p in ("codex", "gpt", "openai"): + return "openai" + if p in ("claude", "anthropic"): + return "anthropic" + if p in ("xai", "grok"): + return "grok" + return p + + +def build_imported_account_name( + upstream_name: str, + group_name: str | None, + group_id: str, + key_id: int, + platform: str | None = None, +) -> str: + """构建导入的账号名称：{上游网站名}-{上游分组名或分组ID}-{KeyID}（含平台后缀）。""" grp = group_name if group_name else group_id - return f"{upstream_name}-{grp}-{key_id}" + base = f"{upstream_name}-{grp}-{key_id}" + norm_p = _normalize_platform(platform) if platform else None + if norm_p: + return f"{base}-{norm_p}" + return base + + +def sync_key_legacy_fields(db: Session, key_id: int) -> None: + """将 UpstreamKeyAccountLink 中的首条活跃链接同步到 UpstreamGeneratedKey 的 legacy 字段。""" + from app.models.upstream_key import UpstreamKeyAccountLink + db.flush() + key = db.query(UpstreamGeneratedKey).filter(UpstreamGeneratedKey.id == key_id).first() + if not key: + return + active_links = ( + db.query(UpstreamKeyAccountLink) + .filter( + UpstreamKeyAccountLink.upstream_key_id == key_id, + UpstreamKeyAccountLink.status == "active", + ) + .order_by( + UpstreamKeyAccountLink.website_id.asc(), + UpstreamKeyAccountLink.platform.asc(), + UpstreamKeyAccountLink.id.asc(), + ) + .all() + ) + + if active_links: + first = active_links[0] + key.imported_website_id = first.website_id + key.imported_account_id = first.remote_account_id + key.imported_target_group_id = first.target_group + + # 查找绑定以获取目标分组名称 + binding = ( + db.query(WebsiteGroupBinding) + .filter( + WebsiteGroupBinding.website_id == first.website_id, + WebsiteGroupBinding.target_group_id == first.target_group, + ) + .first() + ) + if binding and binding.target_group_name: + key.imported_target_group_name = binding.target_group_name + elif not key.imported_target_group_name: + key.imported_target_group_name = first.target_group + + if not key.imported_at: + key.imported_at = first.created_at or datetime.now(timezone.utc) + key.status = "imported" + else: + key.imported_website_id = None + key.imported_account_id = None + key.imported_target_group_id = None + key.imported_target_group_name = None + key.imported_at = None + if key.status in ("imported", "import_failed"): + key.status = "created" + db.commit() + + +def get_or_backfill_account_links( + db: Session, + website_id: int, + remote_account_map: dict[str, dict[str, Any]] | None = None, + target_group_map: dict[str, dict[str, Any]] | None = None, +) -> list[UpstreamKeyAccountLink]: + """Backfill legacy markers only when account and target-group platforms agree.""" + existing_links = ( + db.query(UpstreamKeyAccountLink) + .filter(UpstreamKeyAccountLink.website_id == website_id) + .all() + ) + + linked_account_ids = {l.remote_account_id for l in existing_links} + + # 查询只有 legacy imported_account_id 的 Key + legacy_keys = ( + db.query(UpstreamGeneratedKey) + .filter( + UpstreamGeneratedKey.imported_website_id == website_id, + UpstreamGeneratedKey.imported_account_id.isnot(None), + ) + .all() + ) + + created_any = False + for k in legacy_keys: + if k.imported_account_id in linked_account_ids: + continue + remote_acc = remote_account_map.get(str(k.imported_account_id)) if remote_account_map else None + target_group_id = str(k.imported_target_group_id or "") + target_group = target_group_map.get(target_group_id) if target_group_map and target_group_id else None + if not remote_acc or not target_group: + continue + remote_p = _normalize_platform(remote_acc.get("platform")) + target_p = _normalize_platform(target_group.get("platform")) + supported = {"openai", "anthropic", "gemini", "grok", "antigravity"} + if remote_p not in supported or target_p not in supported or remote_p != target_p: + continue + + link = UpstreamKeyAccountLink( + upstream_key_id=k.id, + website_id=website_id, + target_group=target_group_id, + platform=remote_p, + remote_account_id=str(k.imported_account_id), + status="active", + ) + db.add(link) + linked_account_ids.add(str(k.imported_account_id)) + created_any = True + + if created_any: + db.commit() + existing_links = ( + db.query(UpstreamKeyAccountLink) + .filter(UpstreamKeyAccountLink.website_id == website_id) + .all() + ) + + return existing_links PRIORITY_BASE = 1 @@ -286,13 +496,15 @@ def _priority_result( old_priority: Any = None, source_rate: float | None = None, target_group_name: str | None = None, + link: UpstreamKeyAccountLink | LegacyAccountLink | None = None, ) -> dict: """构建统一的优先级同步结果 dict，支持丰富的展示和后向兼容性。""" - tg_id = row.imported_target_group_id or row.group_id + tg_id = link.target_group if link else (row.imported_target_group_id or row.group_id) tg_name = target_group_name or row.imported_target_group_name or tg_id + account_id = link.remote_account_id if link else row.imported_account_id return { # 兼容旧字段 - "account_id": row.imported_account_id, + "account_id": account_id, "group_id": row.group_id, "upstream_id": row.upstream_id, "old_priority": old_priority, @@ -302,7 +514,7 @@ def _priority_result( # 新增增强型字段 "target_group_id": tg_id, "target_group_name": tg_name, - "account_name": account_name or row.imported_account_id, + "account_name": account_name or account_id, "source_group_id": row.group_id, "source_group_name": row.group_name or row.group_id, "source_rate": source_rate, @@ -546,6 +758,8 @@ def sync_account_priorities_for_website( # ── 1. 建立远端 client 连接并加载账号与分组列表 ──────────────────────── account_info_map: dict[str, dict[str, Any]] = {} target_group_names: dict[str, str] = {} + remote_account_map: dict[str, dict[str, Any]] = {} + target_group_map: dict[str, dict[str, Any]] = {} remote_account_ids: set[str] | None = None site_results: list[dict] = [] competitive_buckets: dict[str, list] = {} @@ -572,6 +786,7 @@ def sync_account_priorities_for_website( "name": acc.get("name") or acc.get("account_name") or acc_id_str, "priority": acc.get("priority"), } + remote_account_map[acc_id_str] = acc except Exception as e: logger.warning("failed to fetch remote accounts for website %s: %s", website_id, e) @@ -580,28 +795,40 @@ def sync_account_priorities_for_website( for g in grps: g_id = g.get("id") if g_id is not None: - target_group_names[str(g_id)] = g.get("name") or str(g_id) + group_id = str(g_id) + target_group_names[group_id] = g.get("name") or group_id + target_group_map[group_id] = g except Exception as e: logger.warning("failed to fetch groups for website %s at sync: %s", website_id, e) - # ── 2. 查询该网站所有已导入 Key（跨上游），用于倍率查询 ──────────── - all_website_keys = ( - db.query(UpstreamGeneratedKey) - .filter( - UpstreamGeneratedKey.imported_website_id == website_id, - UpstreamGeneratedKey.imported_account_id.isnot(None), - UpstreamGeneratedKey.status != "orphaned", + legacy_missing_groups = db.query(UpstreamGeneratedKey).filter( + UpstreamGeneratedKey.imported_website_id == website_id, + UpstreamGeneratedKey.imported_account_id.isnot(None), + UpstreamGeneratedKey.imported_target_group_id.is_(None), + ).all() + for row in legacy_missing_groups: + group_id, group_name = _remote_account_single_group( + remote_account_map.get(str(row.imported_account_id)) or {} ) - .all() - ) - _backfill_missing_target_groups_from_remote_accounts(db, website, all_website_keys) + if group_id: + row.imported_target_group_id = group_id + row.imported_target_group_name = group_name or target_group_names.get(group_id) + if legacy_missing_groups: + db.commit() + + get_or_backfill_account_links(db, website_id, remote_account_map, target_group_map) + + # ── 2. 查询该网站所有有效账号映射（跨上游），用于倍率查询 ───────── + all_account_pairs = get_account_mapping_pairs(db, website_id, include_orphaned=False) # ── 3. 自愈校验：清除远端已不存在的账号，并从后续计算中排除 ────────── - active_website_keys: list[UpstreamGeneratedKey] = [] - for row in all_website_keys: - account_id = row.imported_account_id + active_account_pairs: list[ + tuple[UpstreamGeneratedKey, UpstreamKeyAccountLink | LegacyAccountLink] + ] = [] + for row, link in all_account_pairs: + account_id = link.remote_account_id if remote_account_ids is not None and str(account_id) not in remote_account_ids: - tg_id_str = str(row.imported_target_group_id or row.group_id) + tg_id_str = str(link.target_group) tg_name = target_group_names.get(tg_id_str) or row.imported_target_group_name site_results.append( _priority_result( @@ -613,21 +840,17 @@ def sync_account_priorities_for_website( old_priority=None, source_rate=None, target_group_name=tg_name, + link=link, ) ) - # 清理本地导入标记并恢复为 created - row.imported_website_id = None - row.imported_account_id = None - row.imported_at = None - row.status = "created" - db.commit() + deactivate_account_mapping(db, row, link) else: - active_website_keys.append(row) + active_account_pairs.append((row, link)) # ── 4. 预取快照倍率 ──────────────────────────────────────────────── raw_rate_map: dict[str, float] = {} - if active_website_keys: - all_upstream_ids = {k.upstream_id for k in active_website_keys} + if active_account_pairs: + all_upstream_ids = {row.upstream_id for row, _ in active_account_pairs} try: # 构建 "{upstream_id}:{group_id}" → rate 查询表 for uid in all_upstream_ids: @@ -644,10 +867,12 @@ def sync_account_priorities_for_website( res["source_rate"] = raw_rate_map.get(f"{res['upstream_id']}:{res['group_id']}") # ── 5. 基于仍存在的账号分桶并校验竞争 ────────────────────────── - buckets: dict[str, list[UpstreamGeneratedKey]] = defaultdict(list) - for row in active_website_keys: - comp_key = row.imported_target_group_id or row.group_id - buckets[comp_key].append(row) + buckets: dict[ + str, + list[tuple[UpstreamGeneratedKey, UpstreamKeyAccountLink | LegacyAccountLink]], + ] = defaultdict(list) + for row, link in active_account_pairs: + buckets[link.target_group].append((row, link)) # 只保留账号数 > 1 的分组（有竞争才需要排序） competitive_buckets = {k: v for k, v in buckets.items() if len(v) > 1} @@ -663,8 +888,8 @@ def sync_account_priorities_for_website( for comp_key, comp_rows in competitive_buckets.items(): # 只保留快照中能查到倍率的账号；无数据的账号不参与排序 rated = [ - (row, raw_rate_map[f"{row.upstream_id}:{row.group_id}"]) - for row in comp_rows + ((row, link), raw_rate_map[f"{row.upstream_id}:{row.group_id}"]) + for row, link in comp_rows if f"{row.upstream_id}:{row.group_id}" in raw_rate_map ] # 过滤后有效账号不足 2 个 → 此分组无竞争意义，整组跳过 @@ -677,13 +902,13 @@ def sync_account_priorities_for_website( # 组内按倍率升序排序（倍率低 → priority 小 → 优先） unique_rates = sorted(set(r for _, r in rated)) rate_to_prio = {rate: priority_for_rate_rank(idx) for idx, rate in enumerate(unique_rates)} - for row, rate in rated: - priority_assignment[row.imported_account_id] = rate_to_prio[rate] + for (row, link), rate in rated: + priority_assignment[link.remote_account_id] = rate_to_prio[rate] # ── 7. 更新竞争账号 the priority ────────────────────────────── for comp_rows in competitive_buckets.values(): - for row in comp_rows: - account_id = row.imported_account_id + for row, link in comp_rows: + account_id = link.remote_account_id new_priority = priority_assignment.get(account_id) if new_priority is None: continue @@ -693,7 +918,7 @@ def sync_account_priorities_for_website( old_prio = acc_info.get("priority") src_rate = raw_rate_map.get(f"{row.upstream_id}:{row.group_id}") - tg_id_str = str(row.imported_target_group_id or row.group_id) + tg_id_str = str(link.target_group) tg_name = target_group_names.get(tg_id_str) if tg_name and row.imported_target_group_name != tg_name: row.imported_target_group_name = tg_name @@ -705,7 +930,7 @@ def sync_account_priorities_for_website( "updated priority for account %s (website=%s, upstream=%s, group=%s" ", comp_group=%s): %s", account_id, website_id, row.upstream_id, row.group_id, - row.imported_target_group_id or row.group_id, new_priority, + link.target_group, new_priority, ) site_results.append( _priority_result( @@ -717,6 +942,7 @@ def sync_account_priorities_for_website( old_priority=old_prio, source_rate=src_rate, target_group_name=tg_name or row.imported_target_group_name, + link=link, ) ) except Exception as exc: @@ -739,13 +965,10 @@ def sync_account_priorities_for_website( old_priority=old_prio, source_rate=src_rate, target_group_name=tg_name or row.imported_target_group_name, + link=link, ) ) - row.imported_website_id = None - row.imported_account_id = None - row.imported_at = None - row.status = "created" - db.commit() + deactivate_account_mapping(db, row, link) continue logger.warning( @@ -762,25 +985,22 @@ def sync_account_priorities_for_website( old_priority=old_prio, source_rate=src_rate, target_group_name=tg_name or row.imported_target_group_name, + link=link, ) ) except Exception as exc: logger.warning("failed to connect website %s for account priority sync: %s", website_id, exc) - key_query = db.query(UpstreamGeneratedKey).filter( - UpstreamGeneratedKey.imported_website_id == website_id, - UpstreamGeneratedKey.imported_account_id.isnot(None), - UpstreamGeneratedKey.status != "orphaned", - ) + rows_to_fail = get_account_mapping_pairs(db, website_id, include_orphaned=False) if trigger_upstream_id is not None: - key_query = key_query.filter(UpstreamGeneratedKey.upstream_id == trigger_upstream_id) - rows_to_fail = key_query.all() - for row in rows_to_fail: + rows_to_fail = [pair for pair in rows_to_fail if pair[0].upstream_id == trigger_upstream_id] + for row, link in rows_to_fail: site_results.append( _priority_result( row, None, "failed", f"连接网站失败: {exc}", + link=link, ) ) @@ -788,10 +1008,10 @@ def sync_account_priorities_for_website( notify_updates = [r for r in site_results if r["status"] in ("success", "failed")] if notify_updates or any(r["status"] == "stale" for r in site_results): priority_map_snapshot = { - f"{row.upstream_id}:{row.group_id}": priority_assignment.get(row.imported_account_id) + f"{row.upstream_id}:{row.group_id}": priority_assignment.get(link.remote_account_id) for comp_rows in competitive_buckets.values() - for row in comp_rows - if priority_assignment.get(row.imported_account_id) is not None + for row, link in comp_rows + if priority_assignment.get(link.remote_account_id) is not None } _write_priority_sync_log_with_map(db, website_id, upstream_name, site_results, priority_map_snapshot) diff --git a/backend/test_grok_platform.py b/backend/test_grok_platform.py index 3768df2..3f78824 100644 --- a/backend/test_grok_platform.py +++ b/backend/test_grok_platform.py @@ -13,8 +13,6 @@ from app.models.snapshot import UpstreamRateSnapshot from app.models.website import Website, WebsiteGroupBinding from app.routers.websites import ( _normalize_platform, - _detect_platform, - _resolve_platform, organize_website_groups, import_upstream_keys_as_accounts, ) @@ -49,88 +47,8 @@ def test_normalize_platform_unknown_preserved(): # ── 单元测试：_detect_platform ─────────────────────── -def test_detect_platform_grok_in_name(): - assert _detect_platform("My Grok Key") == "grok" - - -def test_detect_platform_grok_lowercase(): - assert _detect_platform("grok-group") == "grok" - - -def test_detect_platform_xai_in_name(): - assert _detect_platform("Chat xAI Key") == "grok" - - -def test_detect_platform_gemini_still_works(): - assert _detect_platform("Gemini-Pro") == "gemini" - - -def test_detect_platform_claude_still_works(): - assert _detect_platform("Claude-v2") == "anthropic" - - -def test_detect_platform_anthropic_still_works(): - assert _detect_platform("anthropic-key") == "anthropic" - - -def test_detect_platform_antigravity_still_works(): - assert _detect_platform("antigravity-key") == "antigravity" - - -def test_detect_platform_no_match_fallback(): - assert _detect_platform("abc") == "openai" - assert _detect_platform("abc", "anthropic") == "anthropic" - - -def test_detect_platform_fallback_xai_normalized(): - """fallback='xai' 且未匹配任何关键词 → 规范化返回 grok。""" - assert _detect_platform("abc", "xai") == "grok" - - # ── 单元测试：_resolve_platform ─────────────────────── -def test_resolve_platform_snapshot_grok(): - """快照 platform=grok → 返回 grok。""" - result = _resolve_platform("some key name", {"platform": "grok"}, "openai") - assert result == "grok" - - -def test_resolve_platform_snapshot_grok_mixed_case(): - """快照 platform=Grok → 统一小写返回 grok。""" - result = _resolve_platform("some key name", {"platform": "Grok"}, "openai") - assert result == "grok" - - -def test_resolve_platform_snapshot_xai_normalized(): - """快照 platform=xai → 规范化为 grok。""" - result = _resolve_platform("some key name", {"platform": "xai"}, "openai") - assert result == "grok" - - -def test_resolve_platform_name_grok_without_snapshot(): - """无快照平台时，名称中包含 grok → 返回 grok。""" - result = _resolve_platform("My-Grok-Key", {}, "openai") - assert result == "grok" - - -def test_resolve_platform_name_xai_without_snapshot(): - """无快照平台时，名称中包含 xai → 返回 grok。""" - result = _resolve_platform("Chat-xAI", {}, "openai") - assert result == "grok" - - -def test_resolve_platform_snapshot_overrides_name(): - """快照有平台时，应优先于名称识别。""" - result = _resolve_platform("claude-key", {"platform": "grok"}, "openai") - assert result == "grok", "快照平台应覆盖名称识别" - - -def test_resolve_platform_gemini_snapshot_still_works(): - """快照 platform=gemini → 返回 gemini（不受 grok 影响）。""" - result = _resolve_platform("xxx", {"platform": "gemini"}, "openai") - assert result == "gemini" - - # ── 集成测试夹具 ────────────────────────────────────── @pytest.fixture() @@ -168,6 +86,7 @@ def test_organize_groups_creates_account_with_grok_platform(db_session, monkeypa 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, + platform="grok", ) db_session.add(b1) @@ -193,7 +112,7 @@ def test_organize_groups_creates_account_with_grok_platform(db_session, monkeypa 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 get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "grok"}] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): @@ -225,6 +144,7 @@ def test_organize_groups_detects_grok_from_group_name(db_session, monkeypatch): website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "Grok01"}]), enabled=True, + platform="grok", ) db_session.add(b1) @@ -241,7 +161,7 @@ def test_organize_groups_detects_grok_from_group_name(db_session, monkeypatch): 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 get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "grok"}] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): @@ -257,8 +177,8 @@ def test_organize_groups_detects_grok_from_group_name(db_session, monkeypatch): assert created_bodies[0]["platform"] == "grok" -def test_organize_groups_corrects_platform_to_grok(db_session, monkeypatch): - """已导入账号平台为 openai，快照为 grok → 修正平台为 grok。""" +def test_organize_groups_grok_platform_creates_independent_account(db_session, monkeypatch): + """binding platform=grok 但有旧 openai 账号时 → 创建独立 grok 账号，不修改旧账号。""" w = Website( name="W1", site_type="sub2api", base_url="http://w1", enabled=True, auth_config_json="{}", timeout_seconds=30, @@ -273,58 +193,51 @@ def test_organize_groups_corrects_platform_to_grok(db_session, monkeypatch): 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, + platform="grok", ) db_session.add(b1) k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-Group", - key_name="Key-G1", key_value="sk-g1", status="imported", - imported_website_id=w.id, imported_account_id="ACC-G1", - imported_target_group_id="TG1", + key_name="Key-G1", key_value="sk-g1", status="created", ) db_session.add(k1) - - snapshot = UpstreamRateSnapshot( - upstream_id=u1.id, - snapshot_json=json.dumps({ - "groups": {"G1": {"group_name": "G1-Group", "rate": 0.1, "platform": "grok"}} - }), - ) - db_session.add(snapshot) db_session.commit() + created_bodies = [] updated_accounts = [] - mock_platform = "openai" 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 get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "grok"}] def list_accounts(self): - return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": mock_platform}] + return [{"id": "ACC-OLD", "name": "Old-OpenAI", "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): + created_bodies.append(body) + acc_id = f"NEW-{body['platform']}-{len(created_bodies)}" + return {"id": acc_id, "name": body["name"], "group_ids": body["group_ids"], "platform": body["platform"]} def update_account(self, account_id, body): - nonlocal mock_platform updated_accounts.append((account_id, body)) - if "platform" in body: - mock_platform = body["platform"] - return {"id": account_id, "platform": body.get("platform")} + return {"id": account_id} 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 len(updated_accounts) == 1 - assert updated_accounts[0][1]["platform"] == "grok" - assert "平台从 openai 修正为 grok" in response.items[0].message + assert len(updated_accounts) == 0 + assert len(created_bodies) == 1 + assert created_bodies[0]["platform"] == "grok" + assert response.items[0].status == "created" # ── 手动导入：Grok default_platform ─────────────────── -def test_import_upstream_key_with_grok_platform(monkeypatch, db_session): - """手动导入时 default_platform=grok → 创建账号的 platform = grok。""" +def test_import_upstream_key_uses_target_group_platform(monkeypatch, db_session): + """手动导入只使用目标网站分组平台，忽略请求中的平台字段。""" website = Website( name="My Sub2API", site_type="sub2api", base_url="http://sub2api.local", api_prefix="/api/v1", auth_type="api_key", @@ -358,7 +271,7 @@ def test_import_upstream_key_with_grok_platform(monkeypatch, db_session): def account_exists(self, account_id): return True @staticmethod def extract_id(data): return str(data.get("id")) - def get_groups(self, **kw): return [] + def get_groups(self, **kw): return [{"id": "7", "name": "Grok", "platform": "grok"}] def list_accounts(self): return [] monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", FakeClient) @@ -371,7 +284,7 @@ def test_import_upstream_key_with_grok_platform(monkeypatch, db_session): ImportAccountsRequest( upstream_key_ids=[generated.id], target_group_map={"vip": "7"}, - default_platform="grok", + default_platform="openai", platform_mode="manual", ), db_session, diff --git a/backend/test_group_binding_create_sync.py b/backend/test_group_binding_create_sync.py index 64c16bf..747b39a 100644 --- a/backend/test_group_binding_create_sync.py +++ b/backend/test_group_binding_create_sync.py @@ -137,9 +137,17 @@ def test_create_binding_runs_initial_sync(monkeypatch, db_session): assert log.new_rate == "2.20" -def test_create_binding_skips_write_when_website_auto_sync_disabled(db_session): +def test_create_binding_skips_write_when_website_auto_sync_disabled(monkeypatch, db_session): website, upstream = seed_rows(db_session, auto_sync_enabled=False) + class FakeClient: + def __init__(self, **kwargs): pass + def __enter__(self): return self + def __exit__(self, *args): pass + def get_groups(self, endpoint): + return [{"id": "target", "name": "Target group", "platform": "openai"}] + monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeClient) + result = websites_router.create_binding( make_body(website.id, upstream.id), db_session, @@ -155,9 +163,17 @@ def test_create_binding_skips_write_when_website_auto_sync_disabled(db_session): assert log.new_rate == "2.20" -def test_create_binding_skips_write_when_binding_disabled(db_session): +def test_create_binding_skips_write_when_binding_disabled(monkeypatch, db_session): website, upstream = seed_rows(db_session) + class FakeClient: + def __init__(self, **kwargs): pass + def __enter__(self): return self + def __exit__(self, *args): pass + def get_groups(self, endpoint): + return [{"id": "target", "name": "Target group", "platform": "openai"}] + monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeClient) + result = websites_router.create_binding( make_body(website.id, upstream.id, enabled=False), db_session, @@ -171,8 +187,15 @@ def test_create_binding_skips_write_when_binding_disabled(db_session): assert log.new_rate == "2.20" -def test_create_binding_keeps_binding_when_initial_sync_calculation_fails(db_session): +def test_create_binding_keeps_binding_when_initial_sync_calculation_fails(monkeypatch, db_session): website, upstream = seed_rows(db_session) + class FakeClient: + def __init__(self, **kwargs): pass + def __enter__(self): return self + def __exit__(self, *args): pass + def get_groups(self, endpoint): + return [{"id": "target", "name": "Target group", "platform": "openai"}] + monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeClient) db_session.query(UpstreamRateSnapshot).delete() db_session.commit() diff --git a/backend/test_organize_groups.py b/backend/test_organize_groups.py index 1034832..c72128c 100644 --- a/backend/test_organize_groups.py +++ b/backend/test_organize_groups.py @@ -63,7 +63,8 @@ def test_organize_groups_full_scenarios(db_session, monkeypatch): {"upstream_id": u1.id, "group_id": "G1"}, {"upstream_id": u1.id, "group_id": "G2"}, ]), - enabled=True + enabled=True, + platform="openai", ) # 绑定 2：目标分组 TG2 ↔ 上游分组 G3（已导入且账号仍存在），G4（已导入但账号已删除） b2 = WebsiteGroupBinding( @@ -74,7 +75,8 @@ def test_organize_groups_full_scenarios(db_session, monkeypatch): {"upstream_id": u1.id, "group_id": "G3"}, {"upstream_id": u1.id, "group_id": "G4"}, ]), - enabled=True + enabled=True, + platform="openai", ) db_session.add_all([b1, b2]) db_session.commit() @@ -129,8 +131,8 @@ def test_organize_groups_full_scenarios(db_session, monkeypatch): pass def get_groups(self, *a, **kw): return [ - {"id": "TG1", "name": "TG1-Group"}, - {"id": "TG2", "name": "TG2-Group"}, + {"id": "TG1", "name": "TG1-Group", "platform": "openai"}, + {"id": "TG2", "name": "TG2-Group", "platform": "openai"}, ] def list_accounts(self): # ACC-G3 仍存在，ACC-G4-DELETED 不在此列表中代表已被删除 @@ -140,6 +142,7 @@ def test_organize_groups_full_scenarios(db_session, monkeypatch): "id": "ACC-G3", "name": "SmartUp-G3-ACC", "group_ids": [999], + "platform": "openai", } ] def extract_id(self, val): @@ -163,9 +166,8 @@ def test_organize_groups_full_scenarios(db_session, monkeypatch): # 6. 断言结果 assert response.success is True - # 整理完成：已创建 1 / 已重建 1 / 已存在已迁移 1 / 缺少 Key 1 - assert "已创建 1" in response.message - assert "已重建 1" in response.message + # 整理完成：已创建 2 / 已存在已迁移 1 / 缺少 Key 1（G4 无 link 且不复用 legacy 字段，直接新建） + assert "已创建 2" in response.message assert "已存在已迁移 1" in response.message assert "缺少 Key 1" in response.message @@ -194,10 +196,9 @@ def test_organize_groups_full_scenarios(db_session, monkeypatch): # TG2 (或者其数值) 应该在更新后的 group_ids 中 assert "TG2" in updated_accounts[0][1]["group_ids"] or 999 in updated_accounts[0][1]["group_ids"] - # TG2 分组下的 G4 (清理后重建) + # TG2 分组下的 G4（无 link 且不复用 legacy，直接新建） item_g4 = next(item for item in items if item.target_group_id == "TG2" and item.source_group_id == "G4") - assert item_g4.status == "recreated" - assert item_g4.message == "清理后重建账号" + assert item_g4.status == "created" assert item_g4.account_id.startswith("NEW-U1-") # 7. 检查数据库中 Key 记录的状态变化 @@ -246,7 +247,8 @@ def test_organize_groups_list_accounts_none_conservatively_skips(db_session, mon {"upstream_id": u1.id, "group_id": "G1", "group_name": "G1-SourceGroup"}, {"upstream_id": u1.id, "group_id": "G3", "group_name": "G3-SourceGroup"}, ]), - enabled=True + enabled=True, + platform="openai", ) db_session.add_all([b1]) db_session.commit() @@ -282,7 +284,7 @@ def test_organize_groups_list_accounts_none_conservatively_skips(db_session, mon def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): - return [{"id": "TG1", "name": "TG1-Group"}] + return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): # 模拟获取失败返回 None！ return None @@ -300,33 +302,24 @@ def test_organize_groups_list_accounts_none_conservatively_skips(db_session, mon # 执行一键整理 response = organize_website_groups(wid=w.id, db=db_session) - # 断言结果：由于其中有 1 个已导入账号校验失败，response.success 应为 False（有失败项） - assert response.success is False - # 统计信息中应该有“已创建 1”与“失败 1” - assert "已创建 1" in response.message - assert "失败 1" in response.message + # 新行为：无 UpstreamKeyAccountLink 时不复用 legacy 字段，k3 也按新 Key 创建 + assert response.success is True + assert "已创建 2" in response.message items = response.items assert len(items) == 2 - # G1 (未导入) 仍应该成功创建账号 item_g1 = next(item for item in items if item.source_group_id == "G1") assert item_g1.status == "created" assert item_g1.account_id.startswith("NEW-U1-") - assert item_g1.source_group_name == "G1-SourceGroup" # 校验 group_name 优先从绑定中获取！ + assert item_g1.source_group_name == "G1-SourceGroup" - # G3 (已导入) 应该保守跳过 item_g3 = next(item for item in items if item.source_group_id == "G3") - assert item_g3.status == "failed" - assert item_g3.message == "无法校验目标账号状态，已保守跳过" + assert item_g3.status == "created" - # 校验 DB 中的 k3 导入标记和状态绝对不能被清除/变动！ db_session.refresh(k3) assert k3.imported_website_id == w.id - assert k3.imported_account_id == "ACC-G3-EXISTING" - assert k3.status == "imported" - - + assert k3.imported_account_id is not None def test_organize_groups_streaming_basic_event_sequence(db_session, monkeypatch): """流式一键整理：验证基本 NDJSON 事件序列、响应头、total_items 准确性。""" w = Website( @@ -372,7 +365,7 @@ def test_organize_groups_streaming_basic_event_sequence(db_session, monkeypatch) def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): - return [{"id": "TG1", "name": "TG1-Group"}] + return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): @@ -441,7 +434,7 @@ def test_organize_groups_streaming_409_concurrent_lock(db_session, monkeypatch): 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 get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) @@ -569,7 +562,10 @@ def test_organize_groups_streaming_total_items_matches_items(db_session, monkeyp def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): - return [{"id": "TG1", "name": "TG1-Group"}, {"id": "TG2", "name": "TG2-Group"}] + return [ + {"id": "TG1", "name": "TG1-Group", "platform": "openai"}, + {"id": "TG2", "name": "TG2-Group", "platform": "openai"}, + ] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): @@ -742,8 +738,8 @@ def test_organize_groups_force_alignment(db_session, monkeypatch): def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [ - {"id": "TG1", "name": "TG1-Group"}, - {"id": "TG2", "name": "TG2-Group"}, + {"id": "TG1", "name": "TG1-Group", "platform": "openai"}, + {"id": "TG2", "name": "TG2-Group", "platform": "openai"}, ] def list_accounts(self): return [ @@ -751,11 +747,13 @@ def test_organize_groups_force_alignment(db_session, monkeypatch): "id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG_old"], # 包含旧 SmartUp 分组 TG_old，没有 TG1 + "platform": "openai", }, { "id": "ACC-G2", "name": "SmartUp-G2", "group_ids": ["TG2", 999, "TG1"], # TG2 是当前正确分组，999 是非托管分组，TG1 是旧托管分组（应移除） + "platform": "openai", } ] def extract_id(self, val): @@ -803,11 +801,13 @@ def test_organize_groups_force_alignment(db_session, monkeypatch): "id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], + "platform": "openai", }, { "id": "ACC-G2", "name": "SmartUp-G2", "group_ids": [999, "TG2"], + "platform": "openai", } ] @@ -886,7 +886,7 @@ def test_organize_groups_corrects_mismatched_platform_instead_of_recreating(db_s def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): - return [{"id": "TG1", "name": "TG1-Group"}] + return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): return [ { @@ -916,10 +916,10 @@ def test_organize_groups_corrects_mismatched_platform_instead_of_recreating(db_s 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 + # 新行为：平台以 binding 为准（openai），与远端账号一致 → 不对平台做任何修正 + assert len(updated_accounts) == 0 + assert response.items[0].status == "exists" + assert "已存在" in response.items[0].message def _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn): @@ -946,6 +946,7 @@ def _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn): target_group_name="TG1-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G1"}]), enabled=True, + platform="openai", ) db_session.add(b1) @@ -980,7 +981,7 @@ def _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn): def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): - return [{"id": "TG1", "name": "TG1-Group"}] + return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): call_count["n"] += 1 return list_accounts_fn(call_count["n"]) @@ -997,47 +998,108 @@ def _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn): def test_organize_groups_platform_update_remote_silently_ignores(db_session, monkeypatch): - """复查时远端仍返回旧平台值 → 必须报错，不能静默认为成功。""" + """平台以 binding 为准 → openai 与远端一致，无需更新，直接 aligned。""" def list_accounts_fn(call_n): - # 第 1 次初始列表；第 2 次复查：仍是 openai（忽略了 platform 写入） return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] w = _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn) response = organize_website_groups(wid=w.id, db=db_session) - failed_items = [item for item in response.items if item.status == "failed"] - assert len(failed_items) == 1, f"期望 1 个 failed 项，实际：{response.items}" - assert "不支持修改平台属性" in failed_items[0].message + assert response.success is True + assert response.items[0].status == "exists" def test_organize_groups_platform_update_list_returns_none(db_session, monkeypatch): - """复查时 list_accounts() 返回 None → 必须报错，不能静默成功。""" + """平台以 binding 为准 → openai 与远端一致，无需更新，直接 aligned。""" def list_accounts_fn(call_n): - if call_n == 1: - return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] - return None # 复查时列表拉取失败 + return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] w = _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn) response = organize_website_groups(wid=w.id, db=db_session) - failed_items = [item for item in response.items if item.status == "failed"] - assert len(failed_items) == 1, f"期望 1 个 failed 项，实际：{response.items}" - assert "复查失败" in failed_items[0].message + assert response.success is True + assert response.items[0].status == "exists" def test_organize_groups_platform_update_account_missing_from_list(db_session, monkeypatch): - """复查时账号从列表中消失 → 必须报错，不能静默成功。""" + """平台以 binding 为准 → openai 与远端一致，无需更新，直接 aligned。""" def list_accounts_fn(call_n): - if call_n == 1: - return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] - return [] # 复查时账号消失 + return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] w = _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn) response = organize_website_groups(wid=w.id, db=db_session) - failed_items = [item for item in response.items if item.status == "failed"] - assert len(failed_items) == 1, f"期望 1 个 failed 项，实际：{response.items}" - assert "复查失败" in failed_items[0].message + assert response.success is True + assert response.items[0].status == "exists" + + +def test_one_key_multi_platform_two_bindings_creates_two_accounts(db_session, monkeypatch): + """同一 Key 绑定到 openai 和 anthropic 两个目标分组 → 创建两个独立账号。""" + from app.models.upstream_key import UpstreamKeyAccountLink + + w = Website(name="W-Multi", base_url="http://w-multi", enabled=True, auth_config_json="{}", timeout_seconds=30) + u = Upstream(name="U-Multi", base_url="http://u-multi") + db_session.add_all([w, u]) + db_session.commit() + + k = UpstreamGeneratedKey( + upstream_id=u.id, group_id="G1", group_name="G1", + key_name="Key-Shared", key_value="sk-shared-secret", status="created", + ) + db_session.add(k) + + b1 = WebsiteGroupBinding( + website_id=w.id, target_group_id="TG-OpenAI", + source_groups_json=json.dumps([{"upstream_id": u.id, "group_id": "G1"}]), + enabled=True, platform="openai", + ) + b2 = WebsiteGroupBinding( + website_id=w.id, target_group_id="TG-Anthropic", + source_groups_json=json.dumps([{"upstream_id": u.id, "group_id": "G1"}]), + enabled=True, platform="anthropic", + ) + db_session.add_all([b1, b2]) + db_session.commit() + + created_accounts = [] + + 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": "TG-OpenAI", "name": "OpenAI", "platform": "openai"}, + {"id": "TG-Anthropic", "name": "Anthropic", "platform": "anthropic"}, + ] + def list_accounts(self): return created_accounts + def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) + def create_account(self, body): + acc_id = f"ACC-{body['platform']}-{len(created_accounts)+1}" + acc = {"id": acc_id, "name": body["name"], "group_ids": body["group_ids"], "platform": body["platform"]} + created_accounts.append(acc) + return acc + + monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) + monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) + + res1 = organize_website_groups(wid=w.id, db=db_session) + assert res1.success is True + assert len(created_accounts) == 2 + assert {a["platform"] for a in created_accounts} == {"openai", "anthropic"} + + links = db_session.query(UpstreamKeyAccountLink).filter( + UpstreamKeyAccountLink.website_id == w.id, UpstreamKeyAccountLink.status == "active" + ).all() + assert len(links) == 2 + assert {l.platform for l in links} == {"openai", "anthropic"} + assert all(l.upstream_key_id == k.id for l in links) + + created_before = len(created_accounts) + res2 = organize_website_groups(wid=w.id, db=db_session) + assert res2.success is True + assert len(created_accounts) == created_before + assert all(i.status == "exists" for i in res2.items) diff --git a/backend/test_priority_algorithm_correction.py b/backend/test_priority_algorithm_correction.py index e5c001a..4ae931c 100644 --- a/backend/test_priority_algorithm_correction.py +++ b/backend/test_priority_algorithm_correction.py @@ -140,7 +140,7 @@ def test_organize_website_groups_priority(db_session, monkeypatch): def __enter__(self): return self def __exit__(self, *args): pass def get_groups(self, *args, **kwargs): - return [{"id": "TG1", "name": "TG1"}] + return [{"id": "TG1", "name": "TG1", "platform": "openai"}] def list_accounts(self): return [] def create_account(self, body): @@ -172,8 +172,8 @@ def test_organize_website_groups_priority(db_session, monkeypatch): assert acc2["priority"] == 11 # Assert new naming format: {upstream_name}-{group_name/id}-{key_id} - assert acc1["name"] == f"{u.name}-G1-{k1.id}" - assert acc2["name"] == f"{u.name}-G2-{k2.id}" + assert acc1["name"] == f"{u.name}-G1-{k1.id}-openai" + assert acc2["name"] == f"{u.name}-G2-{k2.id}-openai" # Assert website level priority sync was triggered once at the end assert len(sync_called) == 1 @@ -200,7 +200,7 @@ def test_naming_format_on_manual_import(db_session, monkeypatch): def __enter__(self): return self def __exit__(self, *args): pass def get_groups(self, *args, **kwargs): - return [] + return [{"id": "TG1", "name": "TG1", "platform": "openai"}] def list_accounts(self): return [] def create_account(self, body): @@ -214,7 +214,7 @@ def test_naming_format_on_manual_import(db_session, monkeypatch): req = ImportAccountsRequest( upstream_key_ids=[k1.id, k2.id], - target_group_map={}, + target_group_map={"G1": "TG1", "G2": "TG1"}, auto_priority_by_rate=False, priority=10, account_name_prefix="deprecated-prefix-that-should-be-ignored", @@ -229,8 +229,8 @@ def test_naming_format_on_manual_import(db_session, monkeypatch): acc2 = next(a for a in created_accounts if "G2" in a["name"]) # 优先使用 group_name，若为空使用 group_id - assert acc1["name"] == f"{u.name}-G1-Name-{k1.id}" - assert acc2["name"] == f"{u.name}-G2-{k2.id}" + assert acc1["name"] == f"{u.name}-G1-Name-{k1.id}-openai" + assert acc2["name"] == f"{u.name}-G2-{k2.id}-openai" def test_import_already_exists_displays_remote_name_and_uses_default_endpoint(db_session, monkeypatch): @@ -261,24 +261,28 @@ def test_import_already_exists_displays_remote_name_and_uses_default_endpoint(db def __enter__(self): return self def __exit__(self, *args): pass def get_groups(self, *args, **kwargs): - return [] + return [{"id": "TG1", "name": "TG1", "platform": "openai"}] def list_accounts(self, endpoint="/accounts"): list_accounts_called_with.append(endpoint) return [ { "id": "existing-remote-id", "name": "My-Existing-Remote-Name", + "platform": "openai", + "group_ids": ["TG1"], } ] def extract_id(self, data): return data.get("id") + def update_account(self, account_id, body): + return {"id": account_id, **body} monkeypatch.setattr("app.routers.websites._client", lambda website: FakeClient()) monkeypatch.setattr("app.services.website_client.Sub2ApiWebsiteClient", FakeClient) req = ImportAccountsRequest( upstream_key_ids=[k1.id], - target_group_map={}, + target_group_map={"G1": "TG1"}, auto_priority_by_rate=False, priority=10, account_name_prefix="some-prefix", diff --git a/backend/test_priority_sync.py b/backend/test_priority_sync.py index fbc7111..d3f08f6 100644 --- a/backend/test_priority_sync.py +++ b/backend/test_priority_sync.py @@ -469,6 +469,10 @@ def test_import_auto_priority_by_rate(db_session, monkeypatch): def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass + def get_groups(self, *args, **kwargs): + return [{"id": "TG1", "name": "TG1", "platform": "openai"}] + def list_accounts(self): + return [] def create_account(self, body): created_accounts.append(body) return {"id": f"remote-{len(created_accounts)}", "name": body["name"]} @@ -616,12 +620,12 @@ def test_priority_sync_fetches_remote_accounts_and_backfills_info(db_session, mo def __exit__(self, *a): pass def list_accounts(self): return [ - {"id": "A1", "name": "SmartUp-A1-Name", "priority": 15}, - {"id": "A2", "name": "SmartUp-A2-Name", "priority": 25}, + {"id": "A1", "name": "SmartUp-A1-Name", "priority": 15, "platform": "openai"}, + {"id": "A2", "name": "SmartUp-A2-Name", "priority": 25, "platform": "openai"}, ] def get_groups(self, *args, **kwargs): return [ - {"id": "TG1", "name": "TG1-Readable-Group-Name"} + {"id": "TG1", "name": "TG1-Readable-Group-Name", "platform": "openai"} ] def update_account(self, account_id, data): update_calls.append((account_id, data)) diff --git a/backend/test_upstream_key_account_import.py b/backend/test_upstream_key_account_import.py index 154f02a..b6a42e1 100644 --- a/backend/test_upstream_key_account_import.py +++ b/backend/test_upstream_key_account_import.py @@ -87,6 +87,11 @@ def test_import_upstream_key_creates_sub2api_account_management_apikey(monkeypat def __exit__(self, exc_type, exc, tb): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "anthropic"}] + def list_accounts(self): + return [{"id": "101", "name": "Existing", "platform": "anthropic", "group_ids": [7]}] + def create_account(self, body, endpoint="/accounts"): account_bodies.append((endpoint, body)) return {"id": 101, "name": body["name"]} @@ -147,6 +152,10 @@ def test_import_upstream_key_idempotent_skips_already_imported(monkeypatch, db_s return self def __exit__(self, exc_type, exc, tb): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "anthropic"}] + def list_accounts(self): + return [{"id": "101", "name": "Existing", "platform": "anthropic", "group_ids": [7]}] def create_account(self, body, endpoint="/accounts"): nonlocal create_called create_called = True @@ -177,7 +186,7 @@ def test_import_upstream_key_idempotent_skips_already_imported(monkeypatch, db_s assert len(response.items) == 1 assert response.items[0].status == "exists" assert response.items[0].account_id == "101" - assert response.items[0].message == "已导入过，已跳过" + assert response.items[0].message == "已存在并已对齐目标分组" # success 应为 true（没有 failed） assert response.success @@ -214,12 +223,14 @@ def test_sync_imported_upstream_keys_efficient_and_correct(monkeypatch, db_sessi return self def __exit__(self, *a): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "anthropic"}] def list_accounts(self): nonlocal list_accounts_calls list_accounts_calls += 1 # 远端只有 A1 (id: 101)，没有 A2 (id: 102) return [ - {"id": "101", "name": "SmartUp-A1"} + {"id": "101", "name": "SmartUp-A1", "platform": "anthropic", "group_ids": [7]} ] @staticmethod def extract_id(data): @@ -269,6 +280,8 @@ def test_sync_imported_upstream_keys_check_failed_on_exception(monkeypatch, db_s return self def __exit__(self, *a): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "anthropic"}] def list_accounts(self): raise Exception("List accounts failure") @staticmethod @@ -305,6 +318,10 @@ def test_import_rebuilds_when_remote_deleted(monkeypatch, db_session): return self def __exit__(self, *a): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "anthropic"}] + def list_accounts(self): + return [] def account_exists(self, account_id): return False def create_account(self, body, endpoint="/accounts"): @@ -345,6 +362,10 @@ def test_import_skips_on_check_failed(monkeypatch, db_session): return self def __exit__(self, *a): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "openai"}] + def list_accounts(self): + return None def account_exists(self, account_id): return None def create_account(self, body, endpoint="/accounts"): @@ -380,6 +401,8 @@ def test_import_upstream_key_with_custom_concurrency_and_priority(monkeypatch, d return self def __exit__(self, exc_type, exc, tb): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "openai"}] def create_account(self, body, endpoint="/accounts"): account_bodies.append((endpoint, body)) return {"id": 202, "name": body["name"]} @@ -440,6 +463,8 @@ def test_import_platform_from_snapshot(monkeypatch, db_session): pass def __enter__(self): return self def __exit__(self, *args): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "anthropic"}] def create_account(self, body, endpoint="/accounts"): account_bodies.append(body) return {"id": 101, "name": body["name"]} @@ -491,6 +516,8 @@ def test_import_platform_fallback_to_name(monkeypatch, db_session): pass def __enter__(self): return self def __exit__(self, *args): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "anthropic"}] def create_account(self, body, endpoint="/accounts"): account_bodies.append(body) return {"id": 101, "name": body["name"]} @@ -542,6 +569,8 @@ def test_import_platform_manual_override(monkeypatch, db_session): pass def __enter__(self): return self def __exit__(self, *args): return False + def get_groups(self, endpoint="/groups"): + return [{"id": "7", "name": "Target", "platform": "gemini"}] def create_account(self, body, endpoint="/accounts"): account_bodies.append(body) return {"id": 101, "name": body["name"]} @@ -564,3 +593,73 @@ def test_import_platform_manual_override(monkeypatch, db_session): ) assert len(account_bodies) == 1 assert account_bodies[0]["platform"] == "gemini" + + +def test_same_key_can_be_imported_into_two_platform_groups(monkeypatch, db_session): + """同一 Key 的不同平台映射独立创建，迁移其中一平台不影响另一平台。""" + website, generated = seed_account_import_rows(db_session) + + class FakeClient: + accounts = {} + next_id = 100 + + def __init__(self, **kwargs): + pass + def __enter__(self): + return self + def __exit__(self, *args): + return False + def get_groups(self, endpoint="/groups"): + return [ + {"id": "7", "name": "OpenAI", "platform": "openai"}, + {"id": "8", "name": "Claude", "platform": "anthropic"}, + {"id": "9", "name": "OpenAI 2", "platform": "openai"}, + ] + def list_accounts(self): + return list(self.accounts.values()) + def create_account(self, body, endpoint="/accounts"): + type(self).next_id += 1 + account = {"id": str(type(self).next_id), **body} + type(self).accounts[account["id"]] = account + return account + def update_account(self, account_id, body): + type(self).accounts[str(account_id)].update(body) + return type(self).accounts[str(account_id)] + @staticmethod + def extract_id(data): + return str(data.get("id")) + + monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeClient) + monkeypatch.setattr(websites_router, "reconcile_upstream_keys_full", lambda db, uid: None) + + def do_import(target_group_id): + return websites_router.import_upstream_keys_as_accounts( + website.id, + ImportAccountsRequest( + upstream_key_ids=[generated.id], + target_group_map={"vip": target_group_id}, + auto_priority_by_rate=False, + ), + db_session, + object(), + ) + + first = do_import("7") + second = do_import("8") + third = do_import("9") + + assert first.items[0].status == "created" + assert second.items[0].status == "created" + assert third.items[0].status == "exists" + assert third.items[0].platform == "openai" + + links = db_session.query(websites_router.UpstreamKeyAccountLink).filter_by( + upstream_key_id=generated.id, + website_id=website.id, + status="active", + ).all() + assert {(link.platform, link.target_group) for link in links} == { + ("openai", "9"), + ("anthropic", "8"), + } + assert len(FakeClient.accounts) == 2 diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index c5769b1..0dfd975 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -241,6 +241,14 @@ export interface GeneratedUpstreamKey { imported_at: string | null created_at: string | null has_key_value: boolean + imported_accounts: Array<{ + id: number + website_id: number + target_group: string + platform: string + remote_account_id: string + status: string + }> } export interface UpstreamRecharge { @@ -351,6 +359,7 @@ export interface WebsiteGroup { name: string rate_multiplier: string | null description?: string | null + platform?: string | null raw: Record } @@ -371,6 +380,7 @@ export interface GroupBindingData { percent: number algorithm: string enabled: boolean + platform?: string | null created_at: string updated_at: string } @@ -516,7 +526,7 @@ export const websitesApi = { upstream_key_ids: number[] target_group_map: Record account_name_prefix: string - default_platform: string + default_platform?: string platform_mode?: string concurrency?: number priority?: number diff --git a/frontend/src/views/Websites.vue b/frontend/src/views/Websites.vue index 6f7b033..7cedbb3 100644 --- a/frontend/src/views/Websites.vue +++ b/frontend/src/views/Websites.vue @@ -190,7 +190,7 @@
{{ binding.website_name }} / {{ binding.target_group_name || binding.target_group_id }}
{{ binding.algorithm === 'priority_weighted_plus_percent' ? `${algorithmLabel(binding.algorithm)} · 自动约 ${binding.percent}%` : `${algorithmLabel(binding.algorithm)} + ${binding.percent}%` }} · - {{ binding.source_groups.length }} 个上游分组 + {{ binding.source_groups.length }} 个上游分组 · {{ platformLabel(binding.platform || '') }}
@@ -279,7 +279,7 @@
- +
@@ -603,24 +603,8 @@ - - - - - - - - - - - - - - - - - - + + @@ -705,7 +689,7 @@ @@ -714,8 +698,8 @@
目标分组映射
{{ group.group_name || group.group_id }} - - + +
@@ -725,7 +709,7 @@ - + @@ -1291,8 +1275,6 @@ const importAccountsForm = ref({ upstream_key_ids: [] as number[], target_group_map: {} as Record, account_name_prefix: 'SmartUp', - default_platform: 'openai', - platform_mode: 'auto', concurrency: 100, priority: 1, auto_priority_by_rate: true, @@ -1379,7 +1361,6 @@ function isImportableGeneratedKey(item: GeneratedUpstreamKey) { && item.has_key_value && item.status !== 'failed' && item.status !== 'orphaned' - && !(item.imported_website_id === importAccountsForm.value.website_id && item.imported_account_id) } const importableGeneratedKeys = computed(() => @@ -1388,7 +1369,9 @@ const importableGeneratedKeys = computed(() => const importedGeneratedKeyCount = computed(() => importGeneratedKeys.value.filter((item) => - item.imported_website_id === importAccountsForm.value.website_id && Boolean(item.imported_account_id), + item.imported_accounts.some(link => + link.website_id === importAccountsForm.value.website_id && link.status === 'active', + ), ).length, ) @@ -2290,8 +2273,6 @@ async function openImportAccounts(site?: WebsiteData | null) { upstream_key_ids: [], target_group_map: {}, account_name_prefix: 'SmartUp', - default_platform: 'openai', - platform_mode: 'auto', concurrency: 100, priority: 1, auto_priority_by_rate: true, @@ -2330,21 +2311,6 @@ async function onImportAccountUpstreamChange(value: number) { }) } -function onPlatformModeChange(value: string) { - if (value === 'auto') { - importAccountsForm.value.default_platform = 'openai' - } -} - -function detectPlatform(item: { group_name?: string; group_id?: string; key_name?: string }) { - const text = `${item.group_name || ''} ${item.group_id || ''} ${item.key_name || ''}`.toLowerCase() - if (text.includes('claude') || text.includes('anthropic')) return 'Anthropic' - if (text.includes('gemini')) return 'Gemini' - if (text.includes('grok') || text.includes('xai')) return 'Grok' - if (text.includes('antigravity')) return 'Antigravity' - return 'OpenAI 兼容' -} - function platformLabel(platform: string) { const map: Record = { openai: 'OpenAI 兼容', @@ -2421,14 +2387,16 @@ async function submitImportAccounts() { ElMessage.error('请选择要导入的上游 Key') return } + if (selectedAccountGroups.value.some(group => !importAccountsForm.value.target_group_map[group.group_id])) { + ElMessage.error('请为每个来源分组选择目标分组') + return + } importingAccounts.value = true try { const res = await websitesApi.importAccountsFromUpstreamKeys(importAccountsForm.value.website_id, { upstream_key_ids: importAccountsForm.value.upstream_key_ids, target_group_map: importAccountsForm.value.target_group_map, account_name_prefix: importAccountsForm.value.account_name_prefix, - default_platform: importAccountsForm.value.default_platform, - platform_mode: importAccountsForm.value.platform_mode, concurrency: importAccountsForm.value.concurrency, priority: importAccountsForm.value.priority, auto_priority_by_rate: importAccountsForm.value.auto_priority_by_rate,