#!/usr/bin/env python3 """Generate the final plain-text report from scan results""" import json from datetime import datetime, timezone, timedelta with open("/home/kiro/trend-engine/trends/x/account-monitor/scan_results_2026-06-08.json") as f: data = json.load(f) accounts = data["accounts"] DATE_STR = "2026-06-08" # Filter for accounts with qualifying tweets active = [a for a in accounts if a.get("qualifying_count", 0) > 0] # Build all individual tweets list for top 10 all_tweets = [] for a in active: for t in a.get("qualifying", []): t["account_handle"] = a["handle"] t["account_bio"] = a.get("bio", "") # Skip RT-only tweets where likes=0 and views=0 (retweet accounts) all_tweets.append(t) # Sort by score descending all_tweets.sort(key=lambda t: t.get("score", 0), reverse=True) 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)", } # Filter out RT-only spam for top 10 (keep only tweets with actual original content or significant views/likes) # But include everything for the "all accounts" section top_10_eligible = [t for t in all_tweets if not (t.get("is_rt") and t.get("likes", 0) == 0 and t.get("views", 0) < 100)] # Build the report 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 i, t in enumerate(top_10_eligible[:10]): rank = i + 1 handle = t.get("author_handle", t.get("account_handle", "unknown")) tweet_id = t.get("id", "") text = t.get("text", "") views = t.get("views", 0) or 0 likes = t.get("likes", 0) or 0 retweets = t.get("retweets", 0) or 0 date = t.get("created_at", "") # Truncate text for snippet text_clean = text.replace("\n", " ").replace("\r", "").strip() if len(text_clean) > 100: text_clean = text_clean[:97] + "..." url = f"https://x.com/{handle}/status/{tweet_id}" # Format views if views >= 1000000: views_str = f"{views/1000000:.1f}M" elif views >= 1000: views_str = f"{views/1000:.0f}K" else: views_str = str(views) lines.append(f"{rank}, @{handle}, {text_clean} ({url}), {views_str}, {date}") lines.append("") lines.append("") # Per-account breakdown (all active accounts) lines.append("All Accounts with High-Engagement Content (past 3 days)") lines.append("") # Sort accounts by total score descending sorted_active = sorted(active, key=lambda a: a.get("total_score", 0), reverse=True) for a in sorted_active: handle = a["handle"] bio = a.get("bio", "") qualifying = a.get("qualifying", []) if not qualifying: continue # Bio enrichment fallback bio_line = f" -- {BIO_INFO.get(handle, '')}" if bio == "" and handle in BIO_INFO else (f" -- {bio}" if bio else "") total_score = a.get("total_score", 0) best = a.get("best_tweet", {}) best_likes = best.get("likes", 0) if best else 0 best_views = best.get("views", 0) if best else 0 best_date = best.get("created_at", "") if best else "" lines.append(f"@{handle}{bio_line}") lines.append(f" {len(qualifying)} qualifying tweets | Total score: {total_score} | Top: {best_likes} likes, {best_views:,} views on {best_date}") for t in qualifying[:5]: # Show top 5 per account tid = t.get("id", "") th = t.get("author_handle", handle) likes = t.get("likes", 0) or 0 retweets = t.get("retweets", 0) or 0 replies = t.get("replies", 0) or 0 views = t.get("views", 0) or 0 date = t.get("created_at", "") text = t.get("text", "").replace("\n", " ").replace("\r", "").strip() is_rt = t.get("is_rt", False) if len(text) > 80: text = text[:77] + "..." rt_tag = " [RT]" if is_rt else "" lines.append(f" {date} | {likes}L {retweets}RT {replies}R {views:,}V{rt_tag} | {text}") lines.append(f" https://x.com/{th}/status/{tid}") lines.append("") # Summary total_scanned = len(accounts) total_with_content = len(active) total_qualifying = sum(a.get("qualifying_count", 0) for a in accounts) errors = data.get("errors", []) zero_accounts = [a["handle"] for a in accounts if a.get("qualifying_count", 0) == 0] lines.append("Summary") lines.append("") lines.append(f"Total accounts scanned: {total_scanned}") lines.append(f"High-engagement tweets found: {total_qualifying}") lines.append(f"Accounts with high-engagement content: {total_with_content}") lines.append(f"API failures: {len(errors)}") lines.append(f"Accounts with zero qualifying tweets: {len(zero_accounts)}") lines.append("") # Top accounts quick table lines.append("Top Accounts by Engagement (table)") lines.append("#, Account, Tweets, Score, Best Tweet URL, Date, Views") sorted_by_score = sorted(active, key=lambda a: a.get("total_score", 0), reverse=True) for i, a in enumerate(sorted_by_score[:10]): rank = i + 1 handle = a["handle"] tcount = a.get("qualifying_count", 0) tscore = a.get("total_score", 0) best = a.get("best_tweet", {}) best_url = f"https://x.com/{best.get('author_handle', handle)}/status/{best.get('id', '')}" if best and best.get('id') else "N/A" bdate = best.get("created_at", "") if best else "" bviews = f"{best.get('views', 0):,}" if best else "0" lines.append(f"{rank}, @{handle}, {tcount}, {tscore}, {best_url}, {bdate}, {bviews}") report = "\n".join(lines) # Save to disk output_path = f"/home/kiro/trend-engine/trends/x/account-monitor/{DATE_STR}_account-monitor.txt" with open(output_path, "w") as f: f.write(report) print(f"Report saved to {output_path}") print("=" * 60) print(report)