"""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