refactor: Phase 3 拆分 generate 流水线并补全测试

提取 collect/formatters/themes 等 pipeline 模块，新增 wecom 分条、RSS、delta、Agent 工作流测试。

Co-authored-by: Cursor <cursoragent@cursor.com>

daily/generate.py

@@ -2,476 +2,58 @@
 
 from __future__ import annotations
 
-import json
-import re
+import logging
 import sys
-import xml.etree.ElementTree as ET
-from collections import defaultdict
 from datetime import datetime, timezone, timedelta
 from pathlib import Path
-from typing import Any
 
-import certifi
-import httpx
-import logging
-
-logger = logging.getLogger(__name__)
-
-from daily.config import (
-    CACHE_DIR,
-    LOG_DIR,
-    OUTPUT_DIR,
-    SNAPSHOT_FILE,
-    env,
-    env_int,
-    full_desc_limit,
-    news_summary_limit,
-    wecom_skill_desc_limit,
-)
-from daily.format_wecom import build_wecom_report, finalize_wecom_skill_groups, replace_wecom_skill_sections
 from daily.agent_workflow import is_agent_mode, run_agent_workflow
-from daily.delta import compare_depth
+from daily.config import LOG_DIR, OUTPUT_DIR, env
 from daily.cursor_editor import (
     apply_descriptions,
     is_enabled as cursor_editor_enabled,
     run_editorial,
     theme_line_from_editorial,
 )
-from daily.github.auth import github_html_headers
-from daily.github.search import fetch_emerging_repos, fetch_topic_hot_repos
-from daily.github.trending import fetch_github_trending, trending_data_source_note
-from daily.localize import LocalizeJob, localize_descriptions, needs_chinese
+from daily.format_wecom import build_wecom_report, finalize_wecom_skill_groups, replace_wecom_skill_sections
+from daily.github.trending import trending_data_source_note
 from daily.news.fetch import (
-    fetch_ai_news,
-    fetch_cn_ai_news,
     format_cn_news_section,
     format_news_section,
     prepare_wecom_cn_news_items,
     prepare_wecom_news_items,
 )
-from daily.news.rank import apply_news_ranking
-from daily.report_data import (
-    build_full_payload,
-    build_llm_input,
-    data_json_path,
-    save_json,
+from daily.pipeline.collect import collect_report_context
+from daily.pipeline.formatters import (
+    build_highlights,
+    fetch_latest_release_title,
+    format_github_repo_section,
+    format_skill_section,
+    prepare_github_item,
+    prepare_skill_item,
 )
-from daily.skills_board import load_boards
+from daily.pipeline.localize import localize_descriptions_in_place
+from daily.pipeline.snapshot import load_snapshot, save_snapshot
+from daily.pipeline.themes import detect_theme_line, theme_clusters
+from daily.report_data import build_full_payload, data_json_path, save_json
 from daily.skills_group import group_skills_by_source
-from shared.skills_data import format_installs, load_feed
 
-THEME_RULES: list[tuple[str, str, list[str]]] = [
-    ("🎬", "AI 多媒体 / 视频", ["runcomfy", "remotion", "video", "seedance", "inpaint", "lipsync"]),
-    ("🔧", "工程协作 / Skill 元能力", ["grill", "tdd", "architecture", "find-skills", "to-issues"]),
-    ("📱", "飞书 / Lark", ["lark", "feishu"]),
-    ("📣", "内容营销", ["viral", "tiktok", "instagram", "reels"]),
-    ("🎨", "设计 / 前端", ["frontend", "design", "ui-ux", "tailwind"]),
-]
+logger = logging.getLogger(__name__)
 
 
 def _now_cst() -> datetime:
     return datetime.now(timezone(timedelta(hours=8)))
 
 
-def _short_desc(text: str, limit: int = 72) -> str:
-    text = re.sub(r"\s+", " ", text or "").strip()
-    if limit <= 0 or len(text) <= limit:
-        return text
-    return text[: limit - 3] + "..."
-
-
-def _archive_desc(text: str) -> str:
-    return _short_desc(text, full_desc_limit())
-
-
-def _wecom_desc(text: str, limit: int = 36) -> str:
-    return _short_desc(text, limit)
-
-
-def _localize_descriptions_in_place(
-    trending: list[dict[str, Any]],
-    hot: list[dict[str, Any]],
-    github_trending: list[dict[str, Any]],
-    github_emerging: list[dict[str, Any]],
-    github_topic: list[dict[str, Any]],
-    ai_news: dict[str, Any],
-    cn_ai_news: dict[str, Any] | None = None,
-) -> None:
-    full_limit = full_desc_limit()
-    news_limit = news_summary_limit()
-    jobs: list[LocalizeJob] = []
-    seen_skill: set[str] = set()
-    for item in trending + hot:
-        sid = _skill_id(item)
-        if sid in seen_skill:
-            continue
-        seen_skill.add(sid)
-        desc = (item.get("description") or "").strip()
-        if desc:
-            jobs.append(LocalizeJob(f"skill:{sid}", desc, full_limit))
-
-    seen_repo: set[str] = set()
-    for repo_list in (github_trending, github_emerging, github_topic):
-        for item in repo_list:
-            repo = item.get("repo", "")
-            if not repo or repo in seen_repo:
-                continue
-            seen_repo.add(repo)
-            desc = (item.get("description") or "").strip()
-            if desc:
-                jobs.append(LocalizeJob(f"github:{repo}", desc, full_limit))
-
-    if ai_news.get("enabled"):
-        seen_news: set[str] = set()
-        for item in ai_news.get("flat") or []:
-            link = item.get("link", "")
-            if not link or link in seen_news:
-                continue
-            seen_news.add(link)
-            summary = (item.get("summary") or "").strip()
-            if summary:
-                jobs.append(LocalizeJob(f"news:{link}", summary, news_limit))
-
-    zh_map = localize_descriptions(jobs, archive=True)
-    if not zh_map and not jobs:
-        return
-
-    def _apply_zh(mapping: dict[str, str]) -> None:
-        for item in trending + hot:
-            key = f"skill:{_skill_id(item)}"
-            if key in mapping:
-                item["description"] = mapping[key]
-        for repo_list in (github_trending, github_emerging, github_topic):
-            for item in repo_list:
-                key = f"github:{item.get('repo', '')}"
-                if key in mapping:
-                    item["description"] = mapping[key]
-        if ai_news.get("enabled"):
-            for cat in ai_news.get("categories") or []:
-                for item in cat.get("items") or []:
-                    key = f"news:{item.get('link', '')}"
-                    if key in mapping:
-                        item["summary"] = mapping[key]
-            for item in ai_news.get("flat") or []:
-                key = f"news:{item.get('link', '')}"
-                if key in mapping:
-                    item["summary"] = mapping[key]
-
-    _apply_zh(zh_map)
-
-    # 仍为英文的条目再译一轮（长描述或批次失败时）
-    retry_jobs: list[LocalizeJob] = []
-    seen_skill.clear()
-    for item in trending + hot:
-        sid = _skill_id(item)
-        if sid in seen_skill:
-            continue
-        seen_skill.add(sid)
-        desc = (item.get("description") or "").strip()
-        if needs_chinese(desc):
-            retry_jobs.append(LocalizeJob(f"skill:{sid}", desc, full_limit))
-    seen_repo.clear()
-    for repo_list in (github_trending, github_emerging, github_topic):
-        for item in repo_list:
-            repo = item.get("repo", "")
-            if not repo or repo in seen_repo:
-                continue
-            seen_repo.add(repo)
-            desc = (item.get("description") or "").strip()
-            if needs_chinese(desc):
-                retry_jobs.append(LocalizeJob(f"github:{repo}", desc, full_limit))
-    if ai_news.get("enabled"):
-        seen_news.clear()
-        for item in ai_news.get("flat") or []:
-            link = item.get("link", "")
-            if not link or link in seen_news:
-                continue
-            seen_news.add(link)
-            summary = (item.get("summary") or "").strip()
-            if needs_chinese(summary):
-                retry_jobs.append(LocalizeJob(f"news:{link}", summary, news_limit))
-
-    if retry_jobs:
-        _apply_zh(localize_descriptions(retry_jobs, archive=True))
-
-
-def _skill_id(item: dict[str, Any]) -> str:
-    return str(item.get("id") or f"{item.get('source')}/{item.get('title')}")
-
-
-def _load_snapshot() -> set[str]:
-    if not SNAPSHOT_FILE.exists():
-        return set()
-    try:
-        data = json.loads(SNAPSHOT_FILE.read_text(encoding="utf-8"))
-        return set(str(x) for x in (data.get("skill_ids") or []))
-    except (OSError, json.JSONDecodeError):
-        return set()
-
-
-def _save_snapshot(feed: dict[str, Any], date_str: str) -> None:
-    CACHE_DIR.mkdir(parents=True, exist_ok=True)
-    ids: list[str] = []
-    for board in ("topTrending", "topHot"):
-        for item in feed.get(board, [])[:20]:
-            sid = _skill_id(item)
-            if sid not in ids:
-                ids.append(sid)
-    SNAPSHOT_FILE.write_text(
-        json.dumps({"date": date_str, "skill_ids": ids}, ensure_ascii=False, indent=2),
-        encoding="utf-8",
-    )
-
-
-def _prepare_skill_item(item: dict[str, Any], prev_ids: set[str], rank: int) -> dict[str, Any]:
-    badge = ""
-    sid = _skill_id(item)
-    if sid not in prev_ids and prev_ids:
-        badge = "🆕"
-    elif rank == 1:
-        badge = "👑"
-    installs_fmt = item.get("installs_fmt") or format_installs(item.get("installs", 0))
-    title = item.get("source", "?") if item.get("cluster") else item.get("title", "?")
-    desc = item.get("wecom_desc") or item.get("description") or item.get("cluster_titles") or ""
-    limit = wecom_skill_desc_limit()
-    desc_short = desc if item.get("wecom_desc") or limit <= 0 else _wecom_desc(desc, limit)
-    return {
-        "title": title,
-        "source": item.get("source", "?"),
-        "installs_fmt": installs_fmt,
-        "link": item.get("link", ""),
-        "desc_short": desc_short,
-        "badge": badge,
-        "cluster": bool(item.get("cluster")),
-        "cluster_count": item.get("cluster_count"),
-        "cluster_titles": item.get("cluster_titles"),
-    }
-
-
-def _detect_theme_line(feed: dict[str, Any]) -> str:
-    scores: dict[str, int] = defaultdict(int)
-    for board in ("topTrending", "topHot"):
-        for rank, item in enumerate(feed.get(board, [])[:10], 1):
-            haystack = " ".join(
-                [item.get("title", ""), item.get("source", ""), item.get("description", "")]
-            ).lower()
-            for _icon, label, keywords in THEME_RULES:
-                if any(k in haystack for k in keywords):
-                    scores[label] += max(1, 11 - rank)
-                    break
-    if not scores:
-        return "**今日主题**：Agent Skills 生态持续活跃"
-    return f"**今日主题**：{max(scores.items(), key=lambda x: x[1])[0]}"
-
-
-def _build_highlights(
-    trending: list[dict[str, Any]],
-    hot: list[dict[str, Any]],
-    github_trending: list[dict[str, Any]],
-    github_emerging: list[dict[str, Any]],
-    ai_news: dict[str, Any] | None = None,
-    cn_ai_news: dict[str, Any] | None = None,
-) -> list[str]:
-    points: list[str] = []
-    if ai_news and ai_news.get("enabled"):
-        top_news = prepare_wecom_news_items(ai_news)
-        if top_news:
-            n0 = top_news[0]
-            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
-            points.append(
-                f"🌍 AI 时讯 [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
-            )
-        elif ai_news.get("flat"):
-            n0 = ai_news["flat"][0]
-            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
-            points.append(f"🌍 AI 时讯 [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）")
-    if cn_ai_news and cn_ai_news.get("enabled"):
-        top_cn = prepare_wecom_cn_news_items(cn_ai_news)
-        if top_cn:
-            n0 = top_cn[0]
-            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
-            points.append(
-                f"🇨🇳 国内 AI [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
-            )
-        elif cn_ai_news.get("flat"):
-            n0 = cn_ai_news["flat"][0]
-            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
-            points.append(
-                f"🇨🇳 国内 AI [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
-            )
-    if trending:
-        t0 = trending[0]
-        points.append(f"📈 Skills 榜首 **{t0.get('title')}**（{format_installs(t0.get('installs', 0))}）")
-    if github_trending:
-        g0 = github_trending[0]
-        stars = g0.get("stars_today_fmt", "")
-        total = g0.get("total_stars_fmt", "")
-        star_hint = f"+{stars} today · " if stars else (f"⭐{total} · " if total else "")
-        points.append(f"🐙 GitHub Trending [{g0['repo']}]({g0['url']})（{star_hint}{g0.get('language', '')}）")
-    if github_emerging:
-        e0 = github_emerging[0]
-        points.append(f"🌱 新兴 [{e0['repo']}]({e0['url']})（⭐ {e0.get('total_stars_fmt', '?')}）")
-    elif hot:
-        h0 = hot[0]
-        points.append(f"🔥 Skills Hot 榜首 **{h0.get('title')}**（1H {format_installs(h0.get('installs', 0))}）")
-    while len(points) < 3 and len(trending) > len(points):
-        item = trending[len(points)]
-        points.append(f"✨ **{item.get('title')}** · `{item.get('source')}`")
-    return points[:3]
-
-
-def _prepare_github_item(item: dict[str, Any]) -> dict[str, Any]:
-    return {**item, "desc_short": _wecom_desc(item.get("description", ""), 40)}
-
-
-def _fetch_latest_release_title(repo: str) -> str | None:
-    atom_url = f"https://github.com/{repo}/releases.atom"
-    try:
-        with httpx.Client(
-            timeout=12.0,
-            verify=certifi.where(),
-            follow_redirects=True,
-            headers=github_html_headers(),
-        ) as client:
-            resp = client.get(atom_url)
-            if resp.status_code != 200:
-                return None
-            root = ET.fromstring(resp.text)
-            ns = {"a": "http://www.w3.org/2005/Atom"}
-            entry = root.find("a:entry", ns)
-            if entry is None:
-                return None
-            title = entry.find("a:title", ns)
-            return title.text.strip() if title is not None and title.text else None
-    except Exception:
-        return None
-
-
-def _theme_clusters(feed: dict[str, Any], limit: int = 5) -> list[tuple[str, list[str]]]:
-    buckets: dict[str, list[str]] = defaultdict(list)
-    seen: set[str] = set()
-    for board in ("topTrending", "topHot"):
-        for item in feed.get(board, [])[:20]:
-            item_id = _skill_id(item)
-            if item_id in seen:
-                continue
-            seen.add(item_id)
-            haystack = " ".join(
-                [item.get("title", ""), item.get("source", ""), item.get("description", "")]
-            ).lower()
-            for _icon, theme, keywords in THEME_RULES:
-                if any(k in haystack for k in keywords):
-                    label = f"**{item.get('title')}** (`{item.get('source')}`)"
-                    if label not in buckets[theme]:
-                        buckets[theme].append(label)
-                    break
-    return [(theme, examples[:limit]) for theme, examples in buckets.items() if examples]
-
-
-def _format_github_repo_section(repos: list[dict[str, Any]], *, show_created: bool = False) -> list[str]:
-    lines: list[str] = []
-    for i, repo in enumerate(repos, 1):
-        lang = repo.get("language") or "—"
-        stars_today = repo.get("stars_today_fmt") or ""
-        total = repo.get("total_stars_fmt") or ""
-        created = repo.get("created_at") or ""
-        meta_parts = [lang]
-        if stars_today:
-            meta_parts.append(f"+{stars_today} today")
-        if total:
-            meta_parts.append(f"总 ⭐ {total}")
-        if show_created and created:
-            meta_parts.append(f"创建于 {created}")
-        lines.append(f"{i}. **[{repo['repo']}]({repo['url']})** · {' · '.join(meta_parts)}")
-        desc = _archive_desc(repo.get("description", ""))
-        if desc:
-            lines.append(f"   - {desc}")
-        lines.append("")
-    return lines
-
-
-def _format_skill_section(items: list[dict[str, Any]], *, hot: bool = False) -> list[str]:
-    lines: list[str] = []
-    for i, item in enumerate(items, 1):
-        skill_id = item.get("id") or f"{item.get('source', '?')}/{item.get('title', '?')}"
-        link = item.get("link", "")
-        installs = format_installs(item.get("installs", 0))
-        meta = f"1H {installs}" if hot else f"总安装 {installs}"
-        if link:
-            lines.append(f"{i}. **[{skill_id}]({link})** · {meta}")
-        else:
-            lines.append(f"{i}. **{skill_id}** · {meta}")
-        desc = _archive_desc(item.get("description", ""))
-        if desc:
-            lines.append(f"   - {desc}")
-        lines.append("")
-    return lines
-
-
 def generate_report() -> tuple[str, str, Path, Path]:
-    trending_n = env_int("DAILY_TRENDING_LIMIT", 150)
-    hot_n = max(env_int("DAILY_HOT_LIMIT", 150), compare_depth())
-    compare_n = compare_depth()
-    skill_pool = max(10, env_int("DAILY_WECOM_SKILL_POOL", 200))
-    wecom_trending = env_int("DAILY_WECOM_TRENDING", 10)
-    wecom_hot = env_int("DAILY_WECOM_HOT", 10)
-    github_limit = env_int("DAILY_GITHUB_TRENDING_LIMIT", 10)
-    wecom_github = env_int("DAILY_WECOM_GITHUB_TRENDING", env_int("DAILY_WECOM_REPOS", 10))
-    github_fetch_n = max(github_limit, compare_n, wecom_github)
-    emerging_limit = env_int("DAILY_GITHUB_EMERGING_LIMIT", 10)
-    wecom_emerging = env_int("DAILY_WECOM_GITHUB_EMERGING", 10)
-    emerging_fetch_n = max(emerging_limit, compare_n, wecom_emerging)
-    topic_limit = env_int("DAILY_GITHUB_TOPIC_LIMIT", 10)
-    wecom_topic = env_int("DAILY_WECOM_GITHUB_TOPIC", 10)
-    topic_fetch_n = max(topic_limit, compare_n, wecom_topic)
-
-    feed = load_feed(force=True)
-    prev_ids = _load_snapshot()
+    ctx = collect_report_context(_now_cst())
     now = _now_cst()
-    date_str = now.strftime("%Y-%m-%d")
-    time_str = now.strftime("%H:%M") + " (UTC+8)"
-    updated = (feed.get("updatedAt") or "")[:10]
+    prev_ids = load_snapshot()
 
-    trending, hot = load_boards(feed, trending_limit=trending_n, hot_limit=hot_n)
-    github_trending = fetch_github_trending(github_fetch_n)
-    seen_repos = {r["repo"] for r in github_trending}
-    github_emerging = fetch_emerging_repos(emerging_fetch_n, exclude=seen_repos)
-    seen_repos.update(r["repo"] for r in github_emerging)
-    topic_name, github_topic = fetch_topic_hot_repos(topic_fetch_n, exclude=seen_repos)
-    ai_news = fetch_ai_news()
-    cn_ai_news = fetch_cn_ai_news()
-    ai_news = apply_news_ranking(ai_news, date_str=date_str)
-    cn_ai_news = apply_news_ranking(cn_ai_news, date_str=date_str)
-
-    wecom_limits = {
-        "trending": wecom_trending,
-        "hot": wecom_hot,
-        "trending_pool": skill_pool,
-        "hot_pool": skill_pool,
-        "github": wecom_github,
-        "emerging": wecom_emerging,
-        "topic": wecom_topic,
-        "ai_news": env_int("DAILY_WECOM_AI_NEWS", 10),
-        "cn_ai_news": env_int("DAILY_WECOM_CN_AI_NEWS", 8),
-    }
-    llm_input = build_llm_input(
-        date_str=date_str,
-        updated=updated,
-        trending=trending,
-        hot=hot,
-        github_trending=github_trending,
-        github_emerging=github_emerging,
-        github_topic=github_topic,
-        topic_name=topic_name,
-        ai_news=ai_news,
-        cn_ai_news=cn_ai_news,
-        wecom_limits=wecom_limits,
-        agent_mode=is_agent_mode(),
-    )
     save_json(
-        data_json_path(date_str),
+        data_json_path(ctx.date_str),
         build_full_payload(
-            llm_input,
+            ctx.llm_input,
             meta={
                 "generated_at": now.isoformat(),
                 "report_mode": "agent" if is_agent_mode() else "classic",
@@ -483,10 +65,10 @@ def generate_report() -> tuple[str, str, Path, Path]:
     agent_wecom: str | None = None
     if is_agent_mode():
         agent_wecom = run_agent_workflow(
-            llm_input,
-            date_str=date_str,
-            time_str=time_str,
-            updated=updated,
+            ctx.llm_input,
+            date_str=ctx.date_str,
+            time_str=ctx.time_str,
+            updated=ctx.updated,
         )
         if not agent_wecom:
             logger.warning("Agent 工作流失败，回退 classic 模式")
@@ -494,79 +76,88 @@ def generate_report() -> tuple[str, str, Path, Path]:
     editorial_theme: str | None = None
     editorial_highlights: list[str] | None = None
     if agent_wecom is None:
-        editorial = run_editorial(llm_input, date_str=date_str)
+        editorial = run_editorial(ctx.llm_input, date_str=ctx.date_str)
         if editorial:
             apply_descriptions(
-                trending=trending,
-                hot=hot,
-                github_trending=github_trending,
-                github_emerging=github_emerging,
-                github_topic=github_topic,
-                ai_news=ai_news,
-                cn_ai_news=cn_ai_news,
+                trending=ctx.trending,
+                hot=ctx.hot,
+                github_trending=ctx.github_trending,
+                github_emerging=ctx.github_emerging,
+                github_topic=ctx.github_topic,
+                ai_news=ctx.ai_news,
+                cn_ai_news=ctx.cn_ai_news,
                 descriptions=editorial.get("descriptions") or {},
             )
             editorial_theme = theme_line_from_editorial(editorial) or None
             hl = editorial.get("highlights") or []
             editorial_highlights = hl if hl else None
 
-    # 完整版归档：中文化 + 加长摘要（已是中文的条目会跳过翻译）
-    _localize_descriptions_in_place(
-        trending, hot, github_trending, github_emerging, github_topic, ai_news, cn_ai_news
+    localize_descriptions_in_place(
+        ctx.trending,
+        ctx.hot,
+        ctx.github_trending,
+        ctx.github_emerging,
+        ctx.github_topic,
+        ctx.ai_news,
+        ctx.cn_ai_news,
     )
 
-    themes = _theme_clusters(feed)
-
+    limits = ctx.limits
+    themes = theme_clusters(ctx.feed)
     lines = [
-        f"# 早报 · {date_str}",
+        f"# 早报 · {ctx.date_str}",
         "",
         f"> 生成时间：{now.strftime('%Y-%m-%d %H:%M')} (UTC+8)  ",
-        f"> skills 数据更新：{updated}  ",
+        f"> skills 数据更新：{ctx.updated}  ",
         "> 数据来源：[skills.sh/trending](https://skills.sh/trending) · [skills.sh/hot](https://skills.sh/hot) · 国际/国内 AI RSS",
         "",
         "---",
         "",
-        f"## 一、Skills Trending Top {trending_n}",
+        f"## 一、Skills Trending Top {limits.trending_n}",
         "",
-        *_format_skill_section(trending),
+        *format_skill_section(ctx.trending),
         "---",
         "",
-        f"## 二、Skills Hot Top {hot_n}",
+        f"## 二、Skills Hot Top {limits.hot_n}",
         "",
-        *_format_skill_section(hot, hot=True),
+        *format_skill_section(ctx.hot, hot=True),
         "",
         "---",
         "",
-        f"## 三、GitHub Trending Top {github_limit}",
+        f"## 三、GitHub Trending Top {limits.github_limit}",
         "",
         trending_data_source_note(),
         "",
     ]
 
-    if github_trending:
-        lines.extend(_format_github_repo_section(github_trending))
+    if ctx.github_trending:
+        lines.extend(format_github_repo_section(ctx.github_trending))
     else:
         lines.append("*GitHub Trending 获取失败，请检查网络或配置 GITHUB_TOKEN。*")
         lines.append("")
 
-    lines.extend(["---", "", f"## 四、新兴项目 Top {emerging_limit}", "", "> 数据来源：GitHub Search API（需 `GITHUB_TOKEN`）", ""])
-    if github_emerging:
-        lines.extend(_format_github_repo_section(github_emerging, show_created=True))
+    lines.extend(
+        ["---", "", f"## 四、新兴项目 Top {limits.emerging_limit}", "", "> 数据来源：GitHub Search API（需 `GITHUB_TOKEN`）", ""]
+    )
+    if ctx.github_emerging:
+        lines.extend(format_github_repo_section(ctx.github_emerging, show_created=True))
     else:
         lines.append("*新兴项目获取失败或未配置 GITHUB_TOKEN。*")
         lines.append("")
 
-    lines.extend(["---", "", f"## 五、Topic `{topic_name}` Top {topic_limit}", "", "> 数据来源：GitHub Search API（需 `GITHUB_TOKEN`）", ""])
-    if github_topic:
-        lines.extend(_format_github_repo_section(github_topic))
+    lines.extend(
+        ["---", "", f"## 五、Topic `{ctx.topic_name}` Top {limits.topic_limit}", "", "> 数据来源：GitHub Search API（需 `GITHUB_TOKEN`）", ""]
+    )
+    if ctx.github_topic:
+        lines.extend(format_github_repo_section(ctx.github_topic))
     else:
-        lines.append(f"*Topic `{topic_name}` 热点获取失败或未配置 GITHUB_TOKEN。*")
+        lines.append(f"*Topic `{ctx.topic_name}` 热点获取失败或未配置 GITHUB_TOKEN。*")
         lines.append("")
 
     section_no = 6
-    lines.extend(format_news_section(ai_news, section_no=section_no))
+    lines.extend(format_news_section(ctx.ai_news, section_no=section_no))
     section_no += 1
-    lines.extend(format_cn_news_section(cn_ai_news, section_no=section_no))
+    lines.extend(format_cn_news_section(ctx.cn_ai_news, section_no=section_no))
     section_no += 1
 
     watch = (env("GITHUB_REPOS") or "").strip()
@@ -574,7 +165,7 @@ def generate_report() -> tuple[str, str, Path, Path]:
         lines.extend(["---", "", f"## {section_no}、关注仓库 Release", ""])
         section_no += 1
         for repo in [r.strip() for r in watch.split(",") if r.strip()]:
-            release = _fetch_latest_release_title(repo)
+            release = fetch_latest_release_title(repo)
             lines.append(f"- **{repo}**：{release or '暂无 release'}")
         lines.append("")
 
@@ -585,8 +176,8 @@ def generate_report() -> tuple[str, str, Path, Path]:
             lines.append(f"- {ex}")
         lines.append("")
 
-    pick_src = trending[0].get("source", "") if trending else ""
-    pick_name = trending[0].get("title", "") if trending else ""
+    pick_src = ctx.trending[0].get("source", "") if ctx.trending else ""
+    pick_name = ctx.trending[0].get("title", "") if ctx.trending else ""
     pick_command = (
         f"npx skills add {pick_src}/{pick_name}"
         if pick_src and pick_name
@@ -594,56 +185,69 @@ def generate_report() -> tuple[str, str, Path, Path]:
     )
 
     lines.extend(["---", "", "## 安装示例", "", "```bash"])
-    for item in trending[:4]:
+    for item in ctx.trending[:4]:
         src, name = item.get("source", ""), item.get("title", "")
         if src and name:
             lines.append(f"npx skills add {src}/{name}")
-    lines.extend(["```", "", f"*企微短版见 `output/{date_str}.wecom.md`*"])
+    lines.extend(["```", "", f"*企微短版见 `output/{ctx.date_str}.wecom.md`*"])
 
     markdown = "\n".join(lines)
     if agent_wecom:
-        gt = group_skills_by_source(trending, limit=wecom_trending, pool_size=skill_pool)
-        gh = group_skills_by_source(hot, limit=wecom_hot, pool_size=skill_pool)
+        gt = group_skills_by_source(
+            ctx.trending, limit=limits.wecom_trending, pool_size=limits.skill_pool
+        )
+        gh = group_skills_by_source(ctx.hot, limit=limits.wecom_hot, pool_size=limits.skill_pool)
         wecom_md = replace_wecom_skill_sections(agent_wecom, trending=gt, hot=gh)
     else:
         wecom_md = build_wecom_report(
-            date_str=date_str,
-            time_str=time_str,
-            updated=updated,
+            date_str=ctx.date_str,
+            time_str=ctx.time_str,
+            updated=ctx.updated,
             highlights=editorial_highlights
-            or _build_highlights(trending, hot, github_trending, github_emerging, ai_news, cn_ai_news),
-            theme_line=editorial_theme or _detect_theme_line(feed),
-            ai_news=prepare_wecom_news_items(ai_news),
-            cn_ai_news=prepare_wecom_cn_news_items(cn_ai_news),
+            or build_highlights(
+                ctx.trending,
+                ctx.hot,
+                ctx.github_trending,
+                ctx.github_emerging,
+                ctx.ai_news,
+                ctx.cn_ai_news,
+            ),
+            theme_line=editorial_theme or detect_theme_line(ctx.feed),
+            ai_news=prepare_wecom_news_items(ctx.ai_news),
+            cn_ai_news=prepare_wecom_cn_news_items(ctx.cn_ai_news),
             trending=[
-                _prepare_skill_item(item, prev_ids, r)
+                prepare_skill_item(item, prev_ids, r)
                 for r, item in enumerate(
                     finalize_wecom_skill_groups(
-                        group_skills_by_source(trending, limit=wecom_trending, pool_size=skill_pool)
+                        group_skills_by_source(
+                            ctx.trending, limit=limits.wecom_trending, pool_size=limits.skill_pool
+                        )
                     ),
                     1,
                 )
             ],
             hot=[
-                _prepare_skill_item(item, prev_ids, r)
+                prepare_skill_item(item, prev_ids, r)
                 for r, item in enumerate(
                     finalize_wecom_skill_groups(
-                        group_skills_by_source(hot, limit=wecom_hot, pool_size=skill_pool)
+                        group_skills_by_source(
+                            ctx.hot, limit=limits.wecom_hot, pool_size=limits.skill_pool
+                        )
                     ),
                     1,
                 )
             ],
-            repos=[_prepare_github_item(item) for item in github_trending[:wecom_github]],
-            emerging=[_prepare_github_item(item) for item in github_emerging[:wecom_emerging]],
-            topic_name=topic_name,
-            topic_repos=[_prepare_github_item(item) for item in github_topic[:wecom_topic]],
+            repos=[prepare_github_item(item) for item in ctx.github_trending[: limits.wecom_github]],
+            emerging=[prepare_github_item(item) for item in ctx.github_emerging[: limits.wecom_emerging]],
+            topic_name=ctx.topic_name,
+            topic_repos=[prepare_github_item(item) for item in ctx.github_topic[: limits.wecom_topic]],
             pick_command=pick_command,
         )
 
-    _save_snapshot(feed, date_str)
+    save_snapshot(ctx.feed, ctx.date_str)
     OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
-    out_md = OUTPUT_DIR / f"{date_str}.md"
-    out_wecom = OUTPUT_DIR / f"{date_str}.wecom.md"
+    out_md = OUTPUT_DIR / f"{ctx.date_str}.md"
+    out_wecom = OUTPUT_DIR / f"{ctx.date_str}.wecom.md"
     out_md.write_text(markdown, encoding="utf-8")
     out_wecom.write_text(wecom_md, encoding="utf-8")
     return markdown, wecom_md, out_md, out_wecom

daily/pipeline/__init__.py

@@ -0,0 +1 @@
+"""早报生成流水线子模块。"""

daily/pipeline/collect.py

@@ -0,0 +1,145 @@
+"""抓取与结构化输入组装。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from daily.agent_workflow import is_agent_mode
+from daily.config import env_int
+from daily.delta import compare_depth
+from daily.github.search import fetch_emerging_repos, fetch_topic_hot_repos
+from daily.github.trending import fetch_github_trending
+from daily.news.fetch import fetch_ai_news, fetch_cn_ai_news
+from daily.news.rank import apply_news_ranking
+from daily.report_data import build_llm_input
+from daily.skills_board import load_boards
+from shared.skills_data import load_feed
+
+
+@dataclass
+class ReportLimits:
+    trending_n: int
+    hot_n: int
+    skill_pool: int
+    wecom_trending: int
+    wecom_hot: int
+    github_limit: int
+    wecom_github: int
+    emerging_limit: int
+    wecom_emerging: int
+    topic_limit: int
+    wecom_topic: int
+
+
+@dataclass
+class ReportContext:
+    feed: dict[str, Any]
+    date_str: str
+    time_str: str
+    updated: str
+    trending: list[dict[str, Any]]
+    hot: list[dict[str, Any]]
+    github_trending: list[dict[str, Any]]
+    github_emerging: list[dict[str, Any]]
+    github_topic: list[dict[str, Any]]
+    topic_name: str
+    ai_news: dict[str, Any]
+    cn_ai_news: dict[str, Any]
+    limits: ReportLimits
+    wecom_limits: dict[str, int]
+    llm_input: dict[str, Any]
+
+
+def resolve_limits() -> ReportLimits:
+    compare_n = compare_depth()
+    trending_n = env_int("DAILY_TRENDING_LIMIT", 150)
+    hot_n = max(env_int("DAILY_HOT_LIMIT", 150), compare_n)
+    skill_pool = max(10, env_int("DAILY_WECOM_SKILL_POOL", 200))
+    wecom_trending = env_int("DAILY_WECOM_TRENDING", 10)
+    wecom_hot = env_int("DAILY_WECOM_HOT", 10)
+    github_limit = env_int("DAILY_GITHUB_TRENDING_LIMIT", 10)
+    wecom_github = env_int("DAILY_WECOM_GITHUB_TRENDING", env_int("DAILY_WECOM_REPOS", 10))
+    emerging_limit = env_int("DAILY_GITHUB_EMERGING_LIMIT", 10)
+    wecom_emerging = env_int("DAILY_WECOM_GITHUB_EMERGING", 10)
+    topic_limit = env_int("DAILY_GITHUB_TOPIC_LIMIT", 10)
+    wecom_topic = env_int("DAILY_WECOM_GITHUB_TOPIC", 10)
+    return ReportLimits(
+        trending_n=trending_n,
+        hot_n=hot_n,
+        skill_pool=skill_pool,
+        wecom_trending=wecom_trending,
+        wecom_hot=wecom_hot,
+        github_limit=github_limit,
+        wecom_github=wecom_github,
+        emerging_limit=emerging_limit,
+        wecom_emerging=wecom_emerging,
+        topic_limit=topic_limit,
+        wecom_topic=wecom_topic,
+    )
+
+
+def collect_report_context(now: datetime) -> ReportContext:
+    limits = resolve_limits()
+    compare_n = compare_depth()
+    github_fetch_n = max(limits.github_limit, compare_n, limits.wecom_github)
+    emerging_fetch_n = max(limits.emerging_limit, compare_n, limits.wecom_emerging)
+    topic_fetch_n = max(limits.topic_limit, compare_n, limits.wecom_topic)
+
+    feed = load_feed(force=True)
+    date_str = now.strftime("%Y-%m-%d")
+    time_str = now.strftime("%H:%M") + " (UTC+8)"
+    updated = (feed.get("updatedAt") or "")[:10]
+
+    trending, hot = load_boards(feed, trending_limit=limits.trending_n, hot_limit=limits.hot_n)
+    github_trending = fetch_github_trending(github_fetch_n)
+    seen_repos = {r["repo"] for r in github_trending}
+    github_emerging = fetch_emerging_repos(emerging_fetch_n, exclude=seen_repos)
+    seen_repos.update(r["repo"] for r in github_emerging)
+    topic_name, github_topic = fetch_topic_hot_repos(topic_fetch_n, exclude=seen_repos)
+    ai_news = apply_news_ranking(fetch_ai_news(), date_str=date_str)
+    cn_ai_news = apply_news_ranking(fetch_cn_ai_news(), date_str=date_str)
+
+    wecom_limits = {
+        "trending": limits.wecom_trending,
+        "hot": limits.wecom_hot,
+        "trending_pool": limits.skill_pool,
+        "hot_pool": limits.skill_pool,
+        "github": limits.wecom_github,
+        "emerging": limits.wecom_emerging,
+        "topic": limits.wecom_topic,
+        "ai_news": env_int("DAILY_WECOM_AI_NEWS", 10),
+        "cn_ai_news": env_int("DAILY_WECOM_CN_AI_NEWS", 8),
+    }
+    llm_input = build_llm_input(
+        date_str=date_str,
+        updated=updated,
+        trending=trending,
+        hot=hot,
+        github_trending=github_trending,
+        github_emerging=github_emerging,
+        github_topic=github_topic,
+        topic_name=topic_name,
+        ai_news=ai_news,
+        cn_ai_news=cn_ai_news,
+        wecom_limits=wecom_limits,
+        agent_mode=is_agent_mode(),
+    )
+    return ReportContext(
+        feed=feed,
+        date_str=date_str,
+        time_str=time_str,
+        updated=updated,
+        trending=trending,
+        hot=hot,
+        github_trending=github_trending,
+        github_emerging=github_emerging,
+        github_topic=github_topic,
+        topic_name=topic_name,
+        ai_news=ai_news,
+        cn_ai_news=cn_ai_news,
+        limits=limits,
+        wecom_limits=wecom_limits,
+        llm_input=llm_input,
+    )

daily/pipeline/formatters.py

@@ -0,0 +1,180 @@
+"""归档 / 企微格式化与摘要构建。"""
+
+from __future__ import annotations
+
+import re
+import xml.etree.ElementTree as ET
+from typing import Any
+
+import certifi
+import httpx
+
+from daily.config import full_desc_limit, wecom_skill_desc_limit
+from daily.github.auth import github_html_headers
+from daily.news.fetch import prepare_wecom_cn_news_items, prepare_wecom_news_items
+from shared.skills_data import format_installs
+
+from daily.pipeline.snapshot import skill_id
+
+
+def short_desc(text: str, limit: int = 72) -> str:
+    text = re.sub(r"\s+", " ", text or "").strip()
+    if limit <= 0 or len(text) <= limit:
+        return text
+    return text[: limit - 3] + "..."
+
+
+def archive_desc(text: str) -> str:
+    return short_desc(text, full_desc_limit())
+
+
+def wecom_desc(text: str, limit: int = 36) -> str:
+    return short_desc(text, limit)
+
+
+def prepare_skill_item(item: dict[str, Any], prev_ids: set[str], rank: int) -> dict[str, Any]:
+    badge = ""
+    sid = skill_id(item)
+    if sid not in prev_ids and prev_ids:
+        badge = "🆕"
+    elif rank == 1:
+        badge = "👑"
+    installs_fmt = item.get("installs_fmt") or format_installs(item.get("installs", 0))
+    title = item.get("source", "?") if item.get("cluster") else item.get("title", "?")
+    desc = item.get("wecom_desc") or item.get("description") or item.get("cluster_titles") or ""
+    limit = wecom_skill_desc_limit()
+    desc_short = desc if item.get("wecom_desc") or limit <= 0 else wecom_desc(desc, limit)
+    return {
+        "title": title,
+        "source": item.get("source", "?"),
+        "installs_fmt": installs_fmt,
+        "link": item.get("link", ""),
+        "desc_short": desc_short,
+        "badge": badge,
+        "cluster": bool(item.get("cluster")),
+        "cluster_count": item.get("cluster_count"),
+        "cluster_titles": item.get("cluster_titles"),
+    }
+
+
+def prepare_github_item(item: dict[str, Any]) -> dict[str, Any]:
+    return {**item, "desc_short": wecom_desc(item.get("description", ""), 40)}
+
+
+def build_highlights(
+    trending: list[dict[str, Any]],
+    hot: list[dict[str, Any]],
+    github_trending: list[dict[str, Any]],
+    github_emerging: list[dict[str, Any]],
+    ai_news: dict[str, Any] | None = None,
+    cn_ai_news: dict[str, Any] | None = None,
+) -> list[str]:
+    points: list[str] = []
+    if ai_news and ai_news.get("enabled"):
+        top_news = prepare_wecom_news_items(ai_news)
+        if top_news:
+            n0 = top_news[0]
+            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
+            points.append(
+                f"🌍 AI 时讯 [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
+            )
+        elif ai_news.get("flat"):
+            n0 = ai_news["flat"][0]
+            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
+            points.append(f"🌍 AI 时讯 [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）")
+    if cn_ai_news and cn_ai_news.get("enabled"):
+        top_cn = prepare_wecom_cn_news_items(cn_ai_news)
+        if top_cn:
+            n0 = top_cn[0]
+            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
+            points.append(
+                f"🇨🇳 国内 AI [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
+            )
+        elif cn_ai_news.get("flat"):
+            n0 = cn_ai_news["flat"][0]
+            pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
+            points.append(
+                f"🇨🇳 国内 AI [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
+            )
+    if trending:
+        t0 = trending[0]
+        points.append(f"📈 Skills 榜首 **{t0.get('title')}**（{format_installs(t0.get('installs', 0))}）")
+    if github_trending:
+        g0 = github_trending[0]
+        stars = g0.get("stars_today_fmt", "")
+        total = g0.get("total_stars_fmt", "")
+        star_hint = f"+{stars} today · " if stars else (f"⭐{total} · " if total else "")
+        points.append(f"🐙 GitHub Trending [{g0['repo']}]({g0['url']})（{star_hint}{g0.get('language', '')}）")
+    if github_emerging:
+        e0 = github_emerging[0]
+        points.append(f"🌱 新兴 [{e0['repo']}]({e0['url']})（⭐ {e0.get('total_stars_fmt', '?')}）")
+    elif hot:
+        h0 = hot[0]
+        points.append(f"🔥 Skills Hot 榜首 **{h0.get('title')}**（1H {format_installs(h0.get('installs', 0))}）")
+    while len(points) < 3 and len(trending) > len(points):
+        item = trending[len(points)]
+        points.append(f"✨ **{item.get('title')}** · `{item.get('source')}`")
+    return points[:3]
+
+
+def format_github_repo_section(repos: list[dict[str, Any]], *, show_created: bool = False) -> list[str]:
+    lines: list[str] = []
+    for i, repo in enumerate(repos, 1):
+        lang = repo.get("language") or "—"
+        stars_today = repo.get("stars_today_fmt") or ""
+        total = repo.get("total_stars_fmt") or ""
+        created = repo.get("created_at") or ""
+        meta_parts = [lang]
+        if stars_today:
+            meta_parts.append(f"+{stars_today} today")
+        if total:
+            meta_parts.append(f"总 ⭐ {total}")
+        if show_created and created:
+            meta_parts.append(f"创建于 {created}")
+        lines.append(f"{i}. **[{repo['repo']}]({repo['url']})** · {' · '.join(meta_parts)}")
+        desc = archive_desc(repo.get("description", ""))
+        if desc:
+            lines.append(f"   - {desc}")
+        lines.append("")
+    return lines
+
+
+def format_skill_section(items: list[dict[str, Any]], *, hot: bool = False) -> list[str]:
+    lines: list[str] = []
+    for i, item in enumerate(items, 1):
+        sid = item.get("id") or f"{item.get('source', '?')}/{item.get('title', '?')}"
+        link = item.get("link", "")
+        installs = format_installs(item.get("installs", 0))
+        meta = f"1H {installs}" if hot else f"总安装 {installs}"
+        if link:
+            lines.append(f"{i}. **[{sid}]({link})** · {meta}")
+        else:
+            lines.append(f"{i}. **{sid}** · {meta}")
+        desc = archive_desc(item.get("description", ""))
+        if desc:
+            lines.append(f"   - {desc}")
+        lines.append("")
+    return lines
+
+
+def fetch_latest_release_title(repo: str) -> str | None:
+    atom_url = f"https://github.com/{repo}/releases.atom"
+    try:
+        with httpx.Client(
+            timeout=12.0,
+            verify=certifi.where(),
+            follow_redirects=True,
+            headers=github_html_headers(),
+        ) as client:
+            resp = client.get(atom_url)
+            if resp.status_code != 200:
+                return None
+            root = ET.fromstring(resp.text)
+            ns = {"a": "http://www.w3.org/2005/Atom"}
+            entry = root.find("a:entry", ns)
+            if entry is None:
+                return None
+            title = entry.find("a:title", ns)
+            return title.text.strip() if title is not None and title.text else None
+    except Exception:
+        return None

daily/pipeline/localize.py

@@ -0,0 +1,116 @@
+"""归档内容中文化。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from daily.config import full_desc_limit, news_summary_limit
+from daily.localize import LocalizeJob, localize_descriptions, needs_chinese
+
+from daily.pipeline.snapshot import skill_id
+
+
+def localize_descriptions_in_place(
+    trending: list[dict[str, Any]],
+    hot: list[dict[str, Any]],
+    github_trending: list[dict[str, Any]],
+    github_emerging: list[dict[str, Any]],
+    github_topic: list[dict[str, Any]],
+    ai_news: dict[str, Any],
+    cn_ai_news: dict[str, Any] | None = None,
+) -> None:
+    full_limit = full_desc_limit()
+    news_limit = news_summary_limit()
+    jobs: list[LocalizeJob] = []
+    seen_skill: set[str] = set()
+    for item in trending + hot:
+        sid = skill_id(item)
+        if sid in seen_skill:
+            continue
+        seen_skill.add(sid)
+        desc = (item.get("description") or "").strip()
+        if desc:
+            jobs.append(LocalizeJob(f"skill:{sid}", desc, full_limit))
+
+    seen_repo: set[str] = set()
+    for repo_list in (github_trending, github_emerging, github_topic):
+        for item in repo_list:
+            repo = item.get("repo", "")
+            if not repo or repo in seen_repo:
+                continue
+            seen_repo.add(repo)
+            desc = (item.get("description") or "").strip()
+            if desc:
+                jobs.append(LocalizeJob(f"github:{repo}", desc, full_limit))
+
+    if ai_news.get("enabled"):
+        seen_news: set[str] = set()
+        for item in ai_news.get("flat") or []:
+            link = item.get("link", "")
+            if not link or link in seen_news:
+                continue
+            seen_news.add(link)
+            summary = (item.get("summary") or "").strip()
+            if summary:
+                jobs.append(LocalizeJob(f"news:{link}", summary, news_limit))
+
+    zh_map = localize_descriptions(jobs, archive=True)
+    if not zh_map and not jobs:
+        return
+
+    def _apply_zh(mapping: dict[str, str]) -> None:
+        for item in trending + hot:
+            key = f"skill:{skill_id(item)}"
+            if key in mapping:
+                item["description"] = mapping[key]
+        for repo_list in (github_trending, github_emerging, github_topic):
+            for item in repo_list:
+                key = f"github:{item.get('repo', '')}"
+                if key in mapping:
+                    item["description"] = mapping[key]
+        if ai_news.get("enabled"):
+            for cat in ai_news.get("categories") or []:
+                for item in cat.get("items") or []:
+                    key = f"news:{item.get('link', '')}"
+                    if key in mapping:
+                        item["summary"] = mapping[key]
+            for item in ai_news.get("flat") or []:
+                key = f"news:{item.get('link', '')}"
+                if key in mapping:
+                    item["summary"] = mapping[key]
+
+    _apply_zh(zh_map)
+
+    retry_jobs: list[LocalizeJob] = []
+    seen_skill.clear()
+    for item in trending + hot:
+        sid = skill_id(item)
+        if sid in seen_skill:
+            continue
+        seen_skill.add(sid)
+        desc = (item.get("description") or "").strip()
+        if needs_chinese(desc):
+            retry_jobs.append(LocalizeJob(f"skill:{sid}", desc, full_limit))
+    seen_repo.clear()
+    for repo_list in (github_trending, github_emerging, github_topic):
+        for item in repo_list:
+            repo = item.get("repo", "")
+            if not repo or repo in seen_repo:
+                continue
+            seen_repo.add(repo)
+            desc = (item.get("description") or "").strip()
+            if needs_chinese(desc):
+                retry_jobs.append(LocalizeJob(f"github:{repo}", desc, full_limit))
+    if ai_news.get("enabled"):
+        seen_news.clear()
+        for item in ai_news.get("flat") or []:
+            link = item.get("link", "")
+            if not link or link in seen_news:
+                continue
+            seen_news.add(link)
+            summary = (item.get("summary") or "").strip()
+            if needs_chinese(summary):
+                retry_jobs.append(LocalizeJob(f"news:{link}", summary, news_limit))
+
+    if retry_jobs:
+        _apply_zh(localize_descriptions(retry_jobs, archive=True))

daily/pipeline/snapshot.py

@@ -0,0 +1,36 @@
+"""Skills 快照读写。"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from daily.config import CACHE_DIR, SNAPSHOT_FILE
+
+
+def skill_id(item: dict[str, Any]) -> str:
+    return str(item.get("id") or f"{item.get('source')}/{item.get('title')}")
+
+
+def load_snapshot() -> set[str]:
+    if not SNAPSHOT_FILE.exists():
+        return set()
+    try:
+        data = json.loads(SNAPSHOT_FILE.read_text(encoding="utf-8"))
+        return set(str(x) for x in (data.get("skill_ids") or []))
+    except (OSError, json.JSONDecodeError):
+        return set()
+
+
+def save_snapshot(feed: dict[str, Any], date_str: str) -> None:
+    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    ids: list[str] = []
+    for board in ("topTrending", "topHot"):
+        for item in feed.get(board, [])[:20]:
+            sid = skill_id(item)
+            if sid not in ids:
+                ids.append(sid)
+    SNAPSHOT_FILE.write_text(
+        json.dumps({"date": date_str, "skill_ids": ids}, ensure_ascii=False, indent=2),
+        encoding="utf-8",
+    )

daily/pipeline/themes.py

@@ -0,0 +1,53 @@
+"""主题检测与聚类。"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any
+
+THEME_RULES: list[tuple[str, str, list[str]]] = [
+    ("🎬", "AI 多媒体 / 视频", ["runcomfy", "remotion", "video", "seedance", "inpaint", "lipsync"]),
+    ("🔧", "工程协作 / Skill 元能力", ["grill", "tdd", "architecture", "find-skills", "to-issues"]),
+    ("📱", "飞书 / Lark", ["lark", "feishu"]),
+    ("📣", "内容营销", ["viral", "tiktok", "instagram", "reels"]),
+    ("🎨", "设计 / 前端", ["frontend", "design", "ui-ux", "tailwind"]),
+]
+
+
+def detect_theme_line(feed: dict[str, Any]) -> str:
+    scores: dict[str, int] = defaultdict(int)
+    for board in ("topTrending", "topHot"):
+        for rank, item in enumerate(feed.get(board, [])[:10], 1):
+            haystack = " ".join(
+                [item.get("title", ""), item.get("source", ""), item.get("description", "")]
+            ).lower()
+            for _icon, label, keywords in THEME_RULES:
+                if any(k in haystack for k in keywords):
+                    scores[label] += max(1, 11 - rank)
+                    break
+    if not scores:
+        return "**今日主题**：Agent Skills 生态持续活跃"
+    return f"**今日主题**：{max(scores.items(), key=lambda x: x[1])[0]}"
+
+
+def theme_clusters(feed: dict[str, Any], limit: int = 5) -> list[tuple[str, list[str]]]:
+    from daily.pipeline.snapshot import skill_id
+
+    buckets: dict[str, list[str]] = defaultdict(list)
+    seen: set[str] = set()
+    for board in ("topTrending", "topHot"):
+        for item in feed.get(board, [])[:20]:
+            item_id = skill_id(item)
+            if item_id in seen:
+                continue
+            seen.add(item_id)
+            haystack = " ".join(
+                [item.get("title", ""), item.get("source", ""), item.get("description", "")]
+            ).lower()
+            for _icon, theme, keywords in THEME_RULES:
+                if any(k in haystack for k in keywords):
+                    label = f"**{item.get('title')}** (`{item.get('source')}`)"
+                    if label not in buckets[theme]:
+                        buckets[theme].append(label)
+                    break
+    return [(theme, examples[:limit]) for theme, examples in buckets.items() if examples]

tests/test_agent_workflow.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import json
+
+
+def test_analyze_trends_parses_json(monkeypatch, isolated_output):
+    import daily.agent_workflow as agent
+    import daily.config as config
+
+    monkeypatch.setattr(config, "OUTPUT_DIR", isolated_output)
+    monkeypatch.setattr(agent, "OUTPUT_DIR", isolated_output)
+
+    monkeypatch.setattr(
+        agent,
+        "llm_chat",
+        lambda system, user: json.dumps(
+            {
+                "headline": "Skills 视频工具升温",
+                "opening": "今日 remotion 相关技能继续走强。",
+                "themes": [{"name": "视频", "summary": "Remotion 生态活跃"}],
+                "top_picks": [],
+                "signals": [],
+            },
+            ensure_ascii=False,
+        ),
+    )
+
+    llm_input = {"date": "2026-07-03", "skills_trending": []}
+    trends = agent.analyze_trends(llm_input, date_str="2026-07-03")
+
+    assert trends is not None
+    assert trends["headline"] == "Skills 视频工具升温"
+    assert (isolated_output / "2026-07-03.trends.json").exists()
+
+
+def test_write_wecom_report_extracts_markdown_block(monkeypatch):
+    import daily.agent_workflow as agent
+
+    monkeypatch.setattr(
+        agent,
+        "llm_chat",
+        lambda system, user: "```markdown\n📰 **早报 · 2026-07-03**\n\n正文\n```",
+    )
+
+    md = agent.write_wecom_report(
+        {"date": "2026-07-03"},
+        {"headline": "test", "opening": "open"},
+        date_str="2026-07-03",
+        time_str="09:30 (UTC+8)",
+        updated="2026-07-02",
+    )
+
+    assert md is not None
+    assert md.startswith("📰")
+    assert "正文" in md

tests/test_delta.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+
+def test_build_movement_context_detects_new_skill(isolated_output, monkeypatch):
+    import daily.config as config
+    import daily.delta as delta
+
+    monkeypatch.setattr(config, "OUTPUT_DIR", isolated_output)
+    monkeypatch.setattr(delta, "OUTPUT_DIR", isolated_output)
+
+    prev_path = isolated_output / "2026-07-02.data.json"
+    prev_path.write_text(
+        json.dumps(
+            {
+                "data": {
+                    "date": "2026-07-02",
+                    "movement_baseline": {
+                        "skills_trending": [{"id": "a/old", "title": "old"}],
+                        "skills_hot": [],
+                        "github_trending": [],
+                        "github_emerging": [],
+                        "github_topic": [],
+                    },
+                }
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    trending = [
+        {"id": "a/new", "title": "new-skill", "source": "a", "installs": 10, "link": "https://x", "description": ""},
+        {"id": "a/old", "title": "old", "source": "a", "installs": 9, "link": "https://y", "description": ""},
+    ]
+    movement = delta.build_movement_context(
+        date_str="2026-07-03",
+        trending=trending,
+        hot=[],
+        github_trending=[],
+        github_emerging=[],
+        github_topic=[],
+    )
+
+    assert movement["baseline_date"] == "2026-07-02"
+    assert len(movement["skills_trending_moves"]) == 1
+    assert movement["skills_trending_moves"][0]["id"] == "a/new"
+
+
+def test_find_previous_data_skips_missing_days(isolated_output, monkeypatch):
+    import daily.config as config
+    import daily.delta as delta
+
+    monkeypatch.setattr(config, "OUTPUT_DIR", isolated_output)
+    monkeypatch.setattr(delta, "OUTPUT_DIR", isolated_output)
+
+    (isolated_output / "2026-07-01.data.json").write_text(
+        json.dumps({"data": {"date": "2026-07-01", "skills_trending": []}}),
+        encoding="utf-8",
+    )
+
+    found = delta.find_previous_data("2026-07-03")
+    assert found is not None
+    assert found[0] == "2026-07-01"

tests/test_rss_parse.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def test_parse_sample_rss_fixture():
+    from daily.news.feeds import NEWS_CATEGORIES
+    from daily.news.fetch import _parse_feed
+
+    category = NEWS_CATEGORIES[0]
+    feed_name = category.feeds[0].name
+    xml = (FIXTURES / "sample-rss.xml").read_text(encoding="utf-8")
+    items = _parse_feed(xml, feed_name, category)
+
+    assert len(items) == 1
+    assert items[0]["title"] == "Sample AI headline for smoke tests"
+    assert items[0]["link"] == "https://example.com/ai-news/1"
+    assert items[0]["source_name"] == feed_name
+    assert "minimal RSS item" in items[0]["summary"]
+
+
+def test_fetch_ai_news_offline(monkeypatch):
+    from daily.news import fetch as news_fetch
+
+    sample = {
+        "title": "Sample AI headline for smoke tests",
+        "link": "https://example.com/ai-news/1",
+        "summary": "A minimal RSS item used by offline tests.",
+        "published": "Wed, 02 Jul 2026 08:00:00 GMT",
+        "source_name": "OpenAI",
+        "category_id": "official",
+        "category_name": "厂商官方",
+        "category_icon": "🏢",
+    }
+
+    monkeypatch.setattr(
+        news_fetch,
+        "_fetch_news",
+        lambda categories: {
+            "enabled": True,
+            "hours": 72,
+            "categories": [{"id": "official", "name": "厂商官方", "icon": "🏢", "items": [sample]}],
+            "flat": [sample],
+            "stats": {"feeds_total": 1, "feeds_ok": 1, "items_raw": 1, "feeds_failed": []},
+        },
+    )
+
+    payload = news_fetch.fetch_ai_news()
+    assert payload["enabled"] is True
+    assert payload["flat"][0]["title"].startswith("Sample AI")

tests/test_smoke.py

@@ -36,11 +36,12 @@ def test_generate_report_offline(
     monkeypatch,
 ):
     import daily.generate as generate
+    import daily.pipeline.collect as collect
 
-    monkeypatch.setattr(generate, "load_feed", lambda force=False: skills_feed)
-    monkeypatch.setattr(generate, "fetch_github_trending", lambda n: [_sample_github_repo()])
-    monkeypatch.setattr(generate, "fetch_emerging_repos", lambda n, exclude=None: [])
-    monkeypatch.setattr(generate, "fetch_topic_hot_repos", lambda n, exclude=None: ("llm", []))
+    monkeypatch.setattr(collect, "load_feed", lambda force=False: skills_feed)
+    monkeypatch.setattr(collect, "fetch_github_trending", lambda n: [_sample_github_repo()])
+    monkeypatch.setattr(collect, "fetch_emerging_repos", lambda n, exclude=None: [])
+    monkeypatch.setattr(collect, "fetch_topic_hot_repos", lambda n, exclude=None: ("llm", []))
     monkeypatch.setattr(generate, "_now_cst", lambda: fixed_cst)
 
     _, wecom_md, out_md, out_wecom = generate.generate_report()

tests/test_wecom_split.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from daily.wecom_split import split_wecom_messages
+
+
+def test_split_wecom_messages_keeps_short_text():
+    text = "📰 **早报 · 2026-07-03**\n\n短内容"
+    parts = split_wecom_messages(text, limit=4096)
+    assert parts == [text]
+
+
+def test_split_wecom_messages_adds_part_footer():
+    section = "📰 **早报**\n\n" + ("正文行\n" * 200)
+    parts = split_wecom_messages(section, limit=600)
+    assert len(parts) > 1
+    assert all(len(part.encode("utf-8")) <= 600 for part in parts)
+    assert parts[0].endswith("1/" + str(len(parts)))
+
+
+def test_split_sections_on_emoji_headers():
+    from daily.wecom_split import _split_sections
+
+    text = "\n\n".join(
+        [
+            "📰 **早报 · 2026-07-03**",
+            "💡 **今日速览**\n- item one\n- item two",
+            "🌍 **国际 AI 时讯**\n1. [Headline](https://example.com)",
+        ]
+    )
+    sections = _split_sections(text)
+    assert len(sections) == 3
+    assert sections[0].startswith("📰")
+    assert sections[1].startswith("💡")
+    assert sections[2].startswith("🌍")