""" Report generator for X Account Monitor Merges scan_results_CHUNK1 + CHUNK2, formats plain-text report. """ import json, os from datetime import datetime, timezone DATE_STR = datetime.now(timezone.utc).strftime("%Y-%m-%d") OUT_DIR = "/home/kiro/trend-engine/trends/x/account-monitor" REPORT_FILE = os.path.join(OUT_DIR, f"{DATE_STR}_account-monitor.txt") IRRELEVANT_HANDLES = { "ChrisLaubAI", "CINEMA505", "TheAIColony", "GoogleLabs", "GoogleAI", "Ai_Vaidehi", "AITOPIAai", "FaithJaiye2187", "KammelaBala", "MyStageAI", "AlexStudio44", "AIGuideHQ", "4aibsc", "AI_Video_Dumps", "AI_with_jasmin", "AIBuzzNews", } 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)", "azed_ai": "Amira Zairi -- Adobe & Leonardo AI ambassador (~37K)", "JeffSynthesized": "Jeff Synthesized -- NeoCinema AI Studio founder, LA", "ai_filmmaker": "Dmitry Alekseev -- Film director/AI filmmaker", "visiblemaker": "The Visiblemaker (Christian Fleischer) -- AI cinema director", "Magiermogul": "Particle Panic -- AI filmmaker, contest finalist", "WuxiaRocks": "WuxIA Rocks -- AI filmmaker, long-form storytelling", "JiamingSuen": "Jiaming Sun -- AI/VFX creator (CorridorDigital collab)", "moonvalley": "Moonvalley -- AI film model startup", "higgsfield_ai": "Higgsfield AI (secondary/ops account)", "Chandan_Perla": "Chandan Perla -- India AI Film Festival organizer", "AI_Video_Dumps": "[FILTERED] Low-effort generic AI video", "ChrisLaubAI": "[FILTERED] RT-only finance/news", "DonutAI": "AI tool ecosystem", "CharaspowerAI": "Seedance 2 creator -- original anime VFX, kaiju", "AIFrontliner": "AI news aggregator -- ChatGPT/Claude prompts, Seedance", "AIWarper": "AI creator -- Seedance 2.0 prompts, ComfyUI workflows", "AITECHio": "AI/crypto hybrid -- ACN token, AI infrastructure", "4aibsc": "[FILTERED] AI/Web3 hybrid, not filmmaking", "AI_with_jasmin": "[FILTERED] 100% RT aggregator", "AIBuzzNews": "[FILTERED] General AI news aggregator", } def load_chunks(date_str): c1_path = os.path.join(OUT_DIR, f"scan_results_{date_str}_CHUNK1.json") c2_path = os.path.join(OUT_DIR, f"scan_results_{date_str}_CHUNK2.json") chunks = [] for p in [c1_path, c2_path]: if os.path.exists(p): with open(p) as f: chunks.append(json.load(f)) else: print(f"Warning: missing {p}") return chunks def build_report(chunks): all_tweets = [] accounts_with_content = 0 total_qualifying = 0 error_handles = [] empty_handles = [] for chunk in chunks: for acc in chunk["accounts"]: handle = acc["handle"] qc = acc.get("qualifying_count", 0) if qc > 0: accounts_with_content += 1 total_qualifying += qc for t in acc["tweets"]: all_tweets.append(t) elif qc == 0: empty_handles.append(handle) for e in chunk.get("errors", []): error_handles.append(f"{e['handle']}: {e['error']}") all_tweets.sort(key=lambda t: t["score"], reverse=True) # Build deduped top 10: skip irrelevant AND duplicate handles seen_handles = set() deduped_top10 = [] for t in all_tweets: h = t["handle"] if h not in IRRELEVANT_HANDLES and h not in seen_handles: seen_handles.add(h) deduped_top10.append(t) if len(deduped_top10) >= 10: break # Build per-account breakdown account_max = {} for t in all_tweets: h = t["handle"] if h not in account_max or t["score"] > account_max[h]["score"]: account_max[h] = t sorted_accounts = sorted(account_max.items(), key=lambda x: x[1]["score"], reverse=True) lines = [] lines.append(f"Title: X Account Monitor -- {DATE_STR}") lines.append("") lines.append("Top 10 by Engagement Score") lines.append("") lines.append("#, Account, Best Tweet, Views, Date") for rank, t in enumerate(deduped_top10, 1): url = f"https://x.com/{t['handle']}/status/{t['tweet_id']}" snippet = t["text"].replace(",", " ").strip() lines.append(f"{rank}, @{t['handle']}, {snippet} ({url}), {t['views']}, {t['date']}") lines.append("") lines.append("Full Breakdown by Account") lines.append("") for handle, _ in sorted_accounts: bio = BIO_INFO.get(handle, "") marker = " [FILTERED]" if handle in IRRELEVANT_HANDLES else "" lines.append(f"@{handle}{marker} -- {bio}") account_tweets = sorted( [t for t in all_tweets if t["handle"] == handle], key=lambda t: t["score"], reverse=True, ) for t in account_tweets: url = f"https://x.com/{t['handle']}/status/{t['tweet_id']}" is_rt = t.get("full_text", "").startswith("RT @") rt_marker = " [RT]" if is_rt else "" lines.append(f" Score:{t['score']} Likes:{t['likes']} RTs:{t['retweets']} Views:{t['views']} Date:{t['date']}{rt_marker}") lines.append(f" {t['text']}") lines.append(f" {url}") lines.append("") lines.append("") if empty_handles: lines.append("Accounts with No Qualifying Tweets (recent but below thresholds)") lines.append("") for h in sorted(empty_handles): bio = BIO_INFO.get(h, "") lines.append(f" @{h} -- {bio}") lines.append("") if error_handles: lines.append("API Errors") lines.append("") for e in error_handles: lines.append(f" {e}") lines.append("") total_scans = sum(len(c["accounts"]) for c in chunks) lines.append("Summary") lines.append(f"Total scans: {total_scans}") lines.append(f"Accounts with content: {accounts_with_content}") lines.append(f"Total qualifying tweets: {total_qualifying}") lines.append(f"API failures: {len(error_handles)}") lines.append(f"Accounts with no qualifying tweets: {len(empty_handles)}") report_text = "\n".join(lines) os.makedirs(OUT_DIR, exist_ok=True) with open(REPORT_FILE, "w") as f: f.write(report_text) print(report_text) print(f"\n[Report saved to {REPORT_FILE}]") if __name__ == "__main__": chunks = load_chunks(DATE_STR) build_report(chunks)