#!/usr/bin/env python3 """X Account Monitor — Daily cron scan of curated AI filmmaking accounts.""" import os import json import time import re import requests from datetime import datetime, timezone, timedelta # ── Config ────────────────────────────────────────────────────────────────── RELEVANT_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', ] IRRELEVANT_HANDLES = { 'AIBuzzNews', 'AIGuideHQ', 'AITECHio', 'AIFrontliner', 'AIWarper', 'TheAIColony', 'AITOPIAai', 'GoogleLabs', 'GoogleAI', 'DonutAI', 'cladlabs', '4aibsc', 'FaithJaiye2187', 'KammelaBala', 'MyStageAI', 'AlexStudio44', 'AI_Video_Dumps', 'AI_with_jasmin', 'Ai_Vaidehi', 'CharaspowerAI', 'ChrisLaubAI', 'CINEMA505', 'Chandan_Perla', } BIO_INFO = { 'runwayml': 'Runway AI — Gen-4.5 video generation platform', 'CuriousRefuge': 'Curious Refuge — home for AI filmmakers', 'nofilmschool': 'No Film School — filmmaking news & tutorials', 'venturetwins': 'Justine Moore (a16z) — AI video / creator economy VC', 'Kling_ai': 'Kling AI — Kling 3.0 video generation', 'paultrillo': 'Paul Trillo — LA director, AI filmmaking artist', 'c_valenzuelab': 'Cristóbal Valenzuela — Runway CEO/co-founder', 'CorridorDigital': 'Corridor Digital — VFX/sfx studio, AI tool reactions', 'karenxcheng': 'Karen X. Cheng — director, AI brand content', 'higgsfield': 'Higgsfield AI — multi-model AI video workspace', 'pika_labs': 'Pika Labs — AI video for social creators', 'EHuanglu': 'Elsa Huang (el.cine) — film director, AI consultant', 'azed_ai': 'Amira Zairi — Adobe & Leonardo AI ambassador', 'JeffSynthesized': 'Jeff Synthesized — NeoCinema AI Studio founder', 'ai_filmmaker': 'Dmitry Alekseev — AI film director', 'visiblemaker': 'Visiblemaker — AI cinema director (The Great Catspy)', 'Magiermogul': 'Particle Panic — AI filmmaker (Céremony)', 'WuxiaRocks': 'WuxIA Rocks — AI cinematic storytelling', 'JiamingSuen': 'Jiaming Sun — AI/VFX creator', 'moonvalley': 'Moonvalley — AI film model startup', 'higgsfield_ai': 'Higgsfield AI (ops account)', } # ── API Setup ─────────────────────────────────────────────────────────────── 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: raise RuntimeError("GETXAPI_KEY not found in ~/.hermes/.env") BASE_URL = "https://api.getxapi.com/twitter/user/tweets" HEADERS = {"Authorization": f"Bearer {API_KEY}"} # ── Scan Function ─────────────────────────────────────────────────────────── def scan_handle(handle): """Fetch latest 5 tweets for a handle. Returns (tweets[], error|None).""" try: resp = requests.get( BASE_URL, headers=HEADERS, params={"userName": handle, "count": 5}, timeout=60, ) if resp.status_code != 200: return [], f"HTTP {resp.status_code}: {resp.text[:200]}" data = resp.json() tweets = data.get("tweets", []) # Handle case where tweets is a dict keyed by id if isinstance(tweets, dict): tweets = list(tweets.values()) # Filter out retweets of other users tweets = [t for t in tweets if not t.get("retweeted_tweet")] return tweets, None except Exception as e: return [], str(e) # ── Scoring ───────────────────────────────────────────────────────────────── def score_tweet(tweet): likes = tweet.get("likeCount", 0) or 0 retweets = tweet.get("retweetCount", 0) or 0 replies = tweet.get("replyCount", 0) or 0 views = tweet.get("viewCount", 0) or 0 return likes + (retweets * 2) + (replies * 3) + (views / 100) def is_qualifying(tweet): likes = tweet.get("likeCount", 0) or 0 retweets = tweet.get("retweetCount", 0) or 0 views = tweet.get("viewCount", 0) or 0 return likes >= 50 or retweets >= 10 or views >= 1000 # ── Main ──────────────────────────────────────────────────────────────────── today = datetime.now(timezone.utc).strftime("%Y-%m-%d") cutoff = datetime.now(timezone.utc) - timedelta(days=3) cutoff = cutoff.replace(hour=0, minute=0, second=0, microsecond=0) all_results = {} failed = [] qualified_tweets = [] total = len(RELEVANT_HANDLES) for i, handle in enumerate(RELEVANT_HANDLES, 1): tweets, err = scan_handle(handle) if err: failed.append((handle, err)) print(f" [{i}/{total}] @{handle} — FAILED: {err}") else: print(f" [{i}/{total}] @{handle} — {len(tweets)} tweets") for t in tweets: created_str = t.get("createdAt", "") try: created = datetime.strptime(created_str, "%a %b %d %H:%M:%S %z %Y") except (ValueError, TypeError): continue if created < cutoff: continue if is_qualifying(t): qualified_tweets.append({ "handle": handle, "tweet": t, "score": score_tweet(t), "created": created.strftime("%Y-%m-%d %H:%M UTC"), "url": f"https://x.com/{handle}/status/{t.get('id')}", "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, }) all_results[handle] = tweets time.sleep(0.8) # Sort by score descending qualified_tweets.sort(key=lambda x: x["score"], reverse=True) # Save results scan_file = f"/home/kiro/trend-engine/trends/x/raw/scan_results_{today}_FULL.json" with open(scan_file, "w") as f: json.dump({"results": all_results, "failed": failed, "qualified": qualified_tweets, "date": today, "cutoff": cutoff.isoformat()}, f, indent=2) print(f"\nScan complete. {len(qualified_tweets)} qualifying tweets from {total} accounts.") print(f"{len(failed)} failures.") print(f"Results saved to {scan_file}") # ── Generate Report ──────────────────────────────────────────────────────── print("\n" + "=" * 80) print(" X ACCOUNT MONITOR — DAILY REPORT") print(f" Date: {today}") print("=" * 80) if not qualified_tweets: print("\n No qualifying tweets found in the last 3 days.\n") else: print(f"\n TOP 10 TWEETS (by engagement score)\n") print(f" {'#':<3} {'Account':<20} {'Score':<10} {'Likes':<8} {'RTs':<6} {'Views':<10} {'Date':<20}") print(f" {'-'*3} {'-'*20} {'-'*10} {'-'*8} {'-'*6} {'-'*10} {'-'*20}") for idx, tw in enumerate(qualified_tweets[:10], 1): print(f" {idx:<3} @{tw['handle']:<18} {tw['score']:<10.1f} {tw['likes']:<8} {tw['retweets']:<6} {tw['views']:<10} {tw['created']:<20}") print(f" {tw['url']}") # Show truncated tweet text text = tw['tweet'].get('text', '') if text: text_short = text[:120].replace('\n', ' ') print(f" \"{text_short}...\"") print() # Check for inactive high-value accounts tweeted_handles = {tw['handle'] for tw in qualified_tweets} inactive = [h for h in RELEVANT_HANDLES if h not in tweeted_handles] if inactive: print(f"\n INACTIVE ACCOUNTS (no qualifying tweets in 3 days):") for h in inactive: print(f" • @{h} — {BIO_INFO.get(h, '')}") print() # High-value account notes high_value = ['runwayml', 'CuriousRefuge', 'nofilmschool', 'venturetwins', 'Kling_ai', 'c_valenzuelab', 'CorridorDigital', 'EHuanglu'] inactive_high = [h for h in high_value if h not in tweeted_handles] if inactive_high: print(f" ⚠ HIGH-VALUE INACTIVE: @{', @'.join(inactive_high)}") print(f" Consider lowering threshold or removing from watchlist if inactive >6 months.") print(f"\n{'='*80}") print(f"\n Failed ({len(failed)}):") for handle, err in failed: print(f" @{handle}: {err}") print(f"\n{'='*80}")