"""归档 / 企微格式化与摘要构建。""" 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