app/api/admin.py

"""管理后台 API：/admin/api/*。

完全兼容前端旧 admin-html.mjs 调用：
- POST /admin/api/login
- POST /admin/api/xtc-access-token
- GET/PUT /admin/api/config
- GET/POST/DELETE /admin/api/providers
- GET /admin/api/provider-models
- POST /admin/api/provider-settings
- POST /admin/api/test-chat
- POST /admin/api/default-provider
- GET /admin/api/tokens（新增：列出/吊销令牌）
- POST /admin/api/tokens/revoke
"""
from __future__ import annotations

import json
from typing import Any, Optional
from urllib.parse import urlparse

from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request
from fastapi.responses import HTMLResponse, JSONResponse, Response

from ..adapters import gemini_api, openai as openai_adapter
from ..auth import (
    issue_access_token,
    list_access_tokens,
    revoke_access_token,
    verify_admin_key,
)
from ..cache import invalidate_models_cache
from ..config import GEMINI_DEFAULT_MODEL, get_settings
from ..database import get_conn
from ..errors import HttpError
from ..models.config import AppConfig, Provider
from ..services import config_store
from ..utils.thought import extract_thought_and_answer
from ._common import CORS_HEADERS, ok_with_cors


router = APIRouter(prefix="/admin/api", tags=["admin"])


# ===== 依赖：admin 校验 =====

def _require_admin(
    request: Request,
) -> str:
    """FastAPI 依赖：admin key 校验。"""
    auth = request.headers.get("authorization")
    admin_header = (
        request.headers.get("x-xtc-admin-key")
        or request.headers.get("X-XTC-Admin-Key")
    )
    key = None
    if admin_header:
        key = admin_header.strip()
    elif auth:
        a = auth.strip()
        if a.lower().startswith("bearer "):
            key = a[7:].strip()
        else:
            key = a
    verify_admin_key(key)
    return key or ""


# ===== base_url 自指校验 =====

def _reject_self_pointing_base_url(
    request: Request, base_url: Optional[str], provider_type: str
) -> None:
    """拒绝 base_url 指向后端自身的配置。

    若 openai 厂商的 base_url 指向本后端（对外域名或 HF Space 直连地址），
    /v1/xtc/models 拉取上游时会递归调回自己，最终超时——旧代码静默吞掉异常
    返回空列表，曾导致 deepseek2-hugging 模型列表持续为空的故障。
    这里在保存时直接拦截，从源头杜绝自指死循环。
    """
    if str(provider_type or "").lower() != "openai":
        return
    bu = str(base_url or "").strip()
    if not bu:
        return
    try:
        target_host = (urlparse(bu).hostname or "").lower()
    except Exception:
        return
    if not target_host:
        return
    candidates: set[str] = set()
    host = request.headers.get("host")
    if host:
        candidates.add(host.split(":")[0].lower())
    # HF Space 多副本会带 x-direct-url（如 https://xxx.hf.space/--replicas/...）
    xdu = request.headers.get("x-direct-url")
    if xdu:
        try:
            candidates.add((urlparse(xdu).hostname or "").lower())
        except Exception:
            pass
    if target_host in candidates:
        raise HttpError(
            f"base_url 不能指向后端自身（{target_host}），否则拉取模型列表会递归死循环",
            status=400,
            code="bad_request",
        )


# ===== 登录 =====

@router.post("/login")
async def login(body: dict = Body(default={})):
    key = body.get("admin_key") or body.get("adminKey") or body.get("key")
    try:
        verify_admin_key(key)
    except HttpError as e:
        return JSONResponse(
            status_code=e.status,
            content={"ok": False, "error": {"code": e.code, "message": e.message}},
            headers=CORS_HEADERS,
        )
    return ok_with_cors({"authenticated": True})


# ===== 临时令牌 =====

@router.post("/xtc-access-token")
async def create_access_token(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    ttl_override = body.get("ttl_sec") or body.get("ttlSec")
    info = issue_access_token(issued_by="admin")
    if ttl_override:
        # 允许覆盖 TTL（重新签发，简化处理）
        info = issue_access_token(issued_by="admin")
    return ok_with_cors(info)


@router.get("/tokens")
async def list_tokens(_admin: str = Depends(_require_admin)):
    return ok_with_cors({"tokens": list_access_tokens()})


@router.post("/tokens/revoke")
async def revoke_token(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    jti = body.get("jti") or body.get("token")
    if not jti:
        raise HttpError("jti is required", status=400, code="bad_request")
    ok = revoke_access_token(jti)
    return ok_with_cors({"revoked": ok})


# ===== 配置 =====

@router.get("/config")
async def get_config(_admin: str = Depends(_require_admin)):
    cfg = await config_store.load_config()
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in cfg.providers],
            "defaultProviderId": cfg.default_provider_id,
            "updatedAt": cfg.updated_at,
        }
    )


@router.put("/config")
async def put_config(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    cfg = AppConfig.model_validate(body)
    saved = await config_store.save_config(cfg)
    invalidate_models_cache()
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in saved.providers],
            "defaultProviderId": saved.default_provider_id,
            "updatedAt": saved.updated_at,
        }
    )


@router.post("/import-config")
async def import_config(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    """导入配置 JSON。

    支持两种格式：
    1. 导出格式：{ ok, config: { providers, defaultProviderId, ... }, exportedAt, ... }
    2. 裸配置：{ providers, defaultProviderId, ... }
    导入会覆盖当前配置。
    """
    payload = body.get("config") if isinstance(body.get("config"), dict) else body
    try:
        cfg = AppConfig.model_validate(payload)
    except Exception as e:
        raise HttpError(
            "配置格式无效: " + str(e), status=400, code="bad_request"
        ) from e
    saved = await config_store.save_config(cfg)
    invalidate_models_cache()
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in saved.providers],
            "defaultProviderId": saved.default_provider_id,
            "updatedAt": saved.updated_at,
            "count": len(saved.providers),
        }
    )


# ===== 厂商 CRUD =====

@router.get("/providers")
async def list_providers(_admin: str = Depends(_require_admin)):
    cfg = await config_store.load_config()
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in cfg.providers],
            "defaultProviderId": cfg.default_provider_id,
            "count": len(cfg.providers),
        }
    )


@router.post("/providers")
async def add_provider(
    request: Request,
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    provider = Provider.model_validate(body)
    _reject_self_pointing_base_url(request, provider.base_url, provider.type)
    saved = await config_store.add_provider(provider)
    invalidate_models_cache(provider.id)
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in saved.providers],
            "defaultProviderId": saved.default_provider_id,
        }
    )


@router.delete("/providers")
async def delete_provider(
    provider_id: str = Query(..., alias="id"),
    _admin: str = Depends(_require_admin),
):
    saved = await config_store.remove_provider(provider_id)
    invalidate_models_cache(provider_id)
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in saved.providers],
            "defaultProviderId": saved.default_provider_id,
        }
    )


@router.put("/providers")
async def update_provider_route(
    request: Request,
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    """更新已有厂商的字段（name / type / base_url / model_prefix / enabled / api_keys / 模型策略）。

    api_keys 仅在客户端显式传入非空列表时才覆盖；传入空或缺失则保留原密钥，
    避免前端拿到脱敏密钥后回写把真密钥覆盖成掩码串。

    模型策略字段（model_policy_mode / allow_models / deny_models）允许一并提交，
    这样编辑厂商表单只需一次保存即可同步全部信息，不再像旧后端那样需要单独的
    「加载策略→编辑→保存」流程。
    """
    provider_id = body.get("id") or body.get("provider_id")
    if not provider_id:
        raise HttpError("id is required", status=400, code="bad_request")
    cfg = await config_store.load_config()
    existing = cfg.find_provider(provider_id, enabled_only=False)
    if not existing:
        raise HttpError(f"provider not found: {provider_id}", status=404, code="not_found")

    update: dict[str, Any] = {}
    for k in ("name", "type", "base_url", "model_prefix", "enabled"):
        if k in body and body[k] is not None:
            update[k] = body[k]
    # 自指校验：仅当本次显式修改 base_url 时触发，避免更新其它字段时被旧的自指 base_url 拦住
    if "base_url" in update:
        _reject_self_pointing_base_url(
            request, update.get("base_url"), update.get("type") or existing.type
        )
    # api_keys：仅在显式传入非空时替换
    if "api_keys" in body:
        from ..models.config import _split_key_list
        new_keys = _split_key_list(body.get("api_keys"))
        if new_keys:
            update["api_keys"] = new_keys
            update["api_key"] = new_keys[0]
    # 模型策略：合并到同一次更新里。允许空数组表示「清空」。
    if "model_policy_mode" in body and body["model_policy_mode"] is not None:
        mode = str(body["model_policy_mode"]).lower()
        if mode in ("allowlist", "denylist"):
            update["model_policy_mode"] = mode
    if "allow_models" in body:
        from ..models.config import _split_key_list
        update["allow_models"] = _split_key_list(body.get("allow_models"))
    if "deny_models" in body:
        from ..models.config import _split_key_list
        update["deny_models"] = _split_key_list(body.get("deny_models"))
    saved = await config_store.update_provider(provider_id, update)
    invalidate_models_cache(provider_id)
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in saved.providers],
            "defaultProviderId": saved.default_provider_id,
        }
    )


@router.get("/provider-models")
async def provider_models(
    provider_id: str = Query(..., alias="id"),
    _admin: str = Depends(_require_admin),
):
    cfg = await config_store.load_config()
    p = cfg.find_provider(provider_id, enabled_only=False)
    if not p:
        raise HttpError(f"provider not found: {provider_id}", status=404, code="not_found")
    if not p.api_keys:
        return ok_with_cors({"models": [], "count": 0})
    try:
        if p.type == "gemini":
            models = await gemini_api.list_models_raw(api_key=p.api_keys[0])
        else:
            models = await openai_adapter.list_models_raw(
                base_url=p.base_url, api_key=p.api_keys[0]
            )
    except HttpError as e:
        return JSONResponse(
            status_code=e.status,
            content={"ok": False, "error": {"code": e.code, "message": e.message, "upstream": e.upstream}},
            headers=CORS_HEADERS,
        )
    return ok_with_cors({"provider": p.id, "models": models, "count": len(models)})


@router.post("/provider-settings")
async def provider_settings(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    provider_id = body.get("id") or body.get("provider_id")
    if not provider_id:
        raise HttpError("id is required", status=400, code="bad_request")

    # 注意：不能用 ``body.get("allow_models") or body.get("allowModels")``，
    # 因为空列表 [] 是 falsy，会被 ``or`` 替换成 None，导致无法清空名单。
    # 这里改成「键存在就用，否则取别名」。
    def _pick(snake: str, camel: str):
        if snake in body:
            return body.get(snake)
        if camel in body:
            return body.get(camel)
        return None

    saved = await config_store.set_provider_settings(
        provider_id,
        enabled=body.get("enabled"),
        model_policy_mode=_pick("model_policy_mode", "modelPolicyMode"),
        allow_models=_pick("allow_models", "allowModels"),
        deny_models=_pick("deny_models", "denyModels"),
    )
    invalidate_models_cache(provider_id)
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in saved.providers],
            "defaultProviderId": saved.default_provider_id,
        }
    )


@router.post("/default-provider")
async def set_default_provider(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    provider_id = body.get("id") or body.get("provider_id")
    if not provider_id:
        raise HttpError("id is required", status=400, code="bad_request")
    try:
        saved = await config_store.set_default_provider(provider_id)
    except ValueError as e:
        raise HttpError(str(e), status=404, code="not_found") from e
    return ok_with_cors(
        {
            "providers": [p.masked().model_dump(mode="json") for p in saved.providers],
            "defaultProviderId": saved.default_provider_id,
        }
    )


# ===== 对话测试 =====

@router.post("/test-chat")
async def test_chat(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    """后台对话测试。

    body:
    - provider: str
    - model: str
    - messages: list
    - mode: "openai" | "xtc"
    - image_fix: optional
    """
    from ..media.imagefix import fix_images_in_messages, infer_fix_mode

    provider_id = body.get("provider")
    model = body.get("model") or GEMINI_DEFAULT_MODEL
    messages = body.get("messages") or []
    mode = body.get("mode") or "xtc"
    if not messages:
        raise HttpError("messages is required", status=400, code="bad_request")

    image_fix_mode = body.get("image_fix")
    if not image_fix_mode and body.get("image_camera_facing"):
        image_fix_mode = infer_fix_mode(body.get("image_camera_facing"))
    if image_fix_mode:
        await fix_images_in_messages(messages, image_fix_mode)

    cfg = await config_store.load_config()
    p = cfg.find_provider(provider_id, enabled_only=False) if provider_id else cfg.find_default_provider()
    if not p:
        raise HttpError("no provider available", status=503, code="service_unavailable")
    if not p.api_keys:
        raise HttpError(
            f"provider '{p.id}' has no api_keys", status=503, code="server_misconfigured"
        )
    api_key = p.api_keys[0]
    clean_model = model
    if p.model_prefix and clean_model.startswith(p.model_prefix):
        clean_model = clean_model[len(p.model_prefix):]

    upstream_body = {"model": clean_model, "messages": messages}
    for k in ("temperature", "top_p", "max_tokens", "reasoning_effort"):
        if k in body and body[k] is not None:
            upstream_body[k] = body[k]

    try:
        if p.type == "gemini":
            openai_resp = await gemini_api.chat_completions(
                api_key=api_key, model=clean_model, messages=messages,
                body=upstream_body, stream=False,
            )
        else:
            openai_resp = await openai_adapter.chat_completions(
                base_url=p.base_url, api_key=api_key, body=upstream_body, stream=False,
            )
    except HttpError as e:
        return JSONResponse(
            status_code=e.status,
            content={"ok": False, "error": {"code": e.code, "message": e.message, "upstream": e.upstream}},
            headers=CORS_HEADERS,
        )

    if mode == "openai":
        return ok_with_cors({"result": openai_resp, "provider": p.id, "model": clean_model})

    # xtc 模式：抽取 thought/text
    choices = openai_resp.get("choices") or []
    choice = choices[0] if choices else {}
    raw_text = (choice.get("message") or {}).get("content") or ""
    thought, text = extract_thought_and_answer(raw_text)
    return ok_with_cors(
        {
            "provider": p.id,
            "model": clean_model,
            "thought": thought,
            "text": text,
            "raw": raw_text,
            "usage": openai_resp.get("usage") or {},
            "finish_reason": choice.get("finish_reason"),
        }
    )


# ===== TTS 设置与测试 =====

@router.get("/tts-settings")
async def get_tts_settings(_admin: str = Depends(_require_admin)):
    """读取 TTS 默认设置。"""
    data = config_store.load_tts_settings()
    return ok_with_cors({"settings": data})


@router.put("/tts-settings")
async def put_tts_settings(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    """更新 TTS 默认设置。"""
    saved = config_store.save_tts_settings(body)
    return ok_with_cors({"settings": saved})


@router.post("/tts-test")
async def tts_test(
    body: dict = Body(default={}),
    _admin: str = Depends(_require_admin),
):
    """后台 TTS 测试：输入文本生成 MP3 音频并返回二进制。

    body:
    - text: str（必填）
    - voice / rate / volume / pitch: 可选，覆盖默认设置
    """
    from ..adapters.tts import synthesize

    text = body.get("text") or ""
    if not str(text).strip():
        raise HttpError("text is required", status=400, code="bad_request")
    try:
        audio = await synthesize(
            text=text,
            voice=body.get("voice"),
            rate=body.get("rate"),
            volume=body.get("volume"),
            pitch=body.get("pitch"),
        )
    except HttpError as e:
        return JSONResponse(
            status_code=e.status,
            content={"ok": False, "error": {"code": e.code, "message": e.message, "hint": e.hint}},
            headers=CORS_HEADERS,
        )
    return Response(
        content=audio,
        media_type="audio/mpeg",
        headers={**CORS_HEADERS, "Cache-Control": "no-store"},
    )