feat: metadata enrichment for alerts and subscription plugins

Alert backends now populate structured `extra` field with engagement
metrics (views, stars, votes, etc.) instead of embedding them in titles.
Subscription plugins show richer announcements: Twitch viewer counts,
YouTube views/likes/dates, RSS published dates.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-19 10:00:17 +01:00
parent c3b19feb0f
commit 1fe7da9ed8
10 changed files with 614 additions and 52 deletions

View File

@@ -27,6 +27,7 @@ _YT_PLAYER_URL = "https://www.youtube.com/youtubei/v1/player"
_YT_CLIENT_VERSION = "2.20250101.00.00"
_ATOM_NS = "{http://www.w3.org/2005/Atom}"
_YT_NS = "{http://www.youtube.com/xml/schemas/2015}"
_MEDIA_NS = "{http://search.yahoo.com/mrss/}"
_MAX_SEEN = 200
_MAX_ANNOUNCE = 5
_DEFAULT_INTERVAL = 600
@@ -74,6 +75,15 @@ def _truncate(text: str, max_len: int = _MAX_TITLE_LEN) -> str:
return text[: max_len - 3].rstrip() + "..."
def _compact_num(n: int) -> str:
"""Format large numbers compactly: 1234 -> 1.2k, 1234567 -> 1.2M."""
if n >= 1_000_000:
return f"{n / 1_000_000:.1f}M".replace(".0M", "M")
if n >= 1_000:
return f"{n / 1_000:.1f}k".replace(".0k", "k")
return str(n)
def _is_youtube_url(url: str) -> bool:
"""Check if URL is a YouTube domain."""
try:
@@ -213,8 +223,33 @@ def _parse_feed(body: bytes) -> tuple[str, list[dict]]:
link = (link_el.get("href", "") if link_el is not None else "").strip()
if not entry_id:
entry_id = link
# Published date
published = (entry.findtext(f"{_ATOM_NS}published") or "").strip()
date = published[:10] if len(published) >= 10 else ""
# media:statistics views + media:starRating count (likes)
views = 0
likes = 0
group = entry.find(f"{_MEDIA_NS}group")
if group is not None:
community = group.find(f"{_MEDIA_NS}community")
if community is not None:
stats_el = community.find(f"{_MEDIA_NS}statistics")
if stats_el is not None:
try:
views = int(stats_el.get("views", "0"))
except (ValueError, TypeError):
pass
rating_el = community.find(f"{_MEDIA_NS}starRating")
if rating_el is not None:
try:
likes = int(rating_el.get("count", "0"))
except (ValueError, TypeError):
pass
if entry_id:
items.append({"id": entry_id, "title": entry_title, "link": link})
items.append({
"id": entry_id, "title": entry_title, "link": link,
"date": date, "views": views, "likes": likes,
})
return (channel_name, items)
@@ -305,7 +340,21 @@ async def _poll_once(bot, key: str, announce: bool = True) -> None:
for item in shown:
title = _truncate(item["title"]) if item["title"] else "(no title)"
link = item["link"]
# Build metadata suffix
parts = []
views = item.get("views", 0)
likes = item.get("likes", 0)
if views:
parts.append(f"{_compact_num(views)}v")
if likes:
parts.append(f"{_compact_num(likes)}lk")
date = item.get("date", "")
if date:
parts.append(date)
extra = " ".join(parts)
line = f"[{name}] {title}"
if extra:
line += f" | {extra}"
if link:
line += f" -- {link}"
await bot.send(channel, line)