@@ -0,0 +1,371 @@
|
||||
#!/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 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
|
||||
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/"
|
||||
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 fetch_feed(
|
||||
session: requests.Session,
|
||||
source: Source,
|
||||
timeout: float,
|
||||
max_bytes: int,
|
||||
) -> list[FeedItem]:
|
||||
with session.get(source.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"]
|
||||
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)
|
||||
|
||||
session = make_session(user_agent, retries)
|
||||
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)
|
||||
LOG.info("已获取 %s（%d 条）", url, len(fetched[url]))
|
||||
except Exception as exc: # One source should produce a useful aggregated error.
|
||||
errors[url] = str(exc)
|
||||
LOG.error("获取失败 %s: %s", url, exc)
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user