app/api/xtc.py

"""XTC 简化接口：/v1/xtc/providers、/v1/xtc/models、/v1/xtc/chat。

完全兼容前端 XTC-AI/src/common/api.js 调用。
"""
from __future__ import annotations

import json
import re
from typing import Any, AsyncIterator, Optional

from fastapi import APIRouter, Depends, Header, Query, Request
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse

from ..adapters import gemini_api, openai as openai_adapter
from ..auth import require_access_key
from ..cache import get_cached_models, set_cached_models
from ..config import GEMINI_DEFAULT_MODEL, get_settings
from ..errors import HttpError
from ..media.imagefix import fix_images_in_messages, infer_fix_mode
from ..providers.policy import is_model_allowed
from ..services.config_store import load_config
from ..utils.sse import sse_data, sse_done
from ..utils.thought import (
    extract_thought_and_answer,
    extract_thought_from_reasoning_content,
)
from ._common import (
    CORS_HEADERS,
    normalize_chat_body,
    ok_with_cors,
    record_usage,
    select_provider_and_key,
)

router = APIRouter(prefix="/v1/xtc", tags=["xtc"])

# simple 流式：thought 标签识别（提到模块级，避免每次调用重新 import + compile）
_THOUGHT_OPEN_RE = re.compile(r"<thought>\s*$", re.IGNORECASE)
_THOUGHT_CLOSE_RE = re.compile(r"^\s*</thought>", re.IGNORECASE)


@router.get("/providers")
async def list_providers(_key: str = Depends(require_access_key)) -> dict:
    config = await load_config()
    items = [p.public() for p in config.providers if p.enabled]
    return ok_with_cors(
        {
            "defaultProviderId": config.default_provider_id,
            "providers": items,
            "count": len(items),
        }
    )


@router.get("/models")
async def list_models(
    _key: str = Depends(require_access_key),
    provider: Optional[str] = Query(default=None, alias="provider"),
    all: int = Query(default=0, alias="all"),
    x_provider: Optional[str] = Header(default=None, alias="x-provider"),
) -> dict:
    # 前端 api.js 的 fetchModels 通过 x-provider 头指定厂商（与 /v1/xtc/chat
    # 一致），此处需同时识别，否则模型选择器永远只展示默认厂商的模型。
    # 优先取 query 参数（更显式），其次取头。
    provider = provider or (x_provider.strip() if x_provider else x_provider)
    config = await load_config()
    enabled = [p for p in config.providers if p.enabled]
    if all and not provider:
        # 合并所有厂商：单个厂商失败不阻断聚合请求（仅跳过并记日志）
        out: list[dict] = []
        for p in enabled:
            try:
                models = await _fetch_provider_models(p)
            except HttpError as e:
                print(f"[xtc/models] provider {p.id} list failed: {e.message}")
                continue
            for m in models:
                mid = m.get("id") or ""
                if not mid or not is_model_allowed(p, mid):
                    continue
                full_id = f"{p.model_prefix}{mid}" if p.model_prefix else mid
                out.append({"id": full_id, "name": m.get("name") or full_id, "provider": p.id})
        return ok_with_cors({"models": out, "count": len(out), "all": True})

    # 单厂商
    target = None
    if provider:
        target = next((p for p in enabled if p.id == provider), None)
    if not target:
        target = config.find_default_provider()
    if not target:
        raise HttpError(
            "No enabled provider. Add one in /admin.",
            status=503,
            code="service_unavailable",
        )

    models = await _fetch_provider_models(target)
    out = []
    for m in models:
        mid = m.get("id") or ""
        if not mid or not is_model_allowed(target, mid):
            continue
        full_id = f"{target.model_prefix}{mid}" if target.model_prefix else mid
        out.append({"id": full_id, "name": m.get("name") or full_id})
    return ok_with_cors(
        {"provider": target.id, "models": out, "count": len(out)}
    )


async def _fetch_provider_models(provider) -> list[dict]:
    """拉取单个厂商的模型列表。

    上游失败时抛 HttpError（由全局异常处理器转为统一错误响应），不再静默吞掉
    异常返回 []——否则前端无法区分"该厂商无模型"与"上游报错"，也会让 503/超时
    等瞬时故障被误判为永久空列表（与旧后端行为一致：旧后端会向客户端回传
    {ok:false,error:{...}}）。
    仅缓存非空结果，避免上游瞬时返回空列表时缓存被"毒化"30s。
    """
    cached = get_cached_models(provider.id)
    if cached is not None:
        return cached
    if not provider.api_keys:
        raise HttpError(
            f"Provider '{provider.id}' has no api_keys configured",
            status=503,
            code="server_misconfigured",
        )
    if provider.type == "gemini":
        data = await gemini_api.list_models_raw(api_key=provider.api_keys[0])
    else:
        data = await openai_adapter.list_models_raw(
            base_url=provider.base_url, api_key=provider.api_keys[0]
        )
    # 只缓存非空结果，避免上游瞬时返回空列表时缓存被"毒化"
    if data:
        set_cached_models(provider.id, data)
    return data


# ===== /v1/xtc/chat =====

@router.post("/chat")
async def xtc_chat(
    request: Request,
    _key: str = Depends(require_access_key),
    x_provider: Optional[str] = Header(default=None, alias="x-provider"),
    accept: Optional[str] = Header(default=None),
):
    """简化聊天：支持 json 与 multipart。

    关键字段（兼容前端 api.js）：
    - messages: list[{role, content}]
    - model: str
    - provider: str
    - stream: bool
    - stream_mode: "simple" | "openai"
    - image_fix: "mirror_h" | "rotate_180" | "mirror_h_rotate_180"
    - image_camera_facing: "front" | "back"
    - files / file_names: list[str]  附加文本拼到 messages
    - thoughts: bool  是否返回 thought
    - max_tokens / temperature / top_p / seed / stop / n
    """
    content_type = (request.headers.get("content-type") or "").lower()
    if "multipart/form-data" in content_type:
        body = await _parse_multipart(request)
    else:
        try:
            body = await request.json()
        except Exception:
            body = {}

    if not isinstance(body, dict):
        raise HttpError("invalid body", status=400, code="bad_request")

    # 统一归一化：兼容 input+images 简化形态 与 messages OpenAI 形态
    messages, model_from_body, provider_id = normalize_chat_body(body)
    if not messages:
        raise HttpError("input or messages is required", status=400, code="bad_request")

    model = model_from_body or GEMINI_DEFAULT_MODEL
    provider_id = provider_id or x_provider

    # 图片修正
    image_fix_mode = body.get("image_fix")
    camera_facing = body.get("image_camera_facing")
    if not image_fix_mode and camera_facing:
        image_fix_mode = infer_fix_mode(camera_facing)
    image_fixed = False
    image_fix_failed = 0
    image_warning: Optional[str] = None
    if image_fix_mode:
        success, failed, image_warning = await fix_images_in_messages(messages, image_fix_mode)
        image_fixed = success > 0
        image_fix_failed = failed

    config = await load_config()
    provider, api_key, clean_model = await select_provider_and_key(
        config=config,
        provider_id=provider_id,
        model=model,
        fallback_model=GEMINI_DEFAULT_MODEL,
    )

    stream = bool(body.get("stream"))
    stream_mode = str(body.get("stream_mode") or "openai").lower()

    # 构造上游 body（透传 OpenAI 风格参数）
    upstream_body = _build_upstream_body(body, clean_model, messages)

    if stream:
        return _stream_xtc_chat(
            provider,
            api_key,
            clean_model,
            upstream_body,
            messages,
            stream_mode,
            image_fix_mode,
            image_fixed,
            image_fix_failed,
            image_warning,
            _key,
        )

    # 非流式
    try:
        if provider.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=provider.base_url,
                api_key=api_key,
                body=upstream_body,
                stream=False,
            )
        record_usage(
            access_key=_key,
            provider=provider.id,
            model=clean_model,
            usage=(openai_resp or {}).get("usage"),
            ok=True,
        )
        return _build_xtc_response(
            openai_resp,
            provider=provider.id,
            model=clean_model,
            image_fix_mode=image_fix_mode,
            image_fixed=image_fixed,
            image_fix_failed=image_fix_failed,
            image_warning=image_warning,
            camera_facing=camera_facing,
        )
    except HttpError as e:
        record_usage(
            access_key=_key,
            provider=provider.id,
            model=clean_model,
            usage=None,
            ok=False,
            error_code=e.code,
        )
        raise


def _build_upstream_body(body: dict, clean_model: str, messages: list) -> dict:
    """从 XTC body 构造上游 OpenAI 风格 body。"""
    out = {
        "model": clean_model,
        "messages": messages,
    }
    for k in (
        "temperature", "top_p", "top_k", "max_tokens", "max_completion_tokens",
        "presence_penalty", "frequency_penalty", "seed", "stop", "n",
        "response_format", "tools", "tool_choice", "reasoning_effort",
        "stream_options", "extra_body", "google", "user",
    ):
        if k in body and body[k] is not None:
            out[k] = body[k]
    return out


def _build_xtc_response(
    openai_resp: dict,
    *,
    provider: str,
    model: str,
    image_fix_mode: Optional[str],
    image_fixed: bool,
    image_fix_failed: int,
    image_warning: Optional[str],
    camera_facing: Optional[str],
) -> dict:
    """从 OpenAI ChatCompletion 构造 XTC 简化响应。"""
    choices = openai_resp.get("choices") or []
    choice = choices[0] if choices else {}
    message = choice.get("message") or {}
    raw_text = message.get("content") or ""
    reasoning_content = message.get("reasoning_content")
    thought, text = extract_thought_and_answer(raw_text)
    if not thought and reasoning_content:
        thought = extract_thought_from_reasoning_content(reasoning_content)
    if not text:
        text = raw_text

    payload: dict[str, Any] = {
        "ok": True,
        "provider": provider,
        "model": model,
        "thought": thought,
        "text": text,
        "raw": raw_text,
        "usage": openai_resp.get("usage") or {},
        "finish_reason": choice.get("finish_reason"),
    }
    if image_fix_mode:
        payload["image_fix_mode"] = image_fix_mode
        payload["image_fixed"] = image_fixed
        payload["image_fix_failed_count"] = image_fix_failed
        payload["image_camera_facing"] = camera_facing
    if image_warning:
        payload["warning"] = image_warning
    headers = {"x-xtc-provider": provider, "x-xtc-model": model}
    if image_fix_mode:
        headers["x-xtc-image-fix-mode"] = image_fix_mode
    return ok_with_cors(payload, extra_headers=headers)


async def _stream_xtc_chat(
    provider,
    api_key: str,
    clean_model: str,
    upstream_body: dict,
    messages: list,
    stream_mode: str,
    image_fix_mode: Optional[str],
    image_fixed: bool,
    image_fix_failed: int,
    image_warning: Optional[str],
    access_key: str,
):
    """流式：simple 模式输出 {type,delta}；openai 模式直接透传 OpenAI chunk。"""
    from ..utils.sse import passthrough_with_usage

    # 让上游在最后一个流式 chunk 里返回 usage，否则 token 计数永远为 0。
    upstream_body = {
        **upstream_body,
        "stream_options": {**(upstream_body.get("stream_options") or {}), "include_usage": True},
    }

    headers = {
        **CORS_HEADERS,
        "x-xtc-provider": provider.id,
        "x-xtc-model": clean_model,
        "Cache-Control": "no-cache",
        "x-accel-buffering": "no",
    }
    if image_fix_mode:
        headers["x-xtc-image-fix-mode"] = image_fix_mode

    if stream_mode == "simple":
        captured: dict = {}
        return EventSourceResponse(
            _simple_stream_gen(
                provider, api_key, clean_model, upstream_body, messages,
                image_warning, access_key, captured,
            ),
            headers=headers,
        )

    # openai 模式：透传
    if provider.type == "gemini":
        captured_g: dict = {}

        async def gen_openai() -> AsyncIterator[str]:
            try:
                chunk_iter = await gemini_api.chat_completions(
                    api_key=api_key,
                    model=clean_model,
                    messages=messages,
                    body=upstream_body,
                    stream=True,
                )
                async for chunk in chunk_iter:
                    if isinstance(chunk, dict) and chunk.get("usage"):
                        captured_g["usage"] = chunk["usage"]
                    yield sse_data(chunk)
                yield sse_done()
            except HttpError as e:
                record_usage(
                    access_key=access_key, provider=provider.id, model=clean_model,
                    usage=None, ok=False, error_code=e.code,
                )
                yield sse_data({"error": {"code": e.code, "message": e.message, "status": e.status}})
                yield sse_done()
            else:
                record_usage(
                    access_key=access_key, provider=provider.id, model=clean_model,
                    usage=captured_g.get("usage"), ok=True,
                )

        return EventSourceResponse(gen_openai(), headers=headers)

    # OpenAI 厂商透传
    captured_pt: dict = {}

    async def gen_passthrough() -> AsyncIterator[bytes]:
        try:
            chunk_iter = await openai_adapter.chat_completions(
                base_url=provider.base_url, api_key=api_key, body=upstream_body, stream=True
            )
            async for chunk in passthrough_with_usage(chunk_iter, captured_pt):
                yield chunk
        except HttpError as e:
            record_usage(
                access_key=access_key, provider=provider.id, model=clean_model,
                usage=None, ok=False, error_code=e.code,
            )
            err = sse_data({"error": {"code": e.code, "message": e.message, "status": e.status}})
            yield err.encode("utf-8")
            yield sse_done().encode("utf-8")
        else:
            record_usage(
                access_key=access_key, provider=provider.id, model=clean_model,
                usage=captured_pt.get("usage"), ok=True,
            )

    return StreamingResponse(gen_passthrough(), media_type="text/event-stream", headers=headers)


async def _simple_stream_gen(
    provider,
    api_key: str,
    clean_model: str,
    upstream_body: dict,
    messages: list,
    image_warning: Optional[str],
    access_key: str,
    captured: Optional[dict] = None,
) -> AsyncIterator[str]:
    """simple 模式：把 OpenAI chunk 流转换为 {type: text/thought/done, delta}。"""
    in_thought = False
    thought_open_sent = False
    thought_close_sent = False
    open_tag_re = _THOUGHT_OPEN_RE
    close_tag_re = _THOUGHT_CLOSE_RE

    try:
        if provider.type == "gemini":
            chunk_iter = await gemini_api.chat_completions(
                api_key=api_key, model=clean_model, messages=messages,
                body=upstream_body, stream=True,
            )
            async for chunk in chunk_iter:
                if isinstance(chunk, dict) and chunk.get("usage") and captured is not None:
                    captured["usage"] = chunk["usage"]
                event = _simple_from_openai_chunk(chunk)
                if event:
                    yield sse_data(event)
        else:
            # OpenAI 厂商：直接读原始 SSE
            # 必须用 parse_sse_stream 跨 chunk 缓冲解析，否则跨块切断的
            # data: 行会 json.loads 失败被丢弃，导致回复少字、换行被破坏。
            from ..utils.sse import parse_sse_stream
            raw_iter = await openai_adapter.chat_completions(
                base_url=provider.base_url, api_key=api_key, body=upstream_body, stream=True
            )
            async for sse in parse_sse_stream(raw_iter):
                if sse.get("__done__"):
                    yield sse_data({"type": "done", "delta": "", "finish_reason": "stop"})
                    return
                if sse.get("__raw__") is not None:
                    continue
                if isinstance(sse, dict) and sse.get("usage") and captured is not None:
                    captured["usage"] = sse["usage"]
                event = _simple_from_openai_chunk(sse)
                if event:
                    yield sse_data(event)
        yield sse_data({"type": "done", "delta": "", "finish_reason": "stop"})
    except HttpError as e:
        record_usage(
            access_key=access_key, provider=provider.id, model=clean_model,
            usage=None, ok=False, error_code=e.code,
        )
        yield sse_data(
            {
                "type": "error",
                "error": {"code": e.code, "message": e.message, "status": e.status},
            }
        )
        return
    else:
        record_usage(
            access_key=access_key, provider=provider.id, model=clean_model,
            usage=(captured or {}).get("usage"), ok=True,
        )
    if image_warning:
        yield sse_data({"type": "warning", "delta": image_warning})


def _simple_from_openai_chunk(chunk: dict) -> Optional[dict]:
    """把单个 OpenAI chunk 转为 simple 事件。"""
    choices = chunk.get("choices") or []
    if not choices:
        return None
    choice = choices[0]
    delta = choice.get("delta") or {}
    content = delta.get("content")
    finish_reason = choice.get("finish_reason")
    if content is not None:
        return {"type": "text", "delta": content, "finish_reason": finish_reason}
    if finish_reason:
        return {"type": "done", "delta": "", "finish_reason": finish_reason}
    return None


async def _parse_multipart(request: Request) -> dict:
    """解析 multipart/form-data。

    兼容前端 api.js 的 callUpload 路径，该路径发送：
    - 简单字段：access_key / provider / model / input / stream / max_tokens /
      image_fix / file_names（JSON 字符串）/ images（图片 URL 字符串，可多条）
    - 文件字段：name="images" 的图片 UploadFile，name="files" 的文件 UploadFile

    图片 UploadFile 转 data URL 放进 out["images"]；
    文本类文件 UploadFile 读 utf-8 文本放进 out["files"]，文件名进 out["file_names"]。
    最终由 normalize_chat_body 统一组装为 messages。
    """
    import base64

    from starlette.datastructures import UploadFile

    form = await request.form()
    out: dict[str, Any] = {}
    # 简单字段（含 input）
    for key in ("model", "provider", "input", "stream", "stream_mode", "image_fix",
                "image_camera_facing", "max_tokens", "temperature", "top_p",
                "seed", "stop", "n", "thoughts"):
        if key not in form:
            continue
        val = form[key]
        if isinstance(val, UploadFile):
            continue
        val = str(val)
        if val in ("true", "false"):
            out[key] = val == "true"
        elif val.lstrip("-").isdigit():
            out[key] = int(val)
        else:
            try:
                out[key] = float(val)
            except (TypeError, ValueError):
                out[key] = val
    # messages（JSON 字符串）
    if "messages" in form:
        raw = form["messages"]
        if not isinstance(raw, UploadFile):
            try:
                out["messages"] = json.loads(str(raw))
            except Exception:
                out["messages"] = [{"role": "user", "content": str(raw)}]
    # file_names（前端会发 JSON.stringify(names)）
    file_names: list[str] = []
    if "file_names" in form:
        raw = form["file_names"]
        if not isinstance(raw, UploadFile):
            raw = str(raw)
            try:
                parsed = json.loads(raw)
                if isinstance(parsed, list):
                    file_names = [str(x) for x in parsed]
                else:
                    file_names = [raw]
            except Exception:
                file_names = [raw]
    # 遍历所有项，收集图片 URL 字符串 + UploadFile
    image_urls: list[str] = []
    files_text: list[str] = []
    upload_file_names: list[str] = []
    for key, value in form.multi_items():
        if isinstance(value, UploadFile):
            # 检查文件大小限制（10MB）
            # 获取文件大小（需要先检查是否支持获取大小）
            try:
                # 读取文件内容（必须读取，不能仅依赖 value.size）
                content_bytes = await value.read()
                file_size = len(content_bytes)
                # 文件大小限制（10MB）
                if file_size > 10 * 1024 * 1024:  # 10MB limit
                    raise HttpError(
                        f"file too large: {file_size} bytes > 10MB", 
                        status=413, code="too_large"
                    )
                filename = value.filename or key
                if key == "images" or key == "image" or key.startswith("image_"):
                    # 图片文件 → data URL
                    mime = (value.content_type or "image/jpeg").split(";")[0].strip() or "image/jpeg"
                    b64 = base64.b64encode(content_bytes).decode("ascii")
                    image_urls.append(f"data:{mime};base64,{b64}")
                elif key == "files" or key == "file" or key.startswith("file_"):
                    # 文件 → 尝试 utf-8 文本；二进制则记空串占位
                    try:
                        files_text.append(content_bytes.decode("utf-8"))
                    except Exception:
                        files_text.append("")
                    upload_file_names.append(filename)
            except HttpError:
                # 重新抛出HttpError
                raise
            except Exception as e:
                # 处理其他异常，特别是大文件导致的内存问题
                if 'memory' in str(e).lower() or 'out of memory' in str(e).lower():
                    raise HttpError(
                        "Failed to upload inline file.", 
                        status=500, code="internal_error"
                    )
                raise HttpError(
                    "Failed to upload inline file.", 
                    status=500, code="internal_error"
                )
            continue
        # 字符串值
        sval = str(value or "").strip()
        if not sval:
            continue
        if key == "images" or key == "image" or key.startswith("image_"):
            if sval.startswith(("http://", "https://", "data:")):
                image_urls.append(sval)
        elif key == "files" or key == "file" or key.startswith("file_"):
            files_text.append(sval)
            upload_file_names.append(key)
    if image_urls:
        out["images"] = image_urls
    if files_text:
        out["files"] = files_text
        out["file_names"] = (file_names + upload_file_names) or upload_file_names
    return out