study

Paused

study / api /accounts.py
xiaolongmr
区分「上游 CDN 拦截」与「token 失效」，修复导入账号报 403
107e8c6 15 days ago
44.6 kB
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