content-find-agent-new/content_agent/integrations/query_variant.py

from __future__ import annotations

import copy
import os
from pathlib import Path
from typing import Any, Callable, Mapping

import httpx

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations import timeout_config
from content_agent.integrations.query_prompt_config import DEFAULT_PROFILE, load_profile
from content_agent.interfaces import QueryVariantClient, QueryVariantResult

DEFAULT_OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
DEFAULT_QUERY_PROMPT_VERSION = "query_variant.v1"
DEFAULT_QUERY_TIMEOUT_SECONDS = 120.0


def _retryable_status(exc: httpx.HTTPStatusError) -> bool:
    status = getattr(getattr(exc, "response", None), "status_code", None)
    return isinstance(status, int) and (status in (408, 429) or 500 <= status < 600)


# M9D Gate 2:判搜索词是否易搜到中国 50+ 人群喜欢的视频(仅非抖音)。只回 yes/no。
_FIFTY_PLUS_GATE_SYSTEM = (
    "你判断一个中文短视频搜索词,是否容易搜到中国 50 岁以上中老年人群喜欢的视频。"
    "只回答 yes 或 no,不要解释。"
)


class MissingQueryVariantClient:
    def __init__(self, reason: str, detail: dict[str, Any] | None = None) -> None:
        self.reason = reason
        self.detail = detail or {}

    def generate_variant(
        self,
        *,
        seed_term: str,
        evidence_context: dict[str, Any],
    ) -> QueryVariantResult:
        raise ContentAgentError(
            ErrorCode.QUERY_GENERATION_FAILED,
            "query generation failed",
            {
                "reason": self.reason,
                "seed_term": seed_term,
                **self.detail,
            },
        )


class OpenRouterQueryVariantClient:
    def __init__(
        self,
        *,
        api_key: str,
        model: str,
        base_url: str = DEFAULT_OPENROUTER_BASE_URL,
        timeout_seconds: float = DEFAULT_QUERY_TIMEOUT_SECONDS,
        prompt_version: str = DEFAULT_QUERY_PROMPT_VERSION,
        profile: dict[str, Any] | None = None,
        http_post: Callable[..., Any] | None = None,
    ) -> None:
        self.api_key = api_key
        self.model = model
        self.base_url = base_url.rstrip("/")
        self.timeout_seconds = timeout_seconds
        # None → 运行时取 httpx.post(便于测试 monkeypatch httpx.post);也可注入自定义。
        self.http_post = http_post
        self.profile = copy.deepcopy(profile or DEFAULT_PROFILE)
        self.prompt_version = str(self.profile.get("prompt_version") or prompt_version)

    def _timeout(self) -> httpx.Timeout:
        return timeout_config.as_httpx_timeout(
            self.timeout_seconds, read=timeout_config.read_timeout("query_llm")
        )

    def _post(self, *args: Any, **kwargs: Any) -> Any:
        return (self.http_post or httpx.post)(*args, **kwargs)

    def generate_variant(
        self,
        *,
        seed_term: str,
        evidence_context: dict[str, Any],
    ) -> QueryVariantResult:
        # 120s 超时 + 网络/超时/5xx/429 重试一次;解析错误不重试(确定性)。
        for attempt in range(2):
            try:
                response = self._post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json",
                    },
                    json={
                        "model": self.model,
                        "messages": _render_messages(self.profile, seed_term, evidence_context),
                        "temperature": self.profile["temperature"],
                        "max_tokens": self.profile["max_tokens"],
                    },
                    timeout=self._timeout(),
                )
                response.raise_for_status()
                query = _extract_query(response.json())
            except ContentAgentError:
                raise
            except httpx.HTTPStatusError as exc:
                if attempt == 0 and _retryable_status(exc):
                    continue
                raise _generation_error(
                    "openrouter_http_status",
                    seed_term,
                    {"status_code": exc.response.status_code},
                ) from exc
            except httpx.HTTPError as exc:
                if attempt == 0:
                    continue
                raise _generation_error(
                    "openrouter_http_error",
                    seed_term,
                    {"exception_type": type(exc).__name__},
                ) from exc
            except (KeyError, TypeError, ValueError) as exc:
                raise _generation_error(
                    "openrouter_response_invalid",
                    seed_term,
                    {"exception_type": type(exc).__name__},
                ) from exc

            return QueryVariantResult(
                query=query,
                model=self.model,
                prompt_version=self.prompt_version,
                input_evidence=evidence_context,
            )
        # 理论不可达(循环内必 return 或 raise);兜底。
        raise _generation_error("openrouter_http_error", seed_term, {"exception_type": "Unknown"})

    def judge_query_fifty_plus(self, query_text: str) -> bool:
        """M9D Gate 2:返回 True=放行(含拿不准/异常);仅明确 no 才丢弃。"""
        try:
            response = self._post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": self.model,
                    "messages": [
                        {"role": "system", "content": _FIFTY_PLUS_GATE_SYSTEM},
                        {"role": "user", "content": f"搜索词:{query_text}"},
                    ],
                    "temperature": 0,
                    "max_tokens": 4,
                },
                timeout=self._timeout(),
            )
            response.raise_for_status()
            content = response.json()["choices"][0]["message"]["content"]
        except Exception:
            return True
        return not str(content or "").strip().lower().startswith("no")


def query_variant_client_from_env(
    env: Mapping[str, str] | None = None,
    *,
    platform: str = "douyin",
    strategy_version: str = "V1",
    root_dir: Path | str = Path("."),
) -> QueryVariantClient:
    source = os.environ if env is None else env
    # query LLM 也支持切 DashScope/通义千问:优先 CONTENT_AGENT_QUERY_LLM_* / DASHSCOPE,回退 OpenRouter(海外不变)。
    api_key = (
        _env_value(source, "CONTENT_AGENT_QUERY_LLM_API_KEY")
        or _env_value(source, "DASHSCOPE_API_KEY")
        or _env_value(source, "OPENROUTER_API_KEY")
        or _env_value(source, "OPEN_ROUTER_API_KEY")
    )
    model = _env_value(source, "CONTENT_AGENT_QUERY_LLM_MODEL") or _env_value(source, "MODEL")
    base_url = (
        _env_value(source, "CONTENT_AGENT_QUERY_LLM_BASE_URL")
        or _env_value(source, "OPENROUTER_BASE_URL")
        or DEFAULT_OPENROUTER_BASE_URL
    )
    prompt_version = _env_value(source, "CONTENT_AGENT_QUERY_LLM_PROMPT_VERSION") or DEFAULT_QUERY_PROMPT_VERSION
    timeout_seconds = _float_env(
        source,
        "CONTENT_AGENT_QUERY_LLM_TIMEOUT_SECONDS",
        DEFAULT_QUERY_TIMEOUT_SECONDS,
    )

    missing = []
    if not api_key:
        missing.append("CONTENT_AGENT_QUERY_LLM_API_KEY/OPENROUTER_API_KEY")
    if not model:
        missing.append("CONTENT_AGENT_QUERY_LLM_MODEL")
    if missing:
        return MissingQueryVariantClient(
            "query variant LLM config missing",
            {"missing_env_keys": missing},
        )

    return OpenRouterQueryVariantClient(
        api_key=api_key,
        model=model,
        base_url=base_url,
        timeout_seconds=timeout_seconds,
        prompt_version=prompt_version,
        profile=load_profile(platform, strategy_version, root_dir=root_dir),
    )


def _messages(seed_term: str, evidence_context: dict[str, Any]) -> list[dict[str, str]]:
    return _render_messages(DEFAULT_PROFILE, seed_term, evidence_context)


def _render_messages(
    profile: dict[str, Any],
    seed_term: str,
    evidence_context: dict[str, Any],
) -> list[dict[str, str]]:
    return [
        {
            "role": "system",
            "content": str(profile["system"]),
        },
        {
            "role": "user",
            "content": str(profile["user"])
            .replace("{seed_term}", seed_term)
            .replace("{evidence_context}", str(evidence_context)),
        },
    ]


def _extract_query(payload: dict[str, Any]) -> str:
    content = payload["choices"][0]["message"]["content"]
    if not isinstance(content, str):
        raise ValueError("OpenRouter content is not a string")
    query = _normalize_query(content)
    if not query:
        raise ValueError("OpenRouter content is empty")
    if "\n" in query:
        raise ValueError("OpenRouter content has multiple lines")
    return query


def _normalize_query(value: str) -> str:
    query = " ".join(value.split()).strip()
    return query.strip("`'\"“”‘’")


def _generation_error(
    reason: str,
    seed_term: str,
    detail: dict[str, Any] | None = None,
) -> ContentAgentError:
    return ContentAgentError(
        ErrorCode.QUERY_GENERATION_FAILED,
        "query generation failed",
        {
            "reason": reason,
            "seed_term": seed_term,
            **(detail or {}),
        },
    )


def _env_value(env: Mapping[str, str], key: str) -> str:
    return str(env.get(key, "")).strip()


def _float_env(env: Mapping[str, str], key: str, default: float) -> float:
    value = _env_value(env, key)
    if not value:
        return default
    try:
        return float(value)
    except ValueError:
        return default