ozon-seller-kit/server/services/deepseek.py

from __future__ import annotations

import json
import re
from typing import Any

import httpx
from fastapi import HTTPException

from schemas.copy import CopyRequest, CopyResponse, UsageInfo
from services.models_catalog import ModelSpec, get_model_spec, resolve_api_key
from services.prompts.copy_ru import SYSTEM_PROMPT, build_user_prompt

_JSON_BLOCK_RE = re.compile(r"```(?:json)?\s*([\s\S]*?)\s*```", re.IGNORECASE)


def _extract_json_object(content: str) -> dict[str, Any]:
    text = (content or "").strip()
    if not text:
        raise ValueError("模型返回空内容")

    match = _JSON_BLOCK_RE.search(text)
    if match:
        text = match.group(1).strip()

    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        start = text.find("{")
        end = text.rfind("}")
        if start < 0 or end <= start:
            raise ValueError("无法从模型回复中解析 JSON") from None
        data = json.loads(text[start : end + 1])

    if not isinstance(data, dict):
        raise ValueError("模型返回的 JSON 不是对象")
    return data


def _as_str(value: Any, field: str) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    raise ValueError(f"字段 {field} 必须是字符串")


def _as_str_list(value: Any, field: str) -> list[str]:
    if value is None:
        return []
    if isinstance(value, str):
        parts = re.split(r"[,，\n]+", value)
        return [p.strip() for p in parts if p.strip()]
    if isinstance(value, list):
        result: list[str] = []
        for item in value:
            s = str(item).strip()
            if s:
                result.append(s)
        return result
    raise ValueError(f"字段 {field} 必须是字符串数组")


def _as_title_list(value: Any, field: str) -> list[str]:
    """标题本身可能含逗号，不能按标点切分。"""
    if value is None:
        return []
    if isinstance(value, str):
        text = value.strip()
        return [text] if text else []
    if isinstance(value, list):
        return [str(item).strip() for item in value if str(item).strip()]
    raise ValueError(f"字段 {field} 必须是字符串数组")


def _map_copy_payload(data: dict[str, Any], *, model: str, usage: dict[str, Any] | None) -> CopyResponse:
    usage = usage or {}
    return CopyResponse(
        titles_ru=_as_title_list(data.get("titles_ru"), "titles_ru"),
        titles_zh=_as_title_list(data.get("titles_zh"), "titles_zh"),
        description_ru=_as_str(data.get("description_ru"), "description_ru"),
        description_zh=_as_str(data.get("description_zh"), "description_zh"),
        tags_ru=_as_str_list(data.get("tags_ru"), "tags_ru"),
        tags_zh=_as_str_list(data.get("tags_zh"), "tags_zh"),
        model=model,
        usage=UsageInfo(
            prompt_tokens=int(usage.get("prompt_tokens") or 0),
            completion_tokens=int(usage.get("completion_tokens") or 0),
        ),
    )


async def _chat_once(spec: ModelSpec, messages: list[dict[str, str]]) -> tuple[str, dict[str, Any]]:
    api_key = resolve_api_key(spec)
    url = spec.base_url.rstrip("/") + "/chat/completions"
    payload = {
        "model": spec.api_model,
        "messages": messages,
        # 商品事实需要稳定，营销表达仍保留少量变化。
        "temperature": 0.45,
        "max_tokens": spec.max_tokens,
        "response_format": {"type": "json_object"},
        **spec.params,
    }
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    try:
        async with httpx.AsyncClient(timeout=90.0) as client:
            resp = await client.post(url, headers=headers, json=payload)
    except httpx.RequestError as exc:
        raise HTTPException(status_code=502, detail=f"模型网络错误：{exc}") from exc

    if resp.status_code >= 400:
        detail = resp.text[:500]
        raise HTTPException(
            status_code=502,
            detail=f"模型调用失败（HTTP {resp.status_code}）：{detail}",
        )

    body = resp.json()
    try:
        choice = body["choices"][0]
        content = choice["message"]["content"]
    except (KeyError, IndexError, TypeError) as exc:
        raise HTTPException(status_code=502, detail="模型响应格式异常") from exc

    # 思考型模型的思维链也计入 max_tokens，推理过长时正文会是空串。
    if not (content or "").strip() and choice.get("finish_reason") == "length":
        raise HTTPException(
            status_code=502,
            detail=(
                f"模型「{spec.id}」在 max_tokens={spec.max_tokens} 内只输出了思维链、没有正文。"
                "请调高该模型的 max_tokens，或在 config/models.yaml 中为它关闭/降低思维链。"
            ),
        )

    usage = body.get("usage") or {}
    return content, usage


async def generate_copy(req: CopyRequest) -> CopyResponse:
    spec = get_model_spec(req.model or None)
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {
            "role": "user",
            "content": build_user_prompt(
                source_text=req.source_text,
                product_name=req.product_name,
                model_code=req.model_code,
            ),
        },
    ]

    last_error: Exception | None = None
    for attempt in range(2):
        content, usage = await _chat_once(spec, messages)
        try:
            data = _extract_json_object(content)
            result = _map_copy_payload(data, model=spec.id, usage=usage)
            if not result.titles_ru or not result.description_ru:
                raise ValueError("标题或描述俄文为空")
            return result
        except (ValueError, json.JSONDecodeError) as exc:
            last_error = exc
            if attempt == 0:
                messages.append({"role": "assistant", "content": content})
                messages.append(
                    {
                        "role": "user",
                        "content": "上一次输出无法解析为约定 JSON，请仅重新输出合法 JSON 对象，不要其它文字。",
                    }
                )
                continue

    raise HTTPException(
        status_code=502,
        detail=f"模型返回无法解析：{last_error}",
    )