#!/usr/bin/env python3 """ x-ai-monitor: Monitor Twitter/X accounts for new tweets via nitter RSS. Usage: python3 monitor.py # check all accounts, report new tweets python3 monitor.py --reset # clear state (re-check all) python3 monitor.py --list # list monitored accounts python3 monitor.py --add @user # add account to watchlist python3 monitor.py --remove @user # remove account """ import json import sys import os import argparse import time from datetime import datetime, timezone from pathlib import Path import urllib.request import urllib.error import xml.etree.ElementTree as ET # ── Config ──────────────────────────────────────────────────────────────────── SKILL_DIR = Path(__file__).parent.parent STATE_FILE = SKILL_DIR / "state.json" ACCOUNTS_FILE = SKILL_DIR / "references" / "accounts.json" # Nitter instances to try in order NITTER_INSTANCES = [ "https://nitter.perennialte.ch", "https://nitter.poast.org", "https://nitter.privacydev.net", "https://nitter.rawbit.ninja", "https://nitter.1d4.us", ] REQUEST_TIMEOUT = 10 REQUEST_DELAY = 1.5 # seconds between requests (be polite) # ── Helpers ─────────────────────────────────────────────────────────────────── def load_json(path: Path, default): if path.exists(): with open(path) as f: return json.load(f) return default def save_json(path: Path, data): path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w") as f: json.dump(data, f, indent=2) def fetch_url(url: str) -> str | None: try: req = urllib.request.Request( url, headers={"User-Agent": "Mozilla/5.0 (compatible; rss-reader/1.0)"} ) with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp: return resp.read().decode("utf-8", errors="replace") except Exception: return None def fetch_nitter_rss(username: str) -> list[dict] | None: """Fetch recent tweets for a user via nitter RSS. Returns list of tweet dicts.""" username = username.lstrip("@") for instance in NITTER_INSTANCES: url = f"{instance}/{username}/rss" content = fetch_url(url) if not content: continue try: root = ET.fromstring(content) ns = {"atom": "http://www.w3.org/2005/Atom"} channel = root.find("channel") if channel is None: continue items = channel.findall("item") tweets = [] for item in items: title = item.findtext("title", "").strip() link = item.findtext("link", "").strip() pub_date = item.findtext("pubDate", "").strip() description = item.findtext("description", "").strip() # Extract tweet ID from guid (most reliable) or link guid = item.findtext("guid", "").strip() tweet_id = guid if guid.isdigit() else link.rstrip("/").split("/")[-1].split("#")[0] if not tweet_id.isdigit(): continue tweets.append({ "id": tweet_id, "username": username, "title": title, "link": f"https://x.com/{username}/status/{tweet_id}", "pub_date": pub_date, "description": description[:500], }) return tweets except ET.ParseError: continue return None # ── Core logic ──────────────────────────────────────────────────────────────── def check_all(accounts: list[str], state: dict, verbose: bool = True) -> list[dict]: new_tweets = [] for username in accounts: uname = username.lstrip("@").lower() last_id = state.get(uname, {}).get("last_id") if verbose: print(f" Checking @{uname}...", end=" ", flush=True) tweets = fetch_nitter_rss(uname) if tweets is None: if verbose: print("⚠️ fetch failed (all instances down)") continue if not tweets: if verbose: print("no tweets found") continue # Sort by ID descending (newest first) tweets.sort(key=lambda t: int(t["id"]), reverse=True) # Update state with latest ID newest_id = tweets[0]["id"] if uname not in state: state[uname] = {} # First run: just record, don't report everything if last_id is None: state[uname]["last_id"] = newest_id if verbose: print(f"initialized (latest: {newest_id})") else: fresh = [t for t in tweets if int(t["id"]) > int(last_id)] state[uname]["last_id"] = newest_id if fresh: new_tweets.extend(fresh) if verbose: print(f"🆕 {len(fresh)} new tweet(s)") else: if verbose: print("no new tweets") time.sleep(REQUEST_DELAY) return new_tweets # ── CLI ─────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Monitor AI Twitter accounts") parser.add_argument("--reset", action="store_true", help="Clear state") parser.add_argument("--list", action="store_true", help="List accounts") parser.add_argument("--add", metavar="USER", help="Add account") parser.add_argument("--remove", metavar="USER", help="Remove account") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--recent", action="store_true", help="Show latest tweet from each account") args = parser.parse_args() accounts_data = load_json(ACCOUNTS_FILE, {"accounts": []}) accounts = accounts_data.get("accounts", []) if args.list: print(f"Monitoring {len(accounts)} accounts:") for a in accounts: print(f" @{a.lstrip('@')}") return if args.add: uname = args.add.lstrip("@") if uname.lower() not in [a.lower().lstrip("@") for a in accounts]: accounts.append(uname) accounts_data["accounts"] = accounts save_json(ACCOUNTS_FILE, accounts_data) print(f"✅ Added @{uname}") else: print(f"Already monitoring @{uname}") return if args.remove: uname = args.remove.lstrip("@") accounts = [a for a in accounts if a.lower().lstrip("@") != uname.lower()] accounts_data["accounts"] = accounts save_json(ACCOUNTS_FILE, accounts_data) print(f"✅ Removed @{uname}") return if args.recent: print(f"📰 Latest tweet from each account:\n") for username in accounts: uname = username.lstrip("@") tweets = fetch_nitter_rss(uname) if not tweets: print(f"@{uname}: ⚠️ unavailable\n") time.sleep(REQUEST_DELAY) continue tweets.sort(key=lambda t: int(t["id"]), reverse=True) t = tweets[0] print(f"@{uname} · {t['pub_date']}") print(f" {t['title'][:160]}") print(f" {t['link']}\n") time.sleep(REQUEST_DELAY) return state = {} if args.reset else load_json(STATE_FILE, {}) if args.reset: print("🔄 State reset. Re-initializing...") print(f"🔍 Checking {len(accounts)} accounts...") new_tweets = check_all(accounts, state) save_json(STATE_FILE, state) if not new_tweets: if not args.json: print("\n✅ No new tweets.") else: print(json.dumps({"new_tweets": []})) return # Sort by ID (chronological) new_tweets.sort(key=lambda t: int(t["id"])) if args.json: print(json.dumps({"new_tweets": new_tweets}, indent=2)) return print(f"\n🐦 {len(new_tweets)} new tweet(s):\n") for t in new_tweets: print(f"@{t['username']} · {t['pub_date']}") print(f" {t['title'][:120]}") print(f" {t['link']}") print() if __name__ == "__main__": main()