yumao
6ea2a4e4c6
- 新增常驻调度器 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>
126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
"""Skills 榜单：同 source 合并为一条（企微 Top N）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
|
||
def skill_id(item: dict[str, Any]) -> str:
|
||
return str(item.get("id") or f"{item.get('source')}/{item.get('title')}")
|
||
|
||
|
||
def source_from_skill_key(key: str) -> str:
|
||
"""从 skill id（source/title…）还原 source：去掉最后一段 title。"""
|
||
parts = [p for p in str(key or "").split("/") if p]
|
||
if len(parts) >= 2:
|
||
return "/".join(parts[:-1])
|
||
return str(key or "").strip()
|
||
|
||
|
||
def expand_skill_recent_keys(keys: set[str] | None) -> set[str]:
|
||
"""周去重 blocklist：保留原始 key，并展开为 source。"""
|
||
out: set[str] = set()
|
||
for key in keys or set():
|
||
k = str(key or "").strip()
|
||
if not k:
|
||
continue
|
||
out.add(k)
|
||
src = source_from_skill_key(k)
|
||
if src:
|
||
out.add(src)
|
||
return out
|
||
|
||
|
||
def format_installs(n: int | float) -> str:
|
||
if n >= 1_000_000:
|
||
return f"{n / 1_000_000:.1f}M"
|
||
if n >= 1_000:
|
||
return f"{n / 1_000:.1f}K"
|
||
return str(int(n))
|
||
|
||
|
||
def _installs_range(items: list[dict[str, Any]]) -> tuple[int, int, str]:
|
||
values = [int(item.get("installs") or 0) for item in items]
|
||
lo, hi = min(values), max(values)
|
||
if lo == hi:
|
||
return lo, hi, format_installs(hi)
|
||
return lo, hi, f"{format_installs(lo)}–{format_installs(hi)}"
|
||
|
||
|
||
def _best_description(items: list[dict[str, Any]]) -> str:
|
||
for item in items:
|
||
desc = (item.get("description") or "").strip()
|
||
if desc:
|
||
return desc
|
||
return ""
|
||
|
||
|
||
def _cluster_skill(items: list[dict[str, Any]]) -> dict[str, Any]:
|
||
ranked = sorted(items, key=lambda x: int(x.get("installs") or 0), reverse=True)
|
||
top = ranked[0]
|
||
_lo, _hi, installs_fmt = _installs_range(ranked)
|
||
titles = [str(x.get("title") or "") for x in ranked if x.get("title")]
|
||
sample = ", ".join(titles[:4])
|
||
if len(titles) > 4:
|
||
sample = f"{sample}…"
|
||
return {
|
||
"id": skill_id(top),
|
||
"title": top.get("title", ""),
|
||
"source": top.get("source", ""),
|
||
"installs": int(top.get("installs") or 0),
|
||
"installs_min": _lo,
|
||
"installs_max": _hi,
|
||
"installs_fmt": installs_fmt,
|
||
"link": top.get("link", ""),
|
||
"description": _best_description(ranked),
|
||
"cluster": True,
|
||
"cluster_count": len(ranked),
|
||
"cluster_skills": titles,
|
||
"cluster_titles": sample,
|
||
}
|
||
|
||
|
||
def _single_skill(item: dict[str, Any]) -> dict[str, Any]:
|
||
installs = int(item.get("installs") or 0)
|
||
return {
|
||
"id": skill_id(item),
|
||
"title": item.get("title", ""),
|
||
"source": item.get("source", ""),
|
||
"installs": installs,
|
||
"installs_fmt": format_installs(installs),
|
||
"link": item.get("link", ""),
|
||
"description": item.get("description", ""),
|
||
"cluster": False,
|
||
}
|
||
|
||
|
||
def group_skills_by_source(
|
||
items: list[dict[str, Any]],
|
||
*,
|
||
limit: int = 10,
|
||
pool_size: int = 50,
|
||
) -> list[dict[str, Any]]:
|
||
"""按 source 去重合并；保留各 source 在榜内的最佳名次顺序。"""
|
||
if not items or limit <= 0:
|
||
return []
|
||
|
||
pool = items[: max(pool_size, limit)]
|
||
by_source: dict[str, list[dict[str, Any]]] = {}
|
||
first_rank: dict[str, int] = {}
|
||
for rank, item in enumerate(pool, 1):
|
||
source = (item.get("source") or "?").strip() or "?"
|
||
by_source.setdefault(source, []).append(item)
|
||
first_rank.setdefault(source, rank)
|
||
|
||
ordered_sources = sorted(by_source.keys(), key=lambda s: first_rank[s])
|
||
result: list[dict[str, Any]] = []
|
||
for source in ordered_sources:
|
||
group = by_source[source]
|
||
if len(group) == 1:
|
||
result.append(_single_skill(group[0]))
|
||
else:
|
||
result.append(_cluster_skill(group))
|
||
if len(result) >= limit:
|
||
break
|
||
return result
|