#!/usr/bin/env python3 """X Account Monitor - daily cron scan of watchlist via GetXAPI.""" import json import os import re import sys import time from datetime import datetime, timedelta, timezone import requests WATCHLIST = "/home/kiro/trend-engine/trends/x/raw/rising-accounts.md" ENV_PATH = os.path.expanduser("~/.hermes/.env") OUT_DIR = "/home/kiro/trend-engine/trends/x/raw" # Noise words from markdown refs / prose. CRITICAL: nofilmschool + curiousrefuge # are valid Tier-1 handles that appear as plaintext in URLs -- must NOT be skipped. SKIP_WORDS = { "x", "youtube", "facebook", "marketingbrew", "besttwitteraccounts", "theankler", "xo2", "pixflow", "etcjournal", "letsdatascience", "blogs", "nvidia", "blog", "com", "www", "http", "https", "lang", "with", "replies", "highlights", "status", "ai", "film", "gallery", "the", "and", "for", "your", "you", "are", "not", "this", "that", "from", "video", "storytelling", "has", "arrived", "topic", "watch", "posts", "in", "the", "of", "to", "a", "is", "on", "by", "an", "or", "at", } # Auto-discovered / non-core accounts -- report under Filtered section, not top 10 IRRELEVANT_HANDLES = { "AIBuzzNews", "AIGuideHQ", "AITECHio", "AIFrontliner", "AIWarper", "TheAIColony", "AITOPIAai", "GoogleLabs", "GoogleAI", "DonutAI", "cladlabs", "4aibsc", "FaithJaiye2187", "KammelaBala", "MyStageAI", "AlexStudio44", "AI_Video_Dumps", "AI_with_jasmin", "Ai_Vaidehi", "CharaspowerAI", "ChrisLaubAI", "CINEMA505", "Chandan_Perla", } # Handle -> short bio for readable reporting BIO_INFO = { "runwayml": "Runway AI", "CuriousRefuge": "Curious Refuge - AI film hub", "nofilmschool": "No Film School", "venturetwins": "Justine Moore (a16z)", "Kling_ai": "Kling AI", "paultrillo": "Paul Trillo", "c_valenzuelab": "Cristobal Valenzuela (Runway CEO)", "CorridorDigital": "Corridor Digital", "karenxcheng": "Karen X. Cheng", "higgsfield": "Higgsfield AI", "pika_labs": "Pika", "EHuanglu": "el.cine (Elsa Huang)", "azed_ai": "Amira Zairi", "JeffSynthesized": "Jeff Synthesized - NeoCinema", "ai_filmmaker": "Dmitry Alekseev", "visiblemaker": "The Visiblemaker", "Magiermogul": "Particle Panic", "WuxiaRocks": "WuxIA Rocks", "JiamingSuen": "Jiaming Sun", "moonvalley": "Moonvalley", "higgsfield_ai": "Higgsfield (ops)", } def get_api_key(): with open(ENV_PATH) as f: for line in f: if line.startswith("GETXAPI_KEY="): return line.strip().split("=", 1)[1] return None def extract_handles(path): with open(path) as f: text = f.read() handles = set() for m in re.finditer(r"@(\w+)", text): h = m.group(1) if h.lower() in SKIP_WORDS: continue handles.add(h) return handles def scan_handle(handle, api_key): url = f"https://api.getxapi.com/twitter/user/tweets?userName={handle}&count=5" try: resp = requests.get(url, headers={"Authorization": f"Bearer {api_key}"}, timeout=60) if resp.status_code != 200: return None, f"HTTP {resp.status_code}" data = resp.json() tweets = data.get("tweets", []) return tweets, None except Exception as e: return None, str(e) def score_tweet(t): likes = t.get("likeCount", 0) or 0 rts = t.get("retweetCount", 0) or 0 replies = t.get("replyCount", 0) or 0 views = t.get("viewCount", 0) or 0 return likes + (rts * 2) + (replies * 3) + (views / 100), likes, rts, replies, views def parse_date(s): try: return datetime.strptime(s, "%a %b %d %H:%M:%S %z %Y") except Exception: return None def main(): api_key = get_api_key() if not api_key: print("ERROR: GETXAPI_KEY not found in ~/.hermes/.env") sys.exit(1) handles = extract_handles(WATCHLIST) # Normalize to the case used in watchlist but query as-is; keep original casing handle_list = sorted(handles) print(f"Scanning {len(handle_list)} accounts...", file=sys.stderr) today = datetime.now(timezone.utc).strftime("%Y-%m-%d") cutoff = datetime.now(timezone.utc) - timedelta(days=3) cutoff = cutoff.replace(hour=0, minute=0, second=0, microsecond=0) all_results = {} failed = [] for h in handle_list: tweets, err = scan_handle(h, api_key) if err or tweets is None: failed.append((h, err)) else: all_results[h] = tweets time.sleep(0.8) with open(os.path.join(OUT_DIR, f"scan_results_{today}_FULL.json"), "w") as f: json.dump({"results": all_results, "failed": failed}, f) # Score and filter qualifying = [] # (handle, tweet, score) filtered_qualifying = [] # irrelevant but qualifying inactive_core = [] core_handles = set(handle_list) - IRRELEVANT_HANDLES for h, tweets in all_results.items(): best = None best_score = 0 for t in tweets: created = parse_date(t.get("createdAt", "")) if created is None or created < cutoff: continue # skip pure retweets of others if t.get("retweeted_tweet") or t.get("type") == "retweet": continue s, likes, rts, replies, views = score_tweet(t) if likes >= 50 or rts >= 10 or views >= 1000: if s > best_score: best_score = s best = (t, s, likes, rts, replies, views) if best: entry = (h, best[0], best[1], best[2], best[3], best[4], best[5]) if h in IRRELEVANT_HANDLES: filtered_qualifying.append(entry) else: qualifying.append(entry) else: if h in core_handles: inactive_core.append(h) # Sort by score qualifying.sort(key=lambda x: x[2], reverse=True) filtered_qualifying.sort(key=lambda x: x[2], reverse=True) # Build report lines = [] lines.append(f"X ACCOUNT MONITOR - {today}") lines.append(f"Scanned: {len(handle_list)} accounts | Qualifying (3-day window): {len(qualifying)} | Failed: {len(failed)}") lines.append("=" * 70) lines.append("") lines.append("TOP 10 BY ENGAGEMENT SCORE") lines.append("-" * 70) if not qualifying: lines.append("No qualifying tweets in the last 3 days.") for i, (h, t, s, likes, rts, replies, views) in enumerate(qualifying[:10], 1): tid = t.get("id") url = f"https://x.com/{h}/status/{tid}" created = t.get("createdAt", "")[:16] bio = BIO_INFO.get(h, h) txt = (t.get("text") or "").replace("\n", " ")[:80] lines.append(f"{i}. {bio} (@{h})") lines.append(f" Score: {s:.0f} | Likes: {likes} | RTs: {rts} | Replies: {replies} | Views: {views}") lines.append(f" {url}") lines.append(f" [{created}] {txt}") lines.append("") if filtered_qualifying: lines.append("FILTERED (auto-discovered / non-core) - NOT in top 10") lines.append("-" * 70) for h, t, s, likes, rts, replies, views in filtered_qualifying[:8]: tid = t.get("id") url = f"https://x.com/{h}/status/{tid}" lines.append(f"- @{h} | Score: {s:.0f} | Likes: {likes} | Views: {views} | {url}") lines.append("") if inactive_core: lines.append(f"CORE ACCOUNTS INACTIVE (>3 days, no qualifying tweet): {len(inactive_core)}") lines.append("-" * 70) lines.append(", ".join(f"@{h}" for h in inactive_core)) lines.append("") if failed: lines.append(f"FAILED SCANS: {len(failed)}") lines.append("-" * 70) for h, err in failed: lines.append(f"- @{h}: {err}") lines.append("") report = "\n".join(lines) print(report) with open(os.path.join(OUT_DIR, f"x_monitor_report_{today}.txt"), "w") as f: f.write(report) if __name__ == "__main__": main()