#!/usr/bin/env python3 """ Generate .md reports from banger-x parsed.json files. Mimics the reddit trend report format at /home/kiro/trend-engine/trends/reddit/trend_report_*.md Usage: python3 generate_md_report.py # today's report python3 generate_md_report.py 2026-08-03 # specific date python3 generate_md_report.py --all # all dates missing .md """ import json, os, sys from datetime import datetime, timezone from pathlib import Path from collections import Counter RESULTS_DIR = Path("/home/kiro/trend-engine/trends/x/banger-x-results") # Query ID -> category mapping (for grouping in report) QUERY_CATEGORIES = { "1": ("🎬 AI Video & Film", ["1", "2", "3"]), "2": ("🎬 AI Video & Film", ["1", "2", "3"]), "3": ("🎬 AI Video & Film", ["1", "2", "3"]), "4": ("📢 Brand & Marketing", ["4", "5"]), "5": ("📢 Brand & Marketing", ["4", "5"]), "6": ("🤖 AI Agents & Tools", ["6", "7"]), "7": ("🤖 AI Agents & Tools", ["6", "7"]), "8": ("😴 Sleep & Wellness", ["8", "9", "10"]), "9": ("😴 Sleep & Wellness", ["8", "9", "10"]), "10": ("😴 Sleep & Wellness", ["8", "9", "10"]), "11": ("🛒 Ecom & DTC", ["11", "12"]), "12": ("🛒 Ecom & DTC", ["11", "12"]), "14": ("✈️ Travel & eSIM", ["14", "15", "16", "17", "18"]), "15": ("✈️ Travel & eSIM", ["14", "15", "16", "17", "18"]), "16": ("✈️ Travel & eSIM", ["14", "15", "16", "17", "18"]), "17": ("✈️ Travel & eSIM", ["14", "15", "16", "17", "18"]), "18": ("✈️ Travel & eSIM", ["14", "15", "16", "17", "18"]), } def generate_report(date_str=None): """Generate .md report for a given date (YYYY-MM-DD). Defaults to today.""" if date_str is None: date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") parsed_path = RESULTS_DIR / f"{date_str}_banger-x-parsed.json" if not parsed_path.exists(): print(f" SKIP: {parsed_path} not found") return None with open(parsed_path) as f: data = json.load(f) queries = data.get("queries", {}) errors = data.get("errors", []) # Group tweets by category categories = {} for qid, qdata in queries.items(): cat_info = QUERY_CATEGORIES.get(qid) if cat_info is None: continue cat_name = cat_info[0] if cat_name not in categories: categories[cat_name] = {"tweets": [], "query_count": 0, "total_tweets": 0} categories[cat_name]["tweets"].extend(qdata.get("tweets", [])) categories[cat_name]["query_count"] += 1 categories[cat_name]["total_tweets"] += qdata.get("count", 0) # Total stats total_tweets = sum(q.get("count", 0) for q in queries.values()) total_queries = len(queries) # Build markdown md = [] now = datetime.now(timezone.utc) day_name = now.strftime("%A") date_display = now.strftime("%d %B %Y") md.append(f"# 🔵 X/Twitter Trend Report -- {day_name}, {date_display}") md.append("") md.append(f"**{total_tweets} tweets** from **{total_queries} queries** (GetXAPI)") if errors: md.append(f"⚠️ {len(errors)} query errors") md.append("") # Per category cat_order = ["🎬 AI Video & Film", "🤖 AI Agents & Tools", "📢 Brand & Marketing", "🛒 Ecom & DTC", "😴 Sleep & Wellness", "✈️ Travel & eSIM"] for cat_name in cat_order: cat_data = categories.get(cat_name) if not cat_data or not cat_data["tweets"]: continue tweets = cat_data["tweets"] # Deduplicate by tweet ID seen = set() unique = [] for t in tweets: tid = t.get("id", "") if tid and tid not in seen: seen.add(tid) unique.append(t) # Sort by views desc unique.sort(key=lambda x: x.get("views", 0), reverse=True) top = unique[:8] md.append(f"## {cat_name}") md.append(f"**{len(unique)} tweets** across {cat_data['query_count']} queries") md.append("") for i, t in enumerate(top, 1): author = t.get("author", "?") text = (t.get("text", "") or t.get("full_text", ""))[:100].replace("\n", " ") likes = t.get("likes", 0) rts = t.get("retweets", 0) views = t.get("views", 0) tid = t.get("id", "") url = f"https://x.com/i/status/{tid}" if tid else "" md.append(f"{i}. **[{text}]({url})** -- @{author} (❤️{likes} 🔄{rts} 👁️{views})") md.append("") # Top keywords all_text = " ".join([ (t.get("text", "") or t.get("full_text", "")) for cat_data in categories.values() for t in cat_data["tweets"] ]) words = [w.strip(".,!?()[]{}:;\"'").lower() for w in all_text.split() if len(w) > 3] stop_words = {"this", "that", "with", "from", "have", "been", "they", "will", "what", "when", "where", "which", "while", "about", "would", "could", "should", "your", "there", "their", "these", "those", "more", "after", "than", "down", "over", "people", "anyone", "using", "think", "still", "here", "like", "just", "really", "actually"} filtered = [w for w in words if w not in stop_words and not w.startswith("http")] top_keywords = Counter(filtered).most_common(15) if top_keywords: md.append("## 🔥 Top Keywords") md.append("") md.append(" | ".join([f"`{w}` ({c})" for w, c in top_keywords])) md.append("") # Save md_path = RESULTS_DIR / f"{date_str}_banger-x-report.md" with open(md_path, "w") as f: f.write("\n".join(md)) print(f" Report saved: {md_path}") return md_path def main(): if "--all" in sys.argv: # Generate for all dates that have parsed.json but no .md for f in sorted(RESULTS_DIR.glob("*_banger-x-parsed.json")): date_str = f.name.split("_")[0] md_path = RESULTS_DIR / f"{date_str}_banger-x-report.md" if not md_path.exists(): print(f"Generating for {date_str}...") generate_report(date_str) return 0 date_str = sys.argv[1] if len(sys.argv) > 1 else None result = generate_report(date_str) if result is None: return 1 return 0 if __name__ == "__main__": sys.exit(main())