feat: 新增企微 Delta 榜单区块渲染与替换逻辑

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

daily/format_wecom.py

@@ -42,7 +42,7 @@ def _skill_line(rank: int, item: dict[str, Any], *, badge: str = "") -> list[str
         lines = [head]
         if sample or desc:
             hint = desc or sample
-            lines.append(f"   > {hint}")
+            lines.append(f"   {hint}")
         return lines
     title = item.get("title", "?")
     if link:
@@ -51,7 +51,7 @@ def _skill_line(rank: int, item: dict[str, Any], *, badge: str = "") -> list[str
         head = f"{rank}. {badge_prefix}**{title}** · `{source}` · **{installs}**"
     lines = [head]
     if desc:
-        lines.append(f"   > {desc}")
+        lines.append(f"   {desc}")
     return lines
 
 
@@ -178,6 +178,7 @@ def _grouped_skill_to_wecom_item(
         "installs_fmt": installs_fmt,
         "link": item.get("link", ""),
         "desc_short": desc,
+        "badge": item.get("badge", ""),
         "cluster": bool(item.get("cluster")),
         "cluster_count": item.get("cluster_count"),
         "cluster_titles": item.get("cluster_titles"),
@@ -193,19 +194,123 @@ def build_skills_board_section(icon_key: str, board_label: str, items: list[dict
     return "\n".join(lines)
 
 
+def _move_to_wecom_skill_item(move: dict[str, Any]) -> dict[str, Any]:
+    installs = int(move.get("installs") or 0)
+    rank = move.get("rank", "?")
+    board_badge = (move.get("badge") or "").strip()
+    badge = f"[新入 #{rank}]"
+    if board_badge:
+        badge = f"{badge} {board_badge}"
+    return {
+        "title": move.get("title", "?"),
+        "source": move.get("source", "?"),
+        "installs_fmt": move.get("installs_fmt") or str(installs),
+        "link": move.get("link", ""),
+        "description": (move.get("description") or "").strip(),
+        "badge": badge,
+    }
+
+
+def build_skills_delta_sections(
+    trending_moves: list[dict[str, Any]],
+    hot_moves: list[dict[str, Any]],
+) -> str:
+    sections: list[str] = []
+    if trending_moves:
+        prepared = finalize_wecom_skill_groups([_move_to_wecom_skill_item(m) for m in trending_moves])
+        items = [_grouped_skill_to_wecom_item(x) for x in prepared]
+        lines = [f"{ICONS['trending']} **Skills Trending 变化**"]
+        for rank, item in enumerate(items, 1):
+            lines.extend(_skill_line(rank, item, badge=item.get("badge", "")))
+        sections.append("\n".join(lines))
+    if hot_moves:
+        prepared = finalize_wecom_skill_groups([_move_to_wecom_skill_item(m) for m in hot_moves])
+        items = [_grouped_skill_to_wecom_item(x) for x in prepared]
+        lines = [f"{ICONS['hot']} **Skills Hot 变化**"]
+        for rank, item in enumerate(items, 1):
+            lines.extend(_skill_line(rank, item, badge=item.get("badge", "")))
+        sections.append("\n".join(lines))
+    return "\n\n".join(sections)
+
+
+def _github_move_to_repo(move: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "repo": move.get("repo", "?"),
+        "url": move.get("url", ""),
+        "language": move.get("language", ""),
+        "stars_today_fmt": move.get("stars_today_fmt", ""),
+        "total_stars_fmt": move.get("total_stars_fmt", ""),
+        "created_at": move.get("created_at", ""),
+        "description": move.get("description", ""),
+        "desc_short": (move.get("description") or "").strip(),
+        "badge": f"[新入 #{move.get('rank', '?')}]",
+    }
+
+
+def build_github_delta_sections(movement: dict[str, Any], *, topic_name: str) -> str:
+    sections: list[str] = []
+    mapping = [
+        ("github_trending_moves", "github", "GitHub Trending 变化", False),
+        ("github_emerging_moves", "emerging", "GitHub 新兴 变化", True),
+        ("github_topic_moves", "topic", f"Topic `{topic_name}` 变化", False),
+    ]
+    for key, icon_key, label, show_created in mapping:
+        moves = movement.get(key) or []
+        if not moves:
+            continue
+        lines = [f"{ICONS[icon_key]} **{label}**"]
+        for i, move in enumerate(moves, 1):
+            repo = _github_move_to_repo(move)
+            badge = repo.pop("badge", "")
+            chunk = _github_repo_lines([repo], show_created=show_created)
+            if chunk:
+                chunk[0] = f"{i}. {badge} " + chunk[0].split(". ", 1)[-1]
+            lines.extend(chunk)
+        sections.append("\n".join(lines))
+    return "\n\n".join(sections)
+
+
 _SKILL_SECTIONS = re.compile(
     r"📈 \*\*Skills Trending.*?(?=🐙 \*\*GitHub Trending)",
     re.DOTALL,
 )
+_SKILL_TRENDING_BLOCK = re.compile(r"📈 \*\*Skills Trending[^\n]*\n(?:.*?\n)*?(?=\n🔥 \*\*Skills Hot|\n🐙 |\n🌱 |\n🤖 |\Z)", re.DOTALL)
+_SKILL_HOT_BLOCK = re.compile(r"🔥 \*\*Skills Hot[^\n]*\n(?:.*?\n)*?(?=\n🐙 |\n🌱 |\n🤖 |\Z)", re.DOTALL)
+_GITHUB_SECTIONS = re.compile(r"🐙 \*\*GitHub Trending.*", re.DOTALL)
 
 
-def replace_wecom_skill_sections(
+def _strip_board_sections(md: str) -> str:
+    md = _SKILL_TRENDING_BLOCK.sub("", md)
+    md = _SKILL_HOT_BLOCK.sub("", md)
+    if _GITHUB_SECTIONS.search(md):
+        md = _GITHUB_SECTIONS.sub("", md)
+    return re.sub(r"\n{3,}", "\n\n", md).rstrip()
+
+
+def replace_wecom_board_sections(
     md: str,
     *,
+    mode: str,
+    movement: dict[str, Any],
     trending: list[dict[str, Any]],
     hot: list[dict[str, Any]],
+    topic_name: str,
 ) -> str:
-    """用 Python 合并后的 Skills 榜替换或插入 Agent 早报中的对应区块。"""
+    from daily.delta import partition_skill_moves_for_wecom
+
+    if mode == "delta":
+        t_moves, h_moves = partition_skill_moves_for_wecom(
+            movement.get("skills_trending_moves") or [],
+            movement.get("skills_hot_moves") or [],
+        )
+        skills_sec = build_skills_delta_sections(t_moves, h_moves)
+        github_sec = build_github_delta_sections(movement, topic_name=topic_name)
+        board_block = "\n\n".join(x for x in [skills_sec, github_sec] if x)
+        md = _strip_board_sections(md)
+        if board_block:
+            return md + "\n\n" + board_block + "\n"
+        return md + "\n"
+
     trending_sec = build_skills_board_section("trending", "Skills Trending", trending)
     hot_sec = build_skills_board_section("hot", "Skills Hot", hot)
     replacement = f"{trending_sec}\n\n{hot_sec}\n\n"
@@ -218,6 +323,46 @@ def replace_wecom_skill_sections(
     return md.rstrip() + "\n\n" + replacement
 
 
+def replace_wecom_skill_sections(
+    md: str,
+    *,
+    trending: list[dict[str, Any]],
+    hot: list[dict[str, Any]],
+    mode: str = "full",
+    movement: dict[str, Any] | None = None,
+    topic_name: str = "llm",
+) -> str:
+    """用 Python 合并后的 Skills 榜替换或插入 Agent 早报中的对应区块。"""
+    return replace_wecom_board_sections(
+        md,
+        mode=mode,
+        movement=movement or {},
+        trending=trending,
+        hot=hot,
+        topic_name=topic_name,
+    )
+
+
+def _format_pick_link(pick_command: str, *, title: str = "", url: str = "") -> str:
+    cmd = pick_command.strip()
+    if not cmd:
+        return ""
+    label = title.strip()
+    if cmd.startswith("http://") or cmd.startswith("https://"):
+        if not label:
+            m = re.match(r"https?://github\.com/([^/\s#?]+/[^/\s#?]+)", cmd)
+            label = m.group(1) if m else cmd
+        return f"[{label}]({cmd})"
+    m = re.match(r"npx skills add (\S+)", cmd)
+    if m:
+        skill_path = m.group(1)
+        if not label:
+            label = skill_path.split("/")[-1]
+        href = url.strip() or f"https://skills.sh/{skill_path}"
+        return f"[{label}]({href})"
+    return f"`{cmd}`"
+
+
 def build_wecom_report(
     *,
     date_str: str,
@@ -234,6 +379,10 @@ def build_wecom_report(
     ai_news: list[dict[str, Any]] | None = None,
     cn_ai_news: list[dict[str, Any]] | None = None,
     pick_command: str,
+    pick_why: str = "",
+    pick_title: str = "",
+    pick_url: str = "",
+    include_boards: bool = True,
 ) -> str:
     lines = [
         f"{ICONS['header']} **早报 · {date_str}**",
@@ -257,6 +406,7 @@ def build_wecom_report(
         lines.extend(_ai_news_lines(cn_ai_news))
         lines.append("")
 
+    if include_boards:
         lines.append(f"{ICONS['trending']} **Skills Trending Top {len(trending)}**")
         for rank, item in enumerate(trending, 1):
             lines.extend(_skill_line(rank, item, badge=item.get("badge", "")))
@@ -283,6 +433,8 @@ def build_wecom_report(
             lines.append("")
 
     lines.append(f"{ICONS['pick']} **今日首推**")
-    lines.append(f"`{pick_command}`")
+    lines.append(_format_pick_link(pick_command, title=pick_title, url=pick_url))
+    if pick_why:
+        lines.append(f"> {pick_why}")
 
     return "\n".join(lines)

tests/test_wecom_delta.py

@@ -100,3 +100,91 @@ class SkillMovePartitionTests(unittest.TestCase):
             with patch("daily.delta.wecom_mode", return_value="delta"):
                 with patch("daily.delta.delta_baseline_fallback", return_value="full"):
                     self.assertEqual(effective_wecom_mode(date_str="2026-07-10"), "full")
+
+
+class DeltaFormatTests(unittest.TestCase):
+    def test_skills_delta_omits_empty_board(self):
+        from daily.format_wecom import build_skills_delta_sections
+
+        moves = [
+            {
+                "id": "x/y/z",
+                "rank": 3,
+                "title": "z",
+                "source": "x/y",
+                "installs": 10,
+                "installs_fmt": "10",
+                "link": "https://skills.sh/x/y/z",
+                "description": "d",
+                "badge": "Trending #3",
+            }
+        ]
+        text = build_skills_delta_sections(moves, [])
+        self.assertIn("Skills Trending 变化", text)
+        self.assertIn("[新入 #3]", text)
+        self.assertNotIn("Skills Hot 变化", text)
+
+    def test_github_delta_omits_stable_board(self):
+        from daily.format_wecom import build_github_delta_sections
+
+        movement = {
+            "github_trending_moves": [
+                {"repo": "a/b", "url": "https://github.com/a/b", "rank": 1, "language": "Go"}
+            ],
+            "github_emerging_moves": [],
+            "github_topic_moves": [],
+        }
+        text = build_github_delta_sections(movement, topic_name="llm")
+        self.assertIn("GitHub Trending 变化", text)
+        self.assertNotIn("新兴", text)
+
+
+class PushGateTests(unittest.TestCase):
+    def test_push_when_board_has_moves(self):
+        from daily.push_gate import evaluate_push_gate
+
+        gate = evaluate_push_gate(
+            movement={"skills_trending_moves": [{"id": "a/b/c"}], "skills_hot_moves": [], "github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": []},
+            ai_news_items=[],
+            cn_ai_news_items=[],
+            featured_pick=None,
+        )
+        self.assertTrue(gate.should_push)
+        self.assertIn("board_moves", gate.reasons)
+
+    def test_silent_when_all_empty(self):
+        from daily.push_gate import evaluate_push_gate
+
+        gate = evaluate_push_gate(
+            movement={
+                "skills_trending_moves": [],
+                "skills_hot_moves": [],
+                "github_trending_moves": [],
+                "github_emerging_moves": [],
+                "github_topic_moves": [],
+            },
+            ai_news_items=[],
+            cn_ai_news_items=[],
+            featured_pick=None,
+        )
+        self.assertFalse(gate.should_push)
+        self.assertTrue(gate.silent)
+
+    def test_force_push_overrides_silent(self):
+        from daily.push_gate import evaluate_push_gate
+
+        with patch("daily.push_gate.force_push", return_value=True):
+            gate = evaluate_push_gate(
+                movement={
+                    "skills_trending_moves": [],
+                    "skills_hot_moves": [],
+                    "github_trending_moves": [],
+                    "github_emerging_moves": [],
+                    "github_topic_moves": [],
+                },
+                ai_news_items=[],
+                cn_ai_news_items=[],
+                featured_pick=None,
+            )
+        self.assertTrue(gate.should_push)
+        self.assertIn("force_push", gate.reasons)