sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(open): Port web search backends, persistence layer, and env merge from openclaw

Browse Source
This commit is contained in:
Matt Van Horn
2026-02-14 23:35:44 -08:00
parent a09413608d
commit ba330e9a0c
7 changed files with 1843 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/briefing.py
+260
View File
@@ -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()
scripts/lib/brave_search.py
+213
View File
@@ -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
scripts/lib/env.py
+65 -21
View File
@@ -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
scripts/lib/openrouter_search.py
+216
View File
@@ -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
scripts/lib/parallel_search.py
+139
View File
@@ -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
scripts/store.py
+654
View File
@@ -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()
scripts/watchlist.py
+296
View File
@@ -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()