refactor: _theme_clusters 迁入 narrative_axis 并以注入消除反向依赖

theme_clusters/theme_names 经参数注入 THEME_RULES 与 skill_id_fn,
避免 generate.py 的循环 import(T3,为拆分 generate_report 铺路)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-18 13:49:37 +08:00
parent 6ea2a4e4c6
commit 92e0ed9a27
2 changed files with 48 additions and 22 deletions

View File

@@ -65,6 +65,7 @@ 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,
@@ -448,27 +449,6 @@ def _fetch_latest_release_title(repo: str) -> str | None:
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):
@@ -759,7 +739,7 @@ def generate_report() -> tuple[str, str, Path, Path]:
github_topic=github_topic,
)
themes = _theme_clusters(feed)
themes = theme_clusters(feed, theme_rules=THEME_RULES, skill_id_fn=_skill_id)
lines = [
f"# 早报 · {date_str}",

View File

@@ -5,6 +5,7 @@ from __future__ import annotations
import json
import logging
import random
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any
@@ -101,3 +102,48 @@ def enforce_narrative_axis(trends: dict[str, Any], axis: str) -> dict[str, Any]:
out = dict(trends)
out["narrative_axis"] = axis
return out
def theme_clusters(
feed: dict[str, Any],
*,
limit: int = 5,
theme_rules: list[tuple[str, str, list[str]]],
skill_id_fn: Any,
) -> list[tuple[str, list[str]]]:
"""按 THEME_RULES 把 feed 的 topTrending/topHot 聚成 (主题, 示例列表)。
从 generate.py 迁入(原私有 _theme_clusters。theme_rules 与 skill_id_fn
由调用方注入,避免对 generate.py 的反向依赖(防循环 import
"""
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_fn(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 theme_names(
feed: dict[str, Any],
*,
theme_rules: list[tuple[str, str, list[str]]],
skill_id_fn: Any,
limit: int = 3,
) -> list[str]:
"""仅取主题名(不含 markdown 示例),供「今日看点/theme_line」回退文案。"""
return [theme for theme, _ in theme_clusters(
feed, theme_rules=theme_rules, skill_id_fn=skill_id_fn
)][:limit]