"""Mastodon-Helper — RSS pro Researcher-Profil. Kein Auth, keine API-Limits — RSS-Endpoint pro User: `https:///@.rss`. Per SYSTEM-PROMPT § 4: AI-Researcher Cross-Posten oft hier. Start mit Simon Willison. Erweitern in `WATCH_USERS` wenn weitere AI-Builder dort aktiv sind. """ from __future__ import annotations import datetime as dt import email.utils import html import logging import re import urllib.request log = logging.getLogger(__name__) USER_AGENT = "aria-radar/0.1 (+aseckzai@gmail.com)" TIMEOUT_SECONDS = 15 WATCH_USERS = [ "https://fedi.simonwillison.net/@simon.rss", ] ITEM_RE = re.compile(r"(?P.*?)", re.DOTALL) TAG_RE_FACTORY = lambda tag: re.compile(rf"<{tag}[^>]*>(?P.*?)", re.DOTALL) def _extract(body: str, tag: str) -> str: m = TAG_RE_FACTORY(tag).search(body) return m.group("v").strip() if m else "" def _parse_rfc2822(value: str) -> dt.datetime | None: try: return email.utils.parsedate_to_datetime(value) except (TypeError, ValueError): return None def _strip_html(text: str) -> str: return html.unescape(re.sub(r"<[^>]+>", "", text)).strip() def _fetch_feed(feed_url: str, since: dt.datetime, max_items: int) -> list[dict]: req = urllib.request.Request(feed_url, headers={"User-Agent": USER_AGENT, "Accept": "application/rss+xml, application/xml"}) with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp: body = resp.read().decode("utf-8", errors="replace") items: list[dict] = [] for m in ITEM_RE.finditer(body): block = m.group("body") link = _extract(block, "link") or _extract(block, "guid") pub_date_raw = _extract(block, "pubDate") published = _parse_rfc2822(pub_date_raw) or dt.datetime.now(dt.timezone.utc) if published.tzinfo is None: published = published.replace(tzinfo=dt.timezone.utc) if published < since: continue description = _strip_html(_extract(block, "description")) title = description[:80] if description else "(no description)" items.append({ "title": title, "url": link, "date": published.isoformat(), "source": "mastodon", "score": None, "raw": { "feed": feed_url, "text": description[:300], }, }) if len(items) >= max_items: break return items def fetch(*, since: dt.datetime, max_per_user: int = 5) -> list[dict]: """Fetch latest Mastodon posts from WATCH_USERS RSS feeds.""" all_items: list[dict] = [] for feed_url in WATCH_USERS: try: items = _fetch_feed(feed_url, since, max_per_user) except Exception as exc: # noqa: BLE001 log.warning("mastodon %s failed: %s", feed_url, exc) continue all_items.extend(items) log.info("mastodon: %d items from %d users", len(all_items), len(WATCH_USERS)) return all_items if __name__ == "__main__": import json logging.basicConfig(level=logging.INFO) items = fetch(since=dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=2)) print(json.dumps(items, indent=2))