#!/usr/bin/env python3 """X Account Monitor - combined scan + report for cron (2026-09-02).""" import os, re, json, time, requests from datetime import datetime, timezone, timedelta 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: print("ERROR: GETXAPI_KEY not found", flush=True) raise SystemExit(1) WATCHLIST = "/home/kiro/trend-engine/trends/x/raw/rising-accounts.md" with open(WATCHLIST) as f: content = f.read() raw_handles = re.findall(r"@(\w+)", content) # Noise words (markdown reference link-text). Keep nofilmschool + curiousrefuge OUT. SKIP_WORDS = {'x', 'youtube', 'facebook', 'marketingbrew', 'besttwitteraccounts', 'theankler', 'xo2', 'etcjournal', 'letsdatascience', 'pixflow', 'blogs', 'nvidia', 'nofilmschool_com', 'curiousrefuge_com', 'meet', 'blog', 'news', 'official', 'status', 'lang', 'with_replies', 'highlights'} CORE = [ "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", "Chandan_Perla", ] IRRELEVANT = [ "AITOPIAai", "FaithJaiye2187", "KammelaBala", "MyStageAI", "AI_Video_Dumps", "ChrisLaubAI", "GoogleLabs", "TheAIColony", "cladlabs", "AIBuzzNews", "AlexStudio44", "DonutAI", "GoogleAI", "AI_with_jasmin", "CharaspowerAI", "Ai_Vaidehi", "CINEMA505", "AIFrontliner", "AIGuideHQ", "AIWarper", "AITECHio", "4aibsc", ] core_set = set(CORE) irrelevant_set = set(IRRELEVANT) # Build handle list: core first, then irrelevant, then any other watchlist extra handles, seen = [], set() for h in CORE + IRRELEVANT: if h not in seen: handles.append(h); seen.add(h) for h in raw_handles: if h in SKIP_WORDS or h in seen: continue handles.append(h); seen.add(h) print(f"Total handles to scan: {len(handles)} (core {len(CORE)}, irrelevant {len(IRRELEVANT)})", flush=True) def scan_handle(handle): url = f"https://api.getxapi.com/twitter/user/tweets?userName={handle}&count=5" try: resp = requests.get(url, headers={"Authorization": f"Bearer {API_KEY}"}, timeout=60) if resp.status_code != 200: return [], f"HTTP {resp.status_code}" data = resp.json() return data.get("tweets", []), None except Exception as e: return [], str(e) now = datetime.now(timezone.utc) cutoff = (now - timedelta(days=3)).replace(hour=0, minute=0, second=0, microsecond=0) today = now.strftime("%Y-%m-%d") DATE_FMT = "%a %b %d %H:%M:%S %z %Y" def parse_date(s): if not s: return None try: return datetime.strptime(s, DATE_FMT) except Exception: return None all_results = {} failed = [] for handle in handles: tweets, err = scan_handle(handle) if err: failed.append((handle, err)) print(f"FAIL {handle}: {err}", flush=True) else: all_results[handle] = tweets print(f"OK {handle}: {len(tweets)} tweets", flush=True) time.sleep(0.8) json_path = f"/home/kiro/trend-engine/trends/x/scan_results_{today}_FULL.json" with open(json_path, "w") as f: json.dump({"results": all_results, "failed": [x[0] for x in failed]}, f) print(f"Saved {json_path}", flush=True) # ---- scoring ---- qualifying, filtered_qualifying = [], [] for handle, tweets in all_results.items(): if not tweets: continue is_core = handle in core_set for tw in tweets: if not isinstance(tw, dict): continue if tw.get("retweeted_tweet"): continue if (tw.get("text") or "").startswith("RT @"): continue likes = tw.get("likeCount") or 0 rts = tw.get("retweetCount") or 0 replies = tw.get("replyCount") or 0 views = tw.get("viewCount") or 0 score = likes + (rts * 2) + (replies * 3) + (views / 100) dt = parse_date(tw.get("createdAt") or tw.get("created_at")) in_window = dt is not None and dt >= cutoff qualifies = likes >= 50 or rts >= 10 or views >= 1000 if qualifies and in_window: entry = {"handle": handle, "tweet": tw, "score": score, "likes": likes, "rts": rts, "replies": replies, "views": views, "dt": dt, "id": tw.get("id"), "text": tw.get("text") or ""} (qualifying if is_core else filtered_qualifying).append(entry) qualifying.sort(key=lambda x: -x["score"]) filtered_qualifying.sort(key=lambda x: -x["score"]) print(f"Core qualifying: {len(qualifying)} | Filtered qualifying: {len(filtered_qualifying)} | Failed: {len(failed)}", flush=True) # ---- emit JSON summary for report builder ---- scored_path = f"/home/kiro/trend-engine/trends/x/scan_scored_{today}.json" out = { "date": today, "scanned": len(handles), "failed": [h for h, _ in failed], "core": [{k: (e[k].strftime("%Y-%m-%d") if k == "dt" else e[k]) for k in ("handle", "id", "score", "likes", "rts", "replies", "views", "text")} | {"date": e["dt"].strftime("%Y-%m-%d")} for e in qualifying], "filtered": [{k: (e[k].strftime("%Y-%m-%d") if k == "dt" else e[k]) for k in ("handle", "id", "score", "likes", "rts", "replies", "views", "text")} | {"date": e["dt"].strftime("%Y-%m-%d")} for e in filtered_qualifying], } # fix: 'dt' never dumped as string due to comprehension; write cleanly out["core"] = [{"handle": e["handle"], "id": e["id"], "score": e["score"], "likes": e["likes"], "rts": e["rts"], "replies": e["replies"], "views": e["views"], "text": e["text"], "date": e["dt"].strftime("%Y-%m-%d")} for e in qualifying] out["filtered"] = [{"handle": e["handle"], "id": e["id"], "score": e["score"], "likes": e["likes"], "rts": e["rts"], "replies": e["replies"], "views": e["views"], "text": e["text"], "date": e["dt"].strftime("%Y-%m-%d")} for e in filtered_qualifying] with open(scored_path, "w") as f: json.dump(out, f, indent=1) print(f"Saved {scored_path}", flush=True)