#!/usr/bin/env python3 """Merge multiple RSS/Atom feeds into deterministic RSS 2.0 files.""" from __future__ import annotations import argparse import calendar import hashlib import logging import os import re import sys import tempfile from dataclasses import dataclass, field from datetime import datetime, timezone from email.utils import format_datetime from pathlib import Path from typing import Any, Iterable from urllib.parse import parse_qsl, quote, quote_plus, urlencode, urlsplit, urlunsplit import xml.etree.ElementTree as ET import feedparser import requests import yaml from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry LOG = logging.getLogger("rss-merge") ATOM_NS = "http://www.w3.org/2005/Atom" CONTENT_NS = "http://purl.org/rss/1.0/modules/content/" RSSHUB_HOST = "rsshub.yuzhen.pub" RSSHUB_SECRET_ENV = "RSSHUB_ACCESS_KEY" ET.register_namespace("atom", ATOM_NS) ET.register_namespace("content", CONTENT_NS) @dataclass(frozen=True) class Source: url: str name: str = "" @dataclass class FeedItem: title: str link: str guid: str published: datetime | None description: str = "" author: str = "" categories: list[str] = field(default_factory=list) enclosure: dict[str, str] | None = None source_title: str = "" source_url: str = "" class ConfigError(ValueError): pass def load_config(path: Path) -> dict[str, Any]: try: raw = yaml.safe_load(path.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError) as exc: raise ConfigError(f"无法读取配置 {path}: {exc}") from exc if not isinstance(raw, dict): raise ConfigError("feeds.yml 顶层必须是对象") groups = raw.get("groups") if not isinstance(groups, dict) or not groups: raise ConfigError("feeds.yml 必须包含非空的 groups") for name, group in groups.items(): if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", name): raise ConfigError(f"分组名 {name!r} 只能包含字母、数字、下划线和连字符") if not isinstance(group, dict): raise ConfigError(f"分组 {name!r} 必须是对象") sources = group.get("sources") if not isinstance(sources, list) or not sources: raise ConfigError(f"分组 {name!r} 必须包含非空的 sources") parse_sources(sources, name) max_items = group.get("max_items", 200) if not isinstance(max_items, int) or isinstance(max_items, bool) or max_items < 1: raise ConfigError(f"分组 {name!r} 的 max_items 必须是正整数") return raw def parse_sources(values: list[Any], group_name: str) -> list[Source]: result: list[Source] = [] for value in values: if isinstance(value, str): url, name = value.strip(), "" elif isinstance(value, dict) and isinstance(value.get("url"), str): url = value["url"].strip() name = str(value.get("name", "")).strip() else: raise ConfigError(f"分组 {group_name!r} 中的源必须是 URL 字符串或包含 url 的对象") if not url.startswith(("http://", "https://")): raise ConfigError(f"分组 {group_name!r} 中的源不是 HTTP(S) URL: {url!r}") result.append(Source(url=url, name=name)) return result def make_session(user_agent: str, retries: int) -> requests.Session: retry = Retry( total=retries, connect=retries, read=retries, status=retries, backoff_factor=0.8, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(("GET",)), ) session = requests.Session() session.headers.update({"User-Agent": user_agent, "Accept": "application/rss+xml, application/atom+xml, application/xml, text/xml, */*;q=0.5"}) session.mount("http://", HTTPAdapter(max_retries=retry)) session.mount("https://", HTTPAdapter(max_retries=retry)) return session def is_protected_rsshub_url(url: str) -> bool: try: return (urlsplit(url).hostname or "").casefold() == RSSHUB_HOST except ValueError: return False def add_rsshub_access_key(url: str, access_key: str) -> str: """Add or replace the key query parameter without changing the configured URL.""" if not is_protected_rsshub_url(url): return url parsed = urlsplit(url) query = [(name, value) for name, value in parse_qsl(parsed.query, keep_blank_values=True) if name != "key"] query.append(("key", access_key)) # Never transmit the access key over plaintext HTTP. return urlunsplit(("https", parsed.netloc, parsed.path, urlencode(query), parsed.fragment)) def redact_secret(message: str, secret: str) -> str: if not secret: return message for value in {secret, quote(secret, safe=""), quote_plus(secret)}: if value: message = message.replace(value, "***") return message def fetch_feed( session: requests.Session, source: Source, timeout: float, max_bytes: int, rsshub_access_key: str, ) -> list[FeedItem]: request_url = add_rsshub_access_key(source.url, rsshub_access_key) with session.get(request_url, timeout=timeout, stream=True) as response: response.raise_for_status() chunks: list[bytes] = [] size = 0 for chunk in response.iter_content(chunk_size=64 * 1024): if not chunk: continue size += len(chunk) if size > max_bytes: raise ValueError(f"响应超过限制 {max_bytes} bytes") chunks.append(chunk) body = b"".join(chunks) parsed = feedparser.parse(body, response_headers=dict(response.headers)) if parsed.bozo and not parsed.entries: raise ValueError(f"Feed 解析失败: {parsed.bozo_exception}") source_title = source.name or clean_text(parsed.feed.get("title", "")) or source.url return [entry_to_item(entry, source_title, source.url) for entry in parsed.entries] def clean_text(value: Any) -> str: if value is None: return "" return str(value).strip() def parsed_datetime(entry: Any) -> datetime | None: value = entry.get("published_parsed") or entry.get("updated_parsed") or entry.get("created_parsed") if not value: return None try: return datetime.fromtimestamp(calendar.timegm(value), tz=timezone.utc) except (OverflowError, TypeError, ValueError): return None def entry_to_item(entry: Any, source_title: str, source_url: str) -> FeedItem: title = clean_text(entry.get("title")) or "(无标题)" link = clean_text(entry.get("link")) guid = clean_text(entry.get("id")) or link content = entry.get("content") or [] description = "" if content and isinstance(content[0], dict): description = clean_text(content[0].get("value")) description = description or clean_text(entry.get("summary")) or clean_text(entry.get("description")) if not guid: fingerprint = f"{title}\0{description}\0{entry.get('published', '')}" guid = "urn:rss-merge:" + hashlib.sha256(fingerprint.encode("utf-8")).hexdigest() categories = [] for tag in entry.get("tags", []) or []: if isinstance(tag, dict) and clean_text(tag.get("term")): categories.append(clean_text(tag["term"])) enclosure = None enclosures = entry.get("enclosures", []) or [] if enclosures and isinstance(enclosures[0], dict) and enclosures[0].get("href"): raw = enclosures[0] enclosure = { "url": clean_text(raw.get("href")), "type": clean_text(raw.get("type")) or "application/octet-stream", "length": clean_text(raw.get("length")) or "0", } return FeedItem( title=title, link=link, guid=guid, published=parsed_datetime(entry), description=description, author=clean_text(entry.get("author")), categories=categories, enclosure=enclosure, source_title=source_title, source_url=source_url, ) def sort_and_limit(items: Iterable[FeedItem], max_items: int) -> list[FeedItem]: ordered = sorted( items, key=lambda item: (item.published or datetime.min.replace(tzinfo=timezone.utc), item.guid), reverse=True, ) return ordered[:max_items] def add_text(parent: ET.Element, tag: str, text: str | None, **attrs: str) -> ET.Element | None: if text is None or text == "": return None child = ET.SubElement(parent, tag, attrs) child.text = text return child def build_rss(group_name: str, group: dict[str, Any], items: list[FeedItem]) -> bytes: root = ET.Element("rss", {"version": "2.0"}) channel = ET.SubElement(root, "channel") title = clean_text(group.get("title")) or group_name self_url = clean_text(group.get("self_url")) sources = parse_sources(group["sources"], group_name) channel_link = clean_text(group.get("link")) or self_url or sources[0].url add_text(channel, "title", title) add_text(channel, "link", channel_link) add_text(channel, "description", clean_text(group.get("description")) or f"Merged RSS feed: {title}") add_text(channel, "language", clean_text(group.get("language"))) add_text(channel, "generator", "rss-merge") if self_url: ET.SubElement(channel, f"{{{ATOM_NS}}}link", {"href": self_url, "rel": "self", "type": "application/rss+xml"}) dated_items = [item.published for item in items if item.published] if dated_items: add_text(channel, "lastBuildDate", format_datetime(max(dated_items), usegmt=True)) for value in items: node = ET.SubElement(channel, "item") add_text(node, "title", value.title) add_text(node, "link", value.link) guid = add_text(node, "guid", value.guid) if guid is not None: guid.set("isPermaLink", "true" if value.guid == value.link and bool(value.link) else "false") if value.published: add_text(node, "pubDate", format_datetime(value.published, usegmt=True)) add_text(node, "description", value.description) if value.description: add_text(node, f"{{{CONTENT_NS}}}encoded", value.description) add_text(node, "author", value.author) if value.source_title: add_text(node, "source", value.source_title, url=value.source_url) for category in value.categories: add_text(node, "category", category) if value.enclosure: ET.SubElement(node, "enclosure", value.enclosure) ET.indent(root, space=" ") return ET.tostring(root, encoding="utf-8", xml_declaration=True) def safe_output_path(repo_root: Path, group_name: str, group: dict[str, Any]) -> Path: relative = Path(clean_text(group.get("output")) or f"public/{group_name}.xml") if relative.is_absolute(): raise ConfigError(f"分组 {group_name!r} 的 output 必须是相对路径") output = (repo_root / relative).resolve() root = repo_root.resolve() if output == root or root not in output.parents: raise ConfigError(f"分组 {group_name!r} 的 output 超出仓库目录") return output def write_if_changed(path: Path, content: bytes) -> bool: if path.exists() and path.read_bytes() == content: return False path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False) as handle: temp_path = Path(handle.name) handle.write(content) temp_path.replace(path) return True def run(config_path: Path) -> int: config = load_config(config_path) settings = config.get("settings", {}) or {} if not isinstance(settings, dict): raise ConfigError("settings 必须是对象") timeout = float(settings.get("timeout", 20)) retries = int(settings.get("retries", 2)) max_bytes = int(settings.get("max_feed_bytes", 10 * 1024 * 1024)) allow_partial = settings.get("allow_partial", True) if timeout <= 0 or retries < 0 or max_bytes <= 0: raise ConfigError("settings.timeout 和 max_feed_bytes 必须大于 0,retries 不能小于 0") if not isinstance(allow_partial, bool): raise ConfigError("settings.allow_partial 必须是 true 或 false") user_agent = clean_text(settings.get("user_agent")) or ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36" ) groups: dict[str, dict[str, Any]] = config["groups"] rsshub_access_key = os.environ.get(RSSHUB_SECRET_ENV, "") repo_root = config_path.resolve().parent all_sources: dict[str, Source] = {} group_urls: dict[str, list[str]] = {} output_paths: dict[str, Path] = {} claimed_outputs: dict[Path, str] = {} for group_name, group in groups.items(): sources = parse_sources(group["sources"], group_name) group_urls[group_name] = [source.url for source in sources] output = safe_output_path(repo_root, group_name, group) if output in claimed_outputs: raise ConfigError(f"分组 {group_name!r} 和 {claimed_outputs[output]!r} 使用了同一个 output") claimed_outputs[output] = group_name output_paths[group_name] = output for source in sources: all_sources.setdefault(source.url, source) if any(is_protected_rsshub_url(url) for url in all_sources) and not rsshub_access_key: raise ConfigError(f"配置包含 {RSSHUB_HOST},但环境变量 {RSSHUB_SECRET_ENV} 未设置") session = make_session(user_agent, retries) # urllib3 的重试日志可能包含完整查询参数;避免将 access key 写入 Actions 日志。 logging.getLogger("urllib3").setLevel(logging.CRITICAL) fetched: dict[str, list[FeedItem]] = {} errors: dict[str, str] = {} for url, source in all_sources.items(): try: fetched[url] = fetch_feed(session, source, timeout, max_bytes, rsshub_access_key) LOG.info("已获取 %s(%d 条)", url, len(fetched[url])) except Exception as exc: # One source should produce a useful aggregated error. safe_error = redact_secret(str(exc), rsshub_access_key) errors[url] = safe_error LOG.error("获取失败 %s: %s", url, safe_error) if errors and not allow_partial: LOG.error("有 %d 个源失败;为避免临时丢失条目,本次不改写输出。可设置 settings.allow_partial: true 改变此行为。", len(errors)) return 1 changed = 0 generated = 0 for group_name, group in groups.items(): available_urls = [url for url in group_urls[group_name] if url in fetched] if not available_urls: LOG.error("分组 %s 没有成功获取的源,跳过输出", group_name) continue generated += 1 items = [item for url in available_urls for item in fetched[url]] merged = sort_and_limit(items, int(group.get("max_items", 200))) output = output_paths[group_name] if write_if_changed(output, build_rss(group_name, group, merged)): changed += 1 LOG.info("已生成 %s(%d 条)", output, len(merged)) else: LOG.info("无变化 %s(%d 条)", output, len(merged)) if generated == 0: LOG.error("没有任何分组成功生成") return 1 LOG.info("完成:%d 个文件有变化", changed) return 0 if not errors or allow_partial else 1 def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", type=Path, default=Path("feeds.yml"), help="配置文件路径(默认:feeds.yml)") parser.add_argument("--verbose", action="store_true", help="显示调试日志") args = parser.parse_args() logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)s %(message)s") try: return run(args.config) except ConfigError as exc: LOG.error("配置错误:%s", exc) return 2 except Exception: LOG.exception("生成 RSS 时发生未处理错误") return 1 if __name__ == "__main__": sys.exit(main())