#!/usr/bin/env python3 """Full scan with incremental save - X Account Monitor June 8, 2026""" import urllib.request, json, time, sys, os from datetime import datetime, timezone, timedelta API_KEY = "get-x-api-6157f8edb420c3d5e50e062287389aa55635cd7c9630c81f" CUTOFF_DATE = datetime.now(timezone.utc) - timedelta(days=3) 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", "AITOPIAai", "FaithJaiye2187", "KammelaBala", "MyStageAI", "AI_Video_Dumps", "ChrisLaubAI", "GoogleLabs", "TheAIColony", "cladlabs", "AIBuzzNews", "AlexStudio44", "DonutAI", "GoogleAI", "AI_with_jasmin", "CharaspowerAI", "Ai_Vaidehi", "CINEMA505", "AIFrontliner", "AIGuideHQ", "AIWarper", "AITECHio", "4aibsc", ] 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": "Cristobal 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)", } OUTPUT_DIR = "/home/kiro/trend-engine/trends/x/account-monitor" RESULTS_FILE = os.path.join(OUTPUT_DIR, "scan_results_2026-06-08.json") def parse_date(created_at): if not created_at: return None try: dt = datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y") return dt except: return None def fetch_tweets(handle): url = f"https://api.getxapi.com/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) tweets = data.get("tweets", data) if isinstance(tweets, dict): tweets = [tweets] elif not isinstance(tweets, list): tweets = [] return tweets except Exception as e: return {"error": str(e)} def score_tweet(tweet): likes = tweet.get("likeCount", 0) or 0 retweets = tweet.get("retweetCount", 0) or 0 views = tweet.get("viewCount", 0) or 0 replies = tweet.get("replyCount", 0) or 0 return likes + (retweets * 2) + (replies) + (views // 100) # Load existing results if any all_results = {"accounts": [], "errors": [], "scan_date": "2026-06-08 06:00 UTC"} done_handles = set() for a in all_results["accounts"]: done_handles.add(a["handle"]) for i, handle in enumerate(HANDLES): if handle in done_handles: print(f"[SKIP {i+1}/{len(HANDLES)}] @{handle} already scanned", flush=True) continue print(f"[{i+1}/{len(HANDLES)}] Scanning @{handle}...", flush=True) tweets = fetch_tweets(handle) if isinstance(tweets, dict) and "error" in tweets: print(f" ERROR: {tweets['error']}", flush=True) all_results["errors"].append({"handle": handle, "error": tweets["error"]}) # Save incrementally with open(RESULTS_FILE + ".tmp", "w") as f: json.dump(all_results, f) time.sleep(1) continue account_data = { "handle": handle, "tweets": [], "qualifying": [], "bio": BIO_INFO.get(handle, ""), "qualifying_count": 0, "total_score": 0 } for tweet in tweets: if not isinstance(tweet, dict): continue created = parse_date(tweet.get("createdAt", "")) if created is None: continue if created < CUTOFF_DATE: continue likes = tweet.get("likeCount", 0) or 0 retweets = tweet.get("retweetCount", 0) or 0 views = tweet.get("viewCount", 0) or 0 replies = tweet.get("replyCount", 0) or 0 if likes >= 50 or retweets >= 10 or views >= 1000: text = tweet.get("text", "") tweet_id = tweet.get("id", "") author = tweet.get("author", {}) author_handle = author.get("userName", handle) author_followers = author.get("followers", 0) or 0 s = score_tweet(tweet) entry = { "id": tweet_id, "author_handle": author_handle, "author_followers": author_followers, "text": text[:200], "likes": likes, "retweets": retweets, "replies": replies, "views": views, "created_at_str": tweet.get("createdAt", ""), "created_at": created.strftime("%Y-%m-%d"), "is_rt": text.startswith("RT @"), "score": s } account_data["tweets"].append(entry) account_data["qualifying"].append(entry) account_data["qualifying_count"] = len(account_data["qualifying"]) account_data["total_score"] = sum(t["score"] for t in account_data["qualifying"]) account_data["best_tweet"] = max(account_data["qualifying"], key=lambda t: t["score"]) if account_data["qualifying"] else None if account_data["best_tweet"]: account_data["best_score"] = account_data["best_tweet"]["score"] else: account_data["best_score"] = 0 all_results["accounts"].append(account_data) print(f" -> {account_data['qualifying_count']} qualifying tweets (best score: {account_data['best_score']})", flush=True) # Save incrementally with open(RESULTS_FILE + ".tmp", "w") as f: json.dump(all_results, f, indent=2) time.sleep(1) # Save final os.rename(RESULTS_FILE + ".tmp", RESULTS_FILE) print(f"\nFINAL RESULTS:", flush=True) print(f"Total accounts: {len(HANDLES)}", flush=True) print(f"Accounts with content: {sum(1 for a in all_results['accounts'] if a['qualifying_count'] > 0)}", flush=True) print(f"Total qualifying tweets: {sum(a['qualifying_count'] for a in all_results['accounts'])}", flush=True) print(f"Errors: {len(all_results['errors'])}", flush=True) # Print top 20 by total score sorted_accounts = sorted(all_results["accounts"], key=lambda a: a["total_score"], reverse=True) print(f"\nTop 20 by total score:", flush=True) for i, a in enumerate(sorted_accounts[:20]): best = a.get("best_tweet") if best: print(f" {i+1}. @{a['handle']} -- total_score={a['total_score']}, best={best['score']} (likes={best['likes']} RT={best['retweets']} views={best['views']}) date={best['created_at']}", flush=True) else: print(f" {i+1}. @{a['handle']} -- total_score={a['total_score']}, no qualifying tweets", flush=True)