249 lines
8.8 KiB
Python
249 lines
8.8 KiB
Python
"""早报结构化数据：抓取结果 → JSON 中间层。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from daily.config import OUTPUT_DIR, env_int, wecom_mode
|
||
from daily.delta import build_movement_baseline, build_movement_context, compare_depth, effective_wecom_mode
|
||
from daily.news.fetch import prepare_wecom_cn_news_items, prepare_wecom_news_items
|
||
from daily.skills_group import group_skills_by_source
|
||
from daily.text_utils import trim_brief
|
||
|
||
|
||
def skill_id(item: dict[str, Any]) -> str:
|
||
return str(item.get("id") or f"{item.get('source')}/{item.get('title')}")
|
||
|
||
|
||
def _slim_skill(item: dict[str, Any]) -> dict[str, Any]:
|
||
payload = {
|
||
"id": skill_id(item),
|
||
"title": item.get("title", ""),
|
||
"source": item.get("source", ""),
|
||
"installs": item.get("installs", 0),
|
||
"link": item.get("link", ""),
|
||
"description": item.get("description", ""),
|
||
}
|
||
if item.get("cluster"):
|
||
payload.update(
|
||
{
|
||
"cluster": True,
|
||
"cluster_count": item.get("cluster_count", 1),
|
||
"cluster_skills": item.get("cluster_skills", []),
|
||
"cluster_titles": item.get("cluster_titles", ""),
|
||
"installs_fmt": item.get("installs_fmt", ""),
|
||
"installs_min": item.get("installs_min"),
|
||
"installs_max": item.get("installs_max"),
|
||
}
|
||
)
|
||
elif item.get("installs_fmt"):
|
||
payload["installs_fmt"] = item.get("installs_fmt")
|
||
return payload
|
||
|
||
|
||
def _slim_github(item: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"repo": item.get("repo", ""),
|
||
"url": item.get("url", ""),
|
||
"language": item.get("language", ""),
|
||
"stars_today_fmt": item.get("stars_today_fmt", ""),
|
||
"total_stars_fmt": item.get("total_stars_fmt", ""),
|
||
"created_at": item.get("created_at", ""),
|
||
"description": item.get("description", ""),
|
||
}
|
||
|
||
|
||
def _slim_news_items(
|
||
ai_news: dict[str, Any],
|
||
limit: int,
|
||
*,
|
||
prepare=prepare_wecom_news_items,
|
||
date_str: str | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
items: list[dict[str, Any]] = []
|
||
for item in prepare(ai_news, date_str=date_str):
|
||
items.append(
|
||
{
|
||
"link": item.get("link", ""),
|
||
"title": item.get("title", ""),
|
||
"source_name": item.get("source_name", ""),
|
||
"published_fmt": item.get("published_fmt", ""),
|
||
"summary": trim_brief(
|
||
item.get("summary_plain") or item.get("desc_short") or "",
|
||
120,
|
||
),
|
||
}
|
||
)
|
||
if len(items) >= limit:
|
||
break
|
||
if items:
|
||
return items
|
||
for item in (ai_news.get("flat") or [])[:limit]:
|
||
items.append(
|
||
{
|
||
"link": item.get("link", ""),
|
||
"title": item.get("title", ""),
|
||
"source_name": item.get("source_name", ""),
|
||
"published_fmt": item.get("published_fmt", ""),
|
||
"summary": item.get("summary", ""),
|
||
}
|
||
)
|
||
return items
|
||
|
||
|
||
def _wecom_skill_pool() -> int:
|
||
return max(10, env_int("DAILY_WECOM_SKILL_POOL", 200))
|
||
|
||
|
||
def build_llm_input(
|
||
*,
|
||
date_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],
|
||
wecom_limits: dict[str, int],
|
||
research_items: list[dict[str, Any]] | None = None,
|
||
research_tech_items: list[dict[str, Any]] | None = None,
|
||
boards_for_wecom: dict[str, list[dict[str, Any]]] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""供 Cursor 编辑的精简 JSON（不含完整 markdown）。"""
|
||
news_limit = wecom_limits.get("ai_news", 10)
|
||
cn_news_limit = wecom_limits.get("cn_ai_news", 8)
|
||
if research_items is not None:
|
||
slim_research = [
|
||
{
|
||
"link": item.get("link", ""),
|
||
"title": item.get("title", ""),
|
||
"source_name": item.get("source_name", ""),
|
||
"published_fmt": item.get("published_fmt", ""),
|
||
"summary": trim_brief(item.get("desc_short") or "", 120),
|
||
}
|
||
for item in research_items[:news_limit]
|
||
]
|
||
ai_news_payload = slim_research
|
||
tech_news_payload = [
|
||
{
|
||
"link": item.get("link", ""),
|
||
"title": item.get("title", ""),
|
||
"source_name": item.get("source_name", ""),
|
||
"published_fmt": item.get("published_fmt", ""),
|
||
"summary": trim_brief(item.get("desc_short") or "", 120),
|
||
}
|
||
for item in (research_tech_items or [])
|
||
]
|
||
cn_news_payload: list[dict[str, Any]] = []
|
||
ai_news_mode = "research"
|
||
else:
|
||
ai_news_payload = _slim_news_items(ai_news, news_limit, date_str=date_str) if ai_news.get("enabled") else []
|
||
cn_news_payload = (
|
||
_slim_news_items(cn_ai_news, cn_news_limit, prepare=prepare_wecom_cn_news_items, date_str=date_str)
|
||
if cn_ai_news.get("enabled")
|
||
else []
|
||
)
|
||
ai_news_mode = "rss"
|
||
tech_news_payload: list[dict[str, Any]] = []
|
||
depth = compare_depth()
|
||
trend_cmp = trending[:depth]
|
||
hot_cmp = hot[:depth]
|
||
github_cmp = github_trending[:depth]
|
||
emerging_cmp = github_emerging[:depth]
|
||
topic_cmp = github_topic[:depth]
|
||
|
||
if boards_for_wecom:
|
||
trending_slice = boards_for_wecom.get("skills_trending") or []
|
||
hot_slice = boards_for_wecom.get("skills_hot") or []
|
||
github_slice = boards_for_wecom.get("github_trending") or []
|
||
emerging_slice = boards_for_wecom.get("github_emerging") or []
|
||
topic_slice = boards_for_wecom.get("github_topic") or []
|
||
else:
|
||
trending_slice = group_skills_by_source(
|
||
trending,
|
||
limit=wecom_limits.get("trending", 10),
|
||
pool_size=wecom_limits.get("trending_pool", _wecom_skill_pool()),
|
||
)
|
||
hot_slice = group_skills_by_source(
|
||
hot,
|
||
limit=wecom_limits.get("hot", 10),
|
||
pool_size=wecom_limits.get("hot_pool", _wecom_skill_pool()),
|
||
)
|
||
github_slice = github_trending[: wecom_limits.get("github", 5)]
|
||
emerging_slice = github_emerging[: wecom_limits.get("emerging", 3)]
|
||
topic_slice = github_topic[: wecom_limits.get("topic", 3)]
|
||
|
||
movement = build_movement_context(
|
||
date_str=date_str,
|
||
trending=[_slim_skill(x) for x in trend_cmp],
|
||
hot=[_slim_skill(x) for x in hot_cmp],
|
||
github_trending=[_slim_github(x) for x in github_cmp],
|
||
github_emerging=[_slim_github(x) for x in emerging_cmp],
|
||
github_topic=[_slim_github(x) for x in topic_cmp],
|
||
topic_name=topic_name,
|
||
)
|
||
movement_baseline = build_movement_baseline(
|
||
trending=[_slim_skill(x) for x in trend_cmp],
|
||
hot=[_slim_skill(x) for x in hot_cmp],
|
||
github_trending=[_slim_github(x) for x in github_cmp],
|
||
github_emerging=[_slim_github(x) for x in emerging_cmp],
|
||
github_topic=[_slim_github(x) for x in topic_cmp],
|
||
depth=depth,
|
||
)
|
||
eff_mode = effective_wecom_mode(date_str=date_str)
|
||
|
||
return {
|
||
"date": date_str,
|
||
"data_updated": updated,
|
||
"wecom_mode": wecom_mode(),
|
||
"effective_wecom_mode": eff_mode,
|
||
"skills_trending": [_slim_skill(x) for x in trending_slice],
|
||
"skills_hot": [_slim_skill(x) for x in hot_slice],
|
||
"github_trending": [_slim_github(x) for x in github_slice],
|
||
"github_emerging": [_slim_github(x) for x in emerging_slice],
|
||
"github_topic": {
|
||
"topic": topic_name,
|
||
"repos": [_slim_github(x) for x in topic_slice],
|
||
},
|
||
"ai_news": ai_news_payload,
|
||
"tech_ai_news": tech_news_payload,
|
||
"cn_ai_news": cn_news_payload,
|
||
"ai_news_mode": ai_news_mode,
|
||
"movement": movement,
|
||
"movement_baseline": movement_baseline,
|
||
}
|
||
|
||
|
||
def build_full_payload(
|
||
llm_input: dict[str, Any],
|
||
*,
|
||
meta: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
return {"meta": meta, "data": llm_input}
|
||
|
||
|
||
def data_json_path(date_str: str) -> Path:
|
||
return OUTPUT_DIR / f"{date_str}.data.json"
|
||
|
||
|
||
def editorial_json_path(date_str: str) -> Path:
|
||
return OUTPUT_DIR / f"{date_str}.editorial.json"
|
||
|
||
|
||
def featured_json_path(date_str: str) -> Path:
|
||
return OUTPUT_DIR / f"{date_str}.featured.json"
|
||
|
||
|
||
def save_json(path: Path, data: dict[str, Any]) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
|
||
def load_json(path: Path) -> dict[str, Any]:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|