feat: 早报系统重构与功能增强

- 新增常驻调度器 daily/scheduler.py + run-scheduler.ps1(定时生成/推送)
- 新增 daily/bridge_manager.py:Windows 兼容的 Cursor SDK 桥接
- 新增 skills/daily-featured-pick 首推 Skill 与叙事轴/去重逻辑
- 新闻抓取窗口、GitHub 搜索、企微 delta 模式等多项改进
- 补充设计文档与 superpowers 计划/规范
- 新增对应测试(scheduler、featured_pick、github_search、news_fetch_window 等)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-17 18:12:00 +08:00
parent 6192dd4e2a
commit 6ea2a4e4c6
35 changed files with 4389 additions and 359 deletions

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import json
import logging
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
@@ -15,29 +16,119 @@ logger = logging.getLogger(__name__)
BOARD_KEYS = RECENT_BOARD_KEYS
_GITHUB_REPO_RE = re.compile(r"github\.com/([\w.-]+/[\w.-]+)", re.I)
_SKILL_SH_RE = re.compile(r"skills\.sh/([\w.-]+/[\w.-]+(?:/[\w.-]+)?)", re.I)
# 与企微正文榜单标题对齐;顺序用于切分相邻 section
_WECOM_SECTION_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
("skills_trending", re.compile(r"Skills\s+Trending", re.I)),
("skills_hot", re.compile(r"Skills\s+Hot", re.I)),
("github_trending", re.compile(r"GitHub\s+Trending", re.I)),
("github_emerging", re.compile(r"GitHub\s+新兴", re.I)),
("github_topic", re.compile(r"Topic\s+", re.I)),
)
def extract_shown_keys(board: str, items: list[dict[str, Any]]) -> list[str]:
"""从最终展示 items 抽取稳定 identity key。"""
"""从最终展示 items 抽取稳定 identity key。
Skills 榜同时写入 skill id 与 source便于周去重按仓屏蔽。
"""
keys: list[str] = []
seen: set[str] = set()
for item in items:
if board.startswith("skills_"):
key = skill_id(item)
candidates = [skill_id(item), str(item.get("source") or "").strip()]
else:
key = str(item.get("repo") or "")
if not key or key in seen:
continue
seen.add(key)
keys.append(key)
candidates = [str(item.get("repo") or "")]
for key in candidates:
if not key or key in seen:
continue
seen.add(key)
keys.append(key)
return keys
def _keys_from_board_items(board: str, data: dict[str, Any]) -> set[str]:
if board == "github_topic":
topic = data.get("github_topic") or {}
items = topic.get("repos") if isinstance(topic, dict) else []
else:
items = data.get(board) or []
if not isinstance(items, list):
return set()
return set(extract_shown_keys(board, items))
def parse_wecom_shown_keys(md: str) -> dict[str, set[str]]:
"""从企微 Markdown 按榜单 section 解析已展示 keys冷启动兼容"""
out: dict[str, set[str]] = {board: set() for board in BOARD_KEYS}
if not (md or "").strip():
return out
hits: list[tuple[int, str]] = []
for board, pattern in _WECOM_SECTION_PATTERNS:
for match in pattern.finditer(md):
hits.append((match.start(), board))
if not hits:
return out
hits.sort(key=lambda x: x[0])
for idx, (start, board) in enumerate(hits):
end = hits[idx + 1][0] if idx + 1 < len(hits) else len(md)
chunk = md[start:end]
if board.startswith("skills_"):
out[board].update(_SKILL_SH_RE.findall(chunk))
else:
out[board].update(_GITHUB_REPO_RE.findall(chunk))
return out
def _load_shown_keys_for_day(path: Path, date_str: str) -> dict[str, set[str]] | None:
"""读一日历史:优先 wecom_shown_keys缺省则回退 wecom.md再回退 data 榜字段。"""
empty = {board: set() for board in BOARD_KEYS}
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError) as exc:
logger.warning("读取 wecom_shown_keys %s 失败:%s", path, exc)
return None
data = payload.get("data") if isinstance(payload, dict) else None
if not isinstance(data, dict):
return empty
out: dict[str, set[str]] = {board: set() for board in BOARD_KEYS}
shown = data.get("wecom_shown_keys")
if isinstance(shown, dict):
for board in BOARD_KEYS:
keys = shown.get(board) or []
if isinstance(keys, list):
out[board].update(str(k) for k in keys if k)
if any(out.values()):
return out
wecom_path = OUTPUT_DIR / f"{date_str}.wecom.md"
if wecom_path.exists():
try:
md = wecom_path.read_text(encoding="utf-8")
except OSError as exc:
logger.warning("读取 wecom.md 回退 %s 失败:%s", wecom_path, exc)
else:
parsed = parse_wecom_shown_keys(md)
if any(parsed.values()):
return parsed
for board in BOARD_KEYS:
out[board].update(_keys_from_board_items(board, data))
return out
def load_recent_shown_keys(
date_str: str,
*,
lookback_days: int | None = None,
) -> dict[str, set[str]]:
"""近 N 日 data.wecom_shown_keys 并集(不含当日)。缺省或读失败视为空集。"""
"""近 N 日已展示 keys 并集(不含当日)。缺省或读失败视为空集。"""
empty = {board: set() for board in BOARD_KEYS}
try:
dt = datetime.strptime(date_str, "%Y-%m-%d")
@@ -52,22 +143,11 @@ def load_recent_shown_keys(
path = OUTPUT_DIR / f"{prev_date}.data.json"
if not path.exists():
continue
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError) as exc:
logger.warning("读取 wecom_shown_keys %s 失败:%s", path, exc)
continue
data = payload.get("data")
if not isinstance(data, dict):
continue
shown = data.get("wecom_shown_keys")
if not isinstance(shown, dict):
day_keys = _load_shown_keys_for_day(path, prev_date)
if day_keys is None:
continue
for board in BOARD_KEYS:
keys = shown.get(board) or []
if not isinstance(keys, list):
continue
out[board].update(str(k) for k in keys if k)
out[board].update(day_keys.get(board) or set())
return out