yumao
ce04d5a342
提取 collect/formatters/themes 等 pipeline 模块，新增 wecom 分条、RSS、delta、Agent 工作流测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
181 lines
6.9 KiB
Python
181 lines
6.9 KiB
Python
"""归档 / 企微格式化与摘要构建。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import xml.etree.ElementTree as ET
|
||
from typing import Any
|
||
|
||
import certifi
|
||
import httpx
|
||
|
||
from daily.config import full_desc_limit, wecom_skill_desc_limit
|
||
from daily.github.auth import github_html_headers
|
||
from daily.news.fetch import prepare_wecom_cn_news_items, prepare_wecom_news_items
|
||
from shared.skills_data import format_installs
|
||
|
||
from daily.pipeline.snapshot import skill_id
|
||
|
||
|
||
def short_desc(text: str, limit: int = 72) -> str:
|
||
text = re.sub(r"\s+", " ", text or "").strip()
|
||
if limit <= 0 or len(text) <= limit:
|
||
return text
|
||
return text[: limit - 3] + "..."
|
||
|
||
|
||
def archive_desc(text: str) -> str:
|
||
return short_desc(text, full_desc_limit())
|
||
|
||
|
||
def wecom_desc(text: str, limit: int = 36) -> str:
|
||
return short_desc(text, limit)
|
||
|
||
|
||
def prepare_skill_item(item: dict[str, Any], prev_ids: set[str], rank: int) -> dict[str, Any]:
|
||
badge = ""
|
||
sid = skill_id(item)
|
||
if sid not in prev_ids and prev_ids:
|
||
badge = "🆕"
|
||
elif rank == 1:
|
||
badge = "👑"
|
||
installs_fmt = item.get("installs_fmt") or format_installs(item.get("installs", 0))
|
||
title = item.get("source", "?") if item.get("cluster") else item.get("title", "?")
|
||
desc = item.get("wecom_desc") or item.get("description") or item.get("cluster_titles") or ""
|
||
limit = wecom_skill_desc_limit()
|
||
desc_short = desc if item.get("wecom_desc") or limit <= 0 else wecom_desc(desc, limit)
|
||
return {
|
||
"title": title,
|
||
"source": item.get("source", "?"),
|
||
"installs_fmt": installs_fmt,
|
||
"link": item.get("link", ""),
|
||
"desc_short": desc_short,
|
||
"badge": badge,
|
||
"cluster": bool(item.get("cluster")),
|
||
"cluster_count": item.get("cluster_count"),
|
||
"cluster_titles": item.get("cluster_titles"),
|
||
}
|
||
|
||
|
||
def prepare_github_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
return {**item, "desc_short": wecom_desc(item.get("description", ""), 40)}
|
||
|
||
|
||
def build_highlights(
|
||
trending: list[dict[str, Any]],
|
||
hot: list[dict[str, Any]],
|
||
github_trending: list[dict[str, Any]],
|
||
github_emerging: list[dict[str, Any]],
|
||
ai_news: dict[str, Any] | None = None,
|
||
cn_ai_news: dict[str, Any] | None = None,
|
||
) -> list[str]:
|
||
points: list[str] = []
|
||
if ai_news and ai_news.get("enabled"):
|
||
top_news = prepare_wecom_news_items(ai_news)
|
||
if top_news:
|
||
n0 = top_news[0]
|
||
pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
|
||
points.append(
|
||
f"🌍 AI 时讯 [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
|
||
)
|
||
elif ai_news.get("flat"):
|
||
n0 = ai_news["flat"][0]
|
||
pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
|
||
points.append(f"🌍 AI 时讯 [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）")
|
||
if cn_ai_news and cn_ai_news.get("enabled"):
|
||
top_cn = prepare_wecom_cn_news_items(cn_ai_news)
|
||
if top_cn:
|
||
n0 = top_cn[0]
|
||
pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
|
||
points.append(
|
||
f"🇨🇳 国内 AI [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
|
||
)
|
||
elif cn_ai_news.get("flat"):
|
||
n0 = cn_ai_news["flat"][0]
|
||
pub = f" · {n0['published_fmt']}" if n0.get("published_fmt") else ""
|
||
points.append(
|
||
f"🇨🇳 国内 AI [{n0['title']}]({n0['link']})（`{n0.get('source_name', '?')}`{pub}）"
|
||
)
|
||
if trending:
|
||
t0 = trending[0]
|
||
points.append(f"📈 Skills 榜首 **{t0.get('title')}**（{format_installs(t0.get('installs', 0))}）")
|
||
if github_trending:
|
||
g0 = github_trending[0]
|
||
stars = g0.get("stars_today_fmt", "")
|
||
total = g0.get("total_stars_fmt", "")
|
||
star_hint = f"+{stars} today · " if stars else (f"⭐{total} · " if total else "")
|
||
points.append(f"🐙 GitHub Trending [{g0['repo']}]({g0['url']})（{star_hint}{g0.get('language', '')}）")
|
||
if github_emerging:
|
||
e0 = github_emerging[0]
|
||
points.append(f"🌱 新兴 [{e0['repo']}]({e0['url']})（⭐ {e0.get('total_stars_fmt', '?')}）")
|
||
elif hot:
|
||
h0 = hot[0]
|
||
points.append(f"🔥 Skills Hot 榜首 **{h0.get('title')}**（1H {format_installs(h0.get('installs', 0))}）")
|
||
while len(points) < 3 and len(trending) > len(points):
|
||
item = trending[len(points)]
|
||
points.append(f"✨ **{item.get('title')}** · `{item.get('source')}`")
|
||
return points[:3]
|
||
|
||
|
||
def format_github_repo_section(repos: list[dict[str, Any]], *, show_created: bool = False) -> list[str]:
|
||
lines: list[str] = []
|
||
for i, repo in enumerate(repos, 1):
|
||
lang = repo.get("language") or "—"
|
||
stars_today = repo.get("stars_today_fmt") or ""
|
||
total = repo.get("total_stars_fmt") or ""
|
||
created = repo.get("created_at") or ""
|
||
meta_parts = [lang]
|
||
if stars_today:
|
||
meta_parts.append(f"+{stars_today} today")
|
||
if total:
|
||
meta_parts.append(f"总 ⭐ {total}")
|
||
if show_created and created:
|
||
meta_parts.append(f"创建于 {created}")
|
||
lines.append(f"{i}. **[{repo['repo']}]({repo['url']})** · {' · '.join(meta_parts)}")
|
||
desc = archive_desc(repo.get("description", ""))
|
||
if desc:
|
||
lines.append(f" - {desc}")
|
||
lines.append("")
|
||
return lines
|
||
|
||
|
||
def format_skill_section(items: list[dict[str, Any]], *, hot: bool = False) -> list[str]:
|
||
lines: list[str] = []
|
||
for i, item in enumerate(items, 1):
|
||
sid = item.get("id") or f"{item.get('source', '?')}/{item.get('title', '?')}"
|
||
link = item.get("link", "")
|
||
installs = format_installs(item.get("installs", 0))
|
||
meta = f"1H {installs}" if hot else f"总安装 {installs}"
|
||
if link:
|
||
lines.append(f"{i}. **[{sid}]({link})** · {meta}")
|
||
else:
|
||
lines.append(f"{i}. **{sid}** · {meta}")
|
||
desc = archive_desc(item.get("description", ""))
|
||
if desc:
|
||
lines.append(f" - {desc}")
|
||
lines.append("")
|
||
return lines
|
||
|
||
|
||
def fetch_latest_release_title(repo: str) -> str | None:
|
||
atom_url = f"https://github.com/{repo}/releases.atom"
|
||
try:
|
||
with httpx.Client(
|
||
timeout=12.0,
|
||
verify=certifi.where(),
|
||
follow_redirects=True,
|
||
headers=github_html_headers(),
|
||
) as client:
|
||
resp = client.get(atom_url)
|
||
if resp.status_code != 200:
|
||
return None
|
||
root = ET.fromstring(resp.text)
|
||
ns = {"a": "http://www.w3.org/2005/Atom"}
|
||
entry = root.find("a:entry", ns)
|
||
if entry is None:
|
||
return None
|
||
title = entry.find("a:title", ns)
|
||
return title.text.strip() if title is not None and title.text else None
|
||
except Exception:
|
||
return None
|