app/admin.py

"""管理后台端点（/admin/*）：账号池查看 / 上传 / 删除 / 重载，独立 ``[admin].auth_key`` 鉴权。

鉴权（二选一）：Header ``Authorization: Bearer <key>`` 或 Query ``?auth_key=<key>``。
``admin_auth_key`` 留空时所有 /admin/* 返回 404（关闭状态，隐藏端点存在）。
"""
from __future__ import annotations

import asyncio
import json
import re
import time
import zipfile
from io import BytesIO
from pathlib import Path
from typing import Any

import httpx
from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
from fastapi.responses import HTMLResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel

# 账号文件名只允许 [A-Za-z0-9._-]，防路径穿越（name 来自上传文件名 / JSON 内 name 字段）
_SAFE_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$")

from app.account import Account, AccountPool
from app.api_keys import add_key, delete_key, load_keys, save_keys
from app.config import Settings
from app.upstream import get_provider
from app.upstream.auth import DefaultAuthProvider

_bearer = HTTPBearer(auto_error=False)
router = APIRouter(prefix="/admin")

# 管理面板（单文件 HTML+JS，无构建/无依赖）
_PANEL_PATH = Path(__file__).resolve().parent / "static" / "admin.html"


def _get_admin_key(request: Request) -> str:
    return getattr(request.app.state.settings, "admin_auth_key", "")


def verify_admin_key(
    request: Request,
    auth_key: str | None = Query(None, description="管理后台鉴权 key（query 方式）"),
    cred: HTTPAuthorizationCredentials | None = Depends(_bearer),
) -> None:
    """校验 admin auth key；未配置时隐藏端点（404），错误时 401。"""
    key = _get_admin_key(request)
    if not key:
        raise HTTPException(status_code=404, detail="admin endpoints disabled")
    provided = ""
    if cred is not None and cred.scheme.lower() == "bearer":
        provided = cred.credentials
    if not provided and auth_key:
        provided = auth_key
    if provided != key:
        raise HTTPException(status_code=401, detail="invalid admin auth key")


def _pool(request: Request) -> AccountPool:
    return request.app.state.pool  # type: ignore[no-any-return]


def _settings(request: Request) -> Settings:
    return request.app.state.settings  # type: ignore[no-any-return]


def _http_client(request: Request) -> httpx.AsyncClient:
    return request.app.state.http_client  # type: ignore[no-any-return]


def _providers(request: Request) -> dict[str, Any]:
    return request.app.state.providers  # type: ignore[no-any-return]


def _sync_provider(providers: dict[str, Any], acc: Account, settings: Settings,
                   http_client: httpx.AsyncClient) -> None:
    """为新增/更新账号构造 UpstreamProvider 并注入 providers 字典。"""
    providers[acc.name] = get_provider(acc, settings, http_client)


def _rebuild_providers(pool: AccountPool, settings: Settings,
                       http_client: httpx.AsyncClient) -> dict[str, Any]:
    providers: dict[str, Any] = {}
    for acc in pool.all():
        _sync_provider(providers, acc, settings, http_client)
    return providers


@router.get("", response_class=HTMLResponse, include_in_schema=False)
async def admin_panel(request: Request) -> HTMLResponse:
    """管理面板页面（纯静态壳，无敏感数据；浏览器导航无法带头，故页面本身不鉴权）。

    数据接口（/admin/accounts 等）仍走 ``verify_admin_key`` 鉴权——浏览器打开面板
    页面看到登录表单，输入 key 后 JS 以 ``Authorization: Bearer <key>`` 调用 API。
    auth_key 未配置时与其它 /admin/* 一致返回 404（整站 admin 关闭隐藏）。
    """
    if not _get_admin_key(request):
        raise HTTPException(status_code=404, detail="admin endpoints disabled")
    if not _PANEL_PATH.exists():
        raise HTTPException(status_code=500, detail="admin.html missing")
    return HTMLResponse(_PANEL_PATH.read_text(encoding="utf-8"))


def _account_summary(a: Account, now: float) -> dict[str, Any]:
    """账号摘要（不含凭据）：状态派生 disabled / cooldown / available。"""
    cooling = bool(a.cooldown_until and a.cooldown_until > now)
    available = (not a.disabled) and (not cooling) and a.fail_reason is None
    return {
        "name": a.name,
        "source_email": a.source_email,
        "created_at": a.created_at,
        "disabled": a.disabled,
        "fail_reason": a.fail_reason.value if a.fail_reason else None,
        "cooldown_until": a.cooldown_until or 0.0,
        "cooldown_seconds": max(0, int(a.cooldown_until - now)) if cooling else 0,
        "conversation_id": getattr(a, "conversation_id", "") or "",
        "available": available,
    }


def _find(pool: AccountPool, name: str) -> Account:
    for acc in pool.all():
        if acc.name == name:
            return acc
    raise HTTPException(status_code=404, detail=f"account not found: {name}")


@router.get("/accounts")
async def list_accounts(request: Request, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """列出账号摘要（含派生状态），不暴露凭据等敏感字段。"""
    pool = _pool(request)
    now = time.time()
    data = [_account_summary(a, now) for a in pool.all()]
    return {"object": "list", "data": data}


@router.get("/stats")
async def stats(request: Request, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """账号池汇总统计（面板卡片用）。"""
    pool = _pool(request)
    now = time.time()
    total = active = cooldown = disabled = quota = 0
    for a in pool.all():
        if a.fail_reason and a.fail_reason.value == "quota_exhausted":
            quota += 1
        if a.disabled:
            disabled += 1
        elif a.fail_reason is not None:
            disabled += 1
        elif a.cooldown_until and a.cooldown_until > now:
            cooldown += 1
        else:
            active += 1
        total += 1
    return {
        "total": total, "active": active, "cooldown": cooldown,
        "disabled": disabled, "quota_exhausted": quota,
    }


@router.get("/usage")
async def usage(
    request: Request,
    window: str = Query("24h", description="时间窗：24h / 1d / 3d / 7d / 30d"),
    _: None = Depends(verify_admin_key),
) -> dict[str, Any]:
    """token 用量聚合：分桶时间序列 + 分模型汇总 + 总计（面板图表用）。"""
    from app import usage_store
    return usage_store.aggregate(window)


@router.get("/logs")
async def logs(
    request: Request,
    lines: int = Query(300, ge=1, le=2000, description="返回日志尾部行数"),
    _: None = Depends(verify_admin_key),
) -> dict[str, Any]:
    """返回网关日志文件尾部（面板「日志」页实时查看；含生图耗时埋点）。"""
    settings = _settings(request)
    log_path = Path(settings.log_dir) / settings.log_filename
    if not log_path.is_file():
        return {"file": str(log_path), "lines": [], "exists": False}
    try:
        with log_path.open("r", encoding="utf-8", errors="replace") as f:
            # 只读尾部 N 行（文件可能很大）
            tail = f.readlines()[-lines:]
    except OSError:
        return {"file": str(log_path), "lines": [], "exists": True, "error": "read failed"}
    return {"file": str(log_path), "lines": tail, "exists": True, "count": len(tail)}


@router.get("/accounts/{name}")
async def get_account(request: Request, name: str, _: None = Depends(verify_admin_key)) -> Account:
    """获取单个账号完整信息（含凭据，需 admin 鉴权）。"""
    pool = _pool(request)
    for acc in pool.all():
        if acc.name == name:
            return acc
    raise HTTPException(status_code=404, detail=f"account not found: {name}")


@router.post("/accounts")
async def create_or_update_account(
    request: Request, account: Account, _: None = Depends(verify_admin_key),
) -> Account:
    """上传/新增账号，持久化到 account/<name>.json 并同步内存账号池与 provider 缓存。"""
    pool = _pool(request)
    settings = _settings(request)
    http_client = _http_client(request)
    providers = _providers(request)

    pool.add_or_update(account)
    _sync_provider(providers, account, settings, http_client)
    return account


@router.post("/accounts/upload")
async def upload_accounts(
    request: Request,
    _: None = Depends(verify_admin_key),
) -> dict[str, Any]:
    """上传账号文件（.json 单文件，或 .zip 批量），写入 account_dir 并热加载进池。

    - json：文件内须含 ``name`` 字段，文件名须以 .json 结尾（防止误传）。
    - zip：仅提取 ``*.zip``（含 json），可带子目录。
    - 已存在同名账号 → 覆盖写（整体覆盖，与 /admin/accounts POST 行为一致）。
    - 成功后触发热加载：新号进池 + provider 注入，无需重启或手动 /admin/reload。
    - 凭据只写 account_dir，绝不进日志（http_log 已对上传 body 脱敏）。
    - 多文件：用 ``request.form()`` 读全部同名 ``file`` 字段（fastapi 0.141 的
      ``list[UploadFile]`` 依赖对同名多文件有 bug 返回 422，故绕开）。
    """
    pool = _pool(request)
    settings = _settings(request)
    http_client = _http_client(request)
    providers = _providers(request)

    form = await request.form()
    # 注意：form.getlist 返回的是 starlette.datastructures.UploadFile，与
    # fastapi.UploadFile 在 0.141 里 isinstance 不匹配（同名不同类），故不做类型断言。
    uploaded = [v for v in form.getlist("file") if hasattr(v, "read") and hasattr(v, "filename")]
    if not uploaded:
        raise HTTPException(status_code=400, detail="没有收到文件（form 字段名须为 file）")

    collected: list[tuple[str, bytes]] = []
    for file in uploaded:
        filename = (file.filename or "").split("/")[-1].split("\\")[-1]  # 取 basename
        raw = await file.read()
        if not raw:
            continue
        if filename.endswith(".zip"):
            try:
                zf = zipfile.ZipFile(BytesIO(raw))
            except zipfile.BadZipFile:
                raise HTTPException(status_code=400, detail=f"invalid zip file: {filename}") from None
            for info in zf.infolist():
                if info.is_dir():
                    continue
                if info.filename.endswith(".json"):
                    collected.append((Path(info.filename).name, zf.read(info.filename)))
        elif filename.endswith(".json"):
            collected.append((filename, raw))

    files_list: list[tuple[str, bytes]] = collected
    if not files_list:
        raise HTTPException(status_code=400, detail="仅支持 .json 或 .zip 文件")

    # 写盘目录 = 池的目录（与 pool.reload() 扫描一致；settings.account_dir 在
    # 构造池后才改会不一致，测试/运维均以池目录为准）。
    account_dir = pool.account_dir
    account_dir.mkdir(parents=True, exist_ok=True)
    saved: list[str] = []
    errors: list[dict[str, str]] = []
    for name, content in files_list:
        if not _SAFE_NAME_RE.match(name):
            errors.append({"file": name, "error": "文件名非法（仅允许字母数字 . _ -）"})
            continue
        try:
            data = json.loads(content.decode("utf-8"))
        except (ValueError, UnicodeDecodeError):
            errors.append({"file": name, "error": "不是合法的 JSON"})
            continue
        acc_name = str(data.get("name") or "")
        if not acc_name or not _SAFE_NAME_RE.match(acc_name):
            errors.append({"file": name, "error": "JSON 缺少合法的 name 字段（仅允许字母数字 . _ -）"})
            continue
        target = account_dir / f"{acc_name}.json"
        if not _SAFE_NAME_RE.match(acc_name) or target.resolve().parent != account_dir.resolve():
            errors.append({"file": name, "error": "非法账号名（路径穿越拦截）"})
            continue
        target.write_bytes(content)
        saved.append(acc_name)

    if saved:
        try:
            pool.reload()
            request.app.state.providers = _rebuild_providers(pool, settings, http_client)
        except Exception as e:  # noqa: BLE001
            raise HTTPException(status_code=500, detail=f"账号写入成功但热加载失败: {e}") from e
    return {
        "uploaded": saved,
        "skipped": [e["file"] for e in errors],
        "errors": errors,
        "count": len(saved),
    }


def _api_key_dir(request: Request) -> Path:
    """API key 库文件目录 = log_dir（HF 持久化 /data/logs）。"""
    return Path(_settings(request).log_dir)


def _sync_api_keys(request: Request) -> None:
    """把磁盘 key 库同步进 app.state（verify_api_key 查询用）。"""
    request.app.state.api_keys = load_keys(_api_key_dir(request))


@router.get("/apikeys")
async def list_api_keys(request: Request, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """列出全部 API key（含 config 兼容 key 的说明）。"""
    return {
        "object": "list",
        "data": load_keys(_api_key_dir(request)),
        "legacy_key": bool(_settings(request).gateway_api_key),
    }


@router.post("/apikeys")
async def create_api_key(request: Request, name: str = Query("", description="key 名称（可选）"),
                         _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """生成新 API key（写入库文件，同步内存，立即可用）。"""
    entry = add_key(_api_key_dir(request), name)
    _sync_api_keys(request)
    return entry


@router.post("/apikeys/delete")
async def delete_api_key_bulk(request: Request, body: NamesBody,
                              _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """按 key 值删除 API key（一批）。"""
    deleted, not_found = [], []
    for k in body.names:
        if delete_key(_api_key_dir(request), k):
            deleted.append(k)
        else:
            not_found.append(k)
    _sync_api_keys(request)
    return {"deleted": deleted, "not_found": not_found, "count": len(deleted)}


@router.delete("/accounts/{name}")
async def delete_account(request: Request, name: str, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """删除指定账号。"""
    pool = _pool(request)
    providers = _providers(request)
    if not pool.remove(name):
        raise HTTPException(status_code=404, detail=f"account not found: {name}")
    providers.pop(name, None)
    return {"deleted": True, "name": name}


@router.post("/reload")
async def reload_accounts(request: Request, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """重新从磁盘加载账号池，并重建 provider 缓存。"""
    pool = _pool(request)
    settings = _settings(request)
    http_client = _http_client(request)
    pool.reload()
    request.app.state.providers = _rebuild_providers(pool, settings, http_client)
    return {"reloaded": True, "count": len(pool.all())}


class NamesBody(BaseModel):
    """批量操作：账号名列表。"""
    names: list[str] = []


@router.post("/accounts/{name}/enable")
async def enable_account(request: Request, name: str, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """手动恢复单个账号（清 disabled / fail_reason / cooldown，重新纳入轮换）。"""
    pool = _pool(request)
    acc = _find(pool, name)
    pool.enable(acc)
    return {"ok": True, "name": name, "account": _account_summary(acc, time.time())}


@router.post("/accounts/{name}/disable")
async def disable_account(request: Request, name: str, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """手动禁用单个账号（标记 disabled，剔除轮换直到手动 enable）。"""
    pool = _pool(request)
    acc = _find(pool, name)
    pool.disable(acc)
    return {"ok": True, "name": name, "account": _account_summary(acc, time.time())}


@router.post("/accounts/enable")
async def enable_accounts(request: Request, body: NamesBody, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """批量恢复账号。返回成功与未找到的名单。"""
    pool = _pool(request)
    by_name = {a.name: a for a in pool.all()}
    ok, missing = [], []
    for n in body.names:
        acc = by_name.get(n)
        if acc is None:
            missing.append(n)
            continue
        pool.enable(acc)
        ok.append(n)
    return {"enabled": ok, "not_found": missing, "count": len(ok)}


@router.post("/accounts/disable")
async def disable_accounts(request: Request, body: NamesBody, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """批量禁用账号。返回成功与未找到的名单。"""
    pool = _pool(request)
    by_name = {a.name: a for a in pool.all()}
    ok, missing = [], []
    for n in body.names:
        acc = by_name.get(n)
        if acc is None:
            missing.append(n)
            continue
        pool.disable(acc)
        ok.append(n)
    return {"disabled": ok, "not_found": missing, "count": len(ok)}


@router.post("/accounts/delete")
async def delete_accounts(request: Request, body: NamesBody, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """批量删除账号（磁盘 + 内存池 + provider 缓存）。"""
    pool = _pool(request)
    providers = _providers(request)
    deleted, missing = [], []
    for n in body.names:
        if pool.remove(n):
            providers.pop(n, None)
            deleted.append(n)
        else:
            missing.append(n)
    return {"deleted": deleted, "not_found": missing, "count": len(deleted)}


async def _fetch_balance(acc: Account, settings: Settings,
                         http_client: httpx.AsyncClient) -> dict[str, Any]:
    """查单账号余额；返回 {name, balance, error}。balance=None 表示查不到。"""
    try:
        auth = DefaultAuthProvider(acc, settings, http_client)
        bal = await auth.check_balance()
        # check_balance: None=查不到(401/网络/非200), -1=未配置http/无token(不可判定), >=0=可用额度
        if bal is None:
            return {"name": acc.name, "balance": None, "error": "查询失败(token 失效或网络错误)"}
        if bal < 0:
            return {"name": acc.name, "balance": None, "error": "无 token / 未配置"}
        return {"name": acc.name, "balance": bal, "error": None}
    except Exception as e:  # noqa: BLE001
        return {"name": acc.name, "balance": None, "error": str(e)}


@router.get("/accounts/{name}/balance")
async def account_balance(request: Request, name: str, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """查询单个账号上游可用额度（available_credits）。"""
    pool = _pool(request)
    acc = _find(pool, name)
    return await _fetch_balance(acc, _settings(request), _http_client(request))


@router.post("/accounts/balance")
async def batch_balance(request: Request, body: NamesBody, _: None = Depends(verify_admin_key)) -> dict[str, Any]:
    """批量并发查询余额。names 为空 → 查全部账号。并发上限 8，避免打爆上游。"""
    pool = _pool(request)
    settings = _settings(request)
    http_client = _http_client(request)
    if body.names:
        wanted = set(body.names)
        accounts = [a for a in pool.all() if a.name in wanted]
    else:
        accounts = pool.all()

    sem = asyncio.Semaphore(8)

    async def _one(acc: Account) -> dict[str, Any]:
        async with sem:
            return await _fetch_balance(acc, settings, http_client)

    results = await asyncio.gather(*[_one(a) for a in accounts])
    total = sum(r["balance"] for r in results if isinstance(r["balance"], int))
    return {"object": "list", "data": results, "total_balance": total, "count": len(results)}