diff --git a/scripts/briefing.py b/scripts/briefing.py new file mode 100644 index 0000000..f9ef8d9 --- /dev/null +++ b/scripts/briefing.py @@ -0,0 +1,260 @@ +#!/usr/bin/env python3 +"""Morning briefing generator for last30days. + +Synthesizes accumulated findings into formatted briefings. +The Python script collects the data; the agent (via SKILL.md) does the +beautiful synthesis. This script provides the structured data. + +Usage: + python3 briefing.py generate # Daily briefing data + python3 briefing.py generate --weekly # Weekly digest data + python3 briefing.py show [--date DATE] # Show saved briefing +""" + +import argparse +import json +import sys +from datetime import datetime, timedelta +from pathlib import Path + +SCRIPT_DIR = Path(__file__).parent.resolve() +sys.path.insert(0, str(SCRIPT_DIR)) + +import store + +BRIEFS_DIR = Path.home() / ".local" / "share" / "last30days" / "briefs" + + +def generate_daily(since: str = None) -> dict: + """Generate daily briefing data. + + Returns structured data for the agent to synthesize into a beautiful briefing. + """ + store.init_db() + topics = store.list_topics() + + if not topics: + return { + "status": "no_topics", + "message": "No watchlist topics yet. Add one with: last30days watch add \"your topic\"", + } + + enabled = [t for t in topics if t["enabled"]] + if not enabled: + return { + "status": "no_enabled", + "message": "All topics are paused. Enable a topic to generate briefings.", + } + + # Default: findings since yesterday + if not since: + since = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + + briefing_topics = [] + total_new = 0 + + for topic in enabled: + findings = store.get_new_findings(topic["id"], since) + last_run = topic.get("last_run") + last_status = topic.get("last_status", "unknown") + + # Calculate staleness + stale = False + hours_ago = None + if last_run: + try: + run_dt = datetime.fromisoformat(last_run.replace("Z", "+00:00")) + hours_ago = (datetime.now() - run_dt.replace(tzinfo=None)).total_seconds() / 3600 + stale = hours_ago > 36 # Stale if > 36 hours + except (ValueError, TypeError): + stale = True + + topic_data = { + "name": topic["name"], + "findings": findings, + "new_count": len(findings), + "last_run": last_run, + "last_status": last_status, + "stale": stale, + "hours_ago": round(hours_ago, 1) if hours_ago else None, + } + + # Extract top finding by engagement + if findings: + top = max(findings, key=lambda f: f.get("engagement_score", 0)) + topic_data["top_finding"] = { + "title": top.get("source_title", ""), + "source": top.get("source", ""), + "author": top.get("author", ""), + "engagement": top.get("engagement_score", 0), + "content": top.get("content", "")[:300], + } + + briefing_topics.append(topic_data) + total_new += len(findings) + + # Cost info + daily_cost = store.get_daily_cost() + budget = float(store.get_setting("daily_budget", "5.00")) + + # Find the single top finding across all topics (for TL;DR) + all_findings = [] + for t in briefing_topics: + for f in t["findings"]: + f["_topic"] = t["name"] + all_findings.append(f) + + top_overall = None + if all_findings: + top_overall = max(all_findings, key=lambda f: f.get("engagement_score", 0)) + + result = { + "status": "ok", + "date": datetime.now().strftime("%Y-%m-%d"), + "since": since, + "topics": briefing_topics, + "total_new": total_new, + "total_topics": len(briefing_topics), + "top_finding": { + "title": top_overall.get("source_title", ""), + "topic": top_overall.get("_topic", ""), + "engagement": top_overall.get("engagement_score", 0), + } if top_overall else None, + "cost": { + "daily": daily_cost, + "budget": budget, + }, + "failed_topics": [ + t["name"] for t in briefing_topics if t["last_status"] == "failed" + ], + } + + # Save briefing data + _save_briefing(result) + + return result + + +def generate_weekly() -> dict: + """Generate weekly digest data with trend analysis.""" + store.init_db() + + week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") + two_weeks_ago = (datetime.now() - timedelta(days=14)).strftime("%Y-%m-%d") + + topics = store.list_topics() + if not topics: + return {"status": "no_topics", "message": "No watchlist topics."} + + weekly_topics = [] + + for topic in topics: + if not topic["enabled"]: + continue + + # This week's findings + this_week = store.get_new_findings(topic["id"], week_ago) + + # Last week's findings (for comparison) + conn = store._connect() + try: + last_week_rows = conn.execute( + """SELECT * FROM findings + WHERE topic_id = ? AND first_seen >= ? AND first_seen < ? AND dismissed = 0 + ORDER BY engagement_score DESC""", + (topic["id"], two_weeks_ago, week_ago), + ).fetchall() + last_week = [dict(r) for r in last_week_rows] + finally: + conn.close() + + this_engagement = sum(f.get("engagement_score", 0) for f in this_week) + last_engagement = sum(f.get("engagement_score", 0) for f in last_week) + + # Trend calculation + if last_engagement > 0: + engagement_change = ((this_engagement - last_engagement) / last_engagement) * 100 + else: + engagement_change = 100 if this_engagement > 0 else 0 + + weekly_topics.append({ + "name": topic["name"], + "this_week_count": len(this_week), + "last_week_count": len(last_week), + "this_week_engagement": this_engagement, + "last_week_engagement": last_engagement, + "engagement_change_pct": round(engagement_change, 1), + "top_findings": this_week[:5], # Top 5 by engagement (already sorted) + }) + + result = { + "status": "ok", + "type": "weekly", + "week_of": week_ago, + "topics": weekly_topics, + } + + _save_briefing(result, suffix="-weekly") + + return result + + +def show_briefing(date: str = None) -> dict: + """Load a saved briefing by date.""" + if not date: + date = datetime.now().strftime("%Y-%m-%d") + + path = BRIEFS_DIR / f"{date}.json" + if not path.exists(): + # Try weekly + path = BRIEFS_DIR / f"{date}-weekly.json" + + if not path.exists(): + return {"status": "not_found", "message": f"No briefing found for {date}."} + + with open(path) as f: + return json.load(f) + + +def _save_briefing(data: dict, suffix: str = ""): + """Save briefing data to local archive.""" + BRIEFS_DIR.mkdir(parents=True, exist_ok=True) + date = datetime.now().strftime("%Y-%m-%d") + path = BRIEFS_DIR / f"{date}{suffix}.json" + with open(path, "w") as f: + json.dump(data, f, indent=2, default=str) + + +def main(): + parser = argparse.ArgumentParser(description="Generate last30days briefings") + sub = parser.add_subparsers(dest="command") + + # generate + g = sub.add_parser("generate", help="Generate a briefing") + g.add_argument("--weekly", action="store_true", help="Weekly digest") + g.add_argument("--since", help="Findings since date (YYYY-MM-DD)") + + # show + s = sub.add_parser("show", help="Show a saved briefing") + s.add_argument("--date", help="Date (YYYY-MM-DD, default: today)") + + args = parser.parse_args() + + if args.command == "generate": + if args.weekly: + result = generate_weekly() + else: + result = generate_daily(since=args.since) + print(json.dumps(result, indent=2, default=str)) + + elif args.command == "show": + result = show_briefing(date=args.date) + print(json.dumps(result, indent=2, default=str)) + + else: + parser.print_help() + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/scripts/lib/brave_search.py b/scripts/lib/brave_search.py new file mode 100644 index 0000000..33b64a5 --- /dev/null +++ b/scripts/lib/brave_search.py @@ -0,0 +1,213 @@ +"""Brave Search web search for last30days skill. + +Uses the Brave Search API as a fallback web search backend. +Simple, cheap (free tier: 2,000 queries/month), widely available. + +API docs: https://api-dashboard.search.brave.com/app/documentation/web-search/get-started +""" + +import html +import re +import sys +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional +from urllib.parse import urlencode, urlparse + +from . import http + +ENDPOINT = "https://api.search.brave.com/res/v1/web/search" + +# Freshness codes: pd=24h, pw=7d, pm=31d +FRESHNESS_MAP = {1: "pd", 7: "pw", 31: "pm"} + +# Domains to exclude (handled by Reddit/X search) +EXCLUDED_DOMAINS = { + "reddit.com", "www.reddit.com", "old.reddit.com", + "twitter.com", "www.twitter.com", "x.com", "www.x.com", +} + + +def search_web( + topic: str, + from_date: str, + to_date: str, + api_key: str, + depth: str = "default", +) -> List[Dict[str, Any]]: + """Search the web via Brave Search API. + + Args: + topic: Search topic + from_date: Start date (YYYY-MM-DD) + to_date: End date (YYYY-MM-DD) + api_key: Brave Search API key + depth: 'quick', 'default', or 'deep' + + Returns: + List of result dicts with keys: url, title, snippet, source_domain, date, relevance + + Raises: + http.HTTPError: On API errors + """ + count = {"quick": 8, "default": 15, "deep": 25}.get(depth, 15) + + # Calculate days for freshness filter + days = _days_between(from_date, to_date) + freshness = _brave_freshness(days) + + params = { + "q": topic, + "result_filter": "web,news", + "count": count, + "safesearch": "strict", + "text_decorations": 0, + "spellcheck": 0, + } + if freshness: + params["freshness"] = freshness + + url = f"{ENDPOINT}?{urlencode(params)}" + + sys.stderr.write(f"[Web] Searching Brave for: {topic}\n") + sys.stderr.flush() + + response = http.request( + "GET", + url, + headers={"X-Subscription-Token": api_key}, + timeout=15, + ) + + return _normalize_results(response, from_date, to_date) + + +def _days_between(from_date: str, to_date: str) -> int: + """Calculate days between two YYYY-MM-DD dates.""" + try: + d1 = datetime.strptime(from_date, "%Y-%m-%d") + d2 = datetime.strptime(to_date, "%Y-%m-%d") + return max(1, (d2 - d1).days) + except (ValueError, TypeError): + return 30 + + +def _brave_freshness(days: Optional[int]) -> Optional[str]: + """Convert days to Brave freshness parameter. + + Uses canned codes for <=31d, explicit date range for longer periods. + """ + if days is None: + return None + code = next((v for d, v in sorted(FRESHNESS_MAP.items()) if days <= d), None) + if code: + return code + start = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d") + end = datetime.now(timezone.utc).strftime("%Y-%m-%d") + return f"{start}to{end}" + + +def _normalize_results( + response: Dict[str, Any], + from_date: str, + to_date: str, +) -> List[Dict[str, Any]]: + """Convert Brave Search response to websearch item schema. + + Merges news + web results, cleans HTML entities, filters excluded domains. + """ + items = [] + + # Merge news results (tend to be more recent) with web results + raw_results = ( + response.get("news", {}).get("results", []) + + response.get("web", {}).get("results", []) + ) + + for i, result in enumerate(raw_results): + if not isinstance(result, dict): + continue + + url = result.get("url", "") + if not url: + continue + + # Skip excluded domains + try: + domain = urlparse(url).netloc.lower() + if domain in EXCLUDED_DOMAINS: + continue + if domain.startswith("www."): + domain = domain[4:] + except Exception: + domain = "" + + title = _clean_html(str(result.get("title", "")).strip()) + snippet = _clean_html(str(result.get("description", "")).strip()) + + if not title and not snippet: + continue + + # Parse date from Brave's 'age' field or 'page_age' + date = _parse_brave_date(result.get("age"), result.get("page_age")) + date_confidence = "med" if date else "low" + + items.append({ + "id": f"W{i+1}", + "title": title[:200], + "url": url, + "source_domain": domain, + "snippet": snippet[:500], + "date": date, + "date_confidence": date_confidence, + "relevance": 0.6, # Brave doesn't provide relevance scores + "why_relevant": "", + }) + + sys.stderr.write(f"[Web] Brave: {len(items)} results\n") + sys.stderr.flush() + + return items + + +def _clean_html(text: str) -> str: + """Remove HTML tags and decode entities.""" + text = re.sub(r"<[^>]*>", "", text) + text = html.unescape(text) + return text + + +def _parse_brave_date(age: Optional[str], page_age: Optional[str]) -> Optional[str]: + """Parse Brave's age/page_age fields to YYYY-MM-DD. + + Brave returns dates like "3 hours ago", "2 days ago", "January 24, 2026". + """ + text = age or page_age + if not text: + return None + + text_lower = text.lower().strip() + now = datetime.now() + + # "X hours ago" -> today + if re.search(r'\d+\s*hours?\s*ago', text_lower): + return now.strftime("%Y-%m-%d") + + # "X days ago" + match = re.search(r'(\d+)\s*days?\s*ago', text_lower) + if match: + days = int(match.group(1)) + if days <= 60: + return (now - timedelta(days=days)).strftime("%Y-%m-%d") + + # "X weeks ago" + match = re.search(r'(\d+)\s*weeks?\s*ago', text_lower) + if match: + weeks = int(match.group(1)) + return (now - timedelta(weeks=weeks)).strftime("%Y-%m-%d") + + # ISO format: 2026-01-24T... + match = re.search(r'(\d{4}-\d{2}-\d{2})', text) + if match: + return match.group(1) + + return None diff --git a/scripts/lib/env.py b/scripts/lib/env.py index d549768..81d085e 100644 --- a/scripts/lib/env.py +++ b/scripts/lib/env.py @@ -1,5 +1,6 @@ """Environment and API key management for last30days skill.""" +import json import os from pathlib import Path from typing import Optional, Dict, Any @@ -48,15 +49,22 @@ def get_config() -> Dict[str, Any]: # Load from config file first (if configured) file_env = load_env_file(CONFIG_FILE) if CONFIG_FILE else {} - # Environment variables override file - config = { - 'OPENAI_API_KEY': os.environ.get('OPENAI_API_KEY') or file_env.get('OPENAI_API_KEY'), - 'XAI_API_KEY': os.environ.get('XAI_API_KEY') or file_env.get('XAI_API_KEY'), - 'OPENAI_MODEL_POLICY': os.environ.get('OPENAI_MODEL_POLICY') or file_env.get('OPENAI_MODEL_POLICY', 'auto'), - 'OPENAI_MODEL_PIN': os.environ.get('OPENAI_MODEL_PIN') or file_env.get('OPENAI_MODEL_PIN'), - 'XAI_MODEL_POLICY': os.environ.get('XAI_MODEL_POLICY') or file_env.get('XAI_MODEL_POLICY', 'latest'), - 'XAI_MODEL_PIN': os.environ.get('XAI_MODEL_PIN') or file_env.get('XAI_MODEL_PIN'), - } + # Build config: process.env > .env file + keys = [ + ('OPENAI_API_KEY', None), + ('XAI_API_KEY', None), + ('OPENROUTER_API_KEY', None), + ('PARALLEL_API_KEY', None), + ('BRAVE_API_KEY', None), + ('OPENAI_MODEL_POLICY', 'auto'), + ('OPENAI_MODEL_PIN', None), + ('XAI_MODEL_POLICY', 'latest'), + ('XAI_MODEL_PIN', None), + ] + + config = {} + for key, default in keys: + config[key] = os.environ.get(key) or file_env.get(key, default) return config @@ -69,28 +77,53 @@ def config_exists() -> bool: def get_available_sources(config: Dict[str, Any]) -> str: """Determine which sources are available based on API keys. - Returns: 'both', 'reddit', 'x', or 'web' (fallback when no keys) + Returns: 'all', 'both', 'reddit', 'reddit-web', 'x', 'x-web', 'web', or 'none' """ has_openai = bool(config.get('OPENAI_API_KEY')) has_xai = bool(config.get('XAI_API_KEY')) + has_web = has_web_search_keys(config) if has_openai and has_xai: - return 'both' + return 'all' if has_web else 'both' elif has_openai: - return 'reddit' + return 'reddit-web' if has_web else 'reddit' elif has_xai: - return 'x' + return 'x-web' if has_web else 'x' + elif has_web: + return 'web' else: - return 'web' # Fallback: WebSearch only (no API keys needed) + return 'web' # Fallback: assistant WebSearch (no API keys needed) + + +def has_web_search_keys(config: Dict[str, Any]) -> bool: + """Check if any web search API keys are configured.""" + return bool(config.get('OPENROUTER_API_KEY') or config.get('PARALLEL_API_KEY') or config.get('BRAVE_API_KEY')) + + +def get_web_search_source(config: Dict[str, Any]) -> Optional[str]: + """Determine the best available web search backend. + + Priority: Parallel AI > Brave > OpenRouter/Sonar Pro + + Returns: 'parallel', 'brave', 'openrouter', or None + """ + if config.get('PARALLEL_API_KEY'): + return 'parallel' + if config.get('BRAVE_API_KEY'): + return 'brave' + if config.get('OPENROUTER_API_KEY'): + return 'openrouter' + return None def get_missing_keys(config: Dict[str, Any]) -> str: """Determine which sources are missing (accounting for Bird). - Returns: 'both', 'reddit', 'x', or 'none' + Returns: 'all', 'both', 'reddit', 'x', 'web', or 'none' """ has_openai = bool(config.get('OPENAI_API_KEY')) has_xai = bool(config.get('XAI_API_KEY')) + has_web = has_web_search_keys(config) # Check if Bird provides X access (import here to avoid circular dependency) from . import bird_x @@ -98,14 +131,16 @@ def get_missing_keys(config: Dict[str, Any]) -> str: has_x = has_xai or has_bird - if has_openai and has_x: + if has_openai and has_x and has_web: return 'none' + elif has_openai and has_x: + return 'web' # Missing web search keys elif has_openai: - return 'x' # Missing X source + return 'x' # Missing X source (and possibly web) elif has_x: - return 'reddit' # Missing OpenAI key + return 'reddit' # Missing OpenAI key (and possibly web) else: - return 'both' # Missing both + return 'all' # Missing everything def validate_sources(requested: str, available: str, include_web: bool = False) -> tuple[str, Optional[str]]: @@ -119,14 +154,23 @@ def validate_sources(requested: str, available: str, include_web: bool = False) Returns: Tuple of (effective_sources, error_message) """ - # WebSearch-only mode (no API keys) + # No API keys at all + if available == 'none': + if requested == 'auto': + return 'web', "No API keys configured. The assistant can still search the web if it has a search tool." + elif requested == 'web': + return 'web', None + else: + return 'web', f"No API keys configured. Add keys to ~/.config/last30days/.env for Reddit/X." + + # Web-only mode (only web search API keys) if available == 'web': if requested == 'auto': return 'web', None elif requested == 'web': return 'web', None else: - return 'web', f"No API keys configured. Using WebSearch fallback. Add keys to ~/.config/last30days/.env for Reddit/X." + return 'web', f"Only web search keys configured. Add OPENAI_API_KEY for Reddit, XAI_API_KEY for X." if requested == 'auto': # Add web to sources if include_web is set diff --git a/scripts/lib/openrouter_search.py b/scripts/lib/openrouter_search.py new file mode 100644 index 0000000..5342c2f --- /dev/null +++ b/scripts/lib/openrouter_search.py @@ -0,0 +1,216 @@ +"""Perplexity Sonar Pro web search via OpenRouter for last30days skill. + +Uses OpenRouter's chat completions API with Perplexity's Sonar Pro model, +which has built-in web search and returns citations with URLs, titles, and dates. +This is the recommended web search backend -- highest quality results. + +API docs: https://openrouter.ai/docs/quickstart +Model: perplexity/sonar-pro +""" + +import re +import sys +from typing import Any, Dict, List, Optional +from urllib.parse import urlparse + +from . import http + +ENDPOINT = "https://openrouter.ai/api/v1/chat/completions" +MODEL = "perplexity/sonar-pro" + +# Domains to exclude (handled by Reddit/X search) +EXCLUDED_DOMAINS = { + "reddit.com", "www.reddit.com", "old.reddit.com", + "twitter.com", "www.twitter.com", "x.com", "www.x.com", +} + + +def search_web( + topic: str, + from_date: str, + to_date: str, + api_key: str, + depth: str = "default", +) -> List[Dict[str, Any]]: + """Search the web via Perplexity Sonar Pro on OpenRouter. + + Args: + topic: Search topic + from_date: Start date (YYYY-MM-DD) + to_date: End date (YYYY-MM-DD) + api_key: OpenRouter API key + depth: 'quick', 'default', or 'deep' + + Returns: + List of result dicts with keys: url, title, snippet, source_domain, date, relevance + + Raises: + http.HTTPError: On API errors + """ + max_tokens = {"quick": 1024, "default": 2048, "deep": 4096}.get(depth, 2048) + + prompt = ( + f"Find recent blog posts, news articles, tutorials, and discussions " + f"about {topic} published between {from_date} and {to_date}. " + f"Exclude results from reddit.com, x.com, and twitter.com. " + f"For each result, provide the title, URL, publication date, " + f"and a brief summary of why it's relevant." + ) + + payload = { + "model": MODEL, + "messages": [{"role": "user", "content": prompt}], + "max_tokens": max_tokens, + } + + sys.stderr.write(f"[Web] Searching Sonar Pro via OpenRouter for: {topic}\n") + sys.stderr.flush() + + response = http.post( + ENDPOINT, + json_data=payload, + headers={ + "Authorization": f"Bearer {api_key}", + "HTTP-Referer": "https://github.com/mvanhorn/last30days-openclaw", + "X-Title": "last30days", + }, + timeout=30, + ) + + return _normalize_results(response) + + +def _normalize_results(response: Dict[str, Any]) -> List[Dict[str, Any]]: + """Convert Sonar Pro response to websearch item schema. + + Sonar Pro returns: + - search_results: [{title, url, date}] -- structured source metadata + - citations: [url, ...] -- flat list of cited URLs + - choices[0].message.content -- the synthesized text with [N] references + + We prefer search_results (richer metadata), fall back to citations. + """ + items = [] + + # Try search_results first (has title, url, date) + search_results = response.get("search_results", []) + if isinstance(search_results, list) and search_results: + items = _parse_search_results(search_results) + + # Fall back to citations if no search_results + if not items: + citations = response.get("citations", []) + content = _get_content(response) + if isinstance(citations, list) and citations: + items = _parse_citations(citations, content) + + sys.stderr.write(f"[Web] Sonar Pro: {len(items)} results\n") + sys.stderr.flush() + + return items + + +def _parse_search_results(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Parse the search_results array from Sonar Pro.""" + items = [] + + for i, result in enumerate(results): + if not isinstance(result, dict): + continue + + url = result.get("url", "") + if not url: + continue + + # Skip excluded domains + try: + domain = urlparse(url).netloc.lower() + if domain in EXCLUDED_DOMAINS: + continue + if domain.startswith("www."): + domain = domain[4:] + except Exception: + domain = "" + + title = str(result.get("title", "")).strip() + if not title: + continue + + # Sonar Pro provides dates in search_results + date = result.get("date") + date_confidence = "med" if date else "low" + + items.append({ + "id": f"W{i+1}", + "title": title[:200], + "url": url, + "source_domain": domain, + "snippet": str(result.get("snippet", result.get("description", ""))).strip()[:500], + "date": date, + "date_confidence": date_confidence, + "relevance": 0.7, # Sonar Pro results are generally high quality + "why_relevant": "", + }) + + return items + + +def _parse_citations(citations: List[str], content: str) -> List[Dict[str, Any]]: + """Parse the flat citations array, enriching with content context.""" + items = [] + + for i, url in enumerate(citations): + if not isinstance(url, str) or not url: + continue + + # Skip excluded domains + try: + domain = urlparse(url).netloc.lower() + if domain in EXCLUDED_DOMAINS: + continue + if domain.startswith("www."): + domain = domain[4:] + except Exception: + domain = "" + + # Try to extract title from content references like [1] Title... + title = _extract_title_for_citation(content, i + 1) or domain + + items.append({ + "id": f"W{i+1}", + "title": title[:200], + "url": url, + "source_domain": domain, + "snippet": "", + "date": None, + "date_confidence": "low", + "relevance": 0.6, + "why_relevant": "", + }) + + return items + + +def _get_content(response: Dict[str, Any]) -> str: + """Extract the text content from the chat completion response.""" + try: + return response["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError): + return "" + + +def _extract_title_for_citation(content: str, index: int) -> Optional[str]: + """Try to extract a title near a citation reference [N] in the content.""" + if not content: + return None + + # Look for patterns like [1] Title or [1](url) Title + pattern = rf'\[{index}\][)\s]*([^\[\n]{{5,80}})' + match = re.search(pattern, content) + if match: + title = match.group(1).strip().rstrip('.') + # Clean up markdown artifacts + title = re.sub(r'[*_`]', '', title) + return title if len(title) > 3 else None + + return None diff --git a/scripts/lib/parallel_search.py b/scripts/lib/parallel_search.py new file mode 100644 index 0000000..84b20c5 --- /dev/null +++ b/scripts/lib/parallel_search.py @@ -0,0 +1,139 @@ +"""Parallel AI web search for last30days skill. + +Uses the Parallel AI Search API to find web content (blogs, docs, news, tutorials). +This is the preferred web search backend -- it returns LLM-optimized results +with extended excerpts ranked by relevance. + +API docs: https://docs.parallel.ai/search-api/search-quickstart +""" + +import json +import sys +from typing import Any, Dict, List, Optional +from urllib.parse import urlparse + +from . import http + +ENDPOINT = "https://api.parallel.ai/v1beta/search" + +# Domains to exclude (handled by Reddit/X search) +EXCLUDED_DOMAINS = { + "reddit.com", "www.reddit.com", "old.reddit.com", + "twitter.com", "www.twitter.com", "x.com", "www.x.com", +} + + +def search_web( + topic: str, + from_date: str, + to_date: str, + api_key: str, + depth: str = "default", +) -> List[Dict[str, Any]]: + """Search the web via Parallel AI Search API. + + Args: + topic: Search topic + from_date: Start date (YYYY-MM-DD) + to_date: End date (YYYY-MM-DD) + api_key: Parallel AI API key + depth: 'quick', 'default', or 'deep' + + Returns: + List of result dicts with keys: url, title, snippet, source_domain, date, relevance + + Raises: + http.HTTPError: On API errors + """ + max_results = {"quick": 8, "default": 15, "deep": 25}.get(depth, 15) + + payload = { + "objective": ( + f"Find recent blog posts, tutorials, news articles, and discussions " + f"about {topic} from {from_date} to {to_date}. " + f"Exclude reddit.com, x.com, and twitter.com." + ), + "max_results": max_results, + "max_chars_per_result": 500, + } + + sys.stderr.write(f"[Web] Searching Parallel AI for: {topic}\n") + sys.stderr.flush() + + response = http.post( + ENDPOINT, + json_data=payload, + headers={ + "Authorization": f"Bearer {api_key}", + "parallel-beta": "search-extract-2025-10-10", + }, + timeout=30, + ) + + return _normalize_results(response) + + +def _normalize_results(response: Dict[str, Any]) -> List[Dict[str, Any]]: + """Convert Parallel AI response to websearch item schema. + + Args: + response: Raw API response + + Returns: + List of normalized result dicts + """ + items = [] + + # Handle different response shapes + results = response.get("results", []) + if not isinstance(results, list): + return items + + for i, result in enumerate(results): + if not isinstance(result, dict): + continue + + url = result.get("url", "") + if not url: + continue + + # Skip excluded domains + try: + domain = urlparse(url).netloc.lower() + if domain in EXCLUDED_DOMAINS: + continue + # Clean domain for display + if domain.startswith("www."): + domain = domain[4:] + except Exception: + domain = "" + + title = str(result.get("title", "")).strip() + snippet = str(result.get("excerpt", result.get("snippet", result.get("description", "")))).strip() + + if not title and not snippet: + continue + + # Extract relevance score if provided + relevance = result.get("relevance_score", result.get("relevance", 0.6)) + try: + relevance = min(1.0, max(0.0, float(relevance))) + except (TypeError, ValueError): + relevance = 0.6 + + items.append({ + "id": f"W{i+1}", + "title": title[:200], + "url": url, + "source_domain": domain, + "snippet": snippet[:500], + "date": result.get("published_date", result.get("date")), + "date_confidence": "med" if result.get("published_date") or result.get("date") else "low", + "relevance": relevance, + "why_relevant": str(result.get("summary", "")).strip()[:200], + }) + + sys.stderr.write(f"[Web] Parallel AI: {len(items)} results\n") + sys.stderr.flush() + + return items diff --git a/scripts/store.py b/scripts/store.py new file mode 100644 index 0000000..02552ef --- /dev/null +++ b/scripts/store.py @@ -0,0 +1,654 @@ +#!/usr/bin/env python3 +"""SQLite research accumulator for last30days. + +Stores topics, research runs, and findings with: +- WAL mode for safe concurrent access (cron + user) +- FTS5 full-text search with porter+unicode61 tokenizer +- URL-based dedup with engagement metric updates on re-sighting +- Lightweight schema migrations without external dependencies + +Database location: ~/.local/share/last30days/research.db +""" + +import argparse +import json +import sqlite3 +import sys +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any, Dict, List, Optional + +DB_DIR = Path.home() / ".local" / "share" / "last30days" +DB_PATH = DB_DIR / "research.db" + +# Allow override for testing +_db_override = None + + +def _get_db_path() -> Path: + return _db_override or DB_PATH + + +SCHEMA_V1 = """ +PRAGMA journal_mode=WAL; +PRAGMA synchronous=NORMAL; +PRAGMA cache_size=-64000; + +CREATE TABLE IF NOT EXISTS schema_version ( + version INTEGER PRIMARY KEY, + applied_at TEXT DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS topics ( + id INTEGER PRIMARY KEY, + name TEXT UNIQUE NOT NULL, + search_queries TEXT, + schedule TEXT, + enabled INTEGER DEFAULT 1, + created_at TEXT DEFAULT (datetime('now')), + updated_at TEXT DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS research_runs ( + id INTEGER PRIMARY KEY, + topic_id INTEGER REFERENCES topics(id), + run_date TEXT NOT NULL, + source_mode TEXT, + prompt_tokens INTEGER, + completion_tokens INTEGER, + token_cost REAL, + duration_seconds REAL, + status TEXT DEFAULT 'completed', + error_message TEXT, + findings_new INTEGER DEFAULT 0, + findings_updated INTEGER DEFAULT 0, + created_at TEXT DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS findings ( + id INTEGER PRIMARY KEY, + run_id INTEGER REFERENCES research_runs(id), + topic_id INTEGER REFERENCES topics(id), + source TEXT NOT NULL, + source_url TEXT UNIQUE, + source_title TEXT, + author TEXT, + content TEXT, + summary TEXT, + engagement_score REAL, + relevance_score REAL, + first_seen TEXT DEFAULT (datetime('now')), + last_seen TEXT DEFAULT (datetime('now')), + sighting_count INTEGER DEFAULT 1, + dismissed INTEGER DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS idx_findings_topic ON findings(topic_id, first_seen); +CREATE INDEX IF NOT EXISTS idx_findings_source ON findings(source, topic_id); +CREATE INDEX IF NOT EXISTS idx_findings_url ON findings(source_url); + +CREATE VIRTUAL TABLE IF NOT EXISTS findings_fts USING fts5( + content, summary, source_title, author, + tokenize='porter unicode61', + content='findings', + content_rowid='id' +); + +CREATE TRIGGER IF NOT EXISTS findings_ai AFTER INSERT ON findings BEGIN + INSERT INTO findings_fts(rowid, content, summary, source_title, author) + VALUES (new.id, new.content, new.summary, new.source_title, new.author); +END; + +CREATE TRIGGER IF NOT EXISTS findings_ad AFTER DELETE ON findings BEGIN + INSERT INTO findings_fts(findings_fts, rowid, content, summary, source_title, author) + VALUES ('delete', old.id, old.content, old.summary, old.source_title, old.author); +END; + +CREATE TRIGGER IF NOT EXISTS findings_au AFTER UPDATE ON findings BEGIN + INSERT INTO findings_fts(findings_fts, rowid, content, summary, source_title, author) + VALUES ('delete', old.id, old.content, old.summary, old.source_title, old.author); + INSERT INTO findings_fts(rowid, content, summary, source_title, author) + VALUES (new.id, new.content, new.summary, new.source_title, new.author); +END; + +CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT, + updated_at TEXT DEFAULT (datetime('now')) +); +""" + +SCHEMA_V1_DEFAULTS = """ +INSERT OR IGNORE INTO schema_version (version) VALUES (1); +INSERT OR IGNORE INTO settings (key, value) VALUES ('daily_budget', '5.00'); +INSERT OR IGNORE INTO settings (key, value) VALUES ('delivery_channel', ''); +INSERT OR IGNORE INTO settings (key, value) VALUES ('delivery_mode', 'announce'); +INSERT OR IGNORE INTO settings (key, value) VALUES ('briefing_format', 'concise'); +INSERT OR IGNORE INTO settings (key, value) VALUES ('default_schedule', '0 8 * * *'); +""" + +# Future migrations keyed by version number +MIGRATIONS: Dict[int, str] = { + # 2: "ALTER TABLE findings ADD COLUMN tags TEXT DEFAULT '[]';", +} + + +def _connect(db_path: Optional[Path] = None) -> sqlite3.Connection: + """Open a connection with WAL mode and row factory.""" + path = db_path or _get_db_path() + conn = sqlite3.connect(str(path)) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + conn.execute("PRAGMA foreign_keys=ON") + return conn + + +def init_db(db_path: Optional[Path] = None) -> Path: + """Create database and tables if they don't exist. Returns the DB path.""" + path = db_path or _get_db_path() + path.parent.mkdir(parents=True, exist_ok=True) + + conn = _connect(path) + try: + conn.executescript(SCHEMA_V1) + conn.executescript(SCHEMA_V1_DEFAULTS) + _run_migrations(conn) + conn.commit() + finally: + conn.close() + + return path + + +def _run_migrations(conn: sqlite3.Connection): + """Apply pending schema migrations.""" + current = conn.execute( + "SELECT MAX(version) FROM schema_version" + ).fetchone()[0] or 0 + + for version in sorted(MIGRATIONS.keys()): + if version > current: + conn.executescript(MIGRATIONS[version]) + conn.execute( + "INSERT INTO schema_version (version) VALUES (?)", (version,) + ) + + +# --- Topics --- + + +def add_topic( + name: str, + search_queries: Optional[List[str]] = None, + schedule: str = "0 8 * * *", +) -> Dict[str, Any]: + """Add a topic to the watchlist. Returns the topic dict.""" + init_db() + conn = _connect() + try: + queries_json = json.dumps(search_queries) if search_queries else None + conn.execute( + """INSERT INTO topics (name, search_queries, schedule) + VALUES (?, ?, ?) + ON CONFLICT(name) DO UPDATE SET + search_queries = excluded.search_queries, + schedule = excluded.schedule, + updated_at = datetime('now')""", + (name, queries_json, schedule), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM topics WHERE name = ?", (name,) + ).fetchone() + return dict(row) + finally: + conn.close() + + +def remove_topic(name: str) -> bool: + """Remove a topic from the watchlist. Returns True if found.""" + init_db() + conn = _connect() + try: + row = conn.execute( + "SELECT id FROM topics WHERE name = ?", (name,) + ).fetchone() + if not row: + return False + topic_id = row["id"] + # Delete findings and runs for this topic + conn.execute("DELETE FROM findings WHERE topic_id = ?", (topic_id,)) + conn.execute("DELETE FROM research_runs WHERE topic_id = ?", (topic_id,)) + conn.execute("DELETE FROM topics WHERE id = ?", (topic_id,)) + conn.commit() + return True + finally: + conn.close() + + +def list_topics() -> List[Dict[str, Any]]: + """List all topics with stats.""" + init_db() + conn = _connect() + try: + rows = conn.execute( + """SELECT t.*, + (SELECT COUNT(*) FROM findings WHERE topic_id = t.id) as finding_count, + (SELECT MAX(run_date) FROM research_runs WHERE topic_id = t.id) as last_run, + (SELECT status FROM research_runs WHERE topic_id = t.id + ORDER BY created_at DESC LIMIT 1) as last_status + FROM topics t + ORDER BY t.name""" + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +def get_topic(name: str) -> Optional[Dict[str, Any]]: + """Get a topic by name.""" + init_db() + conn = _connect() + try: + row = conn.execute( + "SELECT * FROM topics WHERE name = ?", (name,) + ).fetchone() + return dict(row) if row else None + finally: + conn.close() + + +# --- Research Runs --- + + +def record_run( + topic_id: int, + source_mode: str = "both", + status: str = "completed", + error_message: Optional[str] = None, + duration_seconds: float = 0, + prompt_tokens: int = 0, + completion_tokens: int = 0, + token_cost: float = 0, +) -> int: + """Record a research run. Returns the run ID.""" + conn = _connect() + try: + cursor = conn.execute( + """INSERT INTO research_runs + (topic_id, run_date, source_mode, status, error_message, + duration_seconds, prompt_tokens, completion_tokens, token_cost) + VALUES (?, datetime('now'), ?, ?, ?, ?, ?, ?, ?)""", + ( + topic_id, source_mode, status, error_message, + duration_seconds, prompt_tokens, completion_tokens, token_cost, + ), + ) + conn.commit() + return cursor.lastrowid + finally: + conn.close() + + +def update_run(run_id: int, **kwargs): + """Update a research run's fields.""" + conn = _connect() + try: + sets = ", ".join(f"{k} = ?" for k in kwargs) + values = list(kwargs.values()) + [run_id] + conn.execute(f"UPDATE research_runs SET {sets} WHERE id = ?", values) + conn.commit() + finally: + conn.close() + + +# --- Findings --- + + +def store_findings( + run_id: int, + topic_id: int, + findings: List[Dict[str, Any]], +) -> Dict[str, int]: + """Store findings with URL-based dedup. Returns counts of new/updated.""" + conn = _connect() + new_count = 0 + updated_count = 0 + + try: + for f in findings: + url = f.get("source_url") or f.get("url") + if not url: + continue + + existing = conn.execute( + "SELECT id, engagement_score, sighting_count FROM findings WHERE source_url = ?", + (url,), + ).fetchone() + + if existing: + # Update engagement and re-sighting info + new_engagement = f.get("engagement_score", 0) + conn.execute( + """UPDATE findings SET + last_seen = datetime('now'), + sighting_count = sighting_count + 1, + engagement_score = ?, + run_id = ? + WHERE id = ?""", + ( + max(new_engagement, existing["engagement_score"] or 0), + run_id, + existing["id"], + ), + ) + updated_count += 1 + else: + # New finding + conn.execute( + """INSERT INTO findings + (run_id, topic_id, source, source_url, source_title, + author, content, summary, engagement_score, relevance_score) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + run_id, + topic_id, + f.get("source", "unknown"), + url, + f.get("source_title") or f.get("title", ""), + f.get("author", ""), + f.get("content") or f.get("text", ""), + f.get("summary", ""), + f.get("engagement_score", 0), + f.get("relevance_score", 0), + ), + ) + new_count += 1 + + # Update run stats + conn.execute( + "UPDATE research_runs SET findings_new = ?, findings_updated = ? WHERE id = ?", + (new_count, updated_count, run_id), + ) + conn.commit() + finally: + conn.close() + + return {"new": new_count, "updated": updated_count} + + +def get_new_findings( + topic_id: int, + since: Optional[str] = None, +) -> List[Dict[str, Any]]: + """Get findings for a topic, optionally since a date.""" + conn = _connect() + try: + if since: + rows = conn.execute( + """SELECT * FROM findings + WHERE topic_id = ? AND first_seen >= ? AND dismissed = 0 + ORDER BY first_seen DESC""", + (topic_id, since), + ).fetchall() + else: + rows = conn.execute( + """SELECT * FROM findings + WHERE topic_id = ? AND dismissed = 0 + ORDER BY first_seen DESC""", + (topic_id,), + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +def search_findings(query: str, limit: int = 20) -> List[Dict[str, Any]]: + """FTS5 search across all findings with BM25 ranking.""" + conn = _connect() + try: + rows = conn.execute( + """SELECT f.*, bm25(findings_fts) as rank, t.name as topic_name + FROM findings_fts + JOIN findings f ON f.id = findings_fts.rowid + LEFT JOIN topics t ON t.id = f.topic_id + WHERE findings_fts MATCH ? + ORDER BY rank + LIMIT ?""", + (query, limit), + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +def update_finding(finding_id: int, **kwargs): + """Update a finding's fields.""" + conn = _connect() + try: + sets = ", ".join(f"{k} = ?" for k in kwargs) + values = list(kwargs.values()) + [finding_id] + conn.execute(f"UPDATE findings SET {sets} WHERE id = ?", values) + conn.commit() + finally: + conn.close() + + +def delete_finding(finding_id: int): + """Delete a finding.""" + conn = _connect() + try: + conn.execute("DELETE FROM findings WHERE id = ?", (finding_id,)) + conn.commit() + finally: + conn.close() + + +def dismiss_finding(finding_id: int): + """Mark a finding as dismissed.""" + update_finding(finding_id, dismissed=1) + + +# --- Cost Tracking --- + + +def get_daily_cost(date: Optional[str] = None) -> float: + """Get total token cost for a given day (default: today).""" + conn = _connect() + try: + if not date: + date = datetime.now().strftime("%Y-%m-%d") + row = conn.execute( + """SELECT COALESCE(SUM(token_cost), 0) as total + FROM research_runs + WHERE date(run_date) = date(?)""", + (date,), + ).fetchone() + return row["total"] + finally: + conn.close() + + +# --- Settings --- + + +def get_setting(key: str, default: Optional[str] = None) -> Optional[str]: + """Get a setting value.""" + init_db() + conn = _connect() + try: + row = conn.execute( + "SELECT value FROM settings WHERE key = ?", (key,) + ).fetchone() + return row["value"] if row else default + finally: + conn.close() + + +def set_setting(key: str, value: str): + """Set a setting value.""" + init_db() + conn = _connect() + try: + conn.execute( + """INSERT INTO settings (key, value, updated_at) + VALUES (?, ?, datetime('now')) + ON CONFLICT(key) DO UPDATE SET + value = excluded.value, + updated_at = datetime('now')""", + (key, value), + ) + conn.commit() + finally: + conn.close() + + +# --- Stats --- + + +def get_stats() -> Dict[str, Any]: + """Get overall database stats.""" + conn = _connect() + try: + topic_count = conn.execute("SELECT COUNT(*) FROM topics WHERE enabled = 1").fetchone()[0] + finding_count = conn.execute("SELECT COUNT(*) FROM findings").fetchone()[0] + + week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d") + runs_7d = conn.execute( + "SELECT COUNT(*) FROM research_runs WHERE run_date >= ?", (week_ago,) + ).fetchone()[0] + successful_7d = conn.execute( + "SELECT COUNT(*) FROM research_runs WHERE run_date >= ? AND status = 'completed'", + (week_ago,), + ).fetchone()[0] + failed_7d = conn.execute( + "SELECT COUNT(*) FROM research_runs WHERE run_date >= ? AND status = 'failed'", + (week_ago,), + ).fetchone()[0] + cost_7d = conn.execute( + "SELECT COALESCE(SUM(token_cost), 0) FROM research_runs WHERE run_date >= ?", + (week_ago,), + ).fetchone()[0] + + # Source breakdown + sources = {} + for row in conn.execute( + "SELECT source, COUNT(*) as cnt FROM findings GROUP BY source" + ).fetchall(): + sources[row["source"]] = row["cnt"] + + db_path = _get_db_path() + db_size = db_path.stat().st_size if db_path.exists() else 0 + + return { + "topics_active": topic_count, + "total_findings": finding_count, + "db_size_bytes": db_size, + "runs_7d": runs_7d, + "successful_7d": successful_7d, + "failed_7d": failed_7d, + "cost_7d": cost_7d, + "sources": sources, + "daily_budget": get_setting("daily_budget", "5.00"), + } + finally: + conn.close() + + +def get_trending(days: int = 7) -> List[Dict[str, Any]]: + """Get topics ranked by recent finding activity.""" + conn = _connect() + try: + since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") + rows = conn.execute( + """SELECT t.name, t.id, + COUNT(f.id) as new_findings, + COALESCE(SUM(f.engagement_score), 0) as total_engagement + FROM topics t + LEFT JOIN findings f ON f.topic_id = t.id AND f.first_seen >= ? + WHERE t.enabled = 1 + GROUP BY t.id + ORDER BY new_findings DESC""", + (since,), + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +# --- CLI interface --- + + +def _cli_query(args): + """Handle CLI query command.""" + topic = get_topic(args.topic) + if not topic: + print(json.dumps({"error": f"Topic not found: {args.topic}"})) + return + + since = None + if args.since: + # Parse duration like "7d", "30d" + days = int(args.since.rstrip("d")) + since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") + + findings = get_new_findings(topic["id"], since) + print(json.dumps({"topic": topic["name"], "findings": findings, "count": len(findings)}, default=str)) + + +def _cli_search(args): + """Handle CLI search command.""" + results = search_findings(args.query, limit=args.limit) + print(json.dumps({"query": args.query, "results": results, "count": len(results)}, default=str)) + + +def _cli_trending(args): + """Handle CLI trending command.""" + results = get_trending(args.days) + print(json.dumps({"trending": results}, default=str)) + + +def _cli_stats(args): + """Handle CLI stats command.""" + stats = get_stats() + print(json.dumps(stats, default=str)) + + +def main(): + parser = argparse.ArgumentParser(description="Query the last30days research database") + sub = parser.add_subparsers(dest="command") + + # query + q = sub.add_parser("query", help="Query findings for a topic") + q.add_argument("topic", help="Topic name") + q.add_argument("--since", help="Duration like '7d' or '30d'") + q.set_defaults(func=_cli_query) + + # search + s = sub.add_parser("search", help="Full-text search across findings") + s.add_argument("query", help="Search query") + s.add_argument("--limit", type=int, default=20, help="Max results") + s.set_defaults(func=_cli_search) + + # trending + t = sub.add_parser("trending", help="Show trending topics") + t.add_argument("--days", type=int, default=7, help="Look back N days") + t.set_defaults(func=_cli_trending) + + # stats + st = sub.add_parser("stats", help="Show database stats") + st.set_defaults(func=_cli_stats) + + args = parser.parse_args() + if not args.command: + parser.print_help() + sys.exit(1) + + # Ensure DB exists + init_db() + args.func(args) + + +if __name__ == "__main__": + main() diff --git a/scripts/watchlist.py b/scripts/watchlist.py new file mode 100644 index 0000000..a058d25 --- /dev/null +++ b/scripts/watchlist.py @@ -0,0 +1,296 @@ +#!/usr/bin/env python3 +"""Topic watchlist management for last30days. + +CLI for adding, removing, and listing watched topics with auto-bootstrap. +On first `add`, creates the SQLite database. + +Usage: + python3 watchlist.py add "AI video tools" [--schedule "0 8 * * *"] + python3 watchlist.py add "NVIDIA news" --weekly + python3 watchlist.py remove "AI video tools" + python3 watchlist.py list + python3 watchlist.py run-all + python3 watchlist.py run-one "AI video tools" + python3 watchlist.py config delivery telegram + python3 watchlist.py config budget 10.00 +""" + +import argparse +import json +import subprocess +import sys +import time +from datetime import datetime +from pathlib import Path + +SCRIPT_DIR = Path(__file__).parent.resolve() +sys.path.insert(0, str(SCRIPT_DIR)) + +import store + + +def cmd_add(args): + """Add a topic to the watchlist.""" + schedule = "0 8 * * 1" if args.weekly else (args.schedule or "0 8 * * *") + queries = args.queries.split(",") if args.queries else None + + topic = store.add_topic(args.topic, search_queries=queries, schedule=schedule) + + sched_desc = "weekly (Mondays 8am)" if args.weekly else f"daily ({schedule})" + result = { + "action": "added", + "topic": topic["name"], + "schedule": sched_desc, + "message": f'Added "{topic["name"]}" to watchlist. Schedule: {sched_desc}.', + } + + print(json.dumps(result, default=str)) + + +def cmd_remove(args): + """Remove a topic from the watchlist.""" + removed = store.remove_topic(args.topic) + + if not removed: + print(json.dumps({"action": "not_found", "topic": args.topic, "message": f'Topic not found: "{args.topic}"'})) + return + + remaining = store.list_topics() + + print(json.dumps({ + "action": "removed", + "topic": args.topic, + "message": f'Removed "{args.topic}" from watchlist.', + "remaining": len(remaining), + })) + + +def cmd_list(args): + """List all watched topics with stats.""" + topics = store.list_topics() + budget_used = store.get_daily_cost() + budget_limit = store.get_setting("daily_budget", "5.00") + + result = { + "topics": topics, + "budget_used": budget_used, + "budget_limit": float(budget_limit), + } + print(json.dumps(result, default=str)) + + +def cmd_run_one(args): + """Run research for a single topic.""" + topic = store.get_topic(args.topic) + if not topic: + print(json.dumps({"error": f'Topic not found: "{args.topic}"'})) + sys.exit(1) + + _run_topic(topic) + + +def cmd_run_all(args): + """Run research for all enabled topics with budget guard.""" + topics = store.list_topics() + enabled = [t for t in topics if t["enabled"]] + + if not enabled: + print(json.dumps({"message": "No enabled topics to research."})) + return + + budget_limit = float(store.get_setting("daily_budget", "5.00")) + results = [] + + for topic in enabled: + # Budget guard + daily_cost = store.get_daily_cost() + if daily_cost >= budget_limit: + results.append({ + "topic": topic["name"], + "status": "skipped", + "reason": f"Budget exceeded: ${daily_cost:.2f}/${budget_limit:.2f}", + }) + continue + + result = _run_topic(topic) + results.append(result) + + print(json.dumps({ + "action": "run_all", + "results": results, + "budget_used": store.get_daily_cost(), + "budget_limit": budget_limit, + }, default=str)) + + +def _run_topic(topic: dict) -> dict: + """Run research for a single topic and store findings.""" + start_time = time.time() + topic_id = topic["id"] + + # Record the run + run_id = store.record_run(topic_id, source_mode="both", status="running") + + try: + # Run the research script + cmd = [ + sys.executable, + str(SCRIPT_DIR / "last30days.py"), + topic["name"], + "--emit=json", + ] + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=300, + ) + + duration = time.time() - start_time + + if result.returncode != 0: + store.update_run( + run_id, + status="failed", + error_message=result.stderr[:500], + duration_seconds=duration, + ) + return { + "topic": topic["name"], + "status": "failed", + "error": result.stderr[:200], + "duration": duration, + } + + # Parse research output + data = json.loads(result.stdout) + + # Convert research items to findings format + findings = [] + for item in data.get("reddit", []): + findings.append({ + "source": "reddit", + "url": item.get("url", ""), + "title": item.get("title", ""), + "author": item.get("author", ""), + "content": item.get("title", ""), + "summary": item.get("top_comments_summary", ""), + "engagement_score": item.get("upvotes", 0), + "relevance_score": item.get("relevance", 0), + }) + for item in data.get("x", []): + findings.append({ + "source": "x", + "url": item.get("url", ""), + "title": item.get("text", "")[:100], + "author": item.get("author_handle", ""), + "content": item.get("text", ""), + "engagement_score": item.get("engagement", {}).get("likes", 0), + "relevance_score": item.get("relevance", 0), + }) + + # Store with dedup + counts = store.store_findings(run_id, topic_id, findings) + + store.update_run( + run_id, + status="completed", + duration_seconds=duration, + findings_new=counts["new"], + findings_updated=counts["updated"], + ) + + return { + "topic": topic["name"], + "status": "completed", + "new": counts["new"], + "updated": counts["updated"], + "duration": duration, + } + + except subprocess.TimeoutExpired: + duration = time.time() - start_time + store.update_run( + run_id, status="failed", + error_message="Research timed out after 300s", + duration_seconds=duration, + ) + return {"topic": topic["name"], "status": "failed", "error": "timeout"} + + except json.JSONDecodeError as e: + duration = time.time() - start_time + store.update_run( + run_id, status="failed", + error_message=f"Invalid JSON output: {e}", + duration_seconds=duration, + ) + return {"topic": topic["name"], "status": "failed", "error": f"parse error: {e}"} + + except Exception as e: + duration = time.time() - start_time + store.update_run( + run_id, status="failed", + error_message=str(e)[:500], + duration_seconds=duration, + ) + return {"topic": topic["name"], "status": "failed", "error": str(e)} + + +def cmd_config(args): + """Configure watchlist settings.""" + if args.setting == "delivery": + store.set_setting("delivery_channel", args.value) + print(json.dumps({"action": "config", "setting": "delivery_channel", "value": args.value})) + elif args.setting == "budget": + store.set_setting("daily_budget", args.value) + print(json.dumps({"action": "config", "setting": "daily_budget", "value": args.value})) + else: + print(json.dumps({"error": f"Unknown setting: {args.setting}. Use 'delivery' or 'budget'."})) + + +def main(): + parser = argparse.ArgumentParser(description="Manage last30days topic watchlist") + sub = parser.add_subparsers(dest="command") + + # add + a = sub.add_parser("add", help="Add a topic to the watchlist") + a.add_argument("topic", help="Topic name") + a.add_argument("--schedule", help="Cron expression (default: 0 8 * * *)") + a.add_argument("--weekly", action="store_true", help="Run weekly instead of daily") + a.add_argument("--queries", help="Comma-separated custom search queries") + a.set_defaults(func=cmd_add) + + # remove + r = sub.add_parser("remove", help="Remove a topic from the watchlist") + r.add_argument("topic", help="Topic name") + r.set_defaults(func=cmd_remove) + + # list + l = sub.add_parser("list", help="List all watched topics") + l.set_defaults(func=cmd_list) + + # run-all + ra = sub.add_parser("run-all", help="Run research for all enabled topics") + ra.set_defaults(func=cmd_run_all) + + # run-one + ro = sub.add_parser("run-one", help="Run research for a single topic") + ro.add_argument("topic", help="Topic name") + ro.set_defaults(func=cmd_run_one) + + # config + c = sub.add_parser("config", help="Configure watchlist settings") + c.add_argument("setting", help="Setting name (delivery, budget)") + c.add_argument("value", help="Setting value") + c.set_defaults(func=cmd_config) + + args = parser.parse_args() + if not args.command: + parser.print_help() + sys.exit(1) + + args.func(args) + + +if __name__ == "__main__": + main()