289 lines
9.6 KiB
Python
289 lines
9.6 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,
|
|
"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)
|
|
|
|
|
|
_SKILL_SECTIONS = re.compile(
|
|
r"📈 \*\*Skills Trending.*?(?=🐙 \*\*GitHub Trending)",
|
|
re.DOTALL,
|
|
)
|
|
|
|
|
|
def replace_wecom_skill_sections(
|
|
md: str,
|
|
*,
|
|
trending: list[dict[str, Any]],
|
|
hot: list[dict[str, Any]],
|
|
) -> str:
|
|
"""用 Python 合并后的 Skills 榜替换或插入 Agent 早报中的对应区块。"""
|
|
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 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,
|
|
) -> 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("")
|
|
|
|
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(f"`{pick_command}`")
|
|
|
|
return "\n".join(lines)
|