#!/usr/bin/env python3 """X Account Monitor - Daily scan of AI filmmaking accounts via GetXAPI""" import os import re import json import time import requests from datetime import datetime, timezone, timedelta # ── Config ────────────────────────────────────────────────────────────── WATCHLIST_FILE = "/home/kiro/trend-engine/trends/x/raw/rising-accounts.md" OUTPUT_DIR = "/home/kiro/trend-engine/trends/x/reports" os.makedirs(OUTPUT_DIR, exist_ok=True) # ── API Key ───────────────────────────────────────────────────────────── env_path = os.path.expanduser("~/.hermes/.env") API_KEY = None with open(env_path) as f: for line in f: if line.startswith("GETXAPI_KEY="): API_KEY = line.strip().split("=", 1)[1] break if not API_KEY: print("[FATAL] GETXAPI_KEY not found in ~/.hermes/.env") exit(1) # ── Irrelevant Handles ────────────────────────────────────────────────── IRRELEVANT_HANDLES = { # AI news aggregators "AIBuzzNews", "AIGuideHQ", "AITECHio", "AIFrontliner", "AIWarper", "TheAIColony", "AITOPIAai", # Non-filmmaking auto-discoveries "GoogleLabs", "GoogleAI", "DonutAI", "cladlabs", "4aibsc", "FaithJaiye2187", "KammelaBala", "MyStageAI", "AlexStudio44", # Content repost accounts "AI_Video_Dumps", "AI_with_jasmin", "Ai_Vaidehi", "CharaspowerAI", "ChrisLaubAI", # Other non-core "CINEMA505", "Chandan_Perla", } # ── BIO_INFO ──────────────────────────────────────────────────────────── BIO_INFO = { "runwayml": "Runway AI — platform of record for professional AI filmmaking", "CuriousRefuge": "Curious Refuge — home for AI filmmakers, breaks viral AI film news", "nofilmschool": "No Film School — #1 filmmaking Twitter account, AI film discourse", "venturetwins": "Justine Moore (a16z) — AI video investment & Kling workflow tutorials", "Kling_ai": "Kling AI — dominant model for AI character swaps & multi-shot sequences", "paultrillo": "Paul Trillo — director/artist, created first official Sora music video", "c_valenzuelab": "Cristóbal Valenzuela — Runway CEO, key exec voice in AI filmmaking", "CorridorDigital": "Corridor Digital — 15yr VFX/filmmaking, open-sourced AI tools", "karenxcheng": "Karen X. Cheng — director, AI video for Adobe/Beats by Dre brands", "higgsfield": "Higgsfield AI — hosts Kling/Seedance in one workspace, $500K contest", "pika_labs": "Pika — best-in-class AI video for social media creators", "EHuanglu": "Elsa Huang (el.cine) — film director, AI consultant, 128K followers", "azed_ai": "Amira Zairi — Adobe & Leonardo AI ambassador, honest AI teardowns", "JeffSynthesized": "Jeff Synthesized — NeoCinema AI Studio founder, Runway/Kling CPP", "ai_filmmaker": "Dmitry Alekseev — film director, surreal AI cinema work", "visiblemaker": "The Visiblemaker — creator of The Great Catspy (50M+ views)", "Magiermogul": "Particle Panic — AI filmmaker, $1M AI Film Award shortlist", "WuxiaRocks": "WuxIA Rocks — AI filmmaker tracking character consistency", "JiamingSuen": "Jiaming Sun — AI/VFX creator, CorridorDigital collaborator", "moonvalley": "Moonvalley — AI film model startup, clean-trained model", "higgsfield_ai": "Higgsfield AI (ops account) — Kling 3.0 announcements", } # ── Extract Handles ───────────────────────────────────────────────────── def load_handles(): noise = {'x', 'youtube', 'facebook', 'marketingbrew', 'etcjournal', 'letsdatascience', 'besttwitteraccounts', 'theankler', 'xo2', 'nofilmschool', 'blogs', 'nvidia', 'pixflow', 'curiousrefuge', 'chatgptricks', 'blogsnvidia'} with open(WATCHLIST_FILE) as f: content = f.read() all_h = set(re.findall(r'@(\w+)', content)) handles = sorted([h for h in (all_h - noise) if len(h) > 1 and not h.isdigit()]) return handles # ── Scan Single Handle ────────────────────────────────────────────────── def scan_handle(handle): url = f"https://api.getxapi.com/twitter/user/tweets?userName={handle}&count=5" headers = {"Authorization": f"Bearer {API_KEY}"} try: resp = requests.get(url, headers=headers, timeout=60) if resp.status_code != 200: return None, f"HTTP {resp.status_code}" data = resp.json() tweets = data.get("tweets", []) if not tweets: return None, "no tweets" return tweets, None except Exception as e: return None, str(e) # ── Score & Filter ────────────────────────────────────────────────────── CUTOFF = datetime.now(timezone.utc) - timedelta(days=3) CUTOFF = CUTOFF.replace(hour=0, minute=0, second=0, microsecond=0) def score_tweet(tweet): likes = int(tweet.get("likeCount", 0)) retweets = int(tweet.get("retweetCount", 0)) replies = int(tweet.get("replyCount", 0)) views = int(tweet.get("viewCount", 0)) return likes, retweets, replies, views, likes + (retweets * 2) + (replies * 3) + (views / 100) def parse_date(date_str): try: return datetime.strptime(date_str, "%a %b %d %H:%M:%S %z %Y") except: return None # ── Main ──────────────────────────────────────────────────────────────── def main(): today = datetime.now(timezone.utc).strftime("%Y-%m-%d") handles = load_handles() print(f"[{today}] Scanning {len(handles)} accounts via GetXAPI...") all_results = {} failed = [] for i, handle in enumerate(handles, 1): tweets, err = scan_handle(handle) if tweets is not None: scored = [] for t in tweets: if t.get("type") == "retweet" and t.get("retweeted_tweet"): continue # skip pure retweets dt = parse_date(t.get("createdAt", "")) if dt and dt < CUTOFF: continue # outside 3-day window likes, rts, replies, views, score = score_tweet(t) # Threshold check if likes >= 50 or rts >= 10 or views >= 1000: scored.append({ "id": t.get("id"), "text": t.get("text", "")[:300], "likes": likes, "retweets": rts, "replies": replies, "views": views, "score": round(score, 1), "createdAt": t.get("createdAt"), "url": f"https://x.com/{handle}/status/{t.get('id')}", }) scored.sort(key=lambda x: x["score"], reverse=True) all_results[handle] = scored[:5] # top 5 per handle status = f"✓ {len(scored)} qualifying tweets" else: failed.append({"handle": handle, "error": err}) all_results[handle] = [] status = f"✗ {err}" print(f" [{i:2d}/{len(handles)}] @{handle}: {status}") time.sleep(0.8) # Save raw results result_data = {"results": all_results, "failed": failed, "scanned_at": today} result_path = os.path.join(OUTPUT_DIR, f"scan_results_{today}.json") with open(result_path, "w") as f: json.dump(result_data, f, indent=2) # ── Build Report ───────────────────────────────────────────────── # Collect all qualifying tweets across all handles all_tweets = [] for handle, tweets in all_results.items(): bio = BIO_INFO.get(handle, "") for t in tweets: all_tweets.append((handle, bio, t)) # Sort by score descending all_tweets.sort(key=lambda x: x[2]["score"], reverse=True) today_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") lines = [] lines.append("=" * 72) lines.append(f" X ACCOUNT MONITOR — TOP 10 ENGAGING POSTS") lines.append(f" {today_str}") lines.append(f" Accounts scanned: {len(handles)} | Qualifying tweets: {len(all_tweets)}") lines.append(f" Failed accounts: {len(failed)}") lines.append("=" * 72) lines.append("") if len(failed) > 0: lines.append(f"⚠ Scan failures ({len(failed)}):") for f_entry in failed: lines.append(f" @{f_entry['handle']}: {f_entry['error']}") lines.append("") if not all_tweets: lines.append(" 📭 No tweets met the engagement threshold in the last 3 days.") lines.append("") else: # Top 10 (or fewer) top_n = min(10, len(all_tweets)) lines.append(f" {'#':>2} | {'Account':22s} | Views | Score | Date") lines.append(f" {'-'*2}-+-{'-'*22}-+-{'-'*8}-+-{'-'*7}-+-{'-'*20}") for rank, (handle, bio, tweet) in enumerate(all_tweets[:top_n], 1): views_str = f"{tweet['views']:,}" if tweet['views'] else "—" score_str = f"{tweet['score']:,.0f}" if tweet['score'] else "—" created = tweet.get("createdAt", "") date_short = created[:10] if created else "—" lines.append(f" {rank:2d} | @{handle:20s} | {views_str:>7s} | {score_str:>6s} | {date_short}") lines.append(f" URL: {tweet['url']}") if bio: lines.append(f" Bio: {bio}") text_preview = tweet.get("text", "")[:120].replace("\n", " ") if text_preview: lines.append(f" „{text_preview}…") lines.append("") # ── Account Activity Summary ───────────────────────────────────── lines.append(f"{'='*72}") lines.append(f" ACCOUNT ACTIVITY SUMMARY (3-day window)") lines.append(f"{'='*72}") lines.append("") lines.append(f" {'Handle':22s} | {'Qualifying':>10s} | {'Top Score':>10s} | {'Status':15s}") lines.append(f" {'-'*22}-+-{'-'*10}-+-{'-'*10}-+-{'-'*15}") active_count = 0 inactive_count = 0 for handle in handles: bio = BIO_INFO.get(handle, "") tweets = all_results.get(handle, []) is_irrelevant = handle in IRRELEVANT_HANDLES label = "[FILTERED]" if is_irrelevant else "active" if tweets: top_score = tweets[0]["score"] lines.append(f" @{handle:20s} | {len(tweets):>10d} | {top_score:>10,.0f} | {label:15s}") active_count += 1 else: # Check if handle had any tweets at all (failed scan = unknown) if any(f_entry["handle"] == handle for f_entry in failed): lines.append(f" @{handle:20s} | {'—':>10s} | {'—':>10s} | {'scan failed':15s}") else: lines.append(f" @{handle:20s} | {'0':>10s} | {'—':>10s} | {'no qualifying':15s}") if not is_irrelevant: inactive_count += 1 lines.append("") lines.append(f" Active: {active_count} | Inactive (no qualifying tweets): {inactive_count}") lines.append(f" Filtered (in report but excluded from Top 10): {len(IRRELEVANT_HANDLES)}") lines.append("") # Inactive suggestions if inactive_count > 0: lines.append(f"{'='*72}") lines.append(f" INACTIVE HANDLES (no qualifying tweets in 3 days)") lines.append(f"{'='*72}") lines.append("") for handle in handles: tweets = all_results.get(handle, []) is_irrelevant = handle in IRRELEVANT_HANDLES if not tweets and not is_irrelevant and not any(f_entry["handle"] == handle for f_entry in failed): bio = BIO_INFO.get(handle, "") lines.append(f" @{handle} — {bio}" if bio else f" @{handle}") lines.append("") lines.append(f" Consider: checking if these accounts are worth keeping, or") lines.append(f" lowering the engagement threshold for occasional posters.") lines.append("") lines.append(f"{'='*72}") lines.append(f" Report generated: {today_str}") lines.append(f"{'='*72}") report = "\n".join(lines) report_path = os.path.join(OUTPUT_DIR, f"report_{today}.txt") with open(report_path, "w") as f: f.write(report) print(f"\nReport saved to: {report_path}") print(report) if __name__ == "__main__": main()