app/deps.py

"""共享 FastAPI 依赖：注入 UpstreamProvider + API key 校验 + 错误分类换号。

每次请求 round-robin 取一个账号的 provider，用 :class:`_RetryingClient` 包一层：
按 :class:`~app.account.FailReason` 分类失效 → ``mark_failed`` → 抛 503，下一次请求自动换号。
v1 的 ``gateway_api_key`` 留空则不校验（无认证）；详见 references/auth-and-errors.md。
"""
from __future__ import annotations

import sys
import time
from collections.abc import AsyncIterator
from typing import Any

import httpx
from fastapi import Depends, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from app.account import Account, AccountPool, FailReason
from app.events import IREvent

_bearer = HTTPBearer(auto_error=False)

# 默认的失效 body 关键词（按目标站定制，见 references/auth-and-errors.md）。
_AUTH_HINTS = ("unauthorized", "invalid token", "not authenticated", "login required")
_BAN_HINTS = ("banned", "suspended", "disabled", "forbidden", "封禁", "封号")
_QUOTA_HINTS = ("quota", "limit reached", "insufficient", "credit", "额度", "配额", "余额不足")
_CF_HINTS = ("cloudflare", "captcha", "turnstile", "challenge", "验证码")


def classify_failure(exc: BaseException) -> FailReason | None:
    """把上游异常映射成 FailReason（默认按 HTTP 状态码 + body 关键词；按目标站定制）。

    返回 None 表示非账号级失效（不换号，原样抛出）。
    """
    status: int | None = None
    body = ""
    if isinstance(exc, httpx.HTTPStatusError):
        status = exc.response.status_code
        try:
            body = exc.response.text.lower()
        except Exception:  # noqa: BLE001
            body = ""
    text = f"{body} {str(exc).lower()}"
    # 2026-08-07 实测：model_tier_required（订阅 tier 限制）不是认证失败，
    # 不应罚账号冷却/停用（否则 terra 等付费模型请求会让整池账号短路）。
    if "model_tier_required" in text:
        return None
    if status in (401, 403) or any(h in text for h in _AUTH_HINTS):
        return FailReason.AUTH_FAILED
    # 2026-08-08 实测：402 insufficient_balance_for_model = lifetime 100 分烧完，
    # 永久无余额（不是可恢复的冷却）→ 直接 disabled 踢出池，避免每次生图撞上
    # 白等。fence（200 只回 tool_call 无 media）才是可恢复的账号级冷却（QUOTA_EXHAUSTED）。
    if status == 402 or "insufficient_balance" in text:
        return FailReason.INSUFFICIENT_BALANCE
    if status == 429 or any(h in text for h in _QUOTA_HINTS):
        return FailReason.QUOTA_EXHAUSTED
    if status == 451 or any(h in text for h in _CF_HINTS):
        return FailReason.CF_CHALLENGE
    if any(h in text for h in _BAN_HINTS):
        return FailReason.BANNED
    return None


class _RetryingClient:
    """duck-type UpstreamProvider：包装 stream，失效时自动换下一个账号重试。

    失败分类（AUTH_FAILED / QUOTA_EXHAUSTED / CF_CHALLENGE / BANNED）→ ``mark_failed``
    标记当前账号 → 内部取下一个可用账号继续重试，最多 ``max_switches`` 次。
    **请求内自动换号**：opencode 等客户端收到 503 不重试，单请求内坏号应自动跳过，
    而非把 503 抛给客户端。全部换完仍失败才抛 503。
    流式已 yield 部分内容后重试会重复输出，故仅在**尚未产出任何事件**时换号重试。
    """

    def __init__(self, pool: AccountPool, account: Account, underlying: Any,
                 providers: dict[str, Any] | None = None) -> None:
        self._pool = pool
        self._account = account
        self._underlying = underlying
        # 全部账号的 provider 映射（换号时取下一个账号的 provider）
        self._providers = providers if providers is not None else {}
        # 最多换号次数（首账号 + 后续 N 次）
        self.max_switches = 5

    def _classify(self, exc: BaseException) -> FailReason | None:
        """优先上游 AuthProvider 分类；未分类再回退通用逻辑。"""
        # 2026-08-08：生图 fence（上游 200 只回 tool_call 围栏、无 media）按额度耗尽
        # 换号——fence 是账号级冷却（短暂，恢复后仍可出图），换到好号即可出图。
        # 避免 FenceError 落进通用 classify（不认识 → None → 原样抛 502，白费一个烧号）。
        if type(exc).__name__ == "FenceError":
            return FailReason.QUOTA_EXHAUSTED
        auth = getattr(self._underlying, "_auth", None)
        reason = auth.classify_failure(exc) if auth is not None else None
        if reason is None:
            reason = classify_failure(exc)
        return reason

    def _switch_account(self) -> Any:
        """取下一个可用账号的 provider；内部抛 RuntimeError（转 503）。"""
        acc = self._pool.next()
        provider = self._providers.get(acc.name)
        if provider is None:
            raise RuntimeError(f"account provider not found: {acc.name}")
        self._account = acc
        self._underlying = provider
        return acc

    async def _maybe_delete_dead(self, acc: Account) -> bool:
        """账号失败后查一次额度：<=0 或查不到（token 死）→ 从池删除账号文件。

        用户 2026-08-08 要求：失败号不该留在池里继续被撞。查询走 auth.check_balance，
        只读 GET（1s 内），失败返回 None（token 失效/网络错）→ 同样删除（死号）。
        返回 True=已删除（调用方应跳过 mark_failed，避免 _save 把账号写回）；False=保留。
        """
        try:
            auth = getattr(self._underlying, "_auth", None)
            if auth is None or not hasattr(auth, "check_balance"):
                return False
            balance = await auth.check_balance()
            if balance is None or balance < 0:
                # 查不到（token 死/网络错）或不可判定（-1，测试/无 http）→ 留池走常规 mark_failed
                return False
            if balance > 0:
                return False  # 还有额度，留池（可能只是瞬时错误）
            self._pool.remove(acc.name)
            print(
                f"[deps] account {acc.name} deleted (balance={balance}, "
                f"reason={acc.fail_reason})",
                file=sys.stderr,
            )
            return True
        except Exception:  # noqa: BLE001
            return False

    def switch_account_and_mark_failed(self) -> Any:
        """供 images adapter 在 fence 后手动换号：标记当前账号冷却并换下一个。

        FenceError 在 adapter 层（_collect）抛出，发生在本类 stream() 的 try 块之外，
        stream() 捕获不到。images 路由捕获 FenceError 后调用本方法换号重试。"""
        self._pool.mark_failed(self._account, FailReason.QUOTA_EXHAUSTED)
        print(
            f"[deps] account {self._account.name} fenced (image); "
            f"switching to next account",
            file=sys.stderr,
        )
        return self._switch_account()

    async def stream(self, *args: Any, **kwargs: Any) -> AsyncIterator[IREvent]:
        last_reason: FailReason | None = None
        last_exc: BaseException | None = None
        for _ in range(self.max_switches):
            emitted = False
            try:
                async for ir in self._underlying.stream(*args, **kwargs):
                    emitted = True
                    yield ir
                return  # 正常结束
            except Exception as e:  # noqa: BLE001
                reason = self._classify(e)
                if reason is None:
                    # 非账号级失效：不换号，原样抛
                    raise
                last_reason = reason
                last_exc = e
                # 2026-08-08 用户要求：账号失败时主动查一次额度，<=0（或查不到，token
                # 已死）→ 直接从池里删除，不给 round-robin 再撞。查余额是只读 GET，1s
                # 内完成，仅坏号才查（不拖正常请求）。删除成功后不再 mark_failed（否则
                # _save 会把已删账号重新写回文件）。
                if not await self._maybe_delete_dead(self._account):
                    self._pool.mark_failed(self._account, reason)
                if emitted:
                    # 已产出部分内容，无法安全换号重试（会重复输出）；抛 503 让客户端自行处理
                    raise HTTPException(
                        status_code=503,
                        detail=f"account failed ({reason.value}) mid-stream; retry request to switch account",
                    ) from e
                print(
                    f"[deps] account {self._account.name} failed ({reason.value}); "
                    f"switching to next account (attempt {_ + 1}/{self.max_switches})",
                    file=sys.stderr,
                )
                try:
                    self._switch_account()
                except RuntimeError as switch_exc:
                    raise HTTPException(status_code=503, detail=str(switch_exc)) from switch_exc
        # 全部账号都失败
        raise HTTPException(
            status_code=503,
            detail=f"all accounts failed ({last_reason}); retry request to switch account",
        ) from last_exc


def get_client(request: Request) -> _RetryingClient:
    """round-robin 取一个账号，返回其 _RetryingClient 包装。"""
    st = request.app.state
    pool: AccountPool = st.pool
    providers: dict[str, Any] = st.providers
    try:
        acc = pool.next()
    except RuntimeError as e:
        # 全部账号 disabled/冷却中：给明确 503（含原因）而非裸 500
        raise HTTPException(status_code=503, detail=str(e)) from e
    return _RetryingClient(pool, acc, providers[acc.name], providers=providers)


def get_image_client(request: Request) -> _RetryingClient:
    """生图专用取号：只在"未失败过或冷却已到期"的可用账号里 round-robin。

    2026-08-08 实测：生图撞到 402（0分，INSUFFICIENT_BALANCE 已 disabled 踢出）或
    fence（账号级冷却）的号，要等上游把整轮输出完才返回（~2 分钟），再换号重试，
    一趟白等 6-8 分钟——cherry 里就是"一直加载"。干净号（d7yoa12=100 等）13 秒
    出图。冷却中的 fence 号到期会自动回到候选（QUOTA_EXHAUSTED 可恢复）。

    2026-08-08 改图修复：改图/生图共用本取号逻辑。改图的 input_images 用**全局最近
    生成的签名 URL**（openai_images.py 的 _RECENT_URLS_FILE，跨账号共享——实测 URL
    不绑账号会话，任意有额度账号都能拿它改图），故此处统一 round-robin 即可，
    不需要为改图特判账号。
    """
    st = request.app.state
    pool: AccountPool = st.pool
    providers: dict[str, Any] = st.providers
    candidates = [
        a for a in pool.all()
        if not a.disabled
        and not (a.cooldown_until and a.cooldown_until > time.time())
    ]
    if not candidates:
        try:
            acc = pool.next()
        except RuntimeError as e:
            raise HTTPException(status_code=503, detail=str(e)) from e
        return _RetryingClient(pool, acc, providers[acc.name], providers=providers)
    acc = candidates[getattr(pool, "_idx", 0) % len(candidates)]
    return _RetryingClient(pool, acc, providers[acc.name], providers=providers)


def verify_api_key(
    request: Request,
    cred: HTTPAuthorizationCredentials | None = Depends(_bearer),
) -> None:
    """v1 gateway key 校验；未配置任何 key 则放行（无认证）。每个 /v1 router 都应 Depends。

    校验集合 = 面板生成的多 key 库（app.state.api_keys）+ config 兼容单 key
    （gateway_api_key）。任一匹配即放行 —— 老客户端 key 无需迁移。
    """
    settings = request.app.state.settings
    legacy = settings.gateway_api_key
    keys = getattr(request.app.state, "api_keys", None)
    if not legacy and not keys:
        return  # 未配置任何 key → 无认证放行
    if cred is None or not _is_api_key_valid(cred.credentials, legacy, keys):
        raise HTTPException(status_code=401, detail="invalid api key")


def _is_api_key_valid(provided: str | None, legacy: str, keys: Any) -> bool:
    from app.api_keys import is_valid_key

    key_list = keys if isinstance(keys, list) else []
    return is_valid_key(provided, key_list, legacy=legacy)