dataClean/app/rss_client.py

"""调用 rssKeeper 外部 API"""
from datetime import datetime, timedelta
from typing import List, Optional, Dict, Any
import logging

import requests

from config import settings

logger = logging.getLogger(__name__)


class RSSKeeperClient:
    """rssKeeper 外部 API 客户端。

    配置以 property 形式运行时从 settings 读取，避免模块 import 时
    固化旧值（settings 在 FastAPI lifespan 启动后才会被数据库配置覆盖）。
    """

    def __init__(self, base_url: Optional[str] = None, timeout: Optional[int] = None):
        self._base_url = base_url
        self._timeout = timeout

    @property
    def base_url(self) -> str:
        return (self._base_url or settings.RSSKEEPER_BASE_URL).rstrip("/")

    @property
    def timeout(self) -> int:
        return self._timeout if self._timeout is not None else 30

    def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        url = f"{self.base_url}{path}"
        try:
            resp = requests.get(url, params=params, timeout=self.timeout)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as exc:
            logger.error("请求 rssKeeper 失败: %s - %s", url, exc)
            raise

    def fetch_recent(
        self,
        hours: int = 24,
        limit: int = 200,
        feed_id: Optional[int] = None,
        category: Optional[str] = None,
        search: Optional[str] = None,
        unread_only: bool = False,
    ) -> List[Dict[str, Any]]:
        """获取最近 N 小时的文章"""
        params = {
            "hours": hours,
            "limit": limit,
            "unread_only": unread_only,
        }
        if feed_id is not None:
            params["feed_id"] = feed_id
        if category is not None:
            params["category"] = category
        if search is not None:
            params["search"] = search

        data = self._get("/api/v1/external/recent", params=params)
        return data.get("articles", [])

    def fetch_by_date(self, date: str, category: Optional[str] = None) -> Dict[str, Any]:
        """获取指定日期的文章聚合"""
        params: Dict[str, Any] = {"date": date}
        if category is not None:
            params["category"] = category
        return self._get("/api/v1/external/summary", params=params)

    def fetch_feeds(
        self,
        health_status: Optional[str] = None,
        category: Optional[str] = None,
        error_type: Optional[str] = None,
        is_active: Optional[bool] = True,
    ) -> List[Dict[str, Any]]:
        """获取 RSS 源列表"""
        params: Dict[str, Any] = {}
        if health_status is not None:
            params["health_status"] = health_status
        if category is not None:
            params["category"] = category
        if error_type is not None:
            params["error_type"] = error_type
        if is_active is not None:
            params["is_active"] = is_active

        data = self._get("/api/v1/external/feeds", params=params)
        return data.get("feeds", [])

    def fulltext_search(
        self,
        q: str,
        limit: int = 50,
        offset: int = 0,
        category: Optional[str] = None,
        feed_id: Optional[int] = None,
    ) -> Dict[str, Any]:
        """全文搜索文章"""
        params: Dict[str, Any] = {
            "q": q,
            "limit": limit,
            "offset": offset,
        }
        if category is not None:
            params["category"] = category
        if feed_id is not None:
            params["feed_id"] = feed_id
        return self._get("/api/v1/external/search", params=params)


rss_client = RSSKeeperClient()