441 lines
15 KiB
Python
441 lines
15 KiB
Python
"""企微早报排版。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
from daily.config import wecom_skill_desc_limit
|
|
from daily.localize import LocalizeJob, localize_brief_descriptions, needs_chinese
|
|
from daily.text_utils import trim_brief
|
|
|
|
ICONS = {
|
|
"header": "📰",
|
|
"highlights": "💡",
|
|
"trending": "📈",
|
|
"hot": "🔥",
|
|
"github": "🐙",
|
|
"emerging": "🌱",
|
|
"topic": "🤖",
|
|
"ainews": "🌍",
|
|
"cnainews": "🇨🇳",
|
|
"pick": "📦",
|
|
"theme": "🎯",
|
|
"file": "📄",
|
|
}
|
|
|
|
|
|
def _skill_line(rank: int, item: dict[str, Any], *, badge: str = "") -> list[str]:
|
|
source = item.get("source", "?")
|
|
installs = item.get("installs_fmt", "?")
|
|
link = item.get("link", "")
|
|
desc = item.get("desc_short", "")
|
|
badge_prefix = f"{badge} " if badge else ""
|
|
if item.get("cluster"):
|
|
count = int(item.get("cluster_count") or 1)
|
|
sample = item.get("cluster_titles") or item.get("title", "")
|
|
label = f"**{source}** · {count} skills · **{installs}**"
|
|
if link:
|
|
head = f"{rank}. {badge_prefix}[{label}]({link})"
|
|
else:
|
|
head = f"{rank}. {badge_prefix}{label}"
|
|
lines = [head]
|
|
if sample or desc:
|
|
hint = desc or sample
|
|
lines.append(f" {hint}")
|
|
return lines
|
|
title = item.get("title", "?")
|
|
if link:
|
|
head = f"{rank}. {badge_prefix}[**{title}**]({link}) · `{source}` · **{installs}**"
|
|
else:
|
|
head = f"{rank}. {badge_prefix}**{title}** · `{source}` · **{installs}**"
|
|
lines = [head]
|
|
if desc:
|
|
lines.append(f" {desc}")
|
|
return lines
|
|
|
|
|
|
def _ai_news_lines(items: list[dict[str, Any]]) -> list[str]:
|
|
lines: list[str] = []
|
|
for i, item in enumerate(items, 1):
|
|
title = item.get("title", "?")
|
|
link = item.get("link", "")
|
|
source = item.get("source_name", "?")
|
|
pub = item.get("published_fmt", "")
|
|
desc = item.get("desc_short", "")
|
|
pub_suffix = f" · {pub}" if pub else ""
|
|
if link:
|
|
head = f"{i}. [**{title}**]({link}) · `{source}`{pub_suffix}"
|
|
else:
|
|
head = f"{i}. **{title}** · `{source}`{pub_suffix}"
|
|
lines.append(head)
|
|
if desc:
|
|
lines.append(f" > {desc}")
|
|
return lines
|
|
|
|
|
|
def _github_repo_lines(repos: list[dict[str, Any]], *, show_created: bool = False) -> list[str]:
|
|
lines: list[str] = []
|
|
for i, repo in enumerate(repos, 1):
|
|
name = repo["repo"]
|
|
url = repo["url"]
|
|
lang = repo.get("language", "")
|
|
stars_today = repo.get("stars_today_fmt", "")
|
|
total = repo.get("total_stars_fmt", "")
|
|
created = repo.get("created_at", "")
|
|
meta_parts: list[str] = []
|
|
if lang:
|
|
meta_parts.append(lang)
|
|
if stars_today:
|
|
meta_parts.append(f"+{stars_today} today")
|
|
elif total:
|
|
meta_parts.append(f"⭐{total}")
|
|
if show_created and created:
|
|
meta_parts.append(f"创建于 {created}")
|
|
meta = f" · {' · '.join(meta_parts)}" if meta_parts else ""
|
|
lines.append(f"{i}. [{name}]({url}){meta}")
|
|
desc = repo.get("desc_short") or repo.get("description", "")
|
|
if desc:
|
|
lines.append(f" > {desc}")
|
|
return lines
|
|
|
|
|
|
def _fallback_skill_desc(item: dict[str, Any]) -> str:
|
|
if item.get("cluster"):
|
|
count = int(item.get("cluster_count") or 1)
|
|
source = item.get("source") or "unknown"
|
|
sample = item.get("cluster_titles") or item.get("title") or ""
|
|
return f"{count} agent skills from {source}, including {sample}"
|
|
title = item.get("title") or "skill"
|
|
source = item.get("source") or "unknown"
|
|
return f"{title} skill from {source}"
|
|
|
|
|
|
def _skill_group_key(item: dict[str, Any]) -> str:
|
|
return str(item.get("id") or f"{item.get('source')}/{item.get('title')}")
|
|
|
|
|
|
def finalize_wecom_skill_groups(
|
|
items: list[dict[str, Any]],
|
|
*,
|
|
desc_limit: int | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""为企微 Skills 榜生成一句简要中文简介（与完整版 .md 长描述分离）。"""
|
|
if desc_limit is None:
|
|
desc_limit = wecom_skill_desc_limit()
|
|
limit = desc_limit if desc_limit > 0 else 48
|
|
copies: list[tuple[str, dict[str, Any]]] = []
|
|
jobs: list[LocalizeJob] = []
|
|
for item in items:
|
|
copy = dict(item)
|
|
desc = (copy.get("description") or "").strip()
|
|
if not desc:
|
|
desc = _fallback_skill_desc(copy)
|
|
key = _skill_group_key(copy)
|
|
job_key = f"wecom:{key}"
|
|
if needs_chinese(desc) or len(desc) > limit:
|
|
jobs.append(LocalizeJob(job_key, desc, limit))
|
|
else:
|
|
copy["wecom_desc"] = desc
|
|
copies.append((job_key, copy))
|
|
|
|
zh_map = localize_brief_descriptions(jobs, archive=True)
|
|
out: list[dict[str, Any]] = []
|
|
for job_key, copy in copies:
|
|
if job_key in zh_map:
|
|
copy["wecom_desc"] = zh_map[job_key]
|
|
elif "wecom_desc" not in copy:
|
|
copy["wecom_desc"] = _brief_fallback_desc(
|
|
(copy.get("description") or "").strip() or _fallback_skill_desc(copy),
|
|
limit,
|
|
)
|
|
out.append(copy)
|
|
return out
|
|
|
|
|
|
def _brief_fallback_desc(text: str, limit: int) -> str:
|
|
return trim_brief(text, limit)
|
|
|
|
|
|
def _grouped_skill_to_wecom_item(
|
|
item: dict[str, Any],
|
|
*,
|
|
desc_limit: int | None = None,
|
|
) -> dict[str, Any]:
|
|
if desc_limit is None:
|
|
desc_limit = wecom_skill_desc_limit()
|
|
limit = desc_limit if desc_limit > 0 else 48
|
|
installs = int(item.get("installs") or 0)
|
|
installs_fmt = item.get("installs_fmt") or str(installs)
|
|
desc = (item.get("wecom_desc") or item.get("description") or "").strip()
|
|
if not desc and item.get("cluster"):
|
|
desc = item.get("cluster_titles") or ""
|
|
if not item.get("wecom_desc"):
|
|
desc = _brief_fallback_desc(desc, limit)
|
|
return {
|
|
"title": item.get("title", ""),
|
|
"source": item.get("source", "?"),
|
|
"installs_fmt": installs_fmt,
|
|
"link": item.get("link", ""),
|
|
"desc_short": desc,
|
|
"badge": item.get("badge", ""),
|
|
"cluster": bool(item.get("cluster")),
|
|
"cluster_count": item.get("cluster_count"),
|
|
"cluster_titles": item.get("cluster_titles"),
|
|
}
|
|
|
|
|
|
def build_skills_board_section(icon_key: str, board_label: str, items: list[dict[str, Any]]) -> str:
|
|
prepared = finalize_wecom_skill_groups(items)
|
|
wecom_items = [_grouped_skill_to_wecom_item(x) for x in prepared]
|
|
lines = [f"{ICONS[icon_key]} **{board_label} Top {len(wecom_items)}**"]
|
|
for rank, item in enumerate(wecom_items, 1):
|
|
lines.extend(_skill_line(rank, item))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _move_to_wecom_skill_item(move: dict[str, Any]) -> dict[str, Any]:
|
|
installs = int(move.get("installs") or 0)
|
|
rank = move.get("rank", "?")
|
|
board_badge = (move.get("badge") or "").strip()
|
|
badge = f"[新入 #{rank}]"
|
|
if board_badge:
|
|
badge = f"{badge} {board_badge}"
|
|
return {
|
|
"title": move.get("title", "?"),
|
|
"source": move.get("source", "?"),
|
|
"installs_fmt": move.get("installs_fmt") or str(installs),
|
|
"link": move.get("link", ""),
|
|
"description": (move.get("description") or "").strip(),
|
|
"badge": badge,
|
|
}
|
|
|
|
|
|
def build_skills_delta_sections(
|
|
trending_moves: list[dict[str, Any]],
|
|
hot_moves: list[dict[str, Any]],
|
|
) -> str:
|
|
sections: list[str] = []
|
|
if trending_moves:
|
|
prepared = finalize_wecom_skill_groups([_move_to_wecom_skill_item(m) for m in trending_moves])
|
|
items = [_grouped_skill_to_wecom_item(x) for x in prepared]
|
|
lines = [f"{ICONS['trending']} **Skills Trending 变化**"]
|
|
for rank, item in enumerate(items, 1):
|
|
lines.extend(_skill_line(rank, item, badge=item.get("badge", "")))
|
|
sections.append("\n".join(lines))
|
|
if hot_moves:
|
|
prepared = finalize_wecom_skill_groups([_move_to_wecom_skill_item(m) for m in hot_moves])
|
|
items = [_grouped_skill_to_wecom_item(x) for x in prepared]
|
|
lines = [f"{ICONS['hot']} **Skills Hot 变化**"]
|
|
for rank, item in enumerate(items, 1):
|
|
lines.extend(_skill_line(rank, item, badge=item.get("badge", "")))
|
|
sections.append("\n".join(lines))
|
|
return "\n\n".join(sections)
|
|
|
|
|
|
def _github_move_to_repo(move: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"repo": move.get("repo", "?"),
|
|
"url": move.get("url", ""),
|
|
"language": move.get("language", ""),
|
|
"stars_today_fmt": move.get("stars_today_fmt", ""),
|
|
"total_stars_fmt": move.get("total_stars_fmt", ""),
|
|
"created_at": move.get("created_at", ""),
|
|
"description": move.get("description", ""),
|
|
"desc_short": (move.get("description") or "").strip(),
|
|
"badge": f"[新入 #{move.get('rank', '?')}]",
|
|
}
|
|
|
|
|
|
def build_github_delta_sections(movement: dict[str, Any], *, topic_name: str) -> str:
|
|
sections: list[str] = []
|
|
mapping = [
|
|
("github_trending_moves", "github", "GitHub Trending 变化", False),
|
|
("github_emerging_moves", "emerging", "GitHub 新兴 变化", True),
|
|
("github_topic_moves", "topic", f"Topic `{topic_name}` 变化", False),
|
|
]
|
|
for key, icon_key, label, show_created in mapping:
|
|
moves = movement.get(key) or []
|
|
if not moves:
|
|
continue
|
|
lines = [f"{ICONS[icon_key]} **{label}**"]
|
|
for i, move in enumerate(moves, 1):
|
|
repo = _github_move_to_repo(move)
|
|
badge = repo.pop("badge", "")
|
|
chunk = _github_repo_lines([repo], show_created=show_created)
|
|
if chunk:
|
|
chunk[0] = f"{i}. {badge} " + chunk[0].split(". ", 1)[-1]
|
|
lines.extend(chunk)
|
|
sections.append("\n".join(lines))
|
|
return "\n\n".join(sections)
|
|
|
|
|
|
_SKILL_SECTIONS = re.compile(
|
|
r"📈 \*\*Skills Trending.*?(?=🐙 \*\*GitHub Trending)",
|
|
re.DOTALL,
|
|
)
|
|
_SKILL_TRENDING_BLOCK = re.compile(r"📈 \*\*Skills Trending[^\n]*\n(?:.*?\n)*?(?=\n🔥 \*\*Skills Hot|\n🐙 |\n🌱 |\n🤖 |\Z)", re.DOTALL)
|
|
_SKILL_HOT_BLOCK = re.compile(r"🔥 \*\*Skills Hot[^\n]*\n(?:.*?\n)*?(?=\n🐙 |\n🌱 |\n🤖 |\Z)", re.DOTALL)
|
|
_GITHUB_SECTIONS = re.compile(r"🐙 \*\*GitHub Trending.*", re.DOTALL)
|
|
|
|
|
|
def _strip_board_sections(md: str) -> str:
|
|
md = _SKILL_TRENDING_BLOCK.sub("", md)
|
|
md = _SKILL_HOT_BLOCK.sub("", md)
|
|
if _GITHUB_SECTIONS.search(md):
|
|
md = _GITHUB_SECTIONS.sub("", md)
|
|
return re.sub(r"\n{3,}", "\n\n", md).rstrip()
|
|
|
|
|
|
def replace_wecom_board_sections(
|
|
md: str,
|
|
*,
|
|
mode: str,
|
|
movement: dict[str, Any],
|
|
trending: list[dict[str, Any]],
|
|
hot: list[dict[str, Any]],
|
|
topic_name: str,
|
|
) -> str:
|
|
from daily.delta import partition_skill_moves_for_wecom
|
|
|
|
if mode == "delta":
|
|
t_moves, h_moves = partition_skill_moves_for_wecom(
|
|
movement.get("skills_trending_moves") or [],
|
|
movement.get("skills_hot_moves") or [],
|
|
)
|
|
skills_sec = build_skills_delta_sections(t_moves, h_moves)
|
|
github_sec = build_github_delta_sections(movement, topic_name=topic_name)
|
|
board_block = "\n\n".join(x for x in [skills_sec, github_sec] if x)
|
|
md = _strip_board_sections(md)
|
|
if board_block:
|
|
return md + "\n\n" + board_block + "\n"
|
|
return md + "\n"
|
|
|
|
trending_sec = build_skills_board_section("trending", "Skills Trending", trending)
|
|
hot_sec = build_skills_board_section("hot", "Skills Hot", hot)
|
|
replacement = f"{trending_sec}\n\n{hot_sec}\n\n"
|
|
if _SKILL_SECTIONS.search(md):
|
|
return _SKILL_SECTIONS.sub(replacement, md)
|
|
github_marker = "🐙 **GitHub Trending"
|
|
idx = md.find(github_marker)
|
|
if idx >= 0:
|
|
return md[:idx] + replacement + md[idx:]
|
|
return md.rstrip() + "\n\n" + replacement
|
|
|
|
|
|
def replace_wecom_skill_sections(
|
|
md: str,
|
|
*,
|
|
trending: list[dict[str, Any]],
|
|
hot: list[dict[str, Any]],
|
|
mode: str = "full",
|
|
movement: dict[str, Any] | None = None,
|
|
topic_name: str = "llm",
|
|
) -> str:
|
|
"""用 Python 合并后的 Skills 榜替换或插入 Agent 早报中的对应区块。"""
|
|
return replace_wecom_board_sections(
|
|
md,
|
|
mode=mode,
|
|
movement=movement or {},
|
|
trending=trending,
|
|
hot=hot,
|
|
topic_name=topic_name,
|
|
)
|
|
|
|
|
|
def _format_pick_link(pick_command: str, *, title: str = "", url: str = "") -> str:
|
|
cmd = pick_command.strip()
|
|
if not cmd:
|
|
return ""
|
|
label = title.strip()
|
|
if cmd.startswith("http://") or cmd.startswith("https://"):
|
|
if not label:
|
|
m = re.match(r"https?://github\.com/([^/\s#?]+/[^/\s#?]+)", cmd)
|
|
label = m.group(1) if m else cmd
|
|
return f"[{label}]({cmd})"
|
|
m = re.match(r"npx skills add (\S+)", cmd)
|
|
if m:
|
|
skill_path = m.group(1)
|
|
if not label:
|
|
label = skill_path.split("/")[-1]
|
|
href = url.strip() or f"https://skills.sh/{skill_path}"
|
|
return f"[{label}]({href})"
|
|
return f"`{cmd}`"
|
|
|
|
|
|
def build_wecom_report(
|
|
*,
|
|
date_str: str,
|
|
time_str: str,
|
|
updated: str,
|
|
highlights: list[str],
|
|
theme_line: str,
|
|
trending: list[dict[str, Any]],
|
|
hot: list[dict[str, Any]],
|
|
repos: list[dict[str, Any]],
|
|
emerging: list[dict[str, Any]],
|
|
topic_name: str,
|
|
topic_repos: list[dict[str, Any]],
|
|
ai_news: list[dict[str, Any]] | None = None,
|
|
cn_ai_news: list[dict[str, Any]] | None = None,
|
|
pick_command: str,
|
|
pick_why: str = "",
|
|
pick_title: str = "",
|
|
pick_url: str = "",
|
|
include_boards: bool = True,
|
|
) -> str:
|
|
lines = [
|
|
f"{ICONS['header']} **早报 · {date_str}**",
|
|
f"> ⏱ {time_str} · 数据截至 {updated}",
|
|
"",
|
|
f"{ICONS['highlights']} **今日速览**",
|
|
]
|
|
for point in highlights[:3]:
|
|
lines.append(f"> {point}")
|
|
lines.append("")
|
|
lines.append(f"{ICONS['theme']} {theme_line}")
|
|
lines.append("")
|
|
|
|
if ai_news:
|
|
lines.append(f"{ICONS['ainews']} **国际 AI 时讯 Top {len(ai_news)}**")
|
|
lines.extend(_ai_news_lines(ai_news))
|
|
lines.append("")
|
|
|
|
if cn_ai_news:
|
|
lines.append(f"{ICONS['cnainews']} **国内 AI 时讯 Top {len(cn_ai_news)}**")
|
|
lines.extend(_ai_news_lines(cn_ai_news))
|
|
lines.append("")
|
|
|
|
if include_boards:
|
|
lines.append(f"{ICONS['trending']} **Skills Trending Top {len(trending)}**")
|
|
for rank, item in enumerate(trending, 1):
|
|
lines.extend(_skill_line(rank, item, badge=item.get("badge", "")))
|
|
lines.append("")
|
|
|
|
lines.append(f"{ICONS['hot']} **Skills Hot Top {len(hot)}**")
|
|
for rank, item in enumerate(hot, 1):
|
|
lines.extend(_skill_line(rank, item, badge=item.get("badge", "")))
|
|
lines.append("")
|
|
|
|
if repos:
|
|
lines.append(f"{ICONS['github']} **GitHub Trending Top {len(repos)}**")
|
|
lines.extend(_github_repo_lines(repos))
|
|
lines.append("")
|
|
|
|
if emerging:
|
|
lines.append(f"{ICONS['emerging']} **新兴项目 Top {len(emerging)}**")
|
|
lines.extend(_github_repo_lines(emerging, show_created=True))
|
|
lines.append("")
|
|
|
|
if topic_repos:
|
|
lines.append(f"{ICONS['topic']} **Topic `{topic_name}` Top {len(topic_repos)}**")
|
|
lines.extend(_github_repo_lines(topic_repos))
|
|
lines.append("")
|
|
|
|
lines.append(f"{ICONS['pick']} **今日首推**")
|
|
lines.append(_format_pick_link(pick_command, title=pick_title, url=pick_url))
|
|
if pick_why:
|
|
lines.append(f"> {pick_why}")
|
|
|
|
return "\n".join(lines)
|