yumao
5742dc47ed
daily 自建 Cursor bridge；DAILY_LLM_PROVIDER 控制后端；config/feeds.yaml 与 sensitive_words 可配置。 Co-authored-by: Cursor <cursoragent@cursor.com>
114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
"""从 config/feeds.yaml 加载 RSS 源（失败时回退内置默认）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from daily.config import ROOT
|
||
from daily.news.feeds_types import NewsCategory, NewsFeed
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_FEEDS_FILE = ROOT / "config" / "feeds.yaml"
|
||
|
||
|
||
def _parse_feed(raw: dict[str, Any]) -> NewsFeed:
|
||
return NewsFeed(
|
||
name=str(raw.get("name") or "").strip(),
|
||
url=str(raw.get("url") or "").strip(),
|
||
slow=bool(raw.get("slow")),
|
||
ai_filter=bool(raw.get("ai_filter")),
|
||
)
|
||
|
||
|
||
def _parse_category(raw: dict[str, Any]) -> NewsCategory | None:
|
||
cat_id = str(raw.get("id") or "").strip()
|
||
if not cat_id:
|
||
return None
|
||
feeds_raw = raw.get("feeds") or []
|
||
feeds: list[NewsFeed] = []
|
||
for item in feeds_raw:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
feed = _parse_feed(item)
|
||
if feed.name and feed.url:
|
||
feeds.append(feed)
|
||
if not feeds:
|
||
return None
|
||
return NewsCategory(
|
||
id=cat_id,
|
||
name=str(raw.get("name") or cat_id),
|
||
icon=str(raw.get("icon") or "📰"),
|
||
feeds=tuple(feeds),
|
||
)
|
||
|
||
|
||
def _parse_categories(items: Any) -> tuple[NewsCategory, ...]:
|
||
if not isinstance(items, list):
|
||
return ()
|
||
categories: list[NewsCategory] = []
|
||
for raw in items:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
cat = _parse_category(raw)
|
||
if cat:
|
||
categories.append(cat)
|
||
return tuple(categories)
|
||
|
||
|
||
def _load_yaml() -> dict[str, Any] | None:
|
||
if not _FEEDS_FILE.exists():
|
||
return None
|
||
try:
|
||
import yaml
|
||
except ImportError:
|
||
logger.warning("未安装 PyYAML，无法读取 %s", _FEEDS_FILE)
|
||
return None
|
||
try:
|
||
data = yaml.safe_load(_FEEDS_FILE.read_text(encoding="utf-8"))
|
||
except OSError as exc:
|
||
logger.warning("读取 feeds 配置失败: %s", exc)
|
||
return None
|
||
except Exception as exc:
|
||
logger.warning("解析 feeds.yaml 失败: %s", exc)
|
||
return None
|
||
return data if isinstance(data, dict) else None
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def _yaml_payload() -> dict[str, Any] | None:
|
||
return _load_yaml()
|
||
|
||
|
||
def load_categories(region: str) -> tuple[NewsCategory, ...]:
|
||
data = _yaml_payload()
|
||
if data:
|
||
block = data.get(region) or {}
|
||
categories = _parse_categories(block.get("categories"))
|
||
if categories:
|
||
return categories
|
||
logger.warning("feeds.yaml 中 %s.categories 为空，使用内置默认", region)
|
||
|
||
from daily.news import feeds_defaults as defaults
|
||
|
||
if region == "cn":
|
||
return defaults.CN_NEWS_CATEGORIES
|
||
return defaults.NEWS_CATEGORIES
|
||
|
||
|
||
def load_cn_title_keywords() -> tuple[str, ...]:
|
||
data = _yaml_payload()
|
||
if data:
|
||
block = data.get("cn") or {}
|
||
keywords = block.get("title_keywords")
|
||
if isinstance(keywords, list):
|
||
cleaned = tuple(str(x).strip() for x in keywords if str(x).strip())
|
||
if cleaned:
|
||
return cleaned
|
||
from daily.news.feeds_defaults import CN_AI_TITLE_KEYWORDS
|
||
|
||
return CN_AI_TITLE_KEYWORDS
|