#!/usr/bin/env python3 """X Account Monitor — Full scan of 44 watchlist accounts via GetXAPI""" import urllib.request import json import time import os from datetime import datetime, timezone, timedelta API_KEY = "get-x-api-6157f8edb420c3d5e50e062287389aa55635cd7c9630c81f" BASE_URL = "https://api.getxapi.com/twitter/user/tweets" # All 44 accounts from the watchlist HANDLES = [ # Tier 1: Essential "runwayml", "CuriousRefuge", "nofilmschool", "venturetwins", "Kling_ai", # Tier 2: Filmmaker Voices "paultrillo", "c_valenzuelab", "CorridorDigital", "karenxcheng", "higgsfield", # Tier 3: Industry "pika_labs", "EHuanglu", # Rising Accounts "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 (first batch) "AI_Video_Dumps", "ChrisLaubAI", "GoogleLabs", "TheAIColony", "cladlabs", # Auto-Discovered 2026-05-04 (second batch) "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", ] # Cutoff: 3 days ago from June 9 2026 06:00 UTC CUTOFF = datetime(2026, 6, 6, 6, 0, 0, tzinfo=timezone.utc) RESULTS_FILE = "/home/kiro/trend-engine/trends/x/account-monitor/scan_results_2026-06-09.json" def parse_date(created_at): """Parse Rails-style date string like 'Sun May 17 16:58:59 +0000 2026'""" 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): """Fetch latest 5 tweets for a handle""" url = f"{BASE_URL}?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, list): return tweets return [] except Exception as e: return {"error": str(e)} def is_recent(dt): """Check if tweet is within the last 3 days""" if dt is None: return False return dt >= CUTOFF def is_original_content(text): """Check if tweet is original content, not a retweet""" if not text: return True return not text.startswith("RT @") def format_date(dt): if dt is None: return "" return dt.strftime("%Y-%m-%d") def main(): # Load existing results for resume support all_results = {"accounts": [], "errors": [], "scan_date": "2026-06-09 06:00 UTC"} done_handles = set() if os.path.exists(RESULTS_FILE): try: with open(RESULTS_FILE) as f: existing = json.load(f) for a in existing.get("accounts", []): all_results["accounts"].append(a) done_handles.add(a["handle"]) all_results["errors"] = existing.get("errors", []) print(f"Resuming from existing results: {len(done_handles)} accounts already scanned") except: pass print(f"Starting scan of {len(HANDLES)} accounts...") for i, handle in enumerate(HANDLES): if handle in done_handles: print(f" [{i+1}/{len(HANDLES)}] {handle} — already scanned, skipping") continue print(f" [{i+1}/{len(HANDLES)}] Scanning @{handle}...") result = fetch_tweets(handle) if isinstance(result, dict) and "error" in result: print(f" ERROR: {result['error']}") all_results["errors"].append({"handle": handle, "error": result["error"]}) # Save incrementally with open(RESULTS_FILE + ".tmp", "w") as f: json.dump(all_results, f, indent=2) time.sleep(1) continue tweets = result if isinstance(result, list) else [] print(f" Got {len(tweets)} tweets") account_data = { "handle": handle, "tweets": [], "total_tweets": len(tweets), "recent_tweets": 0, "qualifying_tweets": 0 } for tweet in tweets: if not isinstance(tweet, dict): continue tweet_id = tweet.get("id", "") text = tweet.get("text", "") created_at = tweet.get("createdAt", "") dt = parse_date(created_at) like_count = tweet.get("likeCount", 0) retweet_count = tweet.get("retweetCount", 0) reply_count = tweet.get("replyCount", 0) view_count = tweet.get("viewCount", 0) bookmark_count = tweet.get("bookmarkCount", 0) author = tweet.get("author", {}) author_handle = author.get("userName", handle) followers = author.get("followers", 0) date_str = format_date(dt) is_recent_flag = is_recent(dt) score = like_count + (retweet_count * 2) + reply_count + (view_count / 100) tweet_entry = { "id": tweet_id, "text": text[:200], "full_text": text, "created_at": created_at, "date": date_str, "is_recent": is_recent_flag, "is_rt": text.startswith("RT @"), "like_count": like_count, "retweet_count": retweet_count, "reply_count": reply_count, "view_count": view_count, "bookmark_count": bookmark_count, "score": round(score, 1), "author_handle": author_handle, "followers": followers, "url": f"https://x.com/{author_handle}/status/{tweet_id}" } if is_recent_flag: account_data["recent_tweets"] += 1 # Qualifies if recent AND (meets threshold OR is original content with any engagement) qualifies = is_recent_flag and ( like_count >= 50 or retweet_count >= 10 or view_count >= 1000 ) if qualifies: account_data["qualifying_tweets"] += 1 account_data["tweets"].append(tweet_entry) all_results["accounts"].append(account_data) # Save incrementally after each account with open(RESULTS_FILE + ".tmp", "w") as f: json.dump(all_results, f, indent=2, default=str) time.sleep(1) # Rename .tmp to final if os.path.exists(RESULTS_FILE + ".tmp"): os.rename(RESULTS_FILE + ".tmp", RESULTS_FILE) # Summary total_scanned = len(all_results["accounts"]) + len(all_results["errors"]) total_with_content = sum(1 for a in all_results["accounts"] if a["total_tweets"] > 0) total_qualifying = sum(a["qualifying_tweets"] for a in all_results["accounts"]) total_errors = len(all_results["errors"]) print(f"\n=== SCAN COMPLETE ===") print(f"Accounts scanned: {total_scanned}") print(f"Accounts with tweets: {total_with_content}") print(f"Qualifying tweets (recent + meets threshold): {total_qualifying}") print(f"API errors: {total_errors}") print(f"Results saved to: {RESULTS_FILE}") if __name__ == "__main__": main()