#!/usr/bin/env python3 """ Twitter Monitor via Grok API Checks for new tweets from watched accounts using Grok's live X search. """ import os import sys import json import argparse from datetime import datetime from pathlib import Path import requests # Paths SKILL_DIR = Path(__file__).parent.parent DATA_DIR = SKILL_DIR / "data" DATA_DIR.mkdir(exist_ok=True) WATCHLIST_FILE = DATA_DIR / "watchlist.json" LAST_SEEN_FILE = DATA_DIR / "last_seen.json" SECRETS_FILE = Path.home() / ".clawdbot" / "secrets" / "xai.env" def get_api_key(): """Load API key from secrets file.""" if SECRETS_FILE.exists(): with open(SECRETS_FILE) as f: for line in f: if line.startswith("XAI_API_KEY="): return line.strip().split("=", 1)[1] return os.environ.get("XAI_API_KEY") def load_watchlist(): """Load list of accounts to monitor.""" if not WATCHLIST_FILE.exists(): default = { "accounts": [], "priority": [], "updated_at": datetime.now().isoformat() } save_watchlist(default) return default with open(WATCHLIST_FILE) as f: return json.load(f) def save_watchlist(data): """Save watchlist.""" data["updated_at"] = datetime.now().isoformat() with open(WATCHLIST_FILE, 'w') as f: json.dump(data, f, indent=2) def query_grok_x_search(prompt: str, api_key: str) -> dict: """Query Grok API with X search enabled.""" url = "https://api.x.ai/v1/responses" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": "grok-4-1-fast-non-reasoning", "tools": [{"type": "x_search"}], "input": prompt } try: response = requests.post(url, headers=headers, json=payload, timeout=120) response.raise_for_status() result = response.json() return result except requests.exceptions.Timeout: return {"error": "Request timed out after 120 seconds"} except requests.exceptions.RequestException as e: return {"error": f"Request failed: {str(e)}"} except json.JSONDecodeError as e: return {"error": f"Invalid JSON response: {str(e)}"} def extract_response_text(result: dict) -> str: """Extract the text response from Grok API result.""" # Check for explicit errors (but not None values) if result.get("error"): return f"Error: {result['error']}" if not result: return "Error: Empty response from API" try: outputs = result.get("output", []) if not outputs: return f"Error: No output in response. Keys: {list(result.keys())}" for output in outputs: if output.get("type") == "message": contents = output.get("content", []) for content in contents: if content.get("type") == "output_text": text = content.get("text", "") if text: return text return "No tweet content found in response" except Exception as e: return f"Error parsing response: {e}" def check_accounts(accounts: list, api_key: str, hours: int = 12) -> str: """Check for new tweets from accounts.""" if not accounts: return "No accounts in watchlist. Add some with --add @username" # Build query for multiple accounts handles = " OR ".join(f"from:{u}" for u in accounts[:15]) # Max 15 per batch prompt = f"""Find the most recent tweets from these accounts in the last {hours} hours: {', '.join('@' + u for u in accounts[:15])} For each account that posted recently, show: - @username - Tweet text (brief summary if long) - **Exact timestamp** (e.g. "2h ago", "Jan 26, 3:45 PM", or similar) - Engagement (likes, views if notable) IMPORTANT: Always include the time each tweet was posted - this is crucial. Focus on original tweets and interesting replies. Skip low-engagement replies. If someone hasn't posted in {hours} hours, skip them. Be concise - this is a digest.""" result = query_grok_x_search(prompt, api_key) return extract_response_text(result) def format_digest(results: str, is_morning: bool) -> str: """Format results as a digest message.""" time_label = "Morning" if is_morning else "Evening" now = datetime.now().strftime("%B %d, %Y") digest = f"🐦 **Twitter {time_label} Digest**\n" digest += f"_{now}_\n\n" digest += results return digest def add_account(username: str): """Add account to watchlist.""" username = username.lstrip("@").lower() watchlist = load_watchlist() if username in watchlist["accounts"]: print(f"@{username} already in watchlist") return watchlist["accounts"].append(username) save_watchlist(watchlist) print(f"✅ Added @{username} to watchlist") def remove_account(username: str): """Remove account from watchlist.""" username = username.lstrip("@").lower() watchlist = load_watchlist() if username not in watchlist["accounts"]: print(f"@{username} not in watchlist") return watchlist["accounts"].remove(username) if username in watchlist.get("priority", []): watchlist["priority"].remove(username) save_watchlist(watchlist) print(f"✅ Removed @{username} from watchlist") def list_accounts(): """List all watched accounts.""" watchlist = load_watchlist() accounts = watchlist.get("accounts", []) priority = watchlist.get("priority", []) if not accounts: print("Watchlist is empty. Add accounts with --add @username") return print(f"📋 Watching {len(accounts)} accounts:\n") for acc in accounts: star = "⭐ " if acc in priority else " " print(f"{star}@{acc}") def main(): parser = argparse.ArgumentParser(description="Twitter monitoring via Grok API") parser.add_argument("--add", help="Add account to watchlist") parser.add_argument("--remove", help="Remove account from watchlist") parser.add_argument("--list", action="store_true", help="List watched accounts") parser.add_argument("--morning", action="store_true", help="Morning digest format") parser.add_argument("--evening", action="store_true", help="Evening digest format") parser.add_argument("--hours", type=int, default=12, help="Hours to look back") args = parser.parse_args() if args.add: add_account(args.add) return if args.remove: remove_account(args.remove) return if args.list: list_accounts() return # Run check api_key = get_api_key() if not api_key: print("Error: No XAI_API_KEY found", file=sys.stderr) sys.exit(1) watchlist = load_watchlist() accounts = watchlist.get("accounts", []) if not accounts: print("No accounts in watchlist. Add some with: --add @username") return results = check_accounts(accounts, api_key, args.hours) if args.morning or args.evening: output = format_digest(results, args.morning) else: output = results print(output) if __name__ == "__main__": main()