api/openui.py

from __future__ import annotations

import os
from pathlib import Path
from typing import Any

import httpx
from fastapi import APIRouter, Header, HTTPException
from pydantic import BaseModel, Field

from api.support import require_identity
from services.config import config
from services.log_service import LoggedCall
from services.protocol import openai_v1_chat_complete

# OpenUI Lang 1.0 系统提示词由 Node CLI 预生成（依赖 @openuidev/lang-core），
# 见 web/src/lib/openui/library.ts + `npx @openuidev/cli@latest generate`。
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
_SYSTEM_PROMPT_PATH = _PROJECT_ROOT / "openui" / "system-prompt.txt"
_system_prompt_cache: str | None = None
_system_prompt_mtime: float | None = None


def load_system_prompt() -> str:
    """读取预生成的 OpenUI 系统提示词（约 29KB）。按 mtime 缓存：文件更新后
    下一次调用自动生效，无需重启后端。"""
    global _system_prompt_cache, _system_prompt_mtime
    try:
        mtime = _SYSTEM_PROMPT_PATH.stat().st_mtime
    except OSError as exc:
        raise HTTPException(
            status_code=500,
            detail={"error": f"缺少 OpenUI 系统提示词文件（{_SYSTEM_PROMPT_PATH}）：{exc}"},
        ) from exc
    if _system_prompt_cache is None or _system_prompt_mtime != mtime:
        _system_prompt_cache = _SYSTEM_PROMPT_PATH.read_text(encoding="utf-8")
        _system_prompt_mtime = mtime
    return _system_prompt_cache


def _first_non_empty(*values: Any) -> str:
    for value in values:
        text = str(value or "").strip()
        if text:
            return text
    return ""


def _external_target() -> tuple[str, str, str] | None:
    """
    可选的外部通道（不配置就不存在）。

    只有在明确配置了 OpenUI Cloud 或第三方 OpenAI 兼容接口时才启用；
    默认情况下生成式 UI 走本站自己的账号池，不需要任何外部 Key。
    """
    settings = config.get()
    cloud_key = _first_non_empty(settings.get("openui_api_key"), os.environ.get("THESYS_API_KEY"))
    if cloud_key:
        base_url = _first_non_empty(
            settings.get("openui_base_url"),
            os.environ.get("THESYS_BASE_URL"),
            "https://api.thesys.dev/v1/embed",
        ).rstrip("/")
        model = _first_non_empty(settings.get("openui_model"), os.environ.get("THESYS_MODEL"), "openai/gpt-5")
        return base_url, cloud_key, model

    base_url = _first_non_empty(settings.get("openui_llm_base_url"), os.environ.get("OPENUI_LLM_BASE_URL")).rstrip("/")
    api_key = _first_non_empty(settings.get("openui_llm_api_key"), os.environ.get("OPENUI_LLM_API_KEY"))
    model = _first_non_empty(settings.get("openui_llm_model"), os.environ.get("OPENUI_LLM_MODEL"))
    if base_url and api_key and model:
        return base_url, api_key, model
    return None


def _text_of(content: Any) -> str:
    """从 content（可能是字符串，也可能是多模态分片数组）里抽出纯文本，用于日志预览。"""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, dict):
                text = item.get("text")
                if isinstance(text, str):
                    parts.append(text)
        return " ".join(parts)
    return str(content or "")


def _build_openui_messages(messages: list[OpenUiMessage]) -> list[dict[str, Any]]:
    """把请求里的消息转成 OpenAI 兼容格式。

    普通消息用纯文本；带图片附件（data:image/ 的 base64）的用户/助手消息，
    转成多模态 content 数组（text + image_url），让模型能真正“看到”图片。
    """
    out: list[dict[str, Any]] = []
    for item in messages:
        if item.role not in ("user", "assistant"):
            continue
        images = [url for url in (item.images or []) if isinstance(url, str) and url.startswith("data:image/")]
        if images:
            content: list[dict[str, Any]] = []
            if item.content.strip():
                content.append({"type": "text", "text": item.content})
            for url in images:
                content.append({"type": "image_url", "image_url": {"url": url}})
            out.append({"role": item.role, "content": content})
        elif item.content.strip():
            out.append({"role": item.role, "content": item.content})
    return out


def _completion_text(data: Any) -> str:
    """从 OpenAI 兼容响应里取出文本（content 可能是字符串，也可能是分片数组）。"""
    if not isinstance(data, dict):
        return ""
    choices = data.get("choices") or []
    if not choices:
        return ""
    message = choices[0].get("message") or {}
    content = message.get("content")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, dict):
                text = item.get("text")
                if isinstance(text, str):
                    parts.append(text)
        return "".join(parts)
    return str(content or "")


class OpenUiMessage(BaseModel):
    role: str
    content: str = ""
    images: list[str] = Field(default_factory=list)


class OpenUiChatRequest(BaseModel):
    messages: list[OpenUiMessage] = Field(default_factory=list)


async def _call_local_pool(identity: dict[str, object], model: str, messages: list[dict[str, Any]]) -> str:
    """走本站账号池（逆向 ChatGPT 通道），与 /v1/chat/completions 同一条链路。"""
    settings = config.get()
    payload: dict[str, Any] = {
        "model": model,
        "stream": False,
        "messages": [{"role": "system", "content": load_system_prompt()}, *messages],
    }
    max_tokens = _first_non_empty(settings.get("openui_max_tokens"), os.environ.get("OPENUI_MAX_TOKENS"))
    payload["max_tokens"] = int(max_tokens) if str(max_tokens).isdigit() else 4096
    last_user = next((_text_of(m["content"]) for m in reversed(messages) if m["role"] == "user"), "")
    call = LoggedCall(identity, "/api/openui/chat", model, "生成式 UI", request_text=str(last_user)[:500])
    result = await call.run(openai_v1_chat_complete.handle, payload, identity)
    text = _completion_text(result)
    if not text.strip():
        raise HTTPException(status_code=502, detail={"error": "模型没有返回生成式 UI 内容"})
    return text


async def _call_external(target: tuple[str, str, str], messages: list[dict[str, Any]]) -> str:
    base_url, api_key, model = target
    request_body: dict[str, Any] = {
        "model": model,
        "stream": False,
        "messages": [{"role": "system", "content": load_system_prompt()}, *messages],
    }
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(180.0, connect=20.0)) as client:
            response = await client.post(
                f"{base_url}/chat/completions",
                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                json=request_body,
            )
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=502, detail={"error": f"无法连接生成式 UI 模型服务：{exc}"}) from exc

    if response.status_code >= 400:
        raise HTTPException(
            status_code=502,
            detail={"error": f"生成式 UI 模型服务返回 {response.status_code}", "detail": response.text[:500]},
        )
    try:
        data = response.json()
    except ValueError as exc:
        raise HTTPException(status_code=502, detail={"error": "生成式 UI 模型返回了非 JSON 响应"}) from exc
    text = _completion_text(data)
    if not text.strip():
        raise HTTPException(status_code=502, detail={"error": "生成式 UI 模型未返回内容"})
    return text


def create_router() -> APIRouter:
    router = APIRouter()

    @router.post("/api/openui/chat")
    async def openui_chat(payload: OpenUiChatRequest, authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)

        messages = _build_openui_messages(payload.messages)
        if not messages:
            raise HTTPException(status_code=400, detail={"error": "缺少可发送的消息"})

        # 默认走本站账号池；只有显式配置了外部通道才改走外部。
        external = _external_target()
        if external:
            source = await _call_external(external, messages)
            return {"source": source, "channel": "external"}

        settings = config.get()
        # 号池侧的模型 id 与 OpenUI Cloud 的 `openui_model` 不是同一个命名空间，
        # 因此本地通道单独用 openui_pool_model，避免配了云端模型名却拿去查号池。
        model = _first_non_empty(
            settings.get("openui_pool_model"),
            os.environ.get("OPENUI_POOL_MODEL"),
            "auto",
        )
        source = await _call_local_pool(identity, model, messages)
        return {"source": source, "channel": "local"}

    return router