feat: 首推与昨日冲突时改推并保证一月不重复
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
536
daily/featured_pick.py
Normal file
536
daily/featured_pick.py
Normal file
@@ -0,0 +1,536 @@
|
||||
"""今日首推：解析 DAILY_FEATURED_PICK → 定人（月去重）→ LLM 检索 → featured JSON。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import random
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
from daily.config import OUTPUT_DIR, ROOT, env, featured_dedup_days
|
||||
from daily.llm_client import extract_json_object, has_llm_configured, llm_chat
|
||||
from daily.report_data import featured_json_path, save_json
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_SKILL_DIR = ROOT / "skills" / "daily-featured-pick"
|
||||
_GITHUB_REPO_RE = re.compile(r"github\.com/([^/\s#?]+/[^/\s#?]+)", re.I)
|
||||
|
||||
|
||||
def featured_identity_key(featured: dict[str, Any] | None) -> str:
|
||||
"""稳定身份：skill→id，github→repo，兜底从 url 解析。"""
|
||||
if not featured:
|
||||
return ""
|
||||
typ = str(featured.get("type") or "").lower()
|
||||
if typ == "skill" or featured.get("id"):
|
||||
sid = str(featured.get("id") or "").strip()
|
||||
if sid:
|
||||
return sid
|
||||
repo = str(featured.get("repo") or "").strip()
|
||||
if repo:
|
||||
return repo
|
||||
for field in ("url", "command", "link"):
|
||||
url = str(featured.get(field) or "")
|
||||
m = _GITHUB_REPO_RE.search(url)
|
||||
if m:
|
||||
return m.group(1)
|
||||
return ""
|
||||
|
||||
|
||||
def load_recent_featured_keys(date_str: str, days: int | None = None) -> set[str]:
|
||||
"""近 N 日 data.featured_pick_key 并集（不含当日）。"""
|
||||
lookback = days if days is not None else featured_dedup_days()
|
||||
try:
|
||||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||||
except ValueError:
|
||||
return set()
|
||||
out: set[str] = set()
|
||||
for day_offset in range(1, lookback + 1):
|
||||
prev = (dt - timedelta(days=day_offset)).strftime("%Y-%m-%d")
|
||||
path = OUTPUT_DIR / f"{prev}.data.json"
|
||||
if not path.exists():
|
||||
continue
|
||||
try:
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError) as exc:
|
||||
logger.warning("读取 featured_pick_key %s 失败：%s", path, exc)
|
||||
continue
|
||||
data = payload.get("data") if isinstance(payload, dict) else None
|
||||
if not isinstance(data, dict):
|
||||
continue
|
||||
key = str(data.get("featured_pick_key") or "").strip()
|
||||
if key:
|
||||
out.add(key)
|
||||
return out
|
||||
|
||||
|
||||
def load_yesterday_featured_key(date_str: str) -> str | None:
|
||||
try:
|
||||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||||
except ValueError:
|
||||
return None
|
||||
prev = (dt - timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
path = OUTPUT_DIR / f"{prev}.data.json"
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
data = payload.get("data") if isinstance(payload, dict) else None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
key = str(data.get("featured_pick_key") or "").strip()
|
||||
return key or None
|
||||
|
||||
|
||||
def _featured_rng(date_str: str) -> random.Random:
|
||||
seed = int(hashlib.sha256(f"{date_str}:featured".encode()).hexdigest()[:16], 16)
|
||||
return random.Random(seed)
|
||||
|
||||
|
||||
def _stub_from_pool_item(item: dict[str, Any]) -> dict[str, Any]:
|
||||
repo = str(item.get("repo") or "").strip()
|
||||
if repo:
|
||||
url = str(item.get("url") or f"https://github.com/{repo}").strip()
|
||||
return {
|
||||
"type": "github",
|
||||
"title": repo.split("/")[-1],
|
||||
"repo": repo,
|
||||
"url": url,
|
||||
"command": url,
|
||||
"summary": str(item.get("description") or "")[:160],
|
||||
"why_today": "",
|
||||
"evidence": [],
|
||||
"tags": [],
|
||||
}
|
||||
sid = str(item.get("id") or "").strip()
|
||||
source = str(item.get("source") or "").strip()
|
||||
title = str(item.get("title") or "").strip()
|
||||
return {
|
||||
"type": "skill",
|
||||
"id": sid,
|
||||
"title": title or sid,
|
||||
"command": _skill_command(item),
|
||||
"url": str(item.get("link") or ""),
|
||||
"summary": str(item.get("description") or "")[:160],
|
||||
"why_today": "",
|
||||
"evidence": [],
|
||||
"tags": [],
|
||||
"source": source,
|
||||
}
|
||||
|
||||
|
||||
def featured_resolve(
|
||||
*,
|
||||
date_str: str,
|
||||
candidate: dict[str, Any] | None,
|
||||
pool_a: list[dict[str, Any]],
|
||||
pool_b: list[dict[str, Any]],
|
||||
recent_featured: set[str] | None = None,
|
||||
yesterday_key: str | None = None,
|
||||
rng: random.Random | None = None,
|
||||
) -> tuple[dict[str, Any] | None, str | None]:
|
||||
"""若与昨日同一身份则改推；返回 (seed_stub, identity_key)，不含完整 why。"""
|
||||
if not candidate:
|
||||
return None, None
|
||||
key = featured_identity_key(candidate)
|
||||
if not yesterday_key or key != yesterday_key:
|
||||
return candidate, key or None
|
||||
|
||||
blocked = set(recent_featured or set()) | {yesterday_key}
|
||||
picker = rng or _featured_rng(date_str)
|
||||
|
||||
def _choices(pool: list[dict[str, Any]]) -> list[tuple[str, dict[str, Any]]]:
|
||||
out: list[tuple[str, dict[str, Any]]] = []
|
||||
seen: set[str] = set()
|
||||
for item in pool:
|
||||
ik = featured_identity_key(item)
|
||||
if not ik or ik in blocked or ik in seen:
|
||||
continue
|
||||
seen.add(ik)
|
||||
out.append((ik, item))
|
||||
return out
|
||||
|
||||
options = _choices(pool_a)
|
||||
if not options:
|
||||
options = _choices(pool_b)
|
||||
if not options:
|
||||
logger.info("featured_fallback_exhausted")
|
||||
return candidate, key
|
||||
|
||||
chosen_key, chosen_item = picker.choice(options)
|
||||
return _stub_from_pool_item(chosen_item), chosen_key
|
||||
|
||||
|
||||
def _config_from_candidate(candidate: dict[str, Any]) -> dict[str, str]:
|
||||
typ = str(candidate.get("type") or "").lower()
|
||||
if typ == "skill" or candidate.get("id"):
|
||||
query = str(candidate.get("id") or candidate.get("title") or "").strip()
|
||||
return {"query": query, "url_hint": str(candidate.get("url") or "")}
|
||||
repo = str(candidate.get("repo") or "").strip()
|
||||
if repo:
|
||||
return {
|
||||
"query": repo,
|
||||
"url_hint": str(candidate.get("url") or f"https://github.com/{repo}"),
|
||||
}
|
||||
query = str(candidate.get("title") or candidate.get("url") or "").strip()
|
||||
return {"query": query or "featured", "url_hint": str(candidate.get("url") or "")}
|
||||
|
||||
|
||||
def _seed_candidate_from_config(
|
||||
config: dict[str, str],
|
||||
llm_input: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
matches = match_in_data(llm_input, config["query"])
|
||||
if matches["skills"]:
|
||||
return _stub_from_pool_item(matches["skills"][0])
|
||||
if matches["github"]:
|
||||
return _stub_from_pool_item(matches["github"][0])
|
||||
url = config.get("url_hint") or ""
|
||||
seed: dict[str, Any] = {
|
||||
"type": "other",
|
||||
"title": config["query"],
|
||||
"url": url,
|
||||
"command": url or config["query"],
|
||||
}
|
||||
m = _GITHUB_REPO_RE.search(url)
|
||||
if m:
|
||||
seed["type"] = "github"
|
||||
seed["repo"] = m.group(1)
|
||||
return seed
|
||||
|
||||
|
||||
def parse_featured_pick() -> dict[str, str] | None:
|
||||
"""解析 DAILY_FEATURED_PICK：query 或 query|url。"""
|
||||
raw = (env("DAILY_FEATURED_PICK") or "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
if "|" in raw:
|
||||
query, url_hint = raw.split("|", 1)
|
||||
query = query.strip()
|
||||
url_hint = url_hint.strip()
|
||||
if not query:
|
||||
return None
|
||||
payload: dict[str, str] = {"query": query}
|
||||
if url_hint:
|
||||
payload["url_hint"] = url_hint
|
||||
return payload
|
||||
return {"query": raw}
|
||||
|
||||
|
||||
def _matches_query(text: str, query: str) -> bool:
|
||||
return query.lower() in (text or "").lower()
|
||||
|
||||
|
||||
def _skill_matches(item: dict[str, Any], query: str) -> bool:
|
||||
for key in ("id", "title", "source"):
|
||||
if _matches_query(str(item.get(key) or ""), query):
|
||||
return True
|
||||
for sub in item.get("cluster_skills") or []:
|
||||
if isinstance(sub, str) and _matches_query(sub, query):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def match_in_data(llm_input: dict[str, Any], query: str) -> dict[str, list[dict[str, Any]]]:
|
||||
"""在榜单数据中模糊匹配 query。"""
|
||||
skills: list[dict[str, Any]] = []
|
||||
seen_skill: set[str] = set()
|
||||
for board in ("skills_trending", "skills_hot"):
|
||||
for item in llm_input.get(board) or []:
|
||||
sid = str(item.get("id") or "")
|
||||
if sid in seen_skill:
|
||||
continue
|
||||
if _skill_matches(item, query):
|
||||
seen_skill.add(sid)
|
||||
skills.append({**item, "board": board})
|
||||
if len(skills) >= 5:
|
||||
break
|
||||
if len(skills) >= 5:
|
||||
break
|
||||
|
||||
github: list[dict[str, Any]] = []
|
||||
seen_repo: set[str] = set()
|
||||
for board in ("github_trending", "github_emerging"):
|
||||
for item in llm_input.get(board) or []:
|
||||
repo = str(item.get("repo") or "")
|
||||
if not repo or repo in seen_repo:
|
||||
continue
|
||||
if _matches_query(repo, query):
|
||||
seen_repo.add(repo)
|
||||
github.append({**item, "board": board})
|
||||
if len(github) >= 5:
|
||||
break
|
||||
|
||||
topic = llm_input.get("github_topic") or {}
|
||||
for item in topic.get("repos") or []:
|
||||
repo = str(item.get("repo") or "")
|
||||
if not repo or repo in seen_repo:
|
||||
continue
|
||||
if _matches_query(repo, query):
|
||||
seen_repo.add(repo)
|
||||
github.append({**item, "board": "github_topic"})
|
||||
if len(github) >= 5:
|
||||
break
|
||||
|
||||
return {"skills": skills, "github": github}
|
||||
|
||||
|
||||
def _load_skill() -> str:
|
||||
path = _SKILL_DIR / "SKILL.md"
|
||||
if path.exists():
|
||||
return path.read_text(encoding="utf-8").strip()
|
||||
return "你是早报编辑。根据输入检索今日首推信息，只输出 JSON。"
|
||||
|
||||
|
||||
def _skill_command(item: dict[str, Any]) -> str:
|
||||
source = str(item.get("source") or "").strip()
|
||||
title = str(item.get("title") or "").strip()
|
||||
if source and title:
|
||||
return f"npx skills add {source}/{title}"
|
||||
sid = str(item.get("id") or "").strip()
|
||||
if sid.count("/") >= 2:
|
||||
parts = sid.split("/", 2)
|
||||
return f"npx skills add {parts[0]}/{parts[1]}/{parts[2]}"
|
||||
if sid.count("/") == 1:
|
||||
return f"npx skills add {sid}"
|
||||
return ""
|
||||
|
||||
|
||||
def _evidence_from_skill(item: dict[str, Any]) -> list[str]:
|
||||
board = item.get("board", "")
|
||||
board_label = {
|
||||
"skills_trending": "Skills Trending",
|
||||
"skills_hot": "Skills Hot",
|
||||
}.get(str(board), str(board))
|
||||
installs = item.get("installs_fmt") or item.get("installs")
|
||||
title = item.get("title") or item.get("id") or "?"
|
||||
if installs:
|
||||
return [f"{board_label} 匹配 · {title} · {installs}"]
|
||||
return [f"{board_label} 匹配 · {title}"]
|
||||
|
||||
|
||||
def _evidence_from_github(item: dict[str, Any]) -> list[str]:
|
||||
board = item.get("board", "")
|
||||
board_label = {
|
||||
"github_trending": "GitHub Trending",
|
||||
"github_emerging": "GitHub 新兴",
|
||||
"github_topic": "GitHub Topic",
|
||||
}.get(str(board), str(board))
|
||||
repo = item.get("repo") or "?"
|
||||
stars = item.get("total_stars_fmt") or ""
|
||||
if stars:
|
||||
return [f"{board_label} 匹配 · {repo} · ⭐{stars}"]
|
||||
return [f"{board_label} 匹配 · {repo}"]
|
||||
|
||||
|
||||
def _fallback_featured(
|
||||
config: dict[str, str],
|
||||
llm_input: dict[str, Any],
|
||||
*,
|
||||
partial: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""LLM 不可用或失败时，用榜单匹配 + 配置回退。"""
|
||||
partial = partial or {}
|
||||
matches = match_in_data(llm_input, config["query"])
|
||||
skill = matches["skills"][0] if matches["skills"] else None
|
||||
gh = matches["github"][0] if matches["github"] else None
|
||||
|
||||
if skill:
|
||||
featured: dict[str, Any] = {
|
||||
"title": str(skill.get("title") or config["query"]),
|
||||
"type": "skill",
|
||||
"command": _skill_command(skill),
|
||||
"url": str(skill.get("link") or config.get("url_hint") or ""),
|
||||
"summary": str(partial.get("summary") or skill.get("description") or "")[:160],
|
||||
"why_today": str(
|
||||
partial.get("why_today")
|
||||
or f"今日 Skills 榜匹配到 **{skill.get('title') or config['query']}**，适合作为首推。"
|
||||
),
|
||||
"evidence": list(partial.get("evidence") or _evidence_from_skill(skill)),
|
||||
"tags": list(partial.get("tags") or []),
|
||||
}
|
||||
if skill.get("id"):
|
||||
featured["id"] = skill["id"]
|
||||
return featured
|
||||
|
||||
if gh:
|
||||
return {
|
||||
"title": str(gh.get("repo") or config["query"]).split("/")[-1],
|
||||
"type": "github",
|
||||
"command": str(gh.get("url") or config.get("url_hint") or ""),
|
||||
"url": str(gh.get("url") or config.get("url_hint") or ""),
|
||||
"summary": str(partial.get("summary") or gh.get("description") or "")[:160],
|
||||
"why_today": str(
|
||||
partial.get("why_today")
|
||||
or f"今日 GitHub 榜匹配到 **{gh.get('repo')}**，适合作为首推。"
|
||||
),
|
||||
"evidence": list(partial.get("evidence") or _evidence_from_github(gh)),
|
||||
"tags": list(partial.get("tags") or []),
|
||||
"repo": gh.get("repo"),
|
||||
}
|
||||
|
||||
url = config.get("url_hint") or ""
|
||||
return {
|
||||
"title": config["query"],
|
||||
"type": "other",
|
||||
"command": url or config["query"],
|
||||
"url": url,
|
||||
"summary": str(partial.get("summary") or "")[:160],
|
||||
"why_today": str(
|
||||
partial.get("why_today")
|
||||
or f"**{config['query']}** 未出现在今日 Top 榜，仍值得单独关注。"
|
||||
),
|
||||
"evidence": list(partial.get("evidence") or ([f"主推 · {config['query']}"])),
|
||||
"tags": list(partial.get("tags") or []),
|
||||
}
|
||||
|
||||
|
||||
def _normalize_featured(
|
||||
raw: dict[str, Any],
|
||||
config: dict[str, str],
|
||||
llm_input: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""补齐 command / url / evidence，并与榜单数字对齐。"""
|
||||
matches = match_in_data(llm_input, config["query"])
|
||||
skill = matches["skills"][0] if matches["skills"] else None
|
||||
gh = matches["github"][0] if matches["github"] else None
|
||||
|
||||
featured = dict(raw)
|
||||
featured.setdefault("title", config["query"])
|
||||
featured.setdefault("type", "other")
|
||||
|
||||
if skill and featured.get("type") in {"skill", "other", ""}:
|
||||
featured.setdefault("id", skill.get("id"))
|
||||
featured.setdefault("command", _skill_command(skill))
|
||||
featured.setdefault("url", skill.get("link") or config.get("url_hint") or "")
|
||||
if not featured.get("evidence"):
|
||||
featured["evidence"] = _evidence_from_skill(skill)
|
||||
featured["type"] = "skill"
|
||||
elif gh and featured.get("type") in {"github", "other", ""}:
|
||||
featured.setdefault("repo", gh.get("repo"))
|
||||
featured.setdefault("url", gh.get("url") or config.get("url_hint") or "")
|
||||
featured.setdefault("command", featured.get("url") or gh.get("url") or "")
|
||||
if not featured.get("evidence"):
|
||||
featured["evidence"] = _evidence_from_github(gh)
|
||||
featured["type"] = "github"
|
||||
|
||||
featured.setdefault("command", config.get("url_hint") or config["query"])
|
||||
featured.setdefault("url", config.get("url_hint") or "")
|
||||
featured.setdefault("summary", "")
|
||||
featured.setdefault("why_today", featured.get("summary") or "")
|
||||
featured.setdefault("evidence", [])
|
||||
featured.setdefault("tags", [])
|
||||
return featured
|
||||
|
||||
|
||||
def research_featured_pick(
|
||||
llm_input: dict[str, Any],
|
||||
*,
|
||||
date_str: str,
|
||||
config: dict[str, str] | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Step 0：检索今日首推；成功返回 featured dict，未配置返回 None。"""
|
||||
config = config or parse_featured_pick()
|
||||
if not config:
|
||||
return None
|
||||
|
||||
matches = match_in_data(llm_input, config["query"])
|
||||
payload = {
|
||||
"query": config["query"],
|
||||
"url_hint": config.get("url_hint"),
|
||||
"cwd": env("DAILY_CURSOR_CWD") or str(ROOT),
|
||||
"data_matches": matches,
|
||||
}
|
||||
|
||||
if not has_llm_configured():
|
||||
featured = _fallback_featured(config, llm_input)
|
||||
save_json(featured_json_path(date_str), featured)
|
||||
logger.info("Featured pick（无 LLM，规则回退）：%s", featured.get("title"))
|
||||
return featured
|
||||
|
||||
skill = _load_skill()
|
||||
system = (
|
||||
f"{skill}\n\n"
|
||||
"当前执行 **Step 0：今日首推检索**。\n"
|
||||
"只输出 featured JSON（title, type, command, url, summary, why_today, evidence, tags），"
|
||||
"不要 Markdown，不要解释。"
|
||||
)
|
||||
user = json.dumps(payload, ensure_ascii=False, indent=2)
|
||||
try:
|
||||
raw = llm_chat(system, user)
|
||||
except Exception as exc:
|
||||
logger.warning("Featured pick LLM 失败，回退规则模式：%s", exc)
|
||||
featured = _fallback_featured(config, llm_input)
|
||||
save_json(featured_json_path(date_str), featured)
|
||||
return featured
|
||||
|
||||
if not raw:
|
||||
featured = _fallback_featured(config, llm_input)
|
||||
save_json(featured_json_path(date_str), featured)
|
||||
return featured
|
||||
|
||||
parsed = extract_json_object(raw)
|
||||
if not parsed.get("why_today") and not parsed.get("summary"):
|
||||
logger.warning("Featured pick JSON 无效，回退规则模式")
|
||||
featured = _fallback_featured(config, llm_input, partial=parsed)
|
||||
save_json(featured_json_path(date_str), featured)
|
||||
return featured
|
||||
|
||||
featured = _normalize_featured(parsed, config, llm_input)
|
||||
save_json(featured_json_path(date_str), featured)
|
||||
logger.info("Featured pick 完成：%s", featured.get("title"))
|
||||
return featured
|
||||
|
||||
|
||||
def apply_featured_pick(
|
||||
llm_input: dict[str, Any],
|
||||
*,
|
||||
date_str: str,
|
||||
pool_a: list[dict[str, Any]] | None = None,
|
||||
pool_b: list[dict[str, Any]] | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""先定人（相对昨日改推 + 月去重），再 research，写入 featured_pick / featured_pick_key。"""
|
||||
config = parse_featured_pick()
|
||||
if not config:
|
||||
return None
|
||||
|
||||
seed = _seed_candidate_from_config(config, llm_input)
|
||||
recent = load_recent_featured_keys(date_str)
|
||||
yesterday = load_yesterday_featured_key(date_str)
|
||||
resolved, identity_key = featured_resolve(
|
||||
date_str=date_str,
|
||||
candidate=seed,
|
||||
pool_a=pool_a or [],
|
||||
pool_b=pool_b or [],
|
||||
recent_featured=recent,
|
||||
yesterday_key=yesterday,
|
||||
rng=_featured_rng(date_str),
|
||||
)
|
||||
research_config = config
|
||||
if resolved and identity_key and featured_identity_key(seed) != identity_key:
|
||||
research_config = _config_from_candidate(resolved)
|
||||
|
||||
featured = research_featured_pick(
|
||||
llm_input, date_str=date_str, config=research_config
|
||||
)
|
||||
if featured:
|
||||
llm_input["featured_pick"] = featured
|
||||
key = identity_key or featured_identity_key(featured)
|
||||
if key:
|
||||
llm_input["featured_pick_key"] = key
|
||||
return featured
|
||||
|
||||
|
||||
def pick_command_from_featured(featured: dict[str, Any] | None) -> str | None:
|
||||
cmd = str((featured or {}).get("command") or "").strip()
|
||||
return cmd or None
|
||||
|
||||
|
||||
def pick_why_from_featured(featured: dict[str, Any] | None) -> str | None:
|
||||
why = str((featured or {}).get("why_today") or "").strip()
|
||||
return why or None
|
||||
@@ -50,6 +50,7 @@ from daily.format_wecom import (
|
||||
from daily.agent_workflow import is_agent_mode, run_agent_workflow
|
||||
from daily.featured_pick import (
|
||||
apply_featured_pick,
|
||||
featured_identity_key,
|
||||
pick_command_from_featured,
|
||||
pick_why_from_featured,
|
||||
)
|
||||
@@ -638,7 +639,44 @@ def generate_report() -> tuple[str, str, Path, Path]:
|
||||
research_tech_items=wecom_tech_news if news_merged else None,
|
||||
boards_for_wecom=boards_for_wecom,
|
||||
)
|
||||
featured = apply_featured_pick(llm_input, date_str=date_str)
|
||||
pool_a: list[dict[str, Any]] = []
|
||||
pool_a_keys: set[str] = set()
|
||||
for board_name, items in boards_for_wecom.items():
|
||||
for item in items:
|
||||
keyed = dict(item)
|
||||
keyed["board"] = board_name
|
||||
pool_a.append(keyed)
|
||||
ik = featured_identity_key(keyed)
|
||||
if ik:
|
||||
pool_a_keys.add(ik)
|
||||
pool_b: list[dict[str, Any]] = []
|
||||
for board_name, raw_items, kind in (
|
||||
("skills_trending", trending, "skill"),
|
||||
("skills_hot", hot, "skill"),
|
||||
("github_trending", github_trending, "github"),
|
||||
("github_emerging", github_emerging, "github"),
|
||||
("github_topic", github_topic, "github"),
|
||||
):
|
||||
deep = board_select(
|
||||
board=board_name,
|
||||
items=raw_items,
|
||||
recent_keys=set(),
|
||||
limit=pool,
|
||||
pool_size=pool,
|
||||
kind=kind, # type: ignore[arg-type]
|
||||
)
|
||||
for item in deep:
|
||||
keyed = dict(item)
|
||||
keyed["board"] = board_name
|
||||
ik = featured_identity_key(keyed)
|
||||
if ik and ik not in pool_a_keys:
|
||||
pool_b.append(keyed)
|
||||
featured = apply_featured_pick(
|
||||
llm_input,
|
||||
date_str=date_str,
|
||||
pool_a=pool_a,
|
||||
pool_b=pool_b,
|
||||
)
|
||||
movement = llm_input["movement"]
|
||||
eff_mode = llm_input["effective_wecom_mode"]
|
||||
if news_merged:
|
||||
|
||||
83
tests/test_featured_resolve.py
Normal file
83
tests/test_featured_resolve.py
Normal file
@@ -0,0 +1,83 @@
|
||||
# tests/test_featured_resolve.py
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
import unittest
|
||||
|
||||
from daily.featured_pick import featured_identity_key, featured_resolve
|
||||
|
||||
|
||||
class FeaturedResolveTests(unittest.TestCase):
|
||||
def test_same_as_yesterday_picks_from_pool_a(self):
|
||||
yesterday_key = "headroomlabs-ai/headroom"
|
||||
pool_a = [
|
||||
{"repo": "headroomlabs-ai/headroom", "board": "github_topic"},
|
||||
{"repo": "ollama/ollama", "board": "github_trending"},
|
||||
]
|
||||
rng = random.Random(0)
|
||||
resolved, key = featured_resolve(
|
||||
date_str="2026-07-14",
|
||||
candidate={
|
||||
"type": "github",
|
||||
"url": "https://github.com/headroomlabs-ai/headroom",
|
||||
"title": "headroom",
|
||||
},
|
||||
pool_a=pool_a,
|
||||
pool_b=[],
|
||||
recent_featured={yesterday_key},
|
||||
yesterday_key=yesterday_key,
|
||||
rng=rng,
|
||||
)
|
||||
self.assertNotEqual(key, yesterday_key)
|
||||
self.assertEqual(key, "ollama/ollama")
|
||||
self.assertIsNotNone(resolved)
|
||||
assert resolved is not None
|
||||
self.assertEqual(resolved.get("repo"), "ollama/ollama")
|
||||
|
||||
def test_pool_a_before_pool_b(self):
|
||||
yesterday_key = "blocked/one"
|
||||
pool_a = [{"repo": "pool-a/repo", "board": "github_trending"}]
|
||||
pool_b = [{"repo": "pool-b/repo", "board": "github_emerging"}]
|
||||
resolved, key = featured_resolve(
|
||||
date_str="2026-07-14",
|
||||
candidate={"type": "github", "repo": "blocked/one", "title": "one"},
|
||||
pool_a=pool_a,
|
||||
pool_b=pool_b,
|
||||
recent_featured={yesterday_key},
|
||||
yesterday_key=yesterday_key,
|
||||
rng=random.Random(1),
|
||||
)
|
||||
self.assertEqual(key, "pool-a/repo")
|
||||
assert resolved is not None
|
||||
self.assertEqual(resolved.get("repo"), "pool-a/repo")
|
||||
|
||||
def test_exhausted_keeps_original(self):
|
||||
yesterday_key = "only/one"
|
||||
candidate = {"type": "github", "repo": "only/one", "title": "one"}
|
||||
resolved, key = featured_resolve(
|
||||
date_str="2026-07-14",
|
||||
candidate=candidate,
|
||||
pool_a=[{"repo": "only/one", "board": "github_trending"}],
|
||||
pool_b=[],
|
||||
recent_featured={yesterday_key},
|
||||
yesterday_key=yesterday_key,
|
||||
rng=random.Random(2),
|
||||
)
|
||||
self.assertEqual(key, yesterday_key)
|
||||
self.assertEqual(resolved, candidate)
|
||||
|
||||
def test_identity_key_skill_and_github(self):
|
||||
self.assertEqual(
|
||||
featured_identity_key({"type": "skill", "id": "a/b/c"}),
|
||||
"a/b/c",
|
||||
)
|
||||
self.assertEqual(
|
||||
featured_identity_key(
|
||||
{"type": "github", "url": "https://github.com/foo/bar"}
|
||||
),
|
||||
"foo/bar",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user