from __future__ import annotations import asyncio import io import json import re import uuid import zipfile from datetime import datetime from threading import Lock from typing import Any, Literal from fastapi import APIRouter, Header, HTTPException, Query from fastapi.concurrency import run_in_threadpool from fastapi.responses import Response from pydantic import AliasChoices, BaseModel, ConfigDict, Field from services.auth_service import auth_service from services.account_email_login_service import account_email_login_import_service from api.support import ( require_admin, require_identity, sanitize_cpa_pool, sanitize_cpa_pools, sanitize_sub2api_server, sanitize_sub2api_servers, ) from services.account_service import account_service from services.chatgpt_session_service import ChatGPTSessionStoreError, chatgpt_session_service from services.oauth_login_service import OAuthLoginError, oauth_login_service from services.cpa_service import cpa_config, cpa_import_service, list_remote_files from services.upstream_account_maintenance_service import upstream_account_maintenance_service from services.sub2api_service import ( list_remote_accounts as sub2api_list_remote_accounts, list_remote_groups as sub2api_list_remote_groups, sub2api_config, sub2api_import_service, ) _refresh_jobs: dict[str, dict[str, Any]] = {} _refresh_jobs_lock = Lock() def _set_refresh_job(progress_id: str, **updates: Any) -> None: with _refresh_jobs_lock: job = _refresh_jobs.get(progress_id) if job is not None: job.update(updates) def _get_refresh_job(progress_id: str) -> dict[str, Any] | None: with _refresh_jobs_lock: job = _refresh_jobs.get(progress_id) return dict(job) if job else None class UserKeyCreateRequest(BaseModel): name: str = "" class UserKeyUpdateRequest(BaseModel): model_config = ConfigDict(populate_by_name=True) name: str | None = None enabled: bool | None = None account_pool_enabled: bool | None = Field( default=None, validation_alias=AliasChoices("account_pool_enabled", "accountPoolEnabled"), ) image_memory_button_enabled: bool | None = Field( default=None, validation_alias=AliasChoices("image_memory_button_enabled", "imageMemoryButtonEnabled"), ) third_party_image_enabled: bool | None = Field( default=None, validation_alias=AliasChoices("third_party_image_enabled", "thirdPartyImageEnabled"), ) third_party_image_group_ids: list[str] | None = Field( default=None, validation_alias=AliasChoices("third_party_image_group_ids", "thirdPartyImageGroupIds"), ) third_party_image_model_ids: list[str] | None = Field( default=None, validation_alias=AliasChoices("third_party_image_model_ids", "thirdPartyImageModelIds"), ) image_model_daily_limits: list[dict[str, Any]] | None = Field( default=None, validation_alias=AliasChoices("image_model_daily_limits", "imageModelDailyLimits"), ) key: str | None = None class UserKeyBatchRequest(BaseModel): ids: list[str] = Field(default_factory=list) action: Literal["enable", "disable", "delete"] class UserImagePermissionBatchItem(BaseModel): id: str third_party_image_enabled: bool | None = None third_party_image_group_ids: list[str] | None = None third_party_image_model_ids: list[str] | None = None class UserImagePermissionBatchRequest(BaseModel): items: list[UserImagePermissionBatchItem] = Field(default_factory=list) class AccountCreateRequest(BaseModel): tokens: list[str] = Field(default_factory=list) accounts: list[dict[str, Any]] = Field(default_factory=list) class AccountDeleteRequest(BaseModel): tokens: list[str] = Field(default_factory=list) class AccountRefreshRequest(BaseModel): access_tokens: list[str] = Field(default_factory=list) class AccountReloginRequest(BaseModel): access_token: str = "" class AccountReloginVerifyRequest(BaseModel): relogin_id: str = "" code: str = "" class AccountUpstreamActionsRequest(BaseModel): access_tokens: list[str] = Field(default_factory=list) disable_memory: bool = False hide_conversations: bool = False delete_files: bool = False file_page_limit: int = Field(default=100, ge=1, le=100) class AccountExportRequest(BaseModel): access_tokens: list[str] = Field(default_factory=list) format: Literal["json", "zip"] = "json" class AccountUpdateRequest(BaseModel): access_token: str = "" type: str | None = None status: str | None = None quota: int | None = None shared: bool | None = None class AccountEmailLoginStartRequest(BaseModel): email: str = "" class AccountEmailLoginVerifyRequest(BaseModel): session_id: str = "" code: str = "" class OAuthLoginStartRequest(BaseModel): email_hint: str = "" class OAuthLoginFinishRequest(BaseModel): session_id: str = "" callback: str = "" class CPAPoolCreateRequest(BaseModel): name: str = "" base_url: str = "" secret_key: str = "" class CPAPoolUpdateRequest(BaseModel): name: str | None = None base_url: str | None = None secret_key: str | None = None class CPAImportRequest(BaseModel): names: list[str] = Field(default_factory=list) class Sub2APIServerCreateRequest(BaseModel): name: str = "" base_url: str = "" email: str = "" password: str = "" api_key: str = "" group_id: str = "" class Sub2APIServerUpdateRequest(BaseModel): name: str | None = None base_url: str | None = None email: str | None = None password: str | None = None api_key: str | None = None group_id: str | None = None class Sub2APIImportRequest(BaseModel): account_ids: list[str] = Field(default_factory=list) def _account_payload_token(item: dict[str, Any]) -> str: return str(item.get("access_token") or item.get("accessToken") or "").strip() def _public_account(item: dict[str, Any]) -> dict[str, Any]: result = dict(item) result["has_password"] = bool(str(result.pop("password", "") or "").strip()) return result def _unique_tokens(tokens: list[str]) -> list[str]: return list(dict.fromkeys(str(token or "").strip() for token in tokens if str(token or "").strip())) def _download_timestamp() -> str: return datetime.now().strftime("%Y%m%d-%H%M%S") def _safe_export_name(value: str, fallback: str) -> str: clean = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip()).strip("-._") return (clean or fallback)[:80] def _account_zip_bytes(items: list[dict[str, str]]) -> bytes: buf = io.BytesIO() used_names: set[str] = set() with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as archive: for index, item in enumerate(items, start=1): raw_name = item.get("email") or item.get("account_id") or f"account-{index:03d}" base_name = _safe_export_name(raw_name, f"account-{index:03d}") name = base_name suffix = 2 while name in used_names: name = f"{base_name}-{suffix}" suffix += 1 used_names.add(name) archive.writestr( f"{name}.json", json.dumps(item, ensure_ascii=False, indent=2) + "\n", ) return buf.getvalue() def _has_personal_account_pool(identity: dict[str, object]) -> bool: return identity.get("role") == "admin" or bool(identity.get("account_pool_enabled")) def _require_personal_account_pool(identity: dict[str, object]) -> None: if _has_personal_account_pool(identity): return raise HTTPException(status_code=403, detail={"error": "管理员尚未为你开启个人号池"}) def _quota_identity(identity: dict[str, object]) -> dict[str, object]: return identity if _has_personal_account_pool(identity) else {"id": "", "role": "user", "name": identity.get("name")} def create_router() -> APIRouter: router = APIRouter() @router.get("/api/auth/users") async def list_user_keys(authorization: str | None = Header(default=None)): require_admin(authorization) return {"items": auth_service.list_keys(role="user")} @router.get("/api/auth/normal-users") async def list_normal_user_keys(authorization: str | None = Header(default=None)): require_admin(authorization) return {"items": auth_service.list_keys(role="normal")} @router.post("/api/auth/users") async def create_user_key(body: UserKeyCreateRequest, authorization: str | None = Header(default=None)): require_admin(authorization) try: item, raw_key = auth_service.create_key(role="user", name=body.name) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc return {"item": item, "key": raw_key, "items": auth_service.list_keys(role="user")} @router.post("/api/auth/normal-users") async def create_normal_user_key(body: UserKeyCreateRequest, authorization: str | None = Header(default=None)): require_admin(authorization) try: item, raw_key = auth_service.create_key(role="normal", name=body.name) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc return {"item": item, "key": raw_key, "items": auth_service.list_keys(role="normal")} @router.post("/api/auth/users/batch") async def batch_user_keys(body: UserKeyBatchRequest, authorization: str | None = Header(default=None)): require_admin(authorization) try: result = await run_in_threadpool( auth_service.batch_mutate_keys, body.ids, action=body.action, role="user", ) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc except Exception as exc: print(f"[auth-api] batch internal users failed: {exc}") raise HTTPException(status_code=503, detail={"error": "批量操作内部用户失败，请检查存储后端日志"}) from exc return {"result": result, "items": auth_service.list_keys(role="user")} @router.post("/api/auth/normal-users/batch") async def batch_normal_user_keys(body: UserKeyBatchRequest, authorization: str | None = Header(default=None)): require_admin(authorization) try: result = await run_in_threadpool( auth_service.batch_mutate_keys, body.ids, action=body.action, role="normal", ) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc except Exception as exc: print(f"[auth-api] batch normal users failed: {exc}") raise HTTPException(status_code=503, detail={"error": "批量操作普通用户失败，请检查存储后端日志"}) from exc return {"result": result, "items": auth_service.list_keys(role="normal")} @router.post("/api/auth/{user_kind}/batch-image-permissions") async def batch_user_image_permissions( user_kind: Literal["users", "normal-users"], body: UserImagePermissionBatchRequest, authorization: str | None = Header(default=None), ): require_admin(authorization) role: Literal["user", "normal"] = "normal" if user_kind == "normal-users" else "user" try: result = await run_in_threadpool( auth_service.batch_update_image_permissions, [item.model_dump(exclude_none=True) for item in body.items], role=role, ) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc except Exception as exc: print(f"[auth-api] batch image permissions failed: {exc}") raise HTTPException(status_code=503, detail={"error": "批量更新图片权限失败，请检查存储后端日志"}) from exc return {"result": result, "items": auth_service.list_keys(role=role)} @router.post("/api/auth/users/{key_id}") async def update_user_key( key_id: str, body: UserKeyUpdateRequest, authorization: str | None = Header(default=None), ): require_admin(authorization) updates = { key: value for key, value in { "name": body.name, "enabled": body.enabled, "account_pool_enabled": body.account_pool_enabled, "image_memory_button_enabled": body.image_memory_button_enabled, "third_party_image_enabled": body.third_party_image_enabled, "third_party_image_group_ids": body.third_party_image_group_ids, "third_party_image_model_ids": body.third_party_image_model_ids, "image_model_daily_limits": body.image_model_daily_limits, "key": body.key, }.items() if value is not None } if not updates: raise HTTPException(status_code=400, detail={"error": "还没有检测到改动，请修改后再保存"}) try: item = auth_service.update_key(key_id, updates, role="user") except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc if item is None: raise HTTPException(status_code=404, detail={"error": "这条用户密钥不存在，可能已经被删除"}) return {"item": item, "items": auth_service.list_keys(role="user")} @router.delete("/api/auth/users/{key_id}") async def delete_user_key(key_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) try: deleted = auth_service.delete_key(key_id, role="user") except Exception as exc: print(f"[auth-api] delete internal user {key_id} failed: {exc}") raise HTTPException( status_code=503, detail={"error": "删除内部用户失败，请检查存储后端日志"}, ) from exc if not deleted: raise HTTPException(status_code=404, detail={"error": "这条用户密钥不存在，可能已经被删除"}) return {"items": auth_service.list_keys(role="user")} @router.post("/api/auth/normal-users/{key_id}") async def update_normal_user_key( key_id: str, body: UserKeyUpdateRequest, authorization: str | None = Header(default=None), ): require_admin(authorization) updates = { key: value for key, value in { "name": body.name, "enabled": body.enabled, "account_pool_enabled": body.account_pool_enabled, "image_memory_button_enabled": body.image_memory_button_enabled, "third_party_image_enabled": body.third_party_image_enabled, "third_party_image_group_ids": body.third_party_image_group_ids, "third_party_image_model_ids": body.third_party_image_model_ids, "image_model_daily_limits": body.image_model_daily_limits, "key": body.key, }.items() if value is not None } if not updates: raise HTTPException(status_code=400, detail={"error": "还没有检测到改动，请修改后再保存"}) try: item = auth_service.update_key(key_id, updates, role="normal") except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc if item is None: raise HTTPException(status_code=404, detail={"error": "这条普通用户密钥不存在，可能已经被删除"}) return {"item": item, "items": auth_service.list_keys(role="normal")} @router.delete("/api/auth/normal-users/{key_id}") async def delete_normal_user_key(key_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) try: deleted = auth_service.delete_key(key_id, role="normal") except Exception as exc: print(f"[auth-api] delete normal user {key_id} failed: {exc}") raise HTTPException( status_code=503, detail={"error": "删除普通用户失败，请检查存储后端日志"}, ) from exc if not deleted: raise HTTPException(status_code=404, detail={"error": "这条普通用户密钥不存在，可能已经被删除"}) return {"items": auth_service.list_keys(role="normal")} @router.get("/api/accounts") async def get_accounts(authorization: str | None = Header(default=None)): identity = require_identity(authorization) _require_personal_account_pool(identity) return {"items": account_service.list_accounts_for_identity(identity)} @router.get("/api/accounts/quota") async def get_quota_summary( strategy: str = Query(default="own_first"), allow_shared: bool = Query(default=True), authorization: str | None = Header(default=None), ): identity = require_identity(authorization) effective_identity = _quota_identity(identity) effective_strategy = strategy if _has_personal_account_pool(identity) else "admin_only" return account_service.image_quota_summary_for_identity( effective_identity, strategy=effective_strategy, allow_shared=allow_shared and _has_personal_account_pool(identity), ) @router.post("/api/accounts") async def create_accounts(body: AccountCreateRequest, authorization: str | None = Header(default=None)): identity = require_identity(authorization) _require_personal_account_pool(identity) existing_account_keys = { _account_payload_token(account): str(account.get("account_key") or "").strip() for account in account_service.list_accounts_for_identity(identity) if _account_payload_token(account) } account_payloads: list[dict[str, Any]] = [] for source in body.accounts: if not isinstance(source, dict): continue item = dict(source) snake_session_token = item.pop("session_token", "") camel_session_token = item.pop("sessionToken", "") session_token = str(snake_session_token or camel_session_token or "").strip() if session_token: access_token = _account_payload_token(item) account_key = str( item.get("account_key") or existing_account_keys.get(access_token) or uuid.uuid4() ).strip() session_expires_at = str( item.get("session_expires_at") or item.get("expires") or "" ).strip() try: chatgpt_session_service.capture_token( account_key, session_token, { "session_expires_at": session_expires_at, "fp": item.get("fp"), "registered_proxy": item.get("registered_proxy"), }, ) except (ValueError, ChatGPTSessionStoreError) as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc item.update( { "account_key": account_key, "source_type": "chatgpt_session", "has_renewable_session": True, "session_expires_at": session_expires_at or None, } ) account_payloads.append(item) payload_tokens = [_account_payload_token(item) for item in account_payloads] tokens = _unique_tokens([*body.tokens, *payload_tokens]) if not tokens: raise HTTPException(status_code=400, detail={"error": "tokens is required"}) if account_payloads: result = account_service.add_account_items(account_payloads, owner=identity) payload_token_set = set(_unique_tokens(payload_tokens)) extra_tokens = [token for token in tokens if token not in payload_token_set] if extra_tokens: extra_result = account_service.add_accounts(extra_tokens, owner=identity) result["added"] = int(result.get("added") or 0) + int(extra_result.get("added") or 0) result["skipped"] = int(result.get("skipped") or 0) + int(extra_result.get("skipped") or 0) else: result = account_service.add_accounts(tokens, owner=identity) allowed_tokens = { str(item.get("access_token") or "") for item in account_service.list_accounts_for_identity(identity) } refresh_targets = [token for token in tokens if token in allowed_tokens] refresh_result = account_service.refresh_accounts(refresh_targets) if refresh_targets else {"refreshed": 0, "errors": []} return { **result, "refreshed": refresh_result.get("refreshed", 0), "errors": refresh_result.get("errors", []), "blocked": refresh_result.get("blocked", 0), "items": account_service.list_accounts_for_identity(identity), } @router.delete("/api/accounts") async def delete_accounts(body: AccountDeleteRequest, authorization: str | None = Header(default=None)): identity = require_identity(authorization) _require_personal_account_pool(identity) tokens = [str(token or "").strip() for token in body.tokens if str(token or "").strip()] if not tokens: raise HTTPException(status_code=400, detail={"error": "tokens is required"}) result = account_service.delete_accounts( tokens, owner_id=str(identity.get("id") or ""), admin=identity.get("role") == "admin", ) return {**result, "items": account_service.list_accounts_for_identity(identity)} @router.post("/api/accounts/refresh") async def refresh_accounts(body: AccountRefreshRequest, authorization: str | None = Header(default=None)): identity = require_identity(authorization) _require_personal_account_pool(identity) access_tokens = [str(token or "").strip() for token in body.access_tokens if str(token or "").strip()] if not access_tokens: access_tokens = [ str(item.get("access_token") or "") for item in account_service.list_accounts_for_identity(identity) ] if not access_tokens: raise HTTPException(status_code=400, detail={"error": "access_tokens is required"}) allowed_tokens = { str(item.get("access_token") or "") for item in account_service.list_accounts_for_identity(identity) } access_tokens = [token for token in access_tokens if token in allowed_tokens] if not access_tokens: raise HTTPException(status_code=403, detail={"error": "no permitted accounts selected"}) progress_id = uuid.uuid4().hex owner_id = str(identity.get("id") or "") with _refresh_jobs_lock: if len(_refresh_jobs) > 200: completed = [key for key, value in _refresh_jobs.items() if value.get("done")] for key in completed[:100]: _refresh_jobs.pop(key, None) _refresh_jobs[progress_id] = { "progress_id": progress_id, "owner_id": owner_id, "total": len(access_tokens), "processed": 0, "refreshed": 0, "failed": 0, "done": False, "error": None, } def update_progress(_token: str, success: bool, _error: str | None) -> None: with _refresh_jobs_lock: job = _refresh_jobs.get(progress_id) if job is None: return job["processed"] = int(job.get("processed") or 0) + 1 key = "refreshed" if success else "failed" job[key] = int(job.get(key) or 0) + 1 async def run_refresh() -> None: try: result = await run_in_threadpool( account_service.refresh_accounts, access_tokens, # A definitive upstream 401 means the saved credential can no # longer authenticate. Surface that state to the operator on # a manual refresh, but never delete an account from this UI # action (transient 403/429/network failures are not treated as # invalid by AccountService). defer_invalid_removal=False, cleanup_action="mark_abnormal", invalid_failure_threshold=1, progress_callback=update_progress, ) _set_refresh_job( progress_id, done=True, processed=len(access_tokens), refreshed=int(result.get("refreshed") or 0), failed=len(result.get("errors") or []), result={**result, "items": account_service.list_accounts_for_identity(identity)}, ) except Exception as exc: _set_refresh_job(progress_id, done=True, error=str(exc)) asyncio.create_task(run_refresh()) return {"progress_id": progress_id, "total": len(access_tokens)} @router.post("/api/accounts/re-login") async def re_login_account( body: AccountReloginRequest, authorization: str | None = Header(default=None), ): identity = require_identity(authorization) _require_personal_account_pool(identity) token = str(body.access_token or "").strip() if not token: raise HTTPException(status_code=400, detail={"error": "access_token is required"}) try: return await run_in_threadpool( account_service.re_login_with_password, token, owner_id=str(identity.get("id") or ""), admin=str(identity.get("role") or "") == "admin", ) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc except RuntimeError as exc: raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc @router.post("/api/accounts/re-login/verify") async def verify_re_login_account( body: AccountReloginVerifyRequest, authorization: str | None = Header(default=None), ): identity = require_identity(authorization) _require_personal_account_pool(identity) try: return await run_in_threadpool( account_service.verify_re_login_with_password, str(body.relogin_id or "").strip(), str(body.code or "").strip(), owner_id=str(identity.get("id") or ""), admin=str(identity.get("role") or "") == "admin", ) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc except RuntimeError as exc: raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc @router.get("/api/accounts/refresh/progress/{progress_id}") async def get_refresh_progress(progress_id: str, authorization: str | None = Header(default=None)): identity = require_identity(authorization) _require_personal_account_pool(identity) progress = _get_refresh_job(progress_id) if progress is None: raise HTTPException(status_code=404, detail={"error": "progress not found"}) if str(progress.get("owner_id") or "") != str(identity.get("id") or "") and identity.get("role") != "admin": raise HTTPException(status_code=403, detail={"error": "no permission for this progress"}) progress.pop("owner_id", None) return progress @router.post("/api/accounts/upstream-actions") async def run_upstream_account_actions( body: AccountUpstreamActionsRequest, authorization: str | None = Header(default=None), ): identity = require_identity(authorization) _require_personal_account_pool(identity) if not any((body.disable_memory, body.hide_conversations, body.delete_files)): raise HTTPException(status_code=400, detail={"error": "at least one upstream action is required"}) visible_accounts = account_service.list_accounts_for_identity(identity) allowed_by_token = { str(item.get("access_token") or "").strip(): item for item in visible_accounts if str(item.get("access_token") or "").strip() } requested_tokens = _unique_tokens(body.access_tokens) if not requested_tokens: raise HTTPException(status_code=400, detail={"error": "access_tokens is required"}) if any(token not in allowed_by_token for token in requested_tokens): raise HTTPException(status_code=403, detail={"error": "no permission for one or more selected accounts"}) if len(requested_tokens) > 200: raise HTTPException(status_code=400, detail={"error": "up to 200 accounts can be maintained at once"}) if body.delete_files and len(requested_tokens) > 20: raise HTTPException(status_code=400, detail={"error": "attachment deletion supports up to 20 accounts at once"}) return await run_in_threadpool( upstream_account_maintenance_service.run, requested_tokens, disable_memory=body.disable_memory, hide_conversations=body.hide_conversations, delete_files=body.delete_files, file_page_limit=body.file_page_limit, ) @router.post("/api/accounts/export") async def export_accounts(body: AccountExportRequest, authorization: str | None = Header(default=None)): identity = require_identity(authorization) _require_personal_account_pool(identity) access_tokens = _unique_tokens(body.access_tokens) allowed_tokens = { str(item.get("access_token") or "") for item in account_service.list_accounts_for_identity(identity) } access_tokens = [token for token in access_tokens if token in allowed_tokens] if access_tokens else list(allowed_tokens) items = account_service.build_export_items(access_tokens) if not items: raise HTTPException( status_code=400, detail={"error": "没有可导出的账号，账号至少需要包含 access_token"}, ) timestamp = _download_timestamp() if body.format == "zip": content = _account_zip_bytes(items) return Response( content, media_type="application/zip", headers={"Content-Disposition": f'attachment; filename="codex-accounts-{timestamp}.zip"'}, ) payload: dict[str, str] | list[dict[str, str]] = items[0] if len(items) == 1 else items return Response( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", media_type="application/json", headers={"Content-Disposition": f'attachment; filename="codex-accounts-{timestamp}.json"'}, ) @router.post("/api/accounts/update") async def update_account(body: AccountUpdateRequest, authorization: str | None = Header(default=None)): identity = require_identity(authorization) _require_personal_account_pool(identity) access_token = str(body.access_token or "").strip() if not access_token: raise HTTPException(status_code=400, detail={"error": "access_token is required"}) role = str(identity.get("role") or "").strip().lower() update_source = {"status": body.status, "shared": body.shared} if role == "admin": update_source.update({"type": body.type, "quota": body.quota}) updates = {key: value for key, value in update_source.items() if value is not None} if not updates: raise HTTPException(status_code=400, detail={"error": "还没有检测到改动，请修改后再保存"}) account = account_service.update_account( access_token, updates, owner_id=str(identity.get("id") or ""), admin=role == "admin", ) if account is None: raise HTTPException(status_code=404, detail={"error": "account not found"}) return {"item": _public_account(account), "items": account_service.list_accounts_for_identity(identity)} @router.post("/api/accounts/email-login/start") async def start_account_email_login_import( body: AccountEmailLoginStartRequest, authorization: str | None = Header(default=None), ): identity = require_identity(authorization) _require_personal_account_pool(identity) try: return await run_in_threadpool( account_email_login_import_service.start, body.email, identity, ) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc except RuntimeError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc @router.post("/api/accounts/oauth/start") async def start_oauth_login( body: OAuthLoginStartRequest, authorization: str | None = Header(default=None), ): identity = require_identity(authorization) _require_personal_account_pool(identity) try: return await run_in_threadpool(oauth_login_service.start, body.email_hint) except OAuthLoginError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc @router.post("/api/accounts/oauth/finish") async def finish_oauth_login( body: OAuthLoginFinishRequest, authorization: str | None = Header(default=None), ): identity = require_identity(authorization) _require_personal_account_pool(identity) try: tokens = await run_in_threadpool( oauth_login_service.finish, body.session_id, body.callback, ) except OAuthLoginError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc payload = { "access_token": tokens["access_token"], "refresh_token": tokens["refresh_token"], "id_token": tokens["id_token"], "source_type": "oauth_login", } add_result = await run_in_threadpool( account_service.add_account_items, [payload], owner=identity, ) refresh_result = await run_in_threadpool( account_service.refresh_accounts, [tokens["access_token"]], ) return { **add_result, "refreshed": refresh_result.get("refreshed", 0), "errors": refresh_result.get("errors", []), "items": account_service.list_accounts_for_identity(identity), } @router.post("/api/accounts/email-login/verify") async def verify_account_email_login_import( body: AccountEmailLoginVerifyRequest, authorization: str | None = Header(default=None), ): identity = require_identity(authorization) _require_personal_account_pool(identity) try: return await run_in_threadpool( account_email_login_import_service.verify, body.session_id, body.code, identity, ) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc except RuntimeError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc @router.get("/api/cpa/pools") async def list_cpa_pools(authorization: str | None = Header(default=None)): require_admin(authorization) return {"pools": sanitize_cpa_pools(cpa_config.list_pools())} @router.post("/api/cpa/pools") async def create_cpa_pool(body: CPAPoolCreateRequest, authorization: str | None = Header(default=None)): require_admin(authorization) if not body.base_url.strip(): raise HTTPException(status_code=400, detail={"error": "base_url is required"}) if not body.secret_key.strip(): raise HTTPException(status_code=400, detail={"error": "secret_key is required"}) pool = cpa_config.add_pool(name=body.name, base_url=body.base_url, secret_key=body.secret_key) return {"pool": sanitize_cpa_pool(pool), "pools": sanitize_cpa_pools(cpa_config.list_pools())} @router.post("/api/cpa/pools/{pool_id}") async def update_cpa_pool(pool_id: str, body: CPAPoolUpdateRequest, authorization: str | None = Header(default=None)): require_admin(authorization) pool = cpa_config.update_pool(pool_id, body.model_dump(exclude_none=True)) if pool is None: raise HTTPException(status_code=404, detail={"error": "pool not found"}) return {"pool": sanitize_cpa_pool(pool), "pools": sanitize_cpa_pools(cpa_config.list_pools())} @router.delete("/api/cpa/pools/{pool_id}") async def delete_cpa_pool(pool_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) if not cpa_config.delete_pool(pool_id): raise HTTPException(status_code=404, detail={"error": "pool not found"}) return {"pools": sanitize_cpa_pools(cpa_config.list_pools())} @router.get("/api/cpa/pools/{pool_id}/files") async def cpa_pool_files(pool_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) pool = cpa_config.get_pool(pool_id) if pool is None: raise HTTPException(status_code=404, detail={"error": "pool not found"}) return {"pool_id": pool_id, "files": await run_in_threadpool(list_remote_files, pool)} @router.post("/api/cpa/pools/{pool_id}/import") async def cpa_pool_import(pool_id: str, body: CPAImportRequest, authorization: str | None = Header(default=None)): require_admin(authorization) pool = cpa_config.get_pool(pool_id) if pool is None: raise HTTPException(status_code=404, detail={"error": "pool not found"}) try: job = cpa_import_service.start_import(pool, body.names) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc return {"import_job": job} @router.get("/api/cpa/pools/{pool_id}/import") async def cpa_pool_import_progress(pool_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) pool = cpa_config.get_pool(pool_id) if pool is None: raise HTTPException(status_code=404, detail={"error": "pool not found"}) return {"import_job": pool.get("import_job")} @router.get("/api/sub2api/servers") async def list_sub2api_servers(authorization: str | None = Header(default=None)): require_admin(authorization) return {"servers": sanitize_sub2api_servers(sub2api_config.list_servers())} @router.post("/api/sub2api/servers") async def create_sub2api_server(body: Sub2APIServerCreateRequest, authorization: str | None = Header(default=None)): require_admin(authorization) if not body.base_url.strip(): raise HTTPException(status_code=400, detail={"error": "base_url is required"}) has_login = body.email.strip() and body.password.strip() has_api_key = bool(body.api_key.strip()) if not has_login and not has_api_key: raise HTTPException(status_code=400, detail={"error": "email+password or api_key is required"}) server = sub2api_config.add_server( name=body.name, base_url=body.base_url, email=body.email, password=body.password, api_key=body.api_key, group_id=body.group_id, ) return {"server": sanitize_sub2api_server(server), "servers": sanitize_sub2api_servers(sub2api_config.list_servers())} @router.post("/api/sub2api/servers/{server_id}") async def update_sub2api_server(server_id: str, body: Sub2APIServerUpdateRequest, authorization: str | None = Header(default=None)): require_admin(authorization) server = sub2api_config.update_server(server_id, body.model_dump(exclude_none=True)) if server is None: raise HTTPException(status_code=404, detail={"error": "server not found"}) return {"server": sanitize_sub2api_server(server), "servers": sanitize_sub2api_servers(sub2api_config.list_servers())} @router.delete("/api/sub2api/servers/{server_id}") async def delete_sub2api_server(server_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) if not sub2api_config.delete_server(server_id): raise HTTPException(status_code=404, detail={"error": "server not found"}) return {"servers": sanitize_sub2api_servers(sub2api_config.list_servers())} @router.get("/api/sub2api/servers/{server_id}/groups") async def sub2api_server_groups(server_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) server = sub2api_config.get_server(server_id) if server is None: raise HTTPException(status_code=404, detail={"error": "server not found"}) try: groups = await run_in_threadpool(sub2api_list_remote_groups, server) except Exception as exc: raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc return {"server_id": server_id, "groups": groups} @router.get("/api/sub2api/servers/{server_id}/accounts") async def sub2api_server_accounts(server_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) server = sub2api_config.get_server(server_id) if server is None: raise HTTPException(status_code=404, detail={"error": "server not found"}) try: accounts = await run_in_threadpool(sub2api_list_remote_accounts, server) except Exception as exc: raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc return {"server_id": server_id, "accounts": accounts} @router.post("/api/sub2api/servers/{server_id}/import") async def sub2api_server_import(server_id: str, body: Sub2APIImportRequest, authorization: str | None = Header(default=None)): require_admin(authorization) server = sub2api_config.get_server(server_id) if server is None: raise HTTPException(status_code=404, detail={"error": "server not found"}) try: job = sub2api_import_service.start_import(server, body.account_ids) except ValueError as exc: raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc return {"import_job": job} @router.get("/api/sub2api/servers/{server_id}/import") async def sub2api_server_import_progress(server_id: str, authorization: str | None = Header(default=None)): require_admin(authorization) server = sub2api_config.get_server(server_id) if server is None: raise HTTPException(status_code=404, detail={"error": "server not found"}) return {"import_job": server.get("import_job")} return router