#!/usr/bin/env python3 """X Account Monitor Scan - June 8, 2026""" import urllib.request, json, time, sys from datetime import datetime, timezone, timedelta API_KEY = "get-x-api-6157f8edb420c3d5e50e062287389aa55635cd7c9630c81f" CUTOFF_DATE = datetime.now(timezone.utc) - timedelta(days=3) HANDLES = [ # Tier 1 "runwayml", "CuriousRefuge", "nofilmschool", "venturetwins", "Kling_ai", # Tier 2 "paultrillo", "c_valenzuelab", "CorridorDigital", "karenxcheng", "higgsfield", # Tier 3 "pika_labs", "EHuanglu", # Rising "azed_ai", "JeffSynthesized", "ai_filmmaker", "visiblemaker", "Magiermogul", "WuxiaRocks", "JiamingSuen", "moonvalley", "higgsfield_ai", "Chandan_Perla", # Auto-discovered 2026-05-03 "AITOPIAai", "FaithJaiye2187", "KammelaBala", "MyStageAI", # Auto-discovered 2026-05-04 "AI_Video_Dumps", "ChrisLaubAI", "GoogleLabs", "TheAIColony", "cladlabs", "AIBuzzNews", "AlexStudio44", "DonutAI", "GoogleAI", # Auto-discovered 2026-05-06 "AI_with_jasmin", "CharaspowerAI", # Auto-discovered 2026-05-07 "Ai_Vaidehi", # Auto-discovered 2026-05-08 "CINEMA505", # Auto-discovered 2026-05-12 "AIFrontliner", "AIGuideHQ", # Auto-discovered 2026-05-13 "AIWarper", # Auto-discovered 2026-05-14 "AITECHio", # Auto-discovered 2026-05-17 "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)", } 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) + (views // 100) results = {"accounts": [], "errors": []} for i, handle in enumerate(HANDLES): 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) results["errors"].append({"handle": handle, "error": tweets["error"]}) time.sleep(1) continue account_data = { "handle": handle, "tweets": [], "qualifying": [], "bio": BIO_INFO.get(handle, "") } 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 # Apply thresholds: 50+ likes OR 10+ retweets OR 1000+ views if likes >= 50 or retweets >= 10 or views >= 1000: is_rt = tweet.get("text", "").startswith("RT @") text = tweet.get("text", "")[:120] tweet_id = tweet.get("id", "") author = tweet.get("author", {}) author_handle = author.get("userName", handle) score = score_tweet(tweet) tweet_entry = { "id": tweet_id, "author_handle": author_handle, "text": text, "likes": likes, "retweets": retweets, "replies": replies, "views": views, "created_at": created.strftime("%Y-%m-%d"), "is_rt": is_rt, "score": score } account_data["tweets"].append(tweet_entry) account_data["qualifying"].append(tweet_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 results["accounts"].append(account_data) # Print status q = account_data["qualifying_count"] print(f" -> {q} qualifying tweets" + (f" (best: {account_data['best_tweet']['score']})" if account_data['best_tweet'] else ""), flush=True) time.sleep(1) # Save raw results print("\nSaving results...", flush=True) with open("/home/kiro/trend-engine/trends/x/account-monitor/scan_results_2026-06-08.json", "w") as f: json.dump(results, f, indent=2) print("\nDONE.", flush=True) print(f"Total accounts: {len(results['accounts'])}", flush=True) print(f"Accounts with content: {sum(1 for a in results['accounts'] if a['qualifying_count'] > 0)}", flush=True) print(f"Total qualifying tweets: {sum(a['qualifying_count'] for a in results['accounts'])}", flush=True) print(f"Errors: {len(results['errors'])}", flush=True) # Also write a summary for quick parsing with open("/home/kiro/trend-engine/trends/x/account-monitor/scan_summary_2026-06-08.txt", "w") as f: for a in sorted(results["accounts"], key=lambda x: x["total_score"], reverse=True): f.write(f"@{a['handle']}|{a['qualifying_count']}|{a['total_score']}|{a['best_tweet']['score'] if a['best_tweet'] else 0}\n")