别再手动攒早报了：一条 Python 流水线，抓取 → AI 摘要 → 自动归档

flowchart LR
    A["① 抓取 fetch<br />本地文件/URL 逐源容错<br />判定:单源失败不影响全局"] --> B["② 解析 parse<br />RSS/XML 取标题/链接/摘要<br />判定:缺字段的条目丢弃"]
    B --> C["③ 去重 dedupe<br />标题归一化比对<br />判定:同名只留第一条"]
    C --> D["④ 摘要 summarize<br />有 Key 走 AI 无 Key 降级<br />判定:模式必须可见不静默"]
    D --> E["⑤ 渲染归档 render<br />生成 Markdown 按日期存<br />判定:输出可复现"]

"""
news_pipeline.py —— 科技早报自动流水线（示例代码，已实测可运行）
流程：抓取源 → 解析去重 → 摘要（AI 或本地降级）→ 渲染 Markdown → 按日期归档
依赖：仅 Python 标准库；AI 摘要需自备 OpenAI 兼容 API Key（可选，无则自动降级）
"""
import argparse
import datetime as dt
import json
import os
import re
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path

DEFAULT_UA = "Mozilla/5.0 (compatible; NewsPipeline/1.0)"


# ---------- 1. 抓取：支持本地文件与 http(s) URL，逐源容错 ----------
def fetch_source(src: str) -> bytes:
    if src.startswith(("http://", "https://")):
        req = urllib.request.Request(src, headers={"User-Agent": DEFAULT_UA})
        with urllib.request.urlopen(req, timeout=15) as resp:
            return resp.read()
    return Path(src).read_bytes()  # 本地文件（离线/内网数据）


# ---------- 2. 解析：RSS 2.0 ----------
def parse_items(raw: bytes, source: str) -> list[dict]:
    root = ET.fromstring(raw)
    items = []
    for it in root.iter("item"):
        def text_of(tag):
            node = it.find(tag)
            return (node.text or "").strip() if node is not None else ""
        items.append({
            "title": text_of("title"),
            "link": text_of("link"),
            "pub": text_of("pubDate"),
            "desc": re.sub(r"<[^>]+>", "", text_of("description")),  # 去 HTML 标签
            "source": source,
        })
    return [i for i in items if i["title"]]


def dedupe(items: list[dict]) -> list[dict]:
    seen, out = set(), []
    for it in items:
        key = re.sub(r"\s+", "", it["title"])  # 标题归一化去重
        if key not in seen:
            seen.add(key)
            out.append(it)
    return out


# ---------- 3. 摘要：有 Key 走 AI，无 Key 本地降级（不静默） ----------
def ai_summarize(text: str, api_key: str, model: str) -> str:
    body = json.dumps({
        "model": model,
        "messages": [
            {"role": "system", "content": "用一句中文概括下面资讯的核心事实，不超过40字，不夸大。"},
            {"role": "user", "content": text},
        ],
        "temperature": 0.2,
    }).encode("utf-8")
    req = urllib.request.Request(
        "https://api.openai.com/v1/chat/completions",  # 换成你的 OpenAI 兼容端点
        data=body,
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"},
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    return data["choices"][0]["message"]["content"].strip()


def fallback_summarize(text: str) -> str:
    sents = re.split(r"[。！？]", text)
    head = next((s for s in sents if s.strip()), "")
    return head[:80] if head else text[:80]


def summarize(item: dict, api_key: str, model: str) -> tuple[str, str]:
    if api_key:
        try:
            return ai_summarize(item["title"] + "。" + item["desc"], api_key, model), "ai"
        except Exception as e:
            print(f"  [warn] AI 摘要失败，本条降级：{e}")
    return fallback_summarize(item["desc"] or item["title"]), "local"


# ---------- 4. 渲染 + 5. 归档 ----------
def render(items: list[dict], summaries: list[tuple[str, str]], out_dir: Path, mode_note: str) -> Path:
    today = dt.date.today().isoformat()
    lines = [f"# 科技早报 {today}", "", f"> 生成方式：{mode_note}", ""]
    for it, (summary, mode) in zip(items, summaries):
        lines += [
            f"## {it['title']}",
            f"- 来源：{it['source']}",
            f"- 链接：{it['link']}",
            f"- 摘要（{mode}）：{summary}",
            "",
        ]
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / f"科技早报-{today}.md"
    path.write_text("\n".join(lines), encoding="utf-8")
    return path


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--sources", required=True, help="逗号分隔的本地 rss 文件或 URL")
    ap.add_argument("--out", default="archive")
    ap.add_argument("--top", type=int, default=10)
    ap.add_argument("--api-key", default=os.environ.get("OPENAI_API_KEY", ""))
    ap.add_argument("--model", default="gpt-4o-mini")
    args = ap.parse_args()

    all_items = []
    for src in args.sources.split(","):
        src = src.strip()
        if not src:
            continue
        try:
            raw = fetch_source(src)
            got = parse_items(raw, src)
            print(f"[ok] 源 {src}：抓到 {len(got)} 条")
            all_items += got
        except Exception as e:
            print(f"[fail] 源 {src} 失败：{e}")  # 不静默：坏源要可见

    items = dedupe(all_items)[: args.top]
    results = [summarize(it, args.api_key, args.model) for it in items]
    mode_note = "AI 摘要（OpenAI 兼容 API）" if args.api_key else "本地降级摘要（未配置 API Key）"
    path = render(items, results, Path(args.out), mode_note)

    print(f"[done] 去重后 {len(items)} 条，模式：{mode_note}")
    print(f"[done] 早报已归档：{path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

$ python news_pipeline.py --sources sources/demo_a.rss --out archive
[ok] 源 sources/demo_a.rss：抓到 3 条
[done] 去重后 3 条，模式：本地降级摘要（未配置 API Key）
[done] 早报已归档：archive\科技早报-2026-10-17.md

# 科技早报 2026-10-17

> 生成方式：本地降级摘要（未配置 API Key）

## 示例：某开源框架发布 2.0 版本，启动速度提升明显
- 来源：sources/demo_a.rss
- 链接：https://example.com/demo-a/news/1
- 摘要（local）：演示条目：该版本重构了底层加载逻辑，官方称冷启动时间缩短约四成，并新增了插件热重载能力

flowchart TD
    E["早报不对劲"] --> Q1{"早报条数明显变少/为空?"}
    Q1 -->|是| C1["看抓取日志<br />某源 [fail] → 该源挂了<br />判定:坏源不静默,先修源或换源"]
    Q1 -->|否| Q2{"同一条新闻出现两遍?"}
    Q2 -->|是| C2["去重没覆盖<br />标题带空格/全半角差异<br />判定:归一化后再比对"]
    Q2 -->|否| Q3{"摘要质量差或全是 local?"}
    Q3 -->|是| C3["AI 模式没生效<br />Key 失效/限流 → 看 [warn]<br />判定:降级可见,先查 Key 再查余额"]
    Q3 -->|否| C4["渲染/归档问题<br />查 out 目录与文件时间<br />判定:输出可复现才算完"]

flowchart LR
    L1["第1级 手动脚本<br />你敲命令它跑<br />判定:代码实测跑通+日志可见"] --> L2["第2级 定时自动<br />WorkBuddy/QClaw 到点执行<br />判定:失败有推送,不静默"]
    L2 --> L3["第3级 沉淀知识库<br />早报进 ima 打标签<br />判定:攒一个月能问答检索"]
    L3 --> L4["第4级 多源+分类扩展<br />加源/加主题分类步骤<br />判定:每步仍可单独验证"]