daily-robots/daily/generate.py

"""生成早报 Markdown（完整版 + 企微短版）。"""

from __future__ import annotations

import json
import re
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,
    board_pool_size,
    env,
    env_bool,
    env_int,
    full_desc_limit,
    news_summary_limit,
    wecom_delta_pad,
    wecom_news_desc_limit,
    wecom_pad_pool_size,
    wecom_skill_desc_limit,
)
from daily.board_history import (
    BOARD_KEYS,
    extract_shown_keys,
    load_recent_shown_keys,
    merge_wecom_shown_into_data,
)
from daily.board_select import board_select
from daily.format_wecom import (
    build_wecom_report,
    finalize_wecom_skill_groups,
    replace_wecom_board_sections,
    replace_wecom_news_sections,
    replace_wecom_skill_sections,
    resolve_wecom_board_items,
)
from daily.agent_workflow import is_agent_mode, run_agent_workflow
from daily.featured_pick import (
    apply_featured_pick,
    featured_identity_key,
    pick_command_from_featured,
    pick_why_from_featured,
)
from daily.delta import compare_depth
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.narrative_axis import theme_clusters, theme_names
from daily.news.fetch import (
    fetch_ai_news,
    fetch_cn_ai_news,
    format_cn_news_section,
    format_news_section,
    finalize_wecom_news_items,
    prepare_wecom_cn_news_items,
    prepare_wecom_news_items,
    sync_wecom_news_rows,
)
from daily.news.pushed_links import record_pushed_links
from daily.news.research import (
    fetch_ai_news_research,
    format_research_news_section,
    is_research_mode,
)
from daily.push_gate import evaluate_push_gate
from daily.report_data import (
    build_full_payload,
    build_llm_input,
    data_json_path,
    save_json,
)
from daily.skills_board import format_installs, load_boards, 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"]),
]


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 if news_limit > 0 else wecom_news_desc_limit(),
                    )
                )

    if cn_ai_news and cn_ai_news.get("enabled"):
        seen_cn: set[str] = set()
        for item in cn_ai_news.get("flat") or []:
            link = item.get("link", "")
            if not link or link in seen_cn:
                continue
            seen_cn.add(link)
            summary = (item.get("summary") or "").strip()
            if summary and needs_chinese(summary):
                jobs.append(
                    LocalizeJob(
                        f"news:{link}",
                        summary,
                        news_limit if news_limit > 0 else wecom_news_desc_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]
        if cn_ai_news and cn_ai_news.get("enabled"):
            for item in cn_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 news_limit > 0 else wecom_news_desc_limit(),
                    )
                )
    if cn_ai_news and cn_ai_news.get("enabled"):
        seen_cn: set[str] = set()
        for item in cn_ai_news.get("flat") or []:
            link = item.get("link", "")
            if not link or link in seen_cn:
                continue
            seen_cn.add(link)
            summary = (item.get("summary") or "").strip()
            if needs_chinese(summary):
                retry_jobs.append(
                    LocalizeJob(
                        f"news:{link}",
                        summary,
                        news_limit if news_limit > 0 else wecom_news_desc_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 _top_line(feed: dict[str, Any]) -> str:
    """今日看点行: 优先评分最高的主题, 回退 _theme_clusters 的主题名(非 markdown 示例)。

    DAILY_WECOM_TOP_LINE=0 关闭时退回 _detect_theme_line 原逻辑。
    与原 _detect_theme_line 的差别仅在「无评分命中」时的兜底文案:
    用 theme_names 的真实主题取代硬编码「Agent Skills 生态持续活跃」。
    """
    if not env_bool("DAILY_WECOM_TOP_LINE", True):
        return _detect_theme_line(feed)
    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 scores:
        return f"**今日主题**：{max(scores.items(), key=lambda x: x[1])[0]}"
    names = theme_names(feed, theme_rules=THEME_RULES, skill_id_fn=_skill_id)
    if names:
        return f"**今日主题**：{' · '.join(names)}"
    return "**今日主题**：Agent Skills 生态持续活跃"


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 _sync_movement_github_descriptions(
    movement: dict[str, Any],
    *,
    github_trending: list[dict[str, Any]],
    github_emerging: list[dict[str, Any]],
    github_topic: list[dict[str, Any]],
) -> None:
    """将已中文化的 GitHub 描述同步到 movement 新入榜条目（供 Delta 企微列表使用）。"""
    by_repo: dict[str, str] = {}
    for item in github_trending + github_emerging + github_topic:
        repo = str(item.get("repo") or "")
        desc = (item.get("description") or "").strip()
        if repo and desc:
            by_repo[repo] = desc
    for key in ("github_trending_moves", "github_emerging_moves", "github_topic_moves"):
        for move in movement.get(key) or []:
            repo = str(move.get("repo") or "")
            if repo in by_repo:
                move["description"] = by_repo[repo]


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 (httpx.HTTPError, ET.ParseError, OSError) as exc:
        logger.warning("读取 %s release 失败：%s", atom_url, exc)
        return None


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 _collect(date_str: str) -> dict[str, Any]:
    """抓取 skills/github/news 数据并归一化,计算 wecom 限额与周去重 recent keys。

    无副作用(不写快照/不记已推)。返回供 _select/_render 消费的 bundle。
    """
    # Hot/Trending 前排同 source 极密，需更深抓取才能凑够展示用的唯一 source
    trending_n = env_int("DAILY_TRENDING_LIMIT", 400)
    hot_n = max(env_int("DAILY_HOT_LIMIT", 400), compare_depth())
    compare_n = compare_depth()
    skill_pool = max(10, env_int("DAILY_WECOM_SKILL_POOL", 400))
    wecom_trending = env_int("DAILY_WECOM_TRENDING", 5)
    wecom_hot = env_int("DAILY_WECOM_HOT", 5)
    pad_pool = wecom_pad_pool_size(max(wecom_trending, wecom_hot, 5))
    skill_pool = max(skill_pool, pad_pool)
    github_limit = env_int("DAILY_GITHUB_TRENDING_LIMIT", 10)
    wecom_github = env_int("DAILY_WECOM_GITHUB_TRENDING", env_int("DAILY_WECOM_REPOS", 5))
    # 周去重后顶刊 sticky，HTML/~30 条不够补满；深池默认 100（Search 已分页）
    github_pool = max(pad_pool, env_int("DAILY_GITHUB_POOL", 100))
    github_fetch_n = max(github_limit, compare_n, wecom_github, github_pool)
    emerging_limit = env_int("DAILY_GITHUB_EMERGING_LIMIT", 10)
    wecom_emerging = env_int("DAILY_WECOM_GITHUB_EMERGING", 5)
    emerging_fetch_n = max(emerging_limit, compare_n, wecom_emerging, github_pool)
    topic_limit = env_int("DAILY_GITHUB_TOPIC_LIMIT", 10)
    wecom_topic = env_int("DAILY_WECOM_GITHUB_TOPIC", 5)
    topic_fetch_n = max(topic_limit, compare_n, wecom_topic, github_pool)

    feed = load_feed(force=True)
    prev_ids = _load_snapshot()
    now = _now_cst()
    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=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)

    news_merged = is_research_mode()
    ai_news_research: dict[str, Any] | None = None
    wecom_news: list[dict[str, Any]] = []
    wecom_tech_news: list[dict[str, Any]] = []
    if news_merged:
        ai_news_research = fetch_ai_news_research(date_str=date_str)
        wecom_news = list(ai_news_research.get("items") or [])
        wecom_tech_news = list(ai_news_research.get("tech_items") or [])
        ai_news = {
            "enabled": ai_news_research.get("enabled", False),
            "mode": "research",
            "hours": ai_news_research.get("hours", 24),
            "flat": ai_news_research.get("flat") or [],
            "categories": [],
            "stats": ai_news_research.get("stats") or {},
        }
        cn_ai_news = {"enabled": False, "categories": [], "flat": [], "stats": {}}
    else:
        ai_news = fetch_ai_news()
        cn_ai_news = fetch_cn_ai_news()

    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", 10),
    }
    pool = max(board_pool_size(), skill_pool, pad_pool)
    recent_shown = load_recent_shown_keys(date_str)
    from daily.skills_group import expand_skill_recent_keys

    skill_recent = expand_skill_recent_keys(
        recent_shown["skills_trending"] | recent_shown["skills_hot"]
    )
    # GitHub 三榜共用周去重：任一类出现过的 repo 各榜都不再展示
    github_recent = (
        recent_shown["github_trending"]
        | recent_shown["github_emerging"]
        | recent_shown["github_topic"]
    )
    return {
        "trending_n": trending_n,
        "hot_n": hot_n,
        "github_limit": github_limit,
        "emerging_limit": emerging_limit,
        "topic_limit": topic_limit,
        "wecom_trending": wecom_trending,
        "wecom_hot": wecom_hot,
        "wecom_github": wecom_github,
        "wecom_emerging": wecom_emerging,
        "wecom_topic": wecom_topic,
        "wecom_limits": wecom_limits,
        "pool": pool,
        "pad_pool": pad_pool,
        "feed": feed,
        "prev_ids": prev_ids,
        "now": now,
        "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,
        "news_merged": news_merged,
        "ai_news": ai_news,
        "cn_ai_news": cn_ai_news,
        "ai_news_research": ai_news_research,
        "wecom_news": wecom_news,
        "wecom_tech_news": wecom_tech_news,
        "recent_shown": recent_shown,
        "skill_recent": skill_recent,
        "github_recent": github_recent,
    }


def _select(c: dict[str, Any]) -> dict[str, Any]:
    """全部「选择 + 数据突变」:9 处 board_select + featured_pick + LLM 步骤 + pad。

    时序约束内聚本段:featured pick 依赖 deep-pool;_localize 在 featured 后、
    pad 前执行;写回 shown keys/llm_input 也在此。_render 不再做任何选择。
    副作用: apply_featured_pick 写回 shown keys(保持原时序)。
    """
    date_str = c["date_str"]
    feed = c["feed"]
    prev_ids = c["prev_ids"]
    updated = c["updated"]
    time_str = c["time_str"]
    trending = c["trending"]
    hot = c["hot"]
    github_trending = c["github_trending"]
    github_emerging = c["github_emerging"]
    github_topic = c["github_topic"]
    topic_name = c["topic_name"]
    news_merged = c["news_merged"]
    ai_news = c["ai_news"]
    cn_ai_news = c["cn_ai_news"]
    wecom_news = c["wecom_news"]
    wecom_tech_news = c["wecom_tech_news"]
    wecom_limits = c["wecom_limits"]
    pool = c["pool"]
    pad_pool = c["pad_pool"]
    recent_shown = c["recent_shown"]
    skill_recent = c["skill_recent"]
    github_recent = c["github_recent"]

    selected_trending = board_select(
        board="skills_trending",
        items=trending,
        recent_keys=skill_recent,
        limit=c["wecom_trending"],
        pool_size=pool,
        kind="skill",
    )
    selected_hot = board_select(
        board="skills_hot",
        items=hot,
        recent_keys=skill_recent,
        limit=c["wecom_hot"],
        pool_size=pool,
        kind="skill",
    )
    selected_github = board_select(
        board="github_trending",
        items=github_trending,
        recent_keys=github_recent,
        limit=c["wecom_github"],
        pool_size=pool,
        kind="github",
    )
    github_recent |= {str(r.get("repo") or "") for r in selected_github if r.get("repo")}
    selected_emerging = board_select(
        board="github_emerging",
        items=github_emerging,
        recent_keys=github_recent,
        limit=c["wecom_emerging"],
        pool_size=pool,
        kind="github",
    )
    github_recent |= {str(r.get("repo") or "") for r in selected_emerging if r.get("repo")}
    selected_topic = board_select(
        board="github_topic",
        items=github_topic,
        recent_keys=github_recent,
        limit=c["wecom_topic"],
        pool_size=pool,
        kind="github",
    )
    boards_for_wecom = {
        "skills_trending": selected_trending,
        "skills_hot": selected_hot,
        "github_trending": selected_github,
        "github_emerging": selected_emerging,
        "github_topic": selected_topic,
    }
    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,
        research_items=wecom_news if news_merged else None,
        research_tech_items=wecom_tech_news if news_merged else None,
        boards_for_wecom=boards_for_wecom,
    )
    pool_a: list[dict[str, Any]] = []
    pool_a_keys: set[str] = set()
    for board_name, items in boards_for_wecom.items():
        for item in items:
            keyed = dict(item)
            keyed["board"] = board_name
            pool_a.append(keyed)
            ik = featured_identity_key(keyed)
            if ik:
                pool_a_keys.add(ik)
    pool_b: list[dict[str, Any]] = []
    for board_name, raw_items, kind in (
        ("skills_trending", trending, "skill"),
        ("skills_hot", hot, "skill"),
        ("github_trending", github_trending, "github"),
        ("github_emerging", github_emerging, "github"),
        ("github_topic", github_topic, "github"),
    ):
        deep = board_select(
            board=board_name,
            items=raw_items,
            recent_keys=set(),
            limit=pool,
            pool_size=pool,
            kind=kind,  # type: ignore[arg-type]
        )
        for item in deep:
            keyed = dict(item)
            keyed["board"] = board_name
            ik = featured_identity_key(keyed)
            if ik and ik not in pool_a_keys:
                pool_b.append(keyed)
    featured = apply_featured_pick(
        llm_input,
        date_str=date_str,
        pool_a=pool_a,
        pool_b=pool_b,
    )
    movement = llm_input["movement"]
    eff_mode = llm_input["effective_wecom_mode"]
    if news_merged:
        wecom_ai: list[dict[str, Any]] = []
        wecom_cn: list[dict[str, Any]] = []
    else:
        wecom_ai = prepare_wecom_news_items(ai_news, date_str=date_str)
        wecom_cn = prepare_wecom_cn_news_items(cn_ai_news, date_str=date_str)
    push_gate = evaluate_push_gate(
        movement=movement,
        ai_news_items=wecom_news if news_merged else wecom_ai,
        cn_ai_news_items=wecom_cn,
        featured_pick=featured,
    )
    llm_input["push_gate"] = {
        "should_push": push_gate.should_push,
        "silent": push_gate.silent,
        "reasons": push_gate.reasons,
    }

    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,
        )
        if not agent_wecom:
            logger.warning("Agent 工作流失败，回退 classic 模式")

    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)
        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,
                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
    )
    if not news_merged:
        sync_wecom_news_rows(wecom_ai, ai_news.get("flat") or [])
        sync_wecom_news_rows(wecom_cn, cn_ai_news.get("flat") or [])
        finalize_wecom_news_items(wecom_ai, force_chinese=True)
        finalize_wecom_news_items(wecom_cn, force_chinese=False)
    _sync_movement_github_descriptions(
        movement,
        github_trending=github_trending,
        github_emerging=github_emerging,
        github_topic=github_topic,
    )

    themes = theme_clusters(feed, theme_rules=THEME_RULES, skill_id_fn=_skill_id)

    pick_src = trending[0].get("source", "") if trending else ""
    pick_name = trending[0].get("title", "") if trending else ""
    pick_command = pick_command_from_featured(featured) or (
        f"npx skills add {pick_src}/{pick_name}"
        if pick_src and pick_name
        else "npx skills add vercel-labs/skills/find-skills"
    )
    pick_why = pick_why_from_featured(featured) or ""
    pick_title = str((featured or {}).get("title") or pick_name or "").strip()
    pick_url = str((featured or {}).get("url") or "").strip()

    gt = selected_trending
    gh = selected_hot
    gt_pad = board_select(
        board="skills_trending",
        items=trending,
        recent_keys=skill_recent,
        limit=pad_pool,
        pool_size=pool,
        kind="skill",
    )
    gh_pad = board_select(
        board="skills_hot",
        items=hot,
        recent_keys=skill_recent,
        limit=pad_pool,
        pool_size=pool,
        kind="skill",
    )
    wecom_github_items = [_prepare_github_item(item) for item in selected_github]
    wecom_emerging_items = [_prepare_github_item(item) for item in selected_emerging]
    wecom_topic_items = [_prepare_github_item(item) for item in selected_topic]
    github_pad_recent = (
        recent_shown["github_trending"]
        | recent_shown["github_emerging"]
        | recent_shown["github_topic"]
    )
    wecom_github_pad = [
        _prepare_github_item(item)
        for item in board_select(
            board="github_trending",
            items=github_trending,
            recent_keys=github_pad_recent,
            limit=pad_pool,
            pool_size=pool,
            kind="github",
        )
    ]
    wecom_emerging_pad = [
        _prepare_github_item(item)
        for item in board_select(
            board="github_emerging",
            items=github_emerging,
            recent_keys=github_pad_recent,
            limit=pad_pool,
            pool_size=pool,
            kind="github",
        )
    ]
    wecom_topic_pad = [
        _prepare_github_item(item)
        for item in board_select(
            board="github_topic",
            items=github_topic,
            recent_keys=github_pad_recent,
            limit=pad_pool,
            pool_size=pool,
            kind="github",
        )
    ]
    delta_pad = eff_mode == "delta" and wecom_delta_pad()
    board_kwargs = {
        "mode": eff_mode,
        "movement": movement,
        "trending": gt,
        "hot": gh,
        "topic_name": topic_name,
        "github_trending": wecom_github_items,
        "github_emerging": wecom_emerging_items,
        "github_topic": wecom_topic_items,
        "wecom_trending": c["wecom_trending"],
        "wecom_hot": c["wecom_hot"],
        "wecom_github": c["wecom_github"],
        "wecom_emerging": c["wecom_emerging"],
        "wecom_topic": c["wecom_topic"],
        "pad": delta_pad,
        "date_str": date_str,
        "trending_pad": gt_pad,
        "hot_pad": gh_pad,
        "github_trending_pad": wecom_github_pad,
        "github_emerging_pad": wecom_emerging_pad,
        "github_topic_pad": wecom_topic_pad,
    }
    return {
        "selected_trending": selected_trending,
        "selected_hot": selected_hot,
        "boards_for_wecom": boards_for_wecom,
        "llm_input": llm_input,
        "featured": featured,
        "movement": movement,
        "eff_mode": eff_mode,
        "push_gate": push_gate,
        "wecom_ai": wecom_ai,
        "wecom_cn": wecom_cn,
        "agent_wecom": agent_wecom,
        "editorial_theme": editorial_theme,
        "editorial_highlights": editorial_highlights,
        "themes": themes,
        "pick_command": pick_command,
        "pick_why": pick_why,
        "pick_title": pick_title,
        "pick_url": pick_url,
        "board_kwargs": board_kwargs,
        "prev_ids": prev_ids,
        "gt": gt,
        "gh": gh,
        "wecom_github_items": wecom_github_items,
        "wecom_emerging_items": wecom_emerging_items,
        "wecom_topic_items": wecom_topic_items,
    }


def _render(c: dict[str, Any], s: dict[str, Any]) -> tuple[str, str, Path, Path]:
    """纯拼装: 拼完整版 markdown + 企微 wecom_md + 写盘。

    零选择逻辑。副作用: record_pushed_links + _save_snapshot + save_json + 写文件,
    全部保持原时序(在拼装完成后执行)。
    """
    feed = c["feed"]
    now = c["now"]
    date_str = c["date_str"]
    time_str = c["time_str"]
    updated = c["updated"]
    trending = c["trending"]
    hot = c["hot"]
    github_trending = c["github_trending"]
    github_emerging = c["github_emerging"]
    github_topic = c["github_topic"]
    topic_name = c["topic_name"]
    news_merged = c["news_merged"]
    ai_news = c["ai_news"]
    cn_ai_news = c["cn_ai_news"]
    ai_news_research = c["ai_news_research"]
    wecom_news = c["wecom_news"]
    wecom_tech_news = c["wecom_tech_news"]
    trending_n = c["trending_n"]
    hot_n = c["hot_n"]
    github_limit = c["github_limit"]
    emerging_limit = c["emerging_limit"]
    topic_limit = c["topic_limit"]
    prev_ids = s["prev_ids"]
    llm_input = s["llm_input"]
    featured = s["featured"]
    eff_mode = s["eff_mode"]
    push_gate = s["push_gate"]
    wecom_ai = s["wecom_ai"]
    wecom_cn = s["wecom_cn"]
    agent_wecom = s["agent_wecom"]
    editorial_theme = s["editorial_theme"]
    editorial_highlights = s["editorial_highlights"]
    themes = s["themes"]
    pick_command = s["pick_command"]
    pick_why = s["pick_why"]
    pick_title = s["pick_title"]
    pick_url = s["pick_url"]
    board_kwargs = s["board_kwargs"]
    gt = s["gt"]
    gh = s["gh"]
    wecom_github_items = s["wecom_github_items"]
    wecom_emerging_items = s["wecom_emerging_items"]
    wecom_topic_items = s["wecom_topic_items"]

    lines = [
        f"# 早报 · {date_str}",
        "",
        f"> 生成时间：{now.strftime('%Y-%m-%d %H:%M')} (UTC+8)  ",
        f"> skills 数据更新：{updated}  ",
        "> 数据来源：[skills.sh/trending](https://skills.sh/trending) · [skills.sh/hot](https://skills.sh/hot)"
        + (" · AI 时讯 Deep Research" if news_merged else " · 国际/国内 AI RSS"),
        "",
        "---",
        "",
        f"## 一、Skills Trending Top {trending_n}",
        "",
        *_format_skill_section(trending),
        "---",
        "",
        f"## 二、Skills Hot Top {hot_n}",
        "",
        *_format_skill_section(hot, hot=True),
        "",
        "---",
        "",
        f"## 三、GitHub Trending Top {github_limit}",
        "",
        trending_data_source_note(),
        "",
    ]

    if github_trending:
        lines.extend(_format_github_repo_section(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))
    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))
    else:
        lines.append(f"*Topic `{topic_name}` 热点获取失败或未配置 GITHUB_TOKEN。*")
        lines.append("")

    section_no = 6
    if news_merged and ai_news_research is not None:
        lines.extend(format_research_news_section(ai_news_research, section_no=section_no))
        section_no += 1
    else:
        lines.extend(format_news_section(ai_news, section_no=section_no))
        section_no += 1
        lines.extend(format_cn_news_section(cn_ai_news, section_no=section_no))
        section_no += 1

    watch = (env("GITHUB_REPOS") or "").strip()
    if watch:
        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)
            lines.append(f"- **{repo}**：{release or '暂无 release'}")
        lines.append("")

    lines.extend(["---", "", f"## {section_no}、主题聚类", ""])
    for theme, examples in themes:
        lines.append(f"### {theme}")
        for ex in examples:
            lines.append(f"- {ex}")
        lines.append("")

    lines.extend(["---", "", "## 安装示例", "", "```bash"])
    for item in 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`*"])

    markdown = "\n".join(lines)

    if agent_wecom:
        wecom_md = replace_wecom_skill_sections(agent_wecom, **board_kwargs)
    else:
        wecom_md = build_wecom_report(
            date_str=date_str,
            time_str=time_str,
            updated=updated,
            highlights=editorial_highlights
            or _build_highlights(trending, hot, github_trending, github_emerging, ai_news, cn_ai_news),
            theme_line=editorial_theme or _top_line(feed),
            ai_news=wecom_ai if not news_merged else None,
            cn_ai_news=wecom_cn if not news_merged else None,
            merged_ai_news=wecom_news if news_merged else None,
            merged_tech_ai_news=wecom_tech_news if news_merged else None,
            trending=[
                _prepare_skill_item(item, prev_ids, r)
                for r, item in enumerate(finalize_wecom_skill_groups(gt), 1)
            ],
            hot=[
                _prepare_skill_item(item, prev_ids, r)
                for r, item in enumerate(finalize_wecom_skill_groups(gh), 1)
            ],
            repos=wecom_github_items,
            emerging=wecom_emerging_items,
            topic_name=topic_name,
            topic_repos=wecom_topic_items,
            pick_command=pick_command,
            pick_why=pick_why,
            pick_title=pick_title,
            pick_url=pick_url,
            include_boards=(eff_mode == "full"),
        )
        wecom_md = replace_wecom_board_sections(wecom_md, **board_kwargs)

    wecom_md = replace_wecom_news_sections(
        wecom_md,
        ai_news=wecom_news if news_merged else wecom_ai,
        cn_ai_news=None if news_merged else wecom_cn,
        tech_ai_news=wecom_tech_news if news_merged else None,
        merged=news_merged,
    )

    if push_gate.should_push:
        if news_merged:
            links = [x["link"] for x in wecom_news + wecom_tech_news if x.get("link")]
        else:
            links = [x["link"] for x in wecom_ai + wecom_cn if x.get("link")]
        record_pushed_links(date_str, links)

    final_boards = resolve_wecom_board_items(**board_kwargs)
    shown_keys = {
        board: extract_shown_keys(board, final_boards.get(board) or [])
        for board in BOARD_KEYS
    }
    llm_input = merge_wecom_shown_into_data(llm_input, shown_keys)

    save_json(
        data_json_path(date_str),
        build_full_payload(
            llm_input,
            meta={
                "generated_at": now.isoformat(),
                "report_mode": "agent" if is_agent_mode() else "classic",
                "ai_news_mode": "research" if news_merged else "rss",
                "cursor_editor": cursor_editor_enabled() and not is_agent_mode(),
                "featured_pick": featured.get("title") if featured else None,
                "effective_wecom_mode": eff_mode,
                "push_gate": {
                    "should_push": push_gate.should_push,
                    "silent": push_gate.silent,
                    "reasons": push_gate.reasons,
                },
            },
        ),
    )

    _save_snapshot(feed, 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.write_text(markdown, encoding="utf-8")
    out_wecom.write_text(wecom_md, encoding="utf-8")
    return markdown, wecom_md, out_md, out_wecom


def generate_report() -> tuple[str, str, Path, Path]:
    """编排三段: 抓取(_collect) → 选择(_select) → 拼装(_render)。"""
    collected = _collect(_now_cst().strftime("%Y-%m-%d"))
    selected = _select(collected)
    return _render(collected, selected)


def main() -> int:
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    log_file = LOG_DIR / f"{_now_cst():%Y-%m-%d}.log"
    try:
        _, wecom_md, out_md, out_wecom = generate_report()
        nbytes = len(wecom_md.encode("utf-8"))
        msg = f"[{_now_cst():%H:%M:%S}] OK -> {out_md}, {out_wecom} ({nbytes} bytes)\n"
        log_file.write_text(msg, encoding="utf-8")
        print(msg.strip())
        return 0
    except Exception as exc:
        msg = f"[{_now_cst():%H:%M:%S}] FAIL: {exc}\n"
        log_file.write_text(msg, encoding="utf-8")
        print(msg.strip(), file=sys.stderr)
        return 1