314 lines
10 KiB
Python
314 lines
10 KiB
Python
"""榜单异动：对比昨日 Top N，仅识别「新入榜」条目（分榜、限条）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any, Callable
|
||
|
||
from daily.config import OUTPUT_DIR, env_int
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
KeyFn = Callable[[dict[str, Any]], str]
|
||
|
||
|
||
def compare_depth() -> int:
|
||
return env_int("DAILY_DELTA_COMPARE_DEPTH", 15)
|
||
|
||
|
||
def wecom_new_limit() -> int:
|
||
return env_int("DAILY_WECOM_NEW_MAX", 10)
|
||
|
||
|
||
def skill_id(item: dict[str, Any]) -> str:
|
||
return str(item.get("id") or f"{item.get('source')}/{item.get('title')}")
|
||
|
||
|
||
def _data_json_path(date_str: str) -> Path:
|
||
return OUTPUT_DIR / f"{date_str}.data.json"
|
||
|
||
|
||
def _load_data_json_file(path: Path) -> dict[str, Any]:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _key_set(items: list[dict[str, Any]], key_fn: KeyFn, *, depth: int) -> set[str]:
|
||
return {key_fn(item) for item in items[:depth] if key_fn(item)}
|
||
|
||
|
||
def find_previous_data(date_str: str) -> tuple[str, dict[str, Any]] | None:
|
||
"""查找最近一份早于 date_str 的 data.json。"""
|
||
try:
|
||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||
except ValueError:
|
||
return None
|
||
lookback = env_int("DAILY_DELTA_LOOKBACK_DAYS", 7)
|
||
for days in range(1, lookback + 1):
|
||
prev_date = (dt - timedelta(days=days)).strftime("%Y-%m-%d")
|
||
path = _data_json_path(prev_date)
|
||
if not path.exists():
|
||
continue
|
||
try:
|
||
payload = _load_data_json_file(path)
|
||
except (OSError, ValueError) as exc:
|
||
logger.warning("读取异动基准 %s 失败：%s", path, exc)
|
||
continue
|
||
data = payload.get("data")
|
||
if isinstance(data, dict) and data.get("date"):
|
||
return prev_date, data
|
||
return None
|
||
|
||
|
||
def _prev_board_items(prev_data: dict[str, Any], board: str, depth: int) -> list[dict[str, Any]]:
|
||
baseline = prev_data.get("movement_baseline") or {}
|
||
if board in baseline and isinstance(baseline[board], list):
|
||
return baseline[board][:depth]
|
||
legacy = {
|
||
"skills_trending": "skills_trending",
|
||
"skills_hot": "skills_hot",
|
||
"github_trending": "github_trending",
|
||
"github_emerging": "github_emerging",
|
||
"github_topic": "github_topic",
|
||
}
|
||
if board == "github_topic":
|
||
topic = prev_data.get("github_topic") or {}
|
||
repos = topic.get("repos") if isinstance(topic, dict) else []
|
||
return (repos or [])[:depth]
|
||
field = legacy.get(board, board)
|
||
items = prev_data.get(field) or []
|
||
return items[:depth] if isinstance(items, list) else []
|
||
|
||
|
||
def _format_new_note(board: str, rank: int, *, topic_name: str = "llm") -> str:
|
||
labels = {
|
||
"trending": "Skills Trending",
|
||
"hot": "Skills Hot",
|
||
"github_trending": "GitHub Trending",
|
||
"github_emerging": "GitHub 新兴",
|
||
"github_topic": f"Topic `{topic_name}`",
|
||
}
|
||
return f"新入 {labels.get(board, board)} #{rank}"
|
||
|
||
|
||
def _build_skill_board_moves(
|
||
*,
|
||
board: str,
|
||
items: list[dict[str, Any]],
|
||
prev_data: dict[str, Any] | None,
|
||
depth: int,
|
||
) -> list[dict[str, Any]]:
|
||
board_key = f"skills_{board}"
|
||
prev_ids = (
|
||
_key_set(_prev_board_items(prev_data, board_key, depth), skill_id, depth=depth)
|
||
if prev_data
|
||
else set()
|
||
)
|
||
moves: list[dict[str, Any]] = []
|
||
for rank, item in enumerate(items[:depth], 1):
|
||
sid = skill_id(item)
|
||
if not sid or not prev_data or sid in prev_ids:
|
||
continue
|
||
moves.append(
|
||
{
|
||
"kind": "skill",
|
||
"board": board,
|
||
"id": sid,
|
||
"title": item.get("title", ""),
|
||
"source": item.get("source", ""),
|
||
"installs": int(item.get("installs") or 0),
|
||
"link": item.get("link", ""),
|
||
"description": item.get("description", ""),
|
||
"rank": rank,
|
||
"is_new": True,
|
||
"note": _format_new_note(board, rank),
|
||
}
|
||
)
|
||
return moves
|
||
|
||
|
||
def _repo_key(item: dict[str, Any]) -> str:
|
||
return str(item.get("repo") or "")
|
||
|
||
|
||
def _build_github_board_moves(
|
||
*,
|
||
board: str,
|
||
items: list[dict[str, Any]],
|
||
prev_data: dict[str, Any] | None,
|
||
depth: int,
|
||
topic_name: str = "llm",
|
||
) -> list[dict[str, Any]]:
|
||
prev_ids = (
|
||
_key_set(_prev_board_items(prev_data, board, depth), _repo_key, depth=depth)
|
||
if prev_data
|
||
else set()
|
||
)
|
||
moves: list[dict[str, Any]] = []
|
||
for rank, item in enumerate(items[:depth], 1):
|
||
repo = _repo_key(item)
|
||
if not repo or not prev_data or repo in prev_ids:
|
||
continue
|
||
moves.append(
|
||
{
|
||
"kind": "github",
|
||
"board": board,
|
||
"repo": 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", ""),
|
||
"rank": rank,
|
||
"is_new": True,
|
||
"note": _format_new_note(board, rank, topic_name=topic_name),
|
||
}
|
||
)
|
||
return moves
|
||
|
||
|
||
def _board_summary(
|
||
*,
|
||
label: str,
|
||
baseline_date: str | None,
|
||
depth: int,
|
||
moves: list[dict[str, Any]],
|
||
capped: list[dict[str, Any]],
|
||
) -> str:
|
||
if not baseline_date:
|
||
return f"无历史基准，无法判断 {label} Top{depth} 新增"
|
||
if not moves:
|
||
return f"较 {baseline_date} Top{depth} 无新增 {label} 条目"
|
||
total = len(moves)
|
||
shown = len(capped)
|
||
if shown < total:
|
||
return f"较 {baseline_date} Top{depth} 新增 {total} 条，企微展示前 {shown} 条"
|
||
return f"较 {baseline_date} Top{depth} 新增 {total} 条"
|
||
|
||
|
||
def build_movement_baseline(
|
||
*,
|
||
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]],
|
||
depth: int | None = None,
|
||
) -> dict[str, Any]:
|
||
n = depth if depth is not None else compare_depth()
|
||
return {
|
||
"compare_depth": n,
|
||
"skills_trending": trending[:n],
|
||
"skills_hot": hot[:n],
|
||
"github_trending": github_trending[:n],
|
||
"github_emerging": github_emerging[:n],
|
||
"github_topic": github_topic[:n],
|
||
}
|
||
|
||
|
||
def build_movement_context(
|
||
*,
|
||
date_str: 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 = "llm",
|
||
) -> dict[str, Any]:
|
||
"""生成 Agent 可用的新增榜上下文：分 Trending/Hot/Topic，企微每榜最多 wecom_new_limit 条。"""
|
||
depth = compare_depth()
|
||
cap = wecom_new_limit()
|
||
baseline = find_previous_data(date_str)
|
||
baseline_date = baseline[0] if baseline else None
|
||
prev_data = baseline[1] if baseline else None
|
||
topic = (topic_name or "llm").strip() or "llm"
|
||
|
||
skills_trending_all = _build_skill_board_moves(
|
||
board="trending", items=trending, prev_data=prev_data, depth=depth
|
||
)
|
||
skills_hot_all = _build_skill_board_moves(
|
||
board="hot", items=hot, prev_data=prev_data, depth=depth
|
||
)
|
||
github_trending_all = _build_github_board_moves(
|
||
board="github_trending", items=github_trending, prev_data=prev_data, depth=depth
|
||
)
|
||
github_emerging_all = _build_github_board_moves(
|
||
board="github_emerging", items=github_emerging, prev_data=prev_data, depth=depth
|
||
)
|
||
github_topic_all = _build_github_board_moves(
|
||
board="github_topic",
|
||
items=github_topic,
|
||
prev_data=prev_data,
|
||
depth=depth,
|
||
topic_name=topic,
|
||
)
|
||
|
||
skills_trending_moves = skills_trending_all[:cap]
|
||
skills_hot_moves = skills_hot_all[:cap]
|
||
github_trending_moves = github_trending_all[:cap]
|
||
github_emerging_moves = github_emerging_all[:cap]
|
||
github_topic_moves = github_topic_all[:cap]
|
||
topic_label = f"Topic `{topic}`"
|
||
|
||
return {
|
||
"baseline_date": baseline_date,
|
||
"compare_depth": depth,
|
||
"wecom_new_limit": cap,
|
||
"selection_mode": "top_n",
|
||
"topic_name": topic,
|
||
"skills_trending_moves": skills_trending_moves,
|
||
"skills_hot_moves": skills_hot_moves,
|
||
"skills_trending_stable": not skills_trending_all,
|
||
"skills_hot_stable": not skills_hot_all,
|
||
"skills_trending_summary": _board_summary(
|
||
label="Skills Trending",
|
||
baseline_date=baseline_date,
|
||
depth=depth,
|
||
moves=skills_trending_all,
|
||
capped=skills_trending_moves,
|
||
),
|
||
"skills_hot_summary": _board_summary(
|
||
label="Skills Hot",
|
||
baseline_date=baseline_date,
|
||
depth=depth,
|
||
moves=skills_hot_all,
|
||
capped=skills_hot_moves,
|
||
),
|
||
"github_trending_moves": github_trending_moves,
|
||
"github_emerging_moves": github_emerging_moves,
|
||
"github_topic_moves": github_topic_moves,
|
||
"github_trending_stable": not github_trending_all,
|
||
"github_emerging_stable": not github_emerging_all,
|
||
"github_topic_stable": not github_topic_all,
|
||
"github_trending_summary": _board_summary(
|
||
label="GitHub Trending",
|
||
baseline_date=baseline_date,
|
||
depth=depth,
|
||
moves=github_trending_all,
|
||
capped=github_trending_moves,
|
||
),
|
||
"github_emerging_summary": _board_summary(
|
||
label="GitHub 新兴",
|
||
baseline_date=baseline_date,
|
||
depth=depth,
|
||
moves=github_emerging_all,
|
||
capped=github_emerging_moves,
|
||
),
|
||
"github_topic_summary": _board_summary(
|
||
label=topic_label,
|
||
baseline_date=baseline_date,
|
||
depth=depth,
|
||
moves=github_topic_all,
|
||
capped=github_topic_moves,
|
||
),
|
||
# 兼容旧字段（合并，仅供调试）
|
||
"skills_moves": skills_trending_moves + skills_hot_moves,
|
||
"github_moves": github_trending_moves + github_emerging_moves + github_topic_moves,
|
||
"skills_stable": not skills_trending_all and not skills_hot_all,
|
||
"github_stable": not github_trending_all and not github_emerging_all and not github_topic_all,
|
||
}
|