#!/usr/bin/env python3 """X Account Monitor - Scan Chunk 1 (core AI filmmaking accounts)""" import urllib.request, json, os, time from datetime import datetime, timezone, timedelta API_KEY = "get-x-api-6157f8edb420c3d5e50e062287389aa55635cd7c9630c81f" BASE = "https://api.getxapi.com" RESULTS_FILE = "/home/kiro/trend-engine/trends/x/account-monitor/scan_results_2026-06-13.json" CUTOFF = datetime(2026, 6, 10, 0, 0, 0, tzinfo=timezone.utc) # 3 days ago SCAN_DATE = "2026-06-13 06:00 UTC" BIO_INFO = { "runwayml": "Runway AI — professional AI filmmaking (~260K)", "CuriousRefuge": "Curious Refuge — home for AI filmmakers", "nofilmschool": "No Film School — filmmaking news (~241.7K)", "venturetwins": "Justine Moore — a16z partner AI video (~149.5K)", "Kling_ai": "Kling AI — AI video generation (~130K)", "paultrillo": "Paul Trillo — LA director/artist (~15K)", "c_valenzuelab": "Cristóbal Valenzuela — Runway CEO (~42.7K)", "CorridorDigital": "Corridor Digital — VFX/filmmaking (15M+ YouTube)", "karenxcheng": "Karen X. Cheng — Director, brand AI video (~58K)", "higgsfield": "Higgsfield AI — multi-model platform (~63K)", "pika_labs": "Pika — AI video for social creators (~143K)", "EHuanglu": "el.cine / Elsa Huang — Film director, AI consultant (~128K)", "azed_ai": "Amira Zairi — Adobe & Leonardo AI ambassador (~37K)", "JeffSynthesized": "Jeff Synthesized — NeoCinema AI Studio founder (~8.5K)", "ai_filmmaker": "Dmitry Alekseev — Film director/AI filmmaker", "visiblemaker": "The Visiblemaker — AI cinema director (brand campaigns)", "Magiermogul": "Particle Panic — AI filmmaker, competition-grade shorts (~475)", "WuxiaRocks": "WuxIA Rocks — AI filmmaker, character consistency (~6.5K)", "JiamingSuen": "Jiaming Sun — AI/VFX creator, collaborates w/ Corridor (~900)", "moonvalley": "Moonvalley — AI film model startup, clean-trained", "higgsfield_ai": "Higgsfield AI — secondary/ops account", "Chandan_Perla": "Chandan Perla — India AI Film Festival organizer", } HANDLES = [ "runwayml", "CuriousRefuge", "nofilmschool", "venturetwins", "Kling_ai", "paultrillo", "c_valenzuelab", "CorridorDigital", "karenxcheng", "higgsfield", "pika_labs", "EHuanglu", "azed_ai", "JeffSynthesized", "ai_filmmaker", "visiblemaker", "Magiermogul", "WuxiaRocks", "JiamingSuen", "moonvalley", "higgsfield_ai", "Chandan_Perla", ] def parse_date(created_at): if not created_at: return None try: return datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y") except: return None def fetch_tweets(handle): url = f"{BASE}/twitter/user/tweets?userName={handle}&count=5" req = urllib.request.Request(url, headers={"Authorization": f"Bearer {API_KEY}"}) try: with urllib.request.urlopen(req, timeout=30) as resp: body = resp.read().decode("utf-8", errors="replace") data = json.loads(body, strict=False) except Exception as e: return {"error": str(e), "tweets": []} tweets = data.get("tweets", data) if isinstance(data, dict) else data if not isinstance(tweets, list): return {"error": f"unexpected response type: {type(data)}", "tweets": []} return {"tweets": tweets} print(f"=== Scan Chunk 1 — {SCAN_DATE} ===") print(f"Handles: {len(HANDLES)}") print(f"Cutoff: {CUTOFF.isoformat()}") print() # Load existing results if any all_results = {"accounts": [], "errors": [], "scan_date": SCAN_DATE} if os.path.exists(RESULTS_FILE): with open(RESULTS_FILE) as f: all_results = json.load(f) print(f"Loaded existing results: {len(all_results['accounts'])} accounts already scanned") done_handles = set(a["handle"] for a in all_results.get("accounts", [])) for i, handle in enumerate(HANDLES): if handle in done_handles: print(f"[{i+1}/{len(HANDLES)}] {handle} — SKIP (already scanned)") continue print(f"[{i+1}/{len(HANDLES)}] {handle}...", end=" ", flush=True) result = fetch_tweets(handle) if "error" in result: print(f"ERROR: {result['error']}") all_results.setdefault("errors", []).append({"handle": handle, "error": result["error"]}) else: raw_tweets = result["tweets"] # Apply date filter recent = [] for t in raw_tweets: ca = t.get("createdAt", "") dt = parse_date(ca) if dt and dt >= CUTOFF: autor = t.get("author", {}) tweet_data = { "id": t.get("id", ""), "text": t.get("text", ""), "createdAt": ca, "date": dt.strftime("%Y-%m-%d") if dt else "", "likes": t.get("likeCount", 0) or 0, "retweets": t.get("retweetCount", 0) or 0, "replies": t.get("replyCount", 0) or 0, "views": t.get("viewCount", 0) or 0, "bookmarks": t.get("bookmarkCount", 0) or 0, "author_name": autor.get("userName", handle), "author_followers": autor.get("followers", 0) or 0, } tweet_data["score"] = ( tweet_data["likes"] + tweet_data["retweets"] * 2 + tweet_data["replies"] + tweet_data["views"] / 100 ) recent.append(tweet_data) # Filter by engagement thresholds qualifying = [t for t in recent if t["likes"] >= 50 or t["retweets"] >= 10 or t["views"] >= 1000] qualifying.sort(key=lambda x: x["score"], reverse=True) bio = BIO_INFO.get(handle, "") account_data = { "handle": handle, "bio": bio, "total_fetched": len(raw_tweets), "recent_count": len(recent), "qualifying_count": len(qualifying), "tweets": qualifying, } all_results["accounts"].append(account_data) # Also try dates from non-qualifying for debugging sample_dates = [t["date"] for t in recent[:3]] if recent else [] print(f"{len(raw_tweets)} fetched, {len(recent)} recent{', dates=' + ','.join(sample_dates) if sample_dates else ''}, {len(qualifying)} qualifying") # Incremental save with open(RESULTS_FILE + ".tmp", "w") as f: json.dump(all_results, f, indent=2) time.sleep(1) # Done — rename with open(RESULTS_FILE, "w") as f: json.dump(all_results, f, indent=2) os.remove(RESULTS_FILE + ".tmp") if os.path.exists(RESULTS_FILE + ".tmp") else None total_q = sum(a["qualifying_count"] for a in all_results["accounts"]) print(f"\n=== Chunk 1 Complete: {len(HANDLES)} scanned, {total_q} total qualifying tweets ===")