from __future__ import annotations import json import logging from datetime import datetime, timezone from typing import List from fastapi import APIRouter, Depends, HTTPException, Query, status 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.website import Website, WebsiteGroupBinding, WebsiteSyncLog from app.schemas.website import ( BindingCreate, BindingResponse, BindingUpdate, ImportAccountItem, ImportAccountsRequest, ImportAccountsResponse, ImportGroupItem, ImportGroupsRequest, ImportGroupsResponse, ReorderPriorityItem, ReorderPriorityRequest, ReorderPriorityResponse, SyncImportStatusRequest, TestResult, WebsiteCreate, WebsiteGroupResponse, WebsiteGroupCreate, WebsiteGroupUpdate, WebsiteResponse, WebsiteSyncLogResponse, WebsiteUpdate, WebsiteBatchSyncResponse, OrganizeGroupsItem, OrganizeGroupsResponse, CleanupInvalidAccountsItem, CleanupInvalidAccountsPreviewResponse, CleanupInvalidAccountsExecuteResponse, SetConcurrencyRequest, SetConcurrencyItem, SetConcurrencyResponse, SyncUpstreamModelsItem, SyncUpstreamModelsResponse, ) from app.services.website_client import Sub2ApiWebsiteClient, _extract_id from app.services.website_sync import ( binding_sources, sync_binding, build_target_group_priority_map, sync_account_priorities_for_website, reconcile_upstream_keys_full, sync_account_priorities_for_upstream, latest_rate_map, build_imported_account_name, ) from app.utils.auth import get_current_user from app.utils.number import fixed_decimal_number router = APIRouter(tags=["websites"]) logger = logging.getLogger(__name__) MASK = "***" SECRET_KEYS = {"password", "token", "key", "secret", "api_key"} ALGORITHMS = {"max_plus_percent", "average_plus_percent", "min_plus_percent"} def _mask(cfg: dict) -> dict: masked = {} for key, value in cfg.items(): masked[key] = MASK if key.lower() in SECRET_KEYS and value else value return masked def _website_response(row: Website) -> WebsiteResponse: return WebsiteResponse( id=row.id, name=row.name, site_type=row.site_type, base_url=row.base_url, api_prefix=row.api_prefix, auth_type=row.auth_type, auth_config_masked=_mask(json.loads(row.auth_config_json or "{}")), groups_endpoint=row.groups_endpoint, group_update_endpoint=row.group_update_endpoint, enabled=row.enabled, auto_sync_enabled=row.auto_sync_enabled, timeout_seconds=row.timeout_seconds, last_status=row.last_status, last_checked_at=row.last_checked_at, last_error=row.last_error, created_at=row.created_at, updated_at=row.updated_at, ) def _binding_response(db: Session, row: WebsiteGroupBinding) -> BindingResponse: website = db.query(Website).filter(Website.id == row.website_id).first() return BindingResponse( id=row.id, website_id=row.website_id, website_name=website.name if website else "", target_group_id=row.target_group_id, target_group_name=row.target_group_name, source_groups=binding_sources(row), percent=float(row.percent or 0), algorithm=row.algorithm, enabled=row.enabled, created_at=row.created_at, updated_at=row.updated_at, ) def _log_response(row: WebsiteSyncLog) -> WebsiteSyncLogResponse: return WebsiteSyncLogResponse( id=row.id, website_id=row.website_id, binding_id=row.binding_id, target_group_id=row.target_group_id, target_group_name=row.target_group_name, algorithm=row.algorithm, percent=float(row.percent or 0), source_rates=json.loads(row.source_rates_json or "[]"), old_rate=row.old_rate, new_rate=row.new_rate, status=row.status, message=row.message, created_at=row.created_at, ) def _ensure_unique_target(db: Session, website_id: int, target_group_id: str, exclude_id: int | None = None) -> None: q = db.query(WebsiteGroupBinding).filter( WebsiteGroupBinding.website_id == website_id, WebsiteGroupBinding.target_group_id == target_group_id, ) if exclude_id is not None: q = q.filter(WebsiteGroupBinding.id != exclude_id) if q.first(): raise HTTPException(400, "同一目标网站分组只能维护一条绑定记录") def _client(row: Website) -> Sub2ApiWebsiteClient: return Sub2ApiWebsiteClient( base_url=row.base_url, api_prefix=row.api_prefix, auth_type=row.auth_type, auth_config=json.loads(row.auth_config_json or "{}"), timeout=float(row.timeout_seconds), target_id=row.id, target_name=row.name, ) def _latest_upstream_groups(db: Session, upstream_id: int) -> list[dict]: row = ( db.query(UpstreamRateSnapshot) .filter(UpstreamRateSnapshot.upstream_id == upstream_id) .order_by(UpstreamRateSnapshot.captured_at.desc()) .first() ) if not row: raise HTTPException(404, "no upstream snapshot found; run upstream check first") snapshot = json.loads(row.snapshot_json or "{}") groups = snapshot.get("groups") or {} if not isinstance(groups, dict): return [] return [item for item in groups.values() if isinstance(item, dict)] def _source_group_id(group: dict) -> str: return str(group.get("group_id") or group.get("id") or group.get("name") or "") def _source_group_name(group: dict, gid: str) -> str: return str(group.get("group_name") or group.get("name") or gid) def _source_group_rate(group: dict) -> float: raw = group.get("rate") or group.get("default_rate") or group.get("rate_multiplier") or 1 try: return float(raw) except (TypeError, ValueError): return 1.0 def _numeric_group_id(value: str | None) -> int | None: if value is None or value == "": return None try: return int(value) except ValueError: return None @router.get("/api/websites", response_model=List[WebsiteResponse]) def list_websites(db: Session = Depends(get_db), _=Depends(get_current_user)): return [_website_response(row) for row in db.query(Website).order_by(Website.id).all()] @router.post("/api/websites", response_model=WebsiteResponse, status_code=201) def create_website(body: WebsiteCreate, db: Session = Depends(get_db), _=Depends(get_current_user)): if body.site_type != "sub2api": raise HTTPException(400, "目前只支持 sub2api") row = Website( name=body.name, site_type=body.site_type, base_url=body.base_url.rstrip("/"), api_prefix=body.api_prefix, auth_type=body.auth_type, auth_config_json=json.dumps(body.auth_config, ensure_ascii=False), groups_endpoint=body.groups_endpoint, group_update_endpoint=body.group_update_endpoint, enabled=body.enabled, auto_sync_enabled=body.auto_sync_enabled, timeout_seconds=body.timeout_seconds, ) db.add(row) db.commit() db.refresh(row) return _website_response(row) @router.put("/api/websites/{wid}", response_model=WebsiteResponse) def update_website(wid: int, body: WebsiteUpdate, db: Session = Depends(get_db), _=Depends(get_current_user)): row = db.query(Website).filter(Website.id == wid).first() if not row: raise HTTPException(404, "website not found") data = body.model_dump(exclude_none=True) if "site_type" in data and data["site_type"] != "sub2api": raise HTTPException(400, "目前只支持 sub2api") if "auth_config" in data: existing = json.loads(row.auth_config_json or "{}") incoming = data.pop("auth_config") for key, value in incoming.items(): if value != MASK: existing[key] = value row.auth_config_json = json.dumps(existing, ensure_ascii=False) if "base_url" in data: data["base_url"] = data["base_url"].rstrip("/") for key, value in data.items(): setattr(row, key, value) row.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(row) return _website_response(row) @router.delete("/api/websites/{wid}", status_code=204) def delete_website(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)): row = db.query(Website).filter(Website.id == wid).first() if not row: raise HTTPException(404, "website not found") db.query(WebsiteSyncLog).filter(WebsiteSyncLog.website_id == wid).delete(synchronize_session=False) db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.website_id == wid).delete(synchronize_session=False) db.delete(row) db.commit() @router.post("/api/websites/{wid}/test", response_model=TestResult) def test_website(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)): row = db.query(Website).filter(Website.id == wid).first() if not row: raise HTTPException(404, "website not found") try: with _client(row) as c: groups = c.get_groups(row.groups_endpoint) row.last_status = "healthy" row.last_error = None row.last_checked_at = datetime.now(timezone.utc) db.commit() return TestResult(success=True, message=f"连接成功，获取到 {len(groups)} 个分组") except Exception as exc: row.last_status = "unhealthy" row.last_error = str(exc) row.last_checked_at = datetime.now(timezone.utc) db.commit() return TestResult(success=False, message="连接失败", detail=str(exc)) @router.get("/api/websites/{wid}/groups", response_model=List[WebsiteGroupResponse]) def list_website_groups(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)): row = db.query(Website).filter(Website.id == wid).first() if not row: raise HTTPException(404, "website not found") try: with _client(row) as c: return c.get_groups(row.groups_endpoint) except Exception as exc: raise HTTPException(502, str(exc)) @router.post("/api/websites/{wid}/groups", response_model=dict) def create_website_group( wid: int, body: WebsiteGroupCreate, db: Session = Depends(get_db), _=Depends(get_current_user), ): row = db.query(Website).filter(Website.id == wid).first() if not row: raise HTTPException(404, "website not found") try: with _client(row) as c: payload = { "name": body.name, "description": body.description, } return c.create_group(payload, row.groups_endpoint) except Exception as exc: raise HTTPException(502, str(exc)) @router.put("/api/websites/{wid}/groups/{group_id:path}", response_model=dict) def update_website_group( wid: int, group_id: str, body: WebsiteGroupUpdate, db: Session = Depends(get_db), _=Depends(get_current_user), ): row = db.query(Website).filter(Website.id == wid).first() if not row: raise HTTPException(404, "website not found") try: with _client(row) as c: payload = { "name": body.name, "description": body.description, } resp = c.update_group(row.group_update_endpoint, group_id, payload) # Sync updated group name to local group bindings cache db.query(WebsiteGroupBinding).filter( WebsiteGroupBinding.website_id == wid, WebsiteGroupBinding.target_group_id == group_id ).update({"target_group_name": body.name}) db.commit() return resp except Exception as exc: raise HTTPException(502, str(exc)) @router.post("/api/websites/{wid}/groups/import-from-upstream/{upstream_id}", response_model=ImportGroupsResponse) def import_groups_from_upstream( wid: int, upstream_id: int, body: ImportGroupsRequest, db: Session = Depends(get_db), _=Depends(get_current_user), ): website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") if website.site_type != "sub2api": raise HTTPException(400, "目前只支持 sub2api") upstream = db.query(Upstream).filter(Upstream.id == upstream_id).first() if not upstream: raise HTTPException(404, "upstream not found") selected = set(body.group_ids) groups = _latest_upstream_groups(db, upstream_id) # 拉取目标网站已有分组，同名则跳过 try: existing_names = set() with _client(website) as c: for eg in c.get_groups(website.groups_endpoint): gname = eg.get("name") or eg.get("group_name") or "" if gname: existing_names.add(gname) except Exception: existing_names = set() items: list[ImportGroupItem] = [] with _client(website) as c: for group in groups: source_gid = _source_group_id(group) if not source_gid or (selected and source_gid not in selected): continue source_name = _source_group_name(group, source_gid) target_name = f"{body.name_prefix}{source_name}" if body.name_prefix else source_name # 检查是否已存在同名分组 if target_name in existing_names: items.append(ImportGroupItem( source_group_id=source_gid, source_group_name=source_name, target_group_name=target_name, status="exists", message="目标分组已存在，已跳过", )) continue 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("rpm_limit") is not None: create_body["rpm_limit"] = group.get("rpm_limit") try: created = c.create_group(create_body) target_id = c.extract_id(created) items.append(ImportGroupItem( source_group_id=source_gid, source_group_name=source_name, target_group_id=target_id or None, target_group_name=str(created.get("name") or target_name), status="created", message="已创建", raw=created, )) except Exception as exc: msg = str(exc) # 捕获 409 等已存在错误 if "已存在" in msg or "already exists" in msg.lower() or "409" in msg or "Conflict" in msg: items.append(ImportGroupItem( source_group_id=source_gid, source_group_name=source_name, target_group_name=target_name, status="exists", message="目标分组已存在（接口返回冲突）", )) else: logger.exception("import website group failed website=%s upstream=%s group=%s", wid, upstream_id, source_gid) items.append(ImportGroupItem( source_group_id=source_gid, source_group_name=source_name, target_group_name=target_name, status="failed", message=msg, )) created_count = len([item for item in items if item.status == "created"]) exists_count = len([item for item in items if item.status == "exists"]) failed_count = len([item for item in items if item.status == "failed"]) msg_parts = [] if created_count: msg_parts.append(f"新建 {created_count}") if exists_count: msg_parts.append(f"已存在 {exists_count}") if failed_count: msg_parts.append(f"失败 {failed_count}") return ImportGroupsResponse( success=failed_count == 0, message="、".join(msg_parts) + f" / 共 {len(items)} 个" if msg_parts else f"共处理 {len(items)} 个分组", items=items, ) 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 "antigravity" in lower: return "antigravity" return fallback @router.post("/api/websites/{wid}/accounts/sync-imported-upstream-keys", response_model=ImportAccountsResponse) def sync_imported_upstream_keys( wid: int, body: SyncImportStatusRequest, db: Session = Depends(get_db), _=Depends(get_current_user), ): """校验已导入的上游 Key 在目标 Sub2API 账号管理中是否仍存在。""" website = db.query(Website).filter(Website.id == wid).first() 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] = [] with _client(website) as c: remote_accounts = None try: remote_accounts = c.list_accounts() except Exception as e: logger.warning("failed to list accounts for website %s at sync: %s", wid, e) if remote_accounts is not None: remote_ids = set() for acc in remote_accounts: acc_id = _extract_id(acc) if acc_id: remote_ids.add(str(acc_id)) else: remote_ids = None for row in rows: platform = _detect_platform(f"{row.group_name} {row.group_id} {row.key_name}", "openai") if not row.imported_account_id: continue old_account_id = row.imported_account_id if remote_ids is None: 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="check_failed", 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="目标账号已删除，已清除导入标记", )) else: 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="exists", message="目标账号仍存在", )) db.commit() cleared_count = len([i for i in items if i.status == "stale_cleared"]) check_failed_count = len([i for i in items if i.status == "check_failed"]) msg_parts = [] if cleared_count: msg_parts.append(f"清除 {cleared_count}") if check_failed_count: msg_parts.append(f"校验失败 {check_failed_count}") return ImportAccountsResponse( success=check_failed_count == 0, message="、".join(msg_parts) + f" / 共 {len(items)} 个" if msg_parts else f"共校验 {len(items)} 个，无变化", items=items, ) @router.post("/api/websites/{wid}/accounts/reorder-priority", response_model=ReorderPriorityResponse) def reorder_account_priorities( wid: int, body: ReorderPriorityRequest, db: Session = Depends(get_db), _=Depends(get_current_user), ): website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") if website.site_type != "sub2api": raise HTTPException(400, "目前只支持 sub2api") upstream = db.query(Upstream).filter(Upstream.id == body.upstream_id).first() if not upstream: raise HTTPException(404, "upstream not found") results = sync_account_priorities_for_upstream(db, body.upstream_id, website_id=wid) failed_count = sum(1 for item in results if item.get("status") == "failed") success_count = sum(1 for item in results if item.get("status") == "success") skipped_count = sum(1 for item in results if item.get("status") == "skipped") stale_count = sum(1 for item in results if item.get("status") == "stale") parts = [] if success_count: parts.append(f"更新 {success_count}") if failed_count: parts.append(f"失败 {failed_count}") if stale_count: parts.append(f"清理失效账号 {stale_count}") if skipped_count: parts.append(f"跳过 {skipped_count}") message = "、".join(parts) + f" / 共 {len(results)} 个" if parts else "没有需要重排的账号" return ReorderPriorityResponse( success=failed_count == 0, message=message, items=[ReorderPriorityItem(**item) for item in results], ) @router.post("/api/websites/{wid}/accounts/import-upstream-keys", response_model=ImportAccountsResponse) def import_upstream_keys_as_accounts( wid: int, body: ImportAccountsRequest, db: Session = Depends(get_db), _=Depends(get_current_user), ): website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") if website.site_type != "sub2api": raise HTTPException(400, "目前只支持 sub2api") if not body.upstream_key_ids: raise HTTPException(400, "请选择要导入的 Key") rows = ( db.query(UpstreamGeneratedKey) .filter(UpstreamGeneratedKey.id.in_(body.upstream_key_ids)) .order_by(UpstreamGeneratedKey.id) .all() ) # 导入前尝试对账（失败不阻塞，仅打日志—避免远端不可达时误删本地 Key） upstream_ids = {row.upstream_id for row in rows} for uid in upstream_ids: try: reconcile_upstream_keys_full(db, uid) except Exception as exc: logger.warning("import reconcile failed for upstream %s: %s", uid, exc) # 重新查询（对账成功时已清理失效 Key） rows = ( db.query(UpstreamGeneratedKey) .filter(UpstreamGeneratedKey.id.in_(body.upstream_key_ids)) .order_by(UpstreamGeneratedKey.id) .all() ) found_ids = {row.id for row in rows} missing_ids = [kid for kid in body.upstream_key_ids if kid not in found_ids] items: list[ImportAccountItem] = [ ImportAccountItem( upstream_key_id=kid, source_group_id="", source_group_name="", platform=body.default_platform, status="failed", message="key not found", ) for kid in missing_ids ] # 预载上游以方便获取名称和 base_url from app.models.upstream import Upstream as _Up upstream_ids = {row.upstream_id for row in rows} upstreams_db = db.query(_Up).filter(_Up.id.in_(upstream_ids)).all() upstreams_map = {u.id: u for u in upstreams_db} # 按倍率自动分配优先级 rate_priority_map: dict[tuple[str, int, str], int] = {} if body.auto_priority_by_rate: try: target_group_sources = {} for row in rows: target_group_id = body.target_group_map.get(row.group_id) if target_group_id: target_group_sources.setdefault(str(target_group_id), []).append((row.upstream_id, row.group_id)) rate_priority_map = build_target_group_priority_map(db, target_group_sources) except Exception: # 没有快照时忽略，后续 fallback 到 body.priority pass with _client(website) as c: target_group_names: dict[str, str] = {} 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) except Exception as e: logger.warning("failed to fetch groups for website %s at import: %s", wid, e) # 预先拉取远端账号列表以避免循环请求，并能拿到已存在账号 of 远端名称 remote_accounts_map: dict[str, dict[str, Any]] = {} remote_accounts_list = None has_list_accounts = hasattr(c, "list_accounts") if has_list_accounts: try: remote_accounts_list = c.list_accounts() if remote_accounts_list is not None: for acc in remote_accounts_list: aid = c.extract_id(acc) if aid: remote_accounts_map[aid] = acc except Exception as e: logger.warning("failed to fetch remote accounts at import: %s", e) for row in rows: # 跳过远端已不存在或导入失败的 Key（对账后标记为 orphaned / import_failed） if row.status in ("orphaned", "failed", "import_failed"): 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, status="failed", message="远端 Key 已不存在，请重新生成", )) continue # 先确定平台（失败项也需要记录） if body.platform_mode == "auto": platform = _detect_platform( f"{row.group_name} {row.group_id} {row.key_name}", body.default_platform, ) else: platform = body.default_platform upstream = upstreams_map.get(row.upstream_id) upstream_name = upstream.name if upstream else "Unknown" upstream_base_url = upstream.base_url if upstream else "" # 构建新格式的预期账号名称 expected_new_name = build_imported_account_name( upstream_name=upstream_name, group_name=row.group_name, group_id=row.group_id, key_id=row.id, ) # 幂等校验：已导入过则检查远端账号是否仍存在 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 "" 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, 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 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, platform=platform, status="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) account_name = expected_new_name account_body = { "name": account_name, "platform": platform, "type": "apikey", "credentials": { "api_key": row.key_value, "base_url": upstream_base_url, }, "group_ids": group_ids, "rate_multiplier": 1, "concurrency": body.concurrency, "priority": rate_priority_map.get((str(target_group_id), row.upstream_id, row.group_id), body.priority) if body.auto_priority_by_rate else body.priority, "notes": f"Imported by SmartUp from upstream key #{row.id}", } 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" row.error = None db.commit() 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=account_id or None, account_name=str(created.get("name") or account_name), platform=platform, upstream_base_url=upstream_base_url, status="created", message="已创建账号", raw=created, )) 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) db.commit() 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_name=account_name, platform=platform, upstream_base_url=upstream_base_url, status="failed", message=str(exc), )) created_count = len([item for item in items if item.status == "created"]) exists_count = len([item for item in items if item.status == "exists"]) failed_count = len([item for item in items if item.status == "failed"]) check_failed_count = len([item for item in items if item.status == "check_failed"]) msg_parts = [] if created_count: msg_parts.append(f"新建 {created_count}") if exists_count: msg_parts.append(f"已存在 {exists_count}") if check_failed_count: msg_parts.append(f"校验失败 {check_failed_count}") if failed_count: msg_parts.append(f"失败 {failed_count}") return ImportAccountsResponse( success=failed_count == 0 and check_failed_count == 0, message="、".join(msg_parts) + f" / 共 {len(items)} 个" if msg_parts else f"共处理 {len(items)} 个", items=items, ) def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClient) -> dict[str, dict[str, Any]] | None: import re # 1. Fetch remote accounts remote_accounts = c.list_accounts() if remote_accounts is None: return None remote_accounts_map = {} for acc in remote_accounts: aid = c.extract_id(acc) if aid: remote_accounts_map[aid] = acc candidates = {} # 2. Condition A: Local orphaned keys orphaned_keys = db.query(UpstreamGeneratedKey).filter( UpstreamGeneratedKey.status == "orphaned", UpstreamGeneratedKey.imported_website_id == website.id, UpstreamGeneratedKey.imported_account_id.isnot(None), ).all() for key in orphaned_keys: acc_id = key.imported_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() upstream_name = upstream.name if upstream else None candidates[acc_id] = { "account_id": acc_id, "account_name": acc_name, "upstream_key_id": key.id, "upstream_name": upstream_name, "source_group_name": key.group_name, "reason": "本地上游 Key 已被标记为 orphaned", "is_local_orphan": True, "db_key": key, } # 3. Condition B: Remote SmartUp-marked orphan accounts pattern = re.compile(r"Imported by SmartUp from upstream key #(\d+)") for aid, acc in remote_accounts_map.items(): if aid in candidates: continue key_id = None for field in ("notes", "description", "remark"): val = acc.get(field) if val and isinstance(val, str): match = pattern.search(val) if match: key_id = int(match.group(1)) break if key_id is None: continue key = db.query(UpstreamGeneratedKey).filter( UpstreamGeneratedKey.id == key_id ).first() is_orphan = False reason = "" if not key: is_orphan = True reason = "本地找不到对应的上游 Key 记录" elif key.imported_website_id != website.id or key.imported_account_id != aid: is_orphan = True reason = "本地 Key 记录存在，但未关联到当前目标站的当前账号" elif key.status == 'orphaned': is_orphan = True reason = "本地上游 Key 已被标记为 orphaned" if is_orphan: upstream_name = None source_group_name = None if key: upstream = db.query(Upstream).filter(Upstream.id == key.upstream_id).first() upstream_name = upstream.name if upstream else None source_group_name = key.group_name candidates[aid] = { "account_id": aid, "account_name": acc.get("name") or aid, "upstream_key_id": key_id, "upstream_name": upstream_name, "source_group_name": source_group_name, "reason": reason, "is_local_orphan": False, "db_key": key, } return candidates def _test_candidate_account(c: Sub2ApiWebsiteClient, account_id: str) -> tuple[str, bool, str]: """返回 (test_status, cleanup_allowed, message)""" try: res = c.test_account(account_id) status = res.get("status") msg = res.get("message", "") if status == "404": return "404", True, msg elif status == "success": return "test_passed", False, msg elif status == "failed": return "failed", True, msg else: return "test_unknown", False, msg except Exception as e: return "test_unknown", False, f"测试时发生异常: {e}" @router.post("/api/websites/{wid}/accounts/cleanup-invalid/preview", response_model=CleanupInvalidAccountsPreviewResponse) def preview_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)): website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") with _client(website) as c: candidates = _get_cleanup_candidates(db, website, c) if candidates is None: return CleanupInvalidAccountsPreviewResponse( success=False, message="拉取远端账号列表失败，无法执行清理", items=[] ) items = [] for aid, cand in candidates.items(): test_status, cleanup_allowed, test_msg = _test_candidate_account(c, aid) items.append(CleanupInvalidAccountsItem( account_id=cand["account_id"], account_name=cand["account_name"], upstream_key_id=cand["upstream_key_id"], upstream_name=cand["upstream_name"], source_group_name=cand["source_group_name"], reason=cand["reason"], test_status=test_status, cleanup_allowed=cleanup_allowed, action_status="previewed", message=test_msg, )) return CleanupInvalidAccountsPreviewResponse( success=True, message=f"已成功加载 {len(items)} 个失效账号候选对象", items=items ) @router.post("/api/websites/{wid}/accounts/cleanup-invalid/execute", response_model=CleanupInvalidAccountsExecuteResponse) def execute_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)): website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") with _client(website) as c: candidates = _get_cleanup_candidates(db, website, c) if candidates is None: return CleanupInvalidAccountsExecuteResponse( success=False, message="拉取远端账号列表失败，无法执行清理", items=[] ) items = [] for aid, cand in candidates.items(): test_status, cleanup_allowed, test_msg = _test_candidate_account(c, aid) action_status = "skipped" msg = "测试通过或未知，已跳过删除" if cleanup_allowed: try: del_res = c.delete_account(aid) status = del_res.get("status") del_msg = del_res.get("message", "") if status in ("deleted", "already_deleted"): action_status = status msg = del_msg # Clear database flags if it's a local orphan key if cand["is_local_orphan"] and cand["db_key"]: key = cand["db_key"] 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) else: action_status = "failed" msg = f"删除失败: {del_msg}" except Exception as e: action_status = "failed" msg = f"删除失败: {e}" else: msg = f"账号状态正常，无需清理 ({test_msg})" items.append(CleanupInvalidAccountsItem( account_id=cand["account_id"], account_name=cand["account_name"], upstream_key_id=cand["upstream_key_id"], upstream_name=cand["upstream_name"], source_group_name=cand["source_group_name"], reason=cand["reason"], test_status=test_status, cleanup_allowed=cleanup_allowed, action_status=action_status, message=msg, )) db.commit() deleted_count = sum(1 for item in items if item.action_status in ("deleted", "already_deleted")) failed_count = sum(1 for item in items if item.action_status == "failed") skipped_count = sum(1 for item in items if item.action_status == "skipped") success = (failed_count == 0) msg_parts = [] if deleted_count: msg_parts.append(f"成功清理 {deleted_count} 个") if failed_count: msg_parts.append(f"失败 {failed_count} 个") if skipped_count: msg_parts.append(f"跳过 {skipped_count} 个") message = "清理执行完毕：" + "，".join(msg_parts) if msg_parts else "没有需要清理的失效账号" return CleanupInvalidAccountsExecuteResponse( success=success, message=message, items=items ) @router.post("/api/websites/{wid}/accounts/set-concurrency", response_model=SetConcurrencyResponse) def set_website_accounts_concurrency( wid: int, body: SetConcurrencyRequest, db: Session = Depends(get_db), _=Depends(get_current_user), ): """批量设置当前网站已导入账号的并发数""" website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") if website.site_type != "sub2api": raise HTTPException(400, "only sub2api site supports account concurrency settings") with _client(website) as c: try: remote_accounts = c.list_accounts() except Exception as e: return SetConcurrencyResponse( success=False, message=f"拉取远端账号列表失败: {e}", items=[] ) if remote_accounts is None: return SetConcurrencyResponse( success=False, message="拉取远端账号列表失败，无法设置并发", items=[] ) remote_map = {} for acc in remote_accounts: aid = c.extract_id(acc) if aid: remote_map[aid] = acc keys = db.query(UpstreamGeneratedKey).filter( UpstreamGeneratedKey.imported_website_id == wid, UpstreamGeneratedKey.imported_account_id.isnot(None) ).all() candidates = {} for key in keys: aid = key.imported_account_id if aid not in candidates: candidates[aid] = { "account_id": aid, "db_key": key } if not candidates: return SetConcurrencyResponse( success=True, message="没有找到 SmartUp 导入的有效账号", items=[] ) bulk_candidates = [] skipped_items = [] for aid, cand in candidates.items(): remote_acc = remote_map.get(aid) if not remote_acc: skipped_items.append(SetConcurrencyItem( account_id=aid, account_name=None, old_concurrency=None, target_concurrency=body.concurrency, status="skipped", message="账号在远端已被删除或不存在" )) continue acc_name = remote_acc.get("name") old_con = remote_acc.get("concurrency") try: int(aid) bulk_candidates.append({ "account_id": aid, "account_name": acc_name, "old_concurrency": old_con }) except ValueError: skipped_items.append(SetConcurrencyItem( account_id=aid, account_name=acc_name, old_concurrency=old_con, target_concurrency=body.concurrency, status="skipped", message="账号 ID 无法转换为数字，跳过批量更新" )) items = [] items.extend(skipped_items) if not bulk_candidates: success = all(item.status == "skipped" for item in items) return SetConcurrencyResponse( success=success, message=f"无可执行批量更新的账号。已跳过 {len(skipped_items)} 个", items=items ) bulk_ids = [item["account_id"] for item in bulk_candidates] bulk_success = False bulk_error_msg = "" try: c.bulk_update_accounts(bulk_ids, {"concurrency": body.concurrency}) bulk_success = True except Exception as e: bulk_error_msg = str(e) logger.warning("bulk_update_accounts failed, falling back to individual updates: %s", e) if bulk_success: for item in bulk_candidates: items.append(SetConcurrencyItem( account_id=item["account_id"], account_name=item["account_name"], old_concurrency=item["old_concurrency"], target_concurrency=body.concurrency, status="success", message="成功" )) else: for item in bulk_candidates: aid = item["account_id"] try: c.update_account(aid, {"concurrency": body.concurrency}) items.append(SetConcurrencyItem( account_id=aid, account_name=item["account_name"], old_concurrency=item["old_concurrency"], target_concurrency=body.concurrency, status="success", message="成功 (逐个更新)" )) except Exception as e: items.append(SetConcurrencyItem( account_id=aid, account_name=item["account_name"], old_concurrency=item["old_concurrency"], target_concurrency=body.concurrency, status="failed", message=f"更新失败: {e}" )) success_count = sum(1 for item in items if item.status == "success") failed_count = sum(1 for item in items if item.status == "failed") skip_count = sum(1 for item in items if item.status == "skipped") success = (failed_count == 0) msg_parts = [] if success_count: msg_parts.append(f"成功 {success_count} 个") if failed_count: msg_parts.append(f"失败 {failed_count} 个") if skip_count: msg_parts.append(f"跳过 {skip_count} 个") message = "设置账号并发执行完毕：" + "，".join(msg_parts) return SetConcurrencyResponse( success=success, message=message, items=items ) @router.post("/api/websites/{wid}/accounts/sync-upstream-models", response_model=SyncUpstreamModelsResponse) def sync_website_accounts_upstream_models( wid: int, db: Session = Depends(get_db), _=Depends(get_current_user), ): """一键同步上游模型""" website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") if website.site_type != "sub2api": raise HTTPException(400, "only sub2api site supports account model syncing") with _client(website) as c: try: remote_accounts = c.list_accounts() except Exception as e: return SyncUpstreamModelsResponse( success=False, message=f"拉取远端账号列表失败: {e}", items=[] ) if remote_accounts is None: return SyncUpstreamModelsResponse( success=False, message="拉取远端账号列表失败，无法同步上游模型", items=[] ) remote_map = {} for acc in remote_accounts: aid = c.extract_id(acc) if aid: remote_map[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() candidates = {} for key in keys: aid = key.imported_account_id if aid not in candidates: candidates[aid] = { "account_id": aid, "upstream_id": key.upstream_id, "db_key": key } if not candidates: return SyncUpstreamModelsResponse( success=True, message="没有找到 SmartUp 导入的有效账号", items=[] ) upstream_ids = {cand["upstream_id"] for cand in candidates.values()} upstreams_map = {up.id: up for up in db.query(Upstream).filter(Upstream.id.in_(upstream_ids)).all()} items = [] for aid, cand in candidates.items(): # 校验是否存在于远端 remote_acc = remote_map.get(aid) if not remote_acc: items.append(SyncUpstreamModelsItem( account_id=aid, account_name=None, model_count=0, models=[], status="skipped", message="账号在远端已被删除或不存在" )) continue acc_name = remote_acc.get("name") # 校验账号 ID 是否为数字 try: int(aid) except ValueError: items.append(SyncUpstreamModelsItem( account_id=aid, account_name=acc_name, model_count=0, models=[], status="skipped", message="账号 ID 非数字，跳过同步" )) continue upstream = upstreams_map.get(cand["upstream_id"]) upstream_base_url = upstream.base_url if upstream else None if not upstream_base_url or not upstream_base_url.strip(): items.append(SyncUpstreamModelsItem( account_id=aid, account_name=acc_name, model_count=0, models=[], status="failed", message="上游 base_url 为空，跳过处理避免继续写坏账号" )) continue # 1. 先安全修复 base_url (不依赖同步模型结果) # 在 remote_acc.get("credentials") 基础上进行修改，保留其他非敏感 credentials (例如 compact_model_mapping) current_creds = dict(remote_acc.get("credentials") or {}) current_creds["base_url"] = upstream_base_url try: c.update_account(aid, {"credentials": current_creds}) except Exception as e: items.append(SyncUpstreamModelsItem( account_id=aid, account_name=acc_name, model_count=0, models=[], status="failed", message=f"修复 Base URL 失败: {e}" )) continue # 2. 调用 sub2api 同步模型 try: raw_models = c.sync_account_upstream_models(aid) # 过滤空、去重、排序 valid_models = sorted(list(set(m.strip() for m in raw_models if m and m.strip()))) if not valid_models: items.append(SyncUpstreamModelsItem( account_id=aid, account_name=acc_name, model_count=0, models=[], status="failed", message="已修复 Base URL，但上游同步返回模型列表为空" )) continue # 3. 模型同步成功后，在 current_creds 基础上替换 model_mapping 并写回 updated_creds = dict(current_creds) updated_creds["model_mapping"] = {m: m for m in valid_models} c.update_account(aid, {"credentials": updated_creds}) items.append(SyncUpstreamModelsItem( account_id=aid, account_name=acc_name, model_count=len(valid_models), models=valid_models, status="success", message=f"已修复 Base URL 并同步 {len(valid_models)} 个模型" )) except Exception as e: items.append(SyncUpstreamModelsItem( account_id=aid, account_name=acc_name, model_count=0, models=[], status="failed", message=f"已修复 Base URL，但模型同步失败: {e}" )) success_count = sum(1 for item in items if item.status == "success") failed_count = sum(1 for item in items if item.status == "failed") skip_count = sum(1 for item in items if item.status == "skipped") success = (failed_count == 0) msg_parts = [] if success_count: msg_parts.append(f"成功 {success_count} 个") if failed_count: msg_parts.append(f"失败 {failed_count} 个") if skip_count: msg_parts.append(f"跳过 {skip_count} 个") message = "同步上游模型执行完毕：" + "，".join(msg_parts) return SyncUpstreamModelsResponse( success=success, message=message, items=items ) @router.post("/api/websites/{wid}/groups/organize", response_model=OrganizeGroupsResponse) def organize_website_groups( wid: int, db: Session = Depends(get_db), _=Depends(get_current_user), ): """一键整理分组：按现有绑定关系导入/自愈账号""" website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") if website.site_type != "sub2api": raise HTTPException(400, "目前只支持 sub2api") # 1. 读取当前网站的所有启用绑定关系 bindings = ( db.query(WebsiteGroupBinding) .filter(WebsiteGroupBinding.website_id == wid, WebsiteGroupBinding.enabled == True) .all() ) smartup_managed_group_ids = {str(b.target_group_id) for b in bindings if b.target_group_id} # 合并已导入 Key 上历史记录的 imported_target_group_id，以确保变更绑定后能清理旧分组 hist_group_ids = ( db.query(UpstreamGeneratedKey.imported_target_group_id) .filter( UpstreamGeneratedKey.imported_website_id == wid, UpstreamGeneratedKey.imported_target_group_id.isnot(None), UpstreamGeneratedKey.imported_target_group_id != "", ) .distinct() .all() ) for row in hist_group_ids: smartup_managed_group_ids.add(str(row[0])) items: list[OrganizeGroupsItem] = [] # 2. 收集所有的 upstream_id 并执行对账，以保证本地 Key 的状态最新 upstream_ids = set() for b in bindings: sources = binding_sources(b) for src in sources: uid = src.get("upstream_id") if uid: upstream_ids.add(int(uid)) for uid in upstream_ids: try: reconcile_upstream_keys_full(db, uid) except Exception as exc: logger.warning("organize reconcile failed for upstream %s: %s", uid, exc) # 3. 按倍率自动分配优先级 (分目标分组进行优先级计算) rate_priority_map: dict[tuple[str, int, str], int] = {} try: target_group_sources = {} for b in bindings: for src in binding_sources(b): uid = src.get("upstream_id") gid = src.get("group_id") if uid and gid: target_group_sources.setdefault(str(b.target_group_id), []).append((int(uid), str(gid))) rate_priority_map = build_target_group_priority_map(db, target_group_sources) except Exception: pass with _client(website) as c: # 获取远端分组名称列表和账号列表 target_group_names: dict[str, str] = {} 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) except Exception as e: logger.warning("failed to fetch groups for website %s at organize: %s", wid, e) remote_account_ids: set[str] | None = None try: remote_accounts = c.list_accounts() if remote_accounts is not None: remote_account_ids = set() for acc in remote_accounts: acc_id = c.extract_id(acc) if acc_id: remote_account_ids.add(str(acc_id)) except Exception as e: logger.warning("failed to fetch remote accounts for website %s at organize: %s", wid, e) # 构建 remote_account_map 用于快速查询账号状态 remote_account_map = {} if remote_accounts is not None: for acc in remote_accounts: acc_id = c.extract_id(acc) if acc_id: remote_account_map[str(acc_id)] = acc # 对每个绑定关系遍历其对应的上游 Key 进行整理 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) sources = binding_sources(b) for src in sources: uid = src.get("upstream_id") gid = src.get("group_id") if not uid or not gid: continue _u = db.query(Upstream).filter(Upstream.id == uid).first() upstream_name = _u.name if _u else f"#{uid}" # 来源名称优先取 src.group_name，再 fallback 快照，再 fallback ID source_group_name = src.get("group_name") or "" if not source_group_name: snapshot_groups = {} try: snapshot_groups = latest_rate_map(db, uid) except Exception: pass if gid in snapshot_groups: g_data = snapshot_groups[gid] if isinstance(g_data, dict): source_group_name = g_data.get("name") or g_data.get("group_name") or "" if not source_group_name: source_group_name = gid # 查询该 (upstream_id, group_id) 的非 orphaned Key keys = ( db.query(UpstreamGeneratedKey) .filter( UpstreamGeneratedKey.upstream_id == uid, UpstreamGeneratedKey.group_id == gid, UpstreamGeneratedKey.status != "orphaned", ) .all() ) if not keys: items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name="", account_id=None, account_name=None, status="missing_key", message="请先生成上游 Key", ) ) continue # 获取来源上游的 Base URL 用于账号创建 upstream_base_url = _u.base_url if _u else "" for row in keys: # 跳过状态为 failed 的 Key if row.status == "failed": items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name=row.key_name, account_id=None, account_name=None, status="failed", message="上游 Key 生成状态为失败", ) ) continue # 检测平台类型 platform = _detect_platform( f"{row.group_name} {row.group_id} {row.key_name}", "openai", ) # 1. 幂等校验：已导入过则检查远端账号是否仍存在 if row.imported_website_id == wid and row.imported_account_id: old_account_id = row.imported_account_id if remote_account_ids is None: # 账号列表获取失败，无法校验状态，保守跳过 items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name=row.key_name, account_id=old_account_id, status="failed", message="无法校验目标账号状态，已保守跳过", ) ) continue remote_acc = remote_account_map.get(str(old_account_id)) if remote_acc is not None: # 目标账号存在：检测并对齐目标分组，移除其余已启用的 SmartUp 托管分组，保留非 SmartUp 托管分组 try: current_group_ids = remote_acc.get("group_ids") or [] if not isinstance(current_group_ids, list): current_group_ids = [current_group_ids] 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} if old_set != new_set: c.update_account(old_account_id, {"group_ids": new_group_ids}) # 顺便更新下 remote_account_map 中的数据，防止同个账号后续在其他绑定中重复处理 remote_acc["group_ids"] = new_group_ids msg = "已存在，已迁移到目标分组" 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() items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name=row.key_name, account_id=old_account_id, account_name=str(remote_acc.get("name") or ""), status="exists", message=msg, ) ) continue except Exception as exc: items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name=row.key_name, account_id=old_account_id, status="failed", message=f"更新绑定失败: {exc}", ) ) continue else: # 远端已删除，清理标记后进行重建 row.imported_website_id = None row.imported_account_id = None row.imported_at = None row.status = "created" db.commit() is_recreated = True else: is_recreated = False # 2. 检查是否有明文 Key if not row.key_value: items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name=row.key_name, status="failed", message="该 Key 无明文值，无法导入", ) ) continue # 3. 创建账号并绑定当前目标分组 group_ids = [] numeric_target = _numeric_group_id(target_group_id) if numeric_target is not None: group_ids.append(numeric_target) else: group_ids.append(target_group_id) account_name = build_imported_account_name( upstream_name=upstream_name, group_name=row.group_name, group_id=row.group_id, key_id=row.id, ) priority_val = rate_priority_map.get((str(target_group_id), row.upstream_id, row.group_id), 1) account_body = { "name": account_name, "platform": platform, "type": "apikey", "credentials": { "api_key": row.key_value, "base_url": upstream_base_url, }, "group_ids": group_ids, "rate_multiplier": 1, "concurrency": 100, "priority": priority_val, "notes": f"Imported by SmartUp from upstream key #{row.id}", } 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" row.error = None db.commit() # 将新创建的账号加入 remote_account_map 缓存，以防同账号的其它绑定关系后续做幂等校验 if account_id: remote_account_map[str(account_id)] = created items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name=row.key_name, account_id=account_id or None, account_name=str(created.get("name") or account_name), status="recreated" if is_recreated else "created", message="清理后重建账号" if is_recreated else "已创建账号", ) ) except Exception as exc: logger.exception("organize create account failed website=%s key=%s", wid, row.id) row.status = "import_failed" row.error = str(exc) db.commit() items.append( OrganizeGroupsItem( target_group_id=str(target_group_id), target_group_name=target_group_name, upstream_name=upstream_name, source_group_id=str(gid), source_group_name=str(source_group_name), key_name=row.key_name, status="failed", message=str(exc), ) ) created_count = sum(1 for item in items if item.status == "created") recreated_count = sum(1 for item in items if item.status == "recreated") aligned_count = sum(1 for item in items if item.status == "exists" and "已对齐" in item.message) migrated_count = sum(1 for item in items if item.status == "exists" and "已迁移" in item.message) missing_key_count = sum(1 for item in items if item.status == "missing_key") failed_count = sum(1 for item in items if item.status == "failed") parts = [] if created_count: parts.append(f"已创建 {created_count}") if recreated_count: parts.append(f"已重建 {recreated_count}") if aligned_count: parts.append(f"已存在且已对齐 {aligned_count}") if migrated_count: parts.append(f"已存在已迁移 {migrated_count}") if missing_key_count: parts.append(f"缺少 Key {missing_key_count}") if failed_count: parts.append(f"失败 {failed_count}") # 整理完成后，调用一次网站级 priority 重排以修正历史和当前账号优先级 try: sync_account_priorities_for_website(db, wid) except Exception as exc: logger.warning("failed to sync priorities after organize for website %s: %s", wid, exc) message = "整理完成：" + " / ".join(parts) if parts else "整理完成：无任何绑定或数据" return OrganizeGroupsResponse( success=failed_count == 0, message=message, items=items, ) @router.get("/api/group-bindings", response_model=List[BindingResponse]) def list_bindings(db: Session = Depends(get_db), _=Depends(get_current_user)): rows = db.query(WebsiteGroupBinding).order_by(WebsiteGroupBinding.id.desc()).all() return [_binding_response(db, row) for row in rows] @router.post("/api/websites/{wid}/group-bindings/sync-now", response_model=WebsiteBatchSyncResponse) def sync_website_group_bindings( wid: int, db: Session = Depends(get_db), _=Depends(get_current_user), ): """一键同步网站下所有分组绑定。""" website = db.query(Website).filter(Website.id == wid).first() if not website: raise HTTPException(404, "website not found") bindings = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.website_id == wid).order_by(WebsiteGroupBinding.id.asc()).all() if not bindings: return WebsiteBatchSyncResponse( total=0, success=0, failed=0, skipped=0, message="暂无绑定可同步", logs=[], ) results: list[WebsiteSyncLog] = [] for binding in bindings: try: log = sync_binding(db, binding, write=True) results.append(log) except Exception as exc: logger.exception("batch sync failed for binding %s", binding.id) # Create and persist a synthetic failed log if sync_binding crashed before creating one synthetic_log = WebsiteSyncLog( website_id=wid, binding_id=binding.id, target_group_id=binding.target_group_id, target_group_name=binding.target_group_name, algorithm=binding.algorithm, percent=binding.percent, source_rates_json="[]", status="failed", message=str(exc), created_at=datetime.now(timezone.utc), ) db.add(synthetic_log) db.commit() db.refresh(synthetic_log) results.append(synthetic_log) success_count = sum(1 for r in results if r.status == "success") failed_count = sum(1 for r in results if r.status == "failed") skipped_count = sum(1 for r in results if r.status == "skipped") msg_parts = [] if success_count: msg_parts.append(f"成功 {success_count}") if failed_count: msg_parts.append(f"失败 {failed_count}") if skipped_count: msg_parts.append(f"跳过 {skipped_count}") return WebsiteBatchSyncResponse( total=len(bindings), success=success_count, failed=failed_count, skipped=skipped_count, message=f"同步完成：{'、'.join(msg_parts)}" if msg_parts else "同步完成", logs=[_log_response(r) for r in results], ) @router.post("/api/group-bindings", response_model=BindingResponse, status_code=201) def create_binding(body: BindingCreate, db: Session = Depends(get_db), _=Depends(get_current_user)): website = db.query(Website).filter(Website.id == body.website_id).first() if not website: raise HTTPException(404, "website not found") if body.algorithm not in ALGORITHMS: raise HTTPException(400, "不支持的算法") _ensure_unique_target(db, body.website_id, 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, 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, ) db.add(row) db.commit() db.refresh(row) try: sync_binding(db, row, write=True) except Exception as exc: logger.exception("initial website sync failed for binding %s: %s", row.id, exc) return _binding_response(db, row) @router.put("/api/group-bindings/{bid}", response_model=BindingResponse) def update_binding(bid: int, body: BindingUpdate, db: Session = Depends(get_db), _=Depends(get_current_user)): row = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.id == bid).first() if not row: raise HTTPException(404, "binding not found") data = body.model_dump(exclude_none=True) if "website_id" in data and not db.query(Website).filter(Website.id == data["website_id"]).first(): raise HTTPException(404, "website not found") if "algorithm" in data and data["algorithm"] not in ALGORITHMS: raise HTTPException(400, "不支持的算法") 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) 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.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(row) try: sync_binding(db, row, write=True) except Exception as exc: logger.exception("sync failed after updating binding %s: %s", row.id, exc) return _binding_response(db, row) @router.delete("/api/group-bindings/{bid}", status_code=204) def delete_binding(bid: int, db: Session = Depends(get_db), _=Depends(get_current_user)): row = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.id == bid).first() if not row: raise HTTPException(404, "binding not found") db.delete(row) db.commit() @router.post("/api/group-bindings/{bid}/sync-now", response_model=WebsiteSyncLogResponse) def sync_now(bid: int, db: Session = Depends(get_db), _=Depends(get_current_user)): row = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.id == bid).first() if not row: raise HTTPException(404, "binding not found") return _log_response(sync_binding(db, row, write=True)) @router.get("/api/website-sync-logs", response_model=List[WebsiteSyncLogResponse]) def list_sync_logs( website_id: int | None = Query(None), binding_id: int | None = Query(None), limit: int = Query(50, le=200), offset: int = Query(0), db: Session = Depends(get_db), _=Depends(get_current_user), ): q = db.query(WebsiteSyncLog) if website_id: q = q.filter(WebsiteSyncLog.website_id == website_id) if binding_id: q = q.filter(WebsiteSyncLog.binding_id == binding_id) rows = q.order_by(WebsiteSyncLog.created_at.desc()).offset(offset).limit(limit).all() return [_log_response(row) for row in rows]