412 lines
16 KiB
Python
412 lines
16 KiB
Python
#!/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())
|