Create-knowledge-find-decode/query_planning/sources/pattern_topic_build.py

"""Read-only adapter for the Pattern topic-build detail API."""
from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

import httpx

from core.config import env_value, load_env_file


class TopicBuildSourceError(RuntimeError):
    """Raised when the upstream topic-build API cannot provide a valid snapshot."""


@dataclass(frozen=True)
class PatternTopicBuildConfig:
    base_url: str = "https://pattern.aiddit.com"
    timeout_seconds: float = 30.0

    @classmethod
    def from_env(cls, env_file: str | Path = ".env") -> "PatternTopicBuildConfig":
        file_env = load_env_file(env_file)
        return cls(
            base_url=env_value(
                "PATTERN_API_BASE_URL",
                file_env,
                "https://pattern.aiddit.com",
            ).rstrip("/"),
            timeout_seconds=float(
                env_value("PATTERN_API_TIMEOUT_SECONDS", file_env, "30")
            ),
        )


class PatternTopicBuildSource:
    def __init__(
        self,
        config: PatternTopicBuildConfig,
        *,
        http_get: Callable[..., Any] = httpx.get,
    ) -> None:
        self.config = config
        self.http_get = http_get

    def fetch(self, topic_build_id: int) -> dict[str, Any]:
        url = f"{self.config.base_url}/api/pattern/topic_builds/{topic_build_id}"
        try:
            response = self.http_get(url, timeout=self.config.timeout_seconds)
            response.raise_for_status()
            payload = response.json()
        except (httpx.HTTPError, ValueError, TypeError) as exc:
            raise TopicBuildSourceError(f"选题接口读取失败：{exc}") from exc
        if not isinstance(payload, dict) or payload.get("success") is not True:
            raise TopicBuildSourceError("选题接口没有返回 success=true")
        topics = payload.get("topics")
        if not isinstance(topics, list) or not topics:
            raise TopicBuildSourceError("这次选题构建没有可用的 topics")
        payload = dict(payload)
        payload["source_url"] = url
        return payload