| 1 | #!/usr/bin/env python3 |
| 2 | """SQLite research accumulator for last30days. |
| 3 | |
| 4 | Stores topics, research runs, and findings with: |
| 5 | - WAL mode for safe concurrent access (cron + user) |
| 6 | - FTS5 full-text search with porter+unicode61 tokenizer |
| 7 | - URL-based dedup with engagement metric updates on re-sighting |
| 8 | - Lightweight schema migrations without external dependencies |
| 9 | |
| 10 | Database location: ~/.local/share/last30days/research.db |
| 11 | """ |
| 12 | |
| 13 | import argparse |
| 14 | import json |
| 15 | import os |
| 16 | import re |
| 17 | import sqlite3 |
| 18 | import sys |
| 19 | from contextlib import contextmanager |
| 20 | from datetime import datetime, timedelta, timezone |
| 21 | from pathlib import Path |
| 22 | from typing import Any, Dict, Iterator, List, Optional |
| 23 | |
| 24 | SCRIPT_DIR = Path(__file__).parent.resolve() |
| 25 | sys.path.insert(0, str(SCRIPT_DIR)) |
| 26 | |
| 27 | from lib import dedupe, entity_extract, schema |
| 28 | |
| 29 | DB_DIR = Path.home() / ".local" / "share" / "last30days" |
| 30 | DB_PATH = DB_DIR / "research.db" |
| 31 | |
| 32 | # Allow override for testing |
| 33 | _db_override = None |
| 34 | |
| 35 | |
| 36 | def _get_db_path() -> Path: |
| 37 | return _db_override or DB_PATH |
| 38 | |
| 39 | |
| 40 | @contextmanager |
| 41 | def scoped_db(db_path: Optional[Path]) -> Iterator[None]: |
| 42 | """Route all store access inside the block to ``db_path``. |
| 43 | |
| 44 | ``None`` keeps the shared store. Scoped runs (``--save-dir``) use this so |
| 45 | their findings land next to their briefs instead of leaking into the |
| 46 | shared research.db that unscoped searches read. |
| 47 | """ |
| 48 | global _db_override |
| 49 | if db_path is None: |
| 50 | yield |
| 51 | return |
| 52 | previous = _db_override |
| 53 | _db_override = Path(db_path) |
| 54 | try: |
| 55 | yield |
| 56 | finally: |
| 57 | _db_override = previous |
| 58 | |
| 59 | |
| 60 | def ensure_private_db_files(db_path: Optional[Path] = None) -> Path: |
| 61 | """Create/harden the research database and SQLite sidecars owner-only.""" |
| 62 | path = db_path or _get_db_path() |
| 63 | path.parent.mkdir(parents=True, exist_ok=True) |
| 64 | if not path.exists(): |
| 65 | try: |
| 66 | fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) |
| 67 | except FileExistsError: |
| 68 | pass |
| 69 | else: |
| 70 | os.close(fd) |
| 71 | for candidate in (path, Path(f"{path}-wal"), Path(f"{path}-shm")): |
| 72 | try: |
| 73 | candidate.chmod(0o600) |
| 74 | except FileNotFoundError: |
| 75 | pass |
| 76 | return path |
| 77 | |
| 78 | |
| 79 | SCHEMA_V1 = """ |
| 80 | PRAGMA journal_mode=WAL; |
| 81 | PRAGMA synchronous=NORMAL; |
| 82 | PRAGMA cache_size=-64000; |
| 83 | |
| 84 | CREATE TABLE IF NOT EXISTS schema_version ( |
| 85 | version INTEGER PRIMARY KEY, |
| 86 | applied_at TEXT DEFAULT (datetime('now')) |
| 87 | ); |
| 88 | |
| 89 | CREATE TABLE IF NOT EXISTS topics ( |
| 90 | id INTEGER PRIMARY KEY, |
| 91 | name TEXT UNIQUE NOT NULL, |
| 92 | search_queries TEXT, |
| 93 | schedule TEXT, |
| 94 | enabled INTEGER DEFAULT 1, |
| 95 | created_at TEXT DEFAULT (datetime('now')), |
| 96 | updated_at TEXT DEFAULT (datetime('now')) |
| 97 | ); |
| 98 | |
| 99 | CREATE TABLE IF NOT EXISTS research_runs ( |
| 100 | id INTEGER PRIMARY KEY, |
| 101 | topic_id INTEGER REFERENCES topics(id), |
| 102 | run_date TEXT NOT NULL, |
| 103 | source_mode TEXT, |
| 104 | prompt_tokens INTEGER, |
| 105 | completion_tokens INTEGER, |
| 106 | token_cost REAL, |
| 107 | duration_seconds REAL, |
| 108 | status TEXT DEFAULT 'completed', |
| 109 | error_message TEXT, |
| 110 | findings_new INTEGER DEFAULT 0, |
| 111 | findings_updated INTEGER DEFAULT 0, |
| 112 | created_at TEXT DEFAULT (datetime('now')) |
| 113 | ); |
| 114 | |
| 115 | CREATE TABLE IF NOT EXISTS findings ( |
| 116 | id INTEGER PRIMARY KEY, |
| 117 | run_id INTEGER REFERENCES research_runs(id), |
| 118 | topic_id INTEGER REFERENCES topics(id), |
| 119 | source TEXT NOT NULL, |
| 120 | source_url TEXT UNIQUE, |
| 121 | source_title TEXT, |
| 122 | author TEXT, |
| 123 | content TEXT, |
| 124 | summary TEXT, |
| 125 | engagement_score REAL, |
| 126 | relevance_score REAL, |
| 127 | first_seen TEXT DEFAULT (datetime('now')), |
| 128 | last_seen TEXT DEFAULT (datetime('now')), |
| 129 | sighting_count INTEGER DEFAULT 1, |
| 130 | dismissed INTEGER DEFAULT 0 |
| 131 | ); |
| 132 | |
| 133 | CREATE INDEX IF NOT EXISTS idx_findings_topic ON findings(topic_id, first_seen); |
| 134 | CREATE INDEX IF NOT EXISTS idx_findings_source ON findings(source, topic_id); |
| 135 | CREATE INDEX IF NOT EXISTS idx_findings_url ON findings(source_url); |
| 136 | |
| 137 | CREATE VIRTUAL TABLE IF NOT EXISTS findings_fts USING fts5( |
| 138 | content, summary, source_title, author, |
| 139 | tokenize='porter unicode61', |
| 140 | content='findings', |
| 141 | content_rowid='id' |
| 142 | ); |
| 143 | |
| 144 | CREATE TRIGGER IF NOT EXISTS findings_ai AFTER INSERT ON findings BEGIN |
| 145 | INSERT INTO findings_fts(rowid, content, summary, source_title, author) |
| 146 | VALUES (new.id, new.content, new.summary, new.source_title, new.author); |
| 147 | END; |
| 148 | |
| 149 | CREATE TRIGGER IF NOT EXISTS findings_ad AFTER DELETE ON findings BEGIN |
| 150 | INSERT INTO findings_fts(findings_fts, rowid, content, summary, source_title, author) |
| 151 | VALUES ('delete', old.id, old.content, old.summary, old.source_title, old.author); |
| 152 | END; |
| 153 | |
| 154 | CREATE TRIGGER IF NOT EXISTS findings_au AFTER UPDATE ON findings BEGIN |
| 155 | INSERT INTO findings_fts(findings_fts, rowid, content, summary, source_title, author) |
| 156 | VALUES ('delete', old.id, old.content, old.summary, old.source_title, old.author); |
| 157 | INSERT INTO findings_fts(rowid, content, summary, source_title, author) |
| 158 | VALUES (new.id, new.content, new.summary, new.source_title, new.author); |
| 159 | END; |
| 160 | |
| 161 | CREATE TABLE IF NOT EXISTS settings ( |
| 162 | key TEXT PRIMARY KEY, |
| 163 | value TEXT, |
| 164 | updated_at TEXT DEFAULT (datetime('now')) |
| 165 | ); |
| 166 | """ |
| 167 | |
| 168 | SCHEMA_V1_DEFAULTS = """ |
| 169 | INSERT OR IGNORE INTO schema_version (version) VALUES (1); |
| 170 | INSERT OR IGNORE INTO settings (key, value) VALUES ('daily_budget', '5.00'); |
| 171 | INSERT OR IGNORE INTO settings (key, value) VALUES ('delivery_channel', ''); |
| 172 | INSERT OR IGNORE INTO settings (key, value) VALUES ('delivery_mode', 'announce'); |
| 173 | INSERT OR IGNORE INTO settings (key, value) VALUES ('briefing_format', 'concise'); |
| 174 | INSERT OR IGNORE INTO settings (key, value) VALUES ('default_schedule', '0 8 * * *'); |
| 175 | """ |
| 176 | |
| 177 | _UPDATABLE_RUN_COLUMNS = frozenset({ |
| 178 | "source_mode", |
| 179 | "prompt_tokens", |
| 180 | "completion_tokens", |
| 181 | "token_cost", |
| 182 | "duration_seconds", |
| 183 | "status", |
| 184 | "error_message", |
| 185 | "findings_new", |
| 186 | "findings_updated", |
| 187 | }) |
| 188 | |
| 189 | _UPDATABLE_FINDING_COLUMNS = frozenset({ |
| 190 | "source", |
| 191 | "source_url", |
| 192 | "source_title", |
| 193 | "author", |
| 194 | "content", |
| 195 | "summary", |
| 196 | "engagement_score", |
| 197 | "relevance_score", |
| 198 | "last_seen", |
| 199 | "sighting_count", |
| 200 | "dismissed", |
| 201 | }) |
| 202 | |
| 203 | # Future migrations keyed by version number |
| 204 | MIGRATIONS: Dict[int, str] = { |
| 205 | 2: """ |
| 206 | CREATE TABLE IF NOT EXISTS finding_sightings ( |
| 207 | id INTEGER PRIMARY KEY, |
| 208 | finding_id INTEGER NOT NULL REFERENCES findings(id) ON DELETE CASCADE, |
| 209 | run_id INTEGER REFERENCES research_runs(id) ON DELETE CASCADE, |
| 210 | topic_id INTEGER REFERENCES topics(id) ON DELETE CASCADE, |
| 211 | source TEXT NOT NULL, |
| 212 | source_url TEXT NOT NULL, |
| 213 | source_title TEXT, |
| 214 | engagement_score REAL, |
| 215 | relevance_score REAL, |
| 216 | seen_at TEXT DEFAULT (datetime('now')), |
| 217 | UNIQUE(run_id, finding_id) |
| 218 | ); |
| 219 | |
| 220 | CREATE INDEX IF NOT EXISTS idx_finding_sightings_run |
| 221 | ON finding_sightings(run_id, topic_id); |
| 222 | CREATE INDEX IF NOT EXISTS idx_finding_sightings_topic_seen |
| 223 | ON finding_sightings(topic_id, seen_at); |
| 224 | CREATE INDEX IF NOT EXISTS idx_finding_sightings_url |
| 225 | ON finding_sightings(source_url); |
| 226 | """, |
| 227 | 3: """ |
| 228 | CREATE TABLE IF NOT EXISTS discovery_topics ( |
| 229 | id INTEGER PRIMARY KEY, |
| 230 | name TEXT NOT NULL, |
| 231 | normalized_name TEXT NOT NULL UNIQUE, |
| 232 | entity_key TEXT, |
| 233 | domain TEXT, |
| 234 | first_surfaced TEXT NOT NULL, |
| 235 | last_surfaced TEXT NOT NULL, |
| 236 | surface_count INTEGER NOT NULL DEFAULT 1, |
| 237 | status TEXT NOT NULL DEFAULT 'surfaced' CHECK(status IN ('surfaced','covered')), |
| 238 | covered_at TEXT, |
| 239 | last_run_ref TEXT |
| 240 | ); |
| 241 | |
| 242 | CREATE INDEX IF NOT EXISTS idx_discovery_topics_status_surfaced |
| 243 | ON discovery_topics(status, last_surfaced); |
| 244 | """, |
| 245 | } |
| 246 | |
| 247 | |
| 248 | def _connect(db_path: Optional[Path] = None) -> sqlite3.Connection: |
| 249 | """Open a connection with WAL mode and row factory.""" |
| 250 | path = db_path or _get_db_path() |
| 251 | conn = sqlite3.connect(str(path)) |
| 252 | conn.row_factory = sqlite3.Row |
| 253 | conn.execute("PRAGMA journal_mode=WAL") |
| 254 | conn.execute("PRAGMA synchronous=NORMAL") |
| 255 | conn.execute("PRAGMA foreign_keys=ON") |
| 256 | # WAL lets readers coexist with one writer, but two writers (cron + user) |
| 257 | # still contend for the write lock. Default busy_timeout is 0, so the loser |
| 258 | # raises "database is locked" instantly; wait instead. |
| 259 | conn.execute("PRAGMA busy_timeout=5000") |
| 260 | return conn |
| 261 | |
| 262 | |
| 263 | def init_db(db_path: Optional[Path] = None) -> Path: |
| 264 | """Create database and tables if they don't exist. Returns the DB path.""" |
| 265 | path = db_path or _get_db_path() |
| 266 | path.parent.mkdir(parents=True, exist_ok=True) |
| 267 | |
| 268 | conn = _connect(path) |
| 269 | try: |
| 270 | conn.executescript(SCHEMA_V1) |
| 271 | conn.executescript(SCHEMA_V1_DEFAULTS) |
| 272 | _run_migrations(conn) |
| 273 | conn.commit() |
| 274 | finally: |
| 275 | conn.close() |
| 276 | |
| 277 | return path |
| 278 | |
| 279 | |
| 280 | def _run_migrations(conn: sqlite3.Connection): |
| 281 | """Apply pending schema migrations.""" |
| 282 | current = conn.execute( |
| 283 | "SELECT MAX(version) FROM schema_version" |
| 284 | ).fetchone()[0] or 0 |
| 285 | |
| 286 | for version in sorted(MIGRATIONS.keys()): |
| 287 | if version > current: |
| 288 | conn.executescript(MIGRATIONS[version]) |
| 289 | conn.execute( |
| 290 | "INSERT INTO schema_version (version) VALUES (?)", (version,) |
| 291 | ) |
| 292 | |
| 293 | |
| 294 | # --- Topics --- |
| 295 | |
| 296 | |
| 297 | def add_topic( |
| 298 | name: str, |
| 299 | search_queries: Optional[List[str]] = None, |
| 300 | schedule: str = "0 8 * * *", |
| 301 | ) -> Dict[str, Any]: |
| 302 | """Add a topic to the watchlist. Returns the topic dict.""" |
| 303 | init_db() |
| 304 | conn = _connect() |
| 305 | try: |
| 306 | queries_json = json.dumps(search_queries) if search_queries else None |
| 307 | conn.execute( |
| 308 | """INSERT INTO topics (name, search_queries, schedule) |
| 309 | VALUES (?, ?, ?) |
| 310 | ON CONFLICT(name) DO UPDATE SET |
| 311 | search_queries = excluded.search_queries, |
| 312 | schedule = excluded.schedule, |
| 313 | updated_at = datetime('now')""", |
| 314 | (name, queries_json, schedule), |
| 315 | ) |
| 316 | conn.commit() |
| 317 | row = conn.execute( |
| 318 | "SELECT * FROM topics WHERE name = ?", (name,) |
| 319 | ).fetchone() |
| 320 | return dict(row) |
| 321 | finally: |
| 322 | conn.close() |
| 323 | |
| 324 | |
| 325 | def remove_topic(name: str) -> bool: |
| 326 | """Remove a topic from the watchlist. Returns True if found.""" |
| 327 | init_db() |
| 328 | conn = _connect() |
| 329 | try: |
| 330 | row = conn.execute( |
| 331 | "SELECT id FROM topics WHERE name = ?", (name,) |
| 332 | ).fetchone() |
| 333 | if not row: |
| 334 | return False |
| 335 | topic_id = row["id"] |
| 336 | # Delete findings and runs for this topic |
| 337 | conn.execute("DELETE FROM findings WHERE topic_id = ?", (topic_id,)) |
| 338 | conn.execute("DELETE FROM research_runs WHERE topic_id = ?", (topic_id,)) |
| 339 | conn.execute("DELETE FROM topics WHERE id = ?", (topic_id,)) |
| 340 | conn.commit() |
| 341 | return True |
| 342 | finally: |
| 343 | conn.close() |
| 344 | |
| 345 | |
| 346 | def list_topics() -> List[Dict[str, Any]]: |
| 347 | """List all topics with stats.""" |
| 348 | init_db() |
| 349 | conn = _connect() |
| 350 | try: |
| 351 | rows = conn.execute( |
| 352 | """SELECT t.*, |
| 353 | (SELECT COUNT(*) FROM findings WHERE topic_id = t.id) as finding_count, |
| 354 | (SELECT MAX(run_date) FROM research_runs WHERE topic_id = t.id) as last_run, |
| 355 | (SELECT status FROM research_runs WHERE topic_id = t.id |
| 356 | ORDER BY created_at DESC LIMIT 1) as last_status |
| 357 | FROM topics t |
| 358 | ORDER BY t.name""" |
| 359 | ).fetchall() |
| 360 | return [dict(r) for r in rows] |
| 361 | finally: |
| 362 | conn.close() |
| 363 | |
| 364 | |
| 365 | def get_topic(name: str) -> Optional[Dict[str, Any]]: |
| 366 | """Get a topic by name.""" |
| 367 | init_db() |
| 368 | conn = _connect() |
| 369 | try: |
| 370 | row = conn.execute( |
| 371 | "SELECT * FROM topics WHERE name = ?", (name,) |
| 372 | ).fetchone() |
| 373 | return dict(row) if row else None |
| 374 | finally: |
| 375 | conn.close() |
| 376 | |
| 377 | |
| 378 | # --- Research Runs --- |
| 379 | |
| 380 | |
| 381 | def record_run( |
| 382 | topic_id: int, |
| 383 | source_mode: str = "both", |
| 384 | status: str = "completed", |
| 385 | error_message: Optional[str] = None, |
| 386 | duration_seconds: float = 0, |
| 387 | prompt_tokens: int = 0, |
| 388 | completion_tokens: int = 0, |
| 389 | token_cost: float = 0, |
| 390 | ) -> int: |
| 391 | """Record a research run. Returns the run ID.""" |
| 392 | conn = _connect() |
| 393 | try: |
| 394 | cursor = conn.execute( |
| 395 | """INSERT INTO research_runs |
| 396 | (topic_id, run_date, source_mode, status, error_message, |
| 397 | duration_seconds, prompt_tokens, completion_tokens, token_cost) |
| 398 | VALUES (?, datetime('now'), ?, ?, ?, ?, ?, ?, ?)""", |
| 399 | ( |
| 400 | topic_id, source_mode, status, error_message, |
| 401 | duration_seconds, prompt_tokens, completion_tokens, token_cost, |
| 402 | ), |
| 403 | ) |
| 404 | conn.commit() |
| 405 | return cursor.lastrowid |
| 406 | finally: |
| 407 | conn.close() |
| 408 | |
| 409 | |
| 410 | def update_run(run_id: int, **kwargs): |
| 411 | """Update a research run's fields.""" |
| 412 | conn = _connect() |
| 413 | try: |
| 414 | invalid_columns = sorted(set(kwargs) - _UPDATABLE_RUN_COLUMNS) |
| 415 | if invalid_columns: |
| 416 | raise ValueError( |
| 417 | f"Invalid run update fields: {', '.join(invalid_columns)}" |
| 418 | ) |
| 419 | sets = ", ".join(f"{k} = ?" for k in kwargs) |
| 420 | values = list(kwargs.values()) + [run_id] |
| 421 | conn.execute(f"UPDATE research_runs SET {sets} WHERE id = ?", values) |
| 422 | conn.commit() |
| 423 | finally: |
| 424 | conn.close() |
| 425 | |
| 426 | |
| 427 | def get_latest_completed_runs(topic_id: int, limit: int = 2) -> List[Dict[str, Any]]: |
| 428 | """Return newest completed runs for a topic.""" |
| 429 | conn = _connect() |
| 430 | try: |
| 431 | rows = conn.execute( |
| 432 | """SELECT * FROM research_runs |
| 433 | WHERE topic_id = ? AND status = 'completed' |
| 434 | ORDER BY datetime(run_date) DESC, id DESC |
| 435 | LIMIT ?""", |
| 436 | (topic_id, limit), |
| 437 | ).fetchall() |
| 438 | return [dict(r) for r in rows] |
| 439 | finally: |
| 440 | conn.close() |
| 441 | |
| 442 | |
| 443 | # --- Findings --- |
| 444 | |
| 445 | |
| 446 | def store_findings( |
| 447 | run_id: int, |
| 448 | topic_id: int, |
| 449 | findings: List[Dict[str, Any]], |
| 450 | ) -> Dict[str, int]: |
| 451 | """Store findings with URL-based dedup. Returns counts of new/updated.""" |
| 452 | # Collect findings that have a URL, preserving order. |
| 453 | with_urls: List[tuple[str, Dict[str, Any]]] = [] |
| 454 | for f in findings: |
| 455 | url = f.get("source_url") or f.get("url") |
| 456 | if url: |
| 457 | with_urls.append((url, f)) |
| 458 | |
| 459 | if not with_urls: |
| 460 | conn = _connect() |
| 461 | try: |
| 462 | conn.execute( |
| 463 | "UPDATE research_runs SET findings_new = 0, findings_updated = 0 WHERE id = ?", |
| 464 | (run_id,), |
| 465 | ) |
| 466 | conn.commit() |
| 467 | finally: |
| 468 | conn.close() |
| 469 | return {"new": 0, "updated": 0} |
| 470 | |
| 471 | conn = _connect() |
| 472 | try: |
| 473 | # Single batch SELECT to find existing findings by URL. |
| 474 | urls = [url for url, _ in with_urls] |
| 475 | placeholders = ",".join("?" for _ in urls) |
| 476 | rows = conn.execute( |
| 477 | f"SELECT id, source_url, engagement_score FROM findings WHERE source_url IN ({placeholders})", |
| 478 | urls, |
| 479 | ).fetchall() |
| 480 | existing_by_url = {row["source_url"]: row for row in rows} |
| 481 | |
| 482 | update_rows: List[tuple] = [] |
| 483 | insert_rows: List[tuple] = [] |
| 484 | |
| 485 | for url, f in with_urls: |
| 486 | existing = existing_by_url.get(url) |
| 487 | new_engagement = f.get("engagement_score") or 0 |
| 488 | if existing: |
| 489 | update_rows.append(( |
| 490 | max(new_engagement, existing["engagement_score"] or 0), |
| 491 | run_id, |
| 492 | existing["id"], |
| 493 | )) |
| 494 | else: |
| 495 | insert_rows.append(( |
| 496 | run_id, |
| 497 | topic_id, |
| 498 | f.get("source", "unknown"), |
| 499 | url, |
| 500 | f.get("source_title") or f.get("title", ""), |
| 501 | f.get("author", ""), |
| 502 | f.get("content") or f.get("text", ""), |
| 503 | f.get("summary", ""), |
| 504 | new_engagement, |
| 505 | f.get("relevance_score", 0), |
| 506 | )) |
| 507 | |
| 508 | if update_rows: |
| 509 | conn.executemany( |
| 510 | """UPDATE findings SET |
| 511 | last_seen = datetime('now'), |
| 512 | sighting_count = sighting_count + 1, |
| 513 | engagement_score = ?, |
| 514 | run_id = ? |
| 515 | WHERE id = ?""", |
| 516 | update_rows, |
| 517 | ) |
| 518 | if insert_rows: |
| 519 | # source_url is UNIQUE. The SELECT above is not atomic with this |
| 520 | # write, so a concurrent run (cron + user) can insert the same URL |
| 521 | # between our read and write. Upsert on conflict instead of letting |
| 522 | # IntegrityError abort the whole batch and lose every finding. |
| 523 | conn.executemany( |
| 524 | """INSERT INTO findings |
| 525 | (run_id, topic_id, source, source_url, source_title, |
| 526 | author, content, summary, engagement_score, relevance_score) |
| 527 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 528 | ON CONFLICT(source_url) DO UPDATE SET |
| 529 | last_seen = datetime('now'), |
| 530 | sighting_count = sighting_count + 1, |
| 531 | engagement_score = max( |
| 532 | engagement_score, excluded.engagement_score), |
| 533 | run_id = excluded.run_id""", |
| 534 | insert_rows, |
| 535 | ) |
| 536 | |
| 537 | new_count = len(insert_rows) |
| 538 | updated_count = len(update_rows) |
| 539 | if insert_rows: |
| 540 | # A row whose URL was inserted by a concurrent run between our SELECT |
| 541 | # and the upsert resolves via ON CONFLICT (an update, not a new row), |
| 542 | # bumping its sighting_count above 1. Re-derive the split so |
| 543 | # research_runs.findings_new isn't inflated by conflict-resolved rows |
| 544 | # (source_url is field index 3 in each insert tuple). |
| 545 | inserted_urls = [row[3] for row in insert_rows] |
| 546 | placeholders = ",".join("?" for _ in inserted_urls) |
| 547 | conflicted = conn.execute( |
| 548 | f"SELECT COUNT(*) FROM findings " |
| 549 | f"WHERE source_url IN ({placeholders}) AND sighting_count > 1", |
| 550 | inserted_urls, |
| 551 | ).fetchone()[0] |
| 552 | new_count -= conflicted |
| 553 | updated_count += conflicted |
| 554 | _record_sightings(conn, run_id, topic_id, with_urls, existing_by_url) |
| 555 | conn.execute( |
| 556 | "UPDATE research_runs SET findings_new = ?, findings_updated = ? WHERE id = ?", |
| 557 | (new_count, updated_count, run_id), |
| 558 | ) |
| 559 | conn.commit() |
| 560 | finally: |
| 561 | conn.close() |
| 562 | |
| 563 | return {"new": new_count, "updated": updated_count} |
| 564 | |
| 565 | |
| 566 | def _record_sightings( |
| 567 | conn: sqlite3.Connection, |
| 568 | run_id: int, |
| 569 | topic_id: int, |
| 570 | findings_with_urls: List[tuple[str, Dict[str, Any]]], |
| 571 | existing_by_url: Optional[Dict[str, sqlite3.Row]] = None, |
| 572 | ) -> None: |
| 573 | """Record the findings observed during this run. |
| 574 | |
| 575 | The aggregate findings table keeps one row per URL and updates that row on |
| 576 | re-sighting. This ledger preserves the run/topic membership needed for |
| 577 | watchlist deltas and dossiers. |
| 578 | """ |
| 579 | if not findings_with_urls: |
| 580 | return |
| 581 | |
| 582 | by_url = {url: finding for url, finding in findings_with_urls} |
| 583 | rows_by_url = dict(existing_by_url or {}) |
| 584 | |
| 585 | missing_urls = [url for url in by_url if url not in rows_by_url] |
| 586 | if missing_urls: |
| 587 | placeholders = ",".join("?" for _ in missing_urls) |
| 588 | rows = conn.execute( |
| 589 | f"SELECT id, source_url FROM findings WHERE source_url IN ({placeholders})", |
| 590 | missing_urls, |
| 591 | ).fetchall() |
| 592 | rows_by_url.update({row["source_url"]: row for row in rows}) |
| 593 | |
| 594 | sighting_rows = [] |
| 595 | for url, finding in by_url.items(): |
| 596 | row = rows_by_url.get(url) |
| 597 | if row is None: |
| 598 | continue |
| 599 | sighting_rows.append(( |
| 600 | row["id"], |
| 601 | run_id, |
| 602 | topic_id, |
| 603 | finding.get("source", "unknown"), |
| 604 | url, |
| 605 | finding.get("source_title") or finding.get("title", ""), |
| 606 | finding.get("engagement_score") if finding.get("engagement_score") is not None else 0, |
| 607 | finding.get("relevance_score") if finding.get("relevance_score") is not None else 0, |
| 608 | )) |
| 609 | |
| 610 | if not sighting_rows: |
| 611 | return |
| 612 | |
| 613 | conn.executemany( |
| 614 | """INSERT INTO finding_sightings |
| 615 | (finding_id, run_id, topic_id, source, source_url, source_title, |
| 616 | engagement_score, relevance_score) |
| 617 | VALUES (?, ?, ?, ?, ?, ?, ?, ?) |
| 618 | ON CONFLICT(run_id, finding_id) DO UPDATE SET |
| 619 | topic_id = excluded.topic_id, |
| 620 | source = excluded.source, |
| 621 | source_url = excluded.source_url, |
| 622 | source_title = excluded.source_title, |
| 623 | engagement_score = excluded.engagement_score, |
| 624 | relevance_score = excluded.relevance_score""", |
| 625 | sighting_rows, |
| 626 | ) |
| 627 | |
| 628 | |
| 629 | def get_sightings_for_run(topic_id: int, run_id: int) -> List[Dict[str, Any]]: |
| 630 | """Return findings observed for a topic during a specific run.""" |
| 631 | conn = _connect() |
| 632 | try: |
| 633 | rows = conn.execute( |
| 634 | """SELECT * FROM finding_sightings |
| 635 | WHERE topic_id = ? AND run_id = ? |
| 636 | ORDER BY id""", |
| 637 | (topic_id, run_id), |
| 638 | ).fetchall() |
| 639 | return [dict(r) for r in rows] |
| 640 | finally: |
| 641 | conn.close() |
| 642 | |
| 643 | |
| 644 | def compute_topic_delta(topic_id: int) -> Dict[str, Any]: |
| 645 | """Compare the latest completed watchlist run with the previous run.""" |
| 646 | runs = get_latest_completed_runs(topic_id, limit=2) |
| 647 | topic = _get_topic_by_id(topic_id) |
| 648 | topic_name = topic["name"] if topic else str(topic_id) |
| 649 | if len(runs) < 2: |
| 650 | return { |
| 651 | "topic": topic_name, |
| 652 | "status": "insufficient_history", |
| 653 | "message": "Need at least two completed runs to compute a delta.", |
| 654 | } |
| 655 | |
| 656 | current_run, previous_run = runs[0], runs[1] |
| 657 | current = _sightings_by_url(get_sightings_for_run(topic_id, current_run["id"])) |
| 658 | previous = _sightings_by_url(get_sightings_for_run(topic_id, previous_run["id"])) |
| 659 | |
| 660 | current_urls = set(current) |
| 661 | previous_urls = set(previous) |
| 662 | new_urls = sorted(current_urls - previous_urls) |
| 663 | continued_urls = sorted(current_urls & previous_urls) |
| 664 | dropped_urls = sorted(previous_urls - current_urls) |
| 665 | |
| 666 | findings = { |
| 667 | "new": [current[url] for url in new_urls], |
| 668 | "continued": [current[url] for url in continued_urls], |
| 669 | "dropped": [previous[url] for url in dropped_urls], |
| 670 | } |
| 671 | |
| 672 | return { |
| 673 | "topic": topic_name, |
| 674 | "status": "ok", |
| 675 | "current_run_id": current_run["id"], |
| 676 | "previous_run_id": previous_run["id"], |
| 677 | "new": len(new_urls), |
| 678 | "continued": len(continued_urls), |
| 679 | "dropped": len(dropped_urls), |
| 680 | "sources": _delta_source_counts(findings), |
| 681 | "findings": findings, |
| 682 | } |
| 683 | |
| 684 | |
| 685 | def _get_topic_by_id(topic_id: int) -> Optional[Dict[str, Any]]: |
| 686 | conn = _connect() |
| 687 | try: |
| 688 | row = conn.execute("SELECT * FROM topics WHERE id = ?", (topic_id,)).fetchone() |
| 689 | return dict(row) if row else None |
| 690 | finally: |
| 691 | conn.close() |
| 692 | |
| 693 | |
| 694 | def _sightings_by_url(sightings: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: |
| 695 | """Index sightings by stable URL identity for run-to-run delta comparisons. |
| 696 | |
| 697 | URL-less sightings are intentionally excluded because there is no stable |
| 698 | cross-run identity to classify them as new, continued, or dropped. |
| 699 | """ |
| 700 | return { |
| 701 | sighting["source_url"]: sighting |
| 702 | for sighting in sightings |
| 703 | if sighting.get("source_url") |
| 704 | } |
| 705 | |
| 706 | |
| 707 | def _delta_source_counts( |
| 708 | findings: Dict[str, List[Dict[str, Any]]] |
| 709 | ) -> Dict[str, Dict[str, int]]: |
| 710 | sources = sorted({ |
| 711 | finding.get("source") or "unknown" |
| 712 | for group in findings.values() |
| 713 | for finding in group |
| 714 | }) |
| 715 | counts = { |
| 716 | source: {"new": 0, "continued": 0, "dropped": 0} |
| 717 | for source in sources |
| 718 | } |
| 719 | for group_name, group in findings.items(): |
| 720 | for finding in group: |
| 721 | source = finding.get("source") or "unknown" |
| 722 | counts[source][group_name] += 1 |
| 723 | return counts |
| 724 | |
| 725 | |
| 726 | def get_new_findings( |
| 727 | topic_id: int, |
| 728 | since: Optional[str] = None, |
| 729 | ) -> List[Dict[str, Any]]: |
| 730 | """Get findings for a topic, optionally since a date.""" |
| 731 | conn = _connect() |
| 732 | try: |
| 733 | if since: |
| 734 | rows = conn.execute( |
| 735 | """SELECT * FROM findings |
| 736 | WHERE topic_id = ? AND first_seen >= ? AND dismissed = 0 |
| 737 | ORDER BY first_seen DESC""", |
| 738 | (topic_id, since), |
| 739 | ).fetchall() |
| 740 | else: |
| 741 | rows = conn.execute( |
| 742 | """SELECT * FROM findings |
| 743 | WHERE topic_id = ? AND dismissed = 0 |
| 744 | ORDER BY first_seen DESC""", |
| 745 | (topic_id,), |
| 746 | ).fetchall() |
| 747 | return [dict(r) for r in rows] |
| 748 | finally: |
| 749 | conn.close() |
| 750 | |
| 751 | |
| 752 | def search_findings(query: str, limit: int = 20) -> List[Dict[str, Any]]: |
| 753 | """FTS5 search across all findings with BM25 ranking.""" |
| 754 | conn = _connect() |
| 755 | try: |
| 756 | rows = conn.execute( |
| 757 | """SELECT f.*, bm25(findings_fts) as rank, t.name as topic_name |
| 758 | FROM findings_fts |
| 759 | JOIN findings f ON f.id = findings_fts.rowid |
| 760 | LEFT JOIN topics t ON t.id = f.topic_id |
| 761 | WHERE findings_fts MATCH ? |
| 762 | ORDER BY rank |
| 763 | LIMIT ?""", |
| 764 | (query, limit), |
| 765 | ).fetchall() |
| 766 | return [dict(r) for r in rows] |
| 767 | finally: |
| 768 | conn.close() |
| 769 | |
| 770 | |
| 771 | def update_finding(finding_id: int, **kwargs): |
| 772 | """Update a finding's fields.""" |
| 773 | conn = _connect() |
| 774 | try: |
| 775 | invalid_columns = sorted(set(kwargs) - _UPDATABLE_FINDING_COLUMNS) |
| 776 | if invalid_columns: |
| 777 | raise ValueError( |
| 778 | f"Invalid finding update fields: {', '.join(invalid_columns)}" |
| 779 | ) |
| 780 | sets = ", ".join(f"{k} = ?" for k in kwargs) |
| 781 | values = list(kwargs.values()) + [finding_id] |
| 782 | conn.execute(f"UPDATE findings SET {sets} WHERE id = ?", values) |
| 783 | conn.commit() |
| 784 | finally: |
| 785 | conn.close() |
| 786 | |
| 787 | |
| 788 | def delete_finding(finding_id: int): |
| 789 | """Delete a finding.""" |
| 790 | conn = _connect() |
| 791 | try: |
| 792 | conn.execute("DELETE FROM findings WHERE id = ?", (finding_id,)) |
| 793 | conn.commit() |
| 794 | finally: |
| 795 | conn.close() |
| 796 | |
| 797 | |
| 798 | def dismiss_finding(finding_id: int): |
| 799 | """Mark a finding as dismissed.""" |
| 800 | update_finding(finding_id, dismissed=1) |
| 801 | |
| 802 | |
| 803 | # --- Discovery topic queue --- |
| 804 | |
| 805 | # Conservative floor for fuzzy queue matching (overlap coefficient of entity |
| 806 | # tokens via entity_extract.entity_overlap). Tunable: raise toward 1.0 for |
| 807 | # stricter matching, lower for looser. Matching is annotate-only - a fuzzy |
| 808 | # match stamps prior-surfacing context onto an incoming topic but NEVER merges |
| 809 | # queue rows, so a too-loose threshold can mislabel a card yet never lose data. |
| 810 | DISCOVERY_QUEUE_OVERLAP_THRESHOLD = 0.6 |
| 811 | |
| 812 | |
| 813 | def _normalize_discovery_name(name: str) -> str: |
| 814 | """Queue identity: lowercased, punctuation-stripped, whitespace-collapsed |
| 815 | (thin alias for dedupe.normalize_text).""" |
| 816 | return dedupe.normalize_text(name) |
| 817 | |
| 818 | |
| 819 | def _discovery_entity_key(name: str) -> str: |
| 820 | """Sorted joined significant tokens, computed once at write time.""" |
| 821 | return " ".join(sorted(entity_extract.extract_text_entities(name))) |
| 822 | |
| 823 | |
| 824 | def _discovery_anchor_entities(name: str) -> set[str]: |
| 825 | """Anchor tokens for fuzzy matching: capitalized, all-caps, or |
| 826 | digit-bearing words minus stopwords (product/person/version anchors). |
| 827 | |
| 828 | Generic lowercase words ("chat", "templates") are excluded so two angles |
| 829 | on the same subject ("Gemma 4 chat templates" / "Gemma 4 tool calling |
| 830 | fixes") cross-match while different subjects sharing filler words don't. |
| 831 | """ |
| 832 | anchors = set() |
| 833 | for word in re.sub(r"[^\w\s]", " ", name).split(): |
| 834 | lower = word.casefold() |
| 835 | if lower in entity_extract.ENTITY_STOPWORDS: |
| 836 | continue |
| 837 | if entity_extract.has_anchor_signal(word): |
| 838 | anchors.add(lower) |
| 839 | return anchors |
| 840 | |
| 841 | |
| 842 | def record_discovery_surfacing( |
| 843 | name: str, |
| 844 | domain: str = "", |
| 845 | run_ref: str = "", |
| 846 | as_of: str = "", |
| 847 | inherit_covered_at: Optional[str] = None, |
| 848 | ) -> Dict[str, Any]: |
| 849 | """Upsert a queue row by normalized name. |
| 850 | |
| 851 | A fresh topic inserts with surface_count 1; re-surfacing the same |
| 852 | normalized name increments the count and refreshes last_surfaced and |
| 853 | last_run_ref (first_surfaced never changes). Returns the resulting row. |
| 854 | |
| 855 | A resurfacing with a blank domain (e.g. a global-trending sweep with no |
| 856 | domain) never blanks a domain recorded by an earlier, domain-scoped |
| 857 | surfacing - the stored domain only changes when the incoming domain is |
| 858 | non-empty. ``domain`` is normalized to "" here (never NULL bound) so the |
| 859 | column's storage convention stays consistent regardless of whether a |
| 860 | caller passes "" or None. |
| 861 | |
| 862 | ``inherit_covered_at`` makes a FRESH row be born covered (status |
| 863 | 'covered', covered_at set to the given date). Callers pass it when this |
| 864 | name fuzzy-matched an already-covered prior row, so a user's covered |
| 865 | mark survives judge naming drift instead of forking into a fresh |
| 866 | uncovered row. An existing row's status/covered_at are never modified |
| 867 | by this function - the ON CONFLICT path deliberately ignores it. |
| 868 | |
| 869 | Idempotency guard: when the existing row's last_run_ref already equals |
| 870 | this call's (non-blank) run_ref, the surfacing was ALREADY counted by |
| 871 | this run identity - a retry (e.g. a --finalize re-run with a corrected |
| 872 | angles file) returns the row unchanged instead of double-counting. |
| 873 | Blank run_refs never guard, so callers without a run identity keep the |
| 874 | every-call-increments behavior. |
| 875 | """ |
| 876 | init_db() |
| 877 | domain = domain or "" |
| 878 | normalized = _normalize_discovery_name(name) |
| 879 | entity_key = _discovery_entity_key(name) |
| 880 | status = "covered" if inherit_covered_at else "surfaced" |
| 881 | conn = _connect() |
| 882 | try: |
| 883 | if run_ref: |
| 884 | existing = conn.execute( |
| 885 | "SELECT * FROM discovery_topics WHERE normalized_name = ?", |
| 886 | (normalized,), |
| 887 | ).fetchone() |
| 888 | if existing is not None and existing["last_run_ref"] == run_ref: |
| 889 | return dict(existing) |
| 890 | conn.execute( |
| 891 | """INSERT INTO discovery_topics |
| 892 | (name, normalized_name, entity_key, domain, first_surfaced, |
| 893 | last_surfaced, surface_count, last_run_ref, status, covered_at) |
| 894 | VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?, ?) |
| 895 | ON CONFLICT(normalized_name) DO UPDATE SET |
| 896 | surface_count = surface_count + 1, |
| 897 | last_surfaced = excluded.last_surfaced, |
| 898 | last_run_ref = excluded.last_run_ref, |
| 899 | domain = CASE WHEN excluded.domain <> '' THEN excluded.domain ELSE domain END""", |
| 900 | (name, normalized, entity_key, domain, as_of, as_of, run_ref, status, inherit_covered_at), |
| 901 | ) |
| 902 | conn.commit() |
| 903 | row = conn.execute( |
| 904 | "SELECT * FROM discovery_topics WHERE normalized_name = ?", |
| 905 | (normalized,), |
| 906 | ).fetchone() |
| 907 | return dict(row) |
| 908 | finally: |
| 909 | conn.close() |
| 910 | |
| 911 | |
| 912 | def match_discovery_topic(name: str) -> Optional[Dict[str, Any]]: |
| 913 | """Find the queue row a topic name refers to, or None. |
| 914 | |
| 915 | Exact normalized-name match wins; otherwise the best entity-overlap match |
| 916 | at or above DISCOVERY_QUEUE_OVERLAP_THRESHOLD. Overlap is the better of |
| 917 | the full entity_key token overlap and the anchor-token overlap (see |
| 918 | _discovery_anchor_entities) - full-token overlap alone dilutes the subject |
| 919 | anchor with generic words, so same-subject near-duplicates would never |
| 920 | clear a conservative floor. Matching NEVER merges rows: a fuzzy match only |
| 921 | annotates the incoming topic with the prior row's context. |
| 922 | """ |
| 923 | init_db() |
| 924 | normalized = _normalize_discovery_name(name) |
| 925 | conn = _connect() |
| 926 | try: |
| 927 | row = conn.execute( |
| 928 | "SELECT * FROM discovery_topics WHERE normalized_name = ?", |
| 929 | (normalized,), |
| 930 | ).fetchone() |
| 931 | if row: |
| 932 | return dict(row) |
| 933 | |
| 934 | entities = entity_extract.extract_text_entities(name) |
| 935 | anchors = _discovery_anchor_entities(name) |
| 936 | if not entities and not anchors: |
| 937 | return None |
| 938 | best: Optional[sqlite3.Row] = None |
| 939 | best_overlap = 0.0 |
| 940 | for candidate in conn.execute("SELECT * FROM discovery_topics").fetchall(): |
| 941 | candidate_entities = set((candidate["entity_key"] or "").split()) |
| 942 | overlap = max( |
| 943 | entity_extract.entity_overlap(entities, candidate_entities), |
| 944 | entity_extract.entity_overlap( |
| 945 | anchors, _discovery_anchor_entities(candidate["name"]) |
| 946 | ), |
| 947 | ) |
| 948 | if overlap > best_overlap: |
| 949 | best, best_overlap = candidate, overlap |
| 950 | if best is not None and best_overlap >= DISCOVERY_QUEUE_OVERLAP_THRESHOLD: |
| 951 | return dict(best) |
| 952 | return None |
| 953 | finally: |
| 954 | conn.close() |
| 955 | |
| 956 | |
| 957 | def list_discovery_queue(status: Optional[str] = None) -> List[Dict[str, Any]]: |
| 958 | """List queue rows, newest surfacing first, optionally filtered by status.""" |
| 959 | init_db() |
| 960 | conn = _connect() |
| 961 | try: |
| 962 | if status: |
| 963 | rows = conn.execute( |
| 964 | """SELECT * FROM discovery_topics WHERE status = ? |
| 965 | ORDER BY last_surfaced DESC, id DESC""", |
| 966 | (status,), |
| 967 | ).fetchall() |
| 968 | else: |
| 969 | rows = conn.execute( |
| 970 | "SELECT * FROM discovery_topics ORDER BY last_surfaced DESC, id DESC" |
| 971 | ).fetchall() |
| 972 | return [dict(r) for r in rows] |
| 973 | finally: |
| 974 | conn.close() |
| 975 | |
| 976 | |
| 977 | def mark_discovery_covered(name: str, as_of: str) -> Optional[Dict[str, Any]]: |
| 978 | """Mark a queued topic covered by EXACT normalized name. |
| 979 | |
| 980 | Returns the updated row, or None when no row matches - callers must error |
| 981 | loudly on None, never silently no-op. Fuzzy matching is deliberately not |
| 982 | offered here: covering mutates state, so it demands the exact name. |
| 983 | """ |
| 984 | init_db() |
| 985 | normalized = _normalize_discovery_name(name) |
| 986 | conn = _connect() |
| 987 | try: |
| 988 | cursor = conn.execute( |
| 989 | """UPDATE discovery_topics |
| 990 | SET status = 'covered', covered_at = ? |
| 991 | WHERE normalized_name = ?""", |
| 992 | (as_of, normalized), |
| 993 | ) |
| 994 | conn.commit() |
| 995 | if cursor.rowcount == 0: |
| 996 | return None |
| 997 | row = conn.execute( |
| 998 | "SELECT * FROM discovery_topics WHERE normalized_name = ?", |
| 999 | (normalized,), |
| 1000 | ).fetchone() |
| 1001 | return dict(row) |
| 1002 | finally: |
| 1003 | conn.close() |
| 1004 | |
| 1005 | |
| 1006 | # --- Cost Tracking --- |
| 1007 | |
| 1008 | |
| 1009 | def get_daily_cost(date: Optional[str] = None) -> float: |
| 1010 | """Get total token cost for a given day (default: today).""" |
| 1011 | conn = _connect() |
| 1012 | try: |
| 1013 | if not date: |
| 1014 | date = datetime.now(timezone.utc).strftime("%Y-%m-%d") |
| 1015 | row = conn.execute( |
| 1016 | """SELECT COALESCE(SUM(token_cost), 0) as total |
| 1017 | FROM research_runs |
| 1018 | WHERE date(run_date) = date(?)""", |
| 1019 | (date,), |
| 1020 | ).fetchone() |
| 1021 | return row["total"] |
| 1022 | finally: |
| 1023 | conn.close() |
| 1024 | |
| 1025 | |
| 1026 | # --- Settings --- |
| 1027 | |
| 1028 | |
| 1029 | def get_setting(key: str, default: Optional[str] = None) -> Optional[str]: |
| 1030 | """Get a setting value.""" |
| 1031 | init_db() |
| 1032 | conn = _connect() |
| 1033 | try: |
| 1034 | row = conn.execute( |
| 1035 | "SELECT value FROM settings WHERE key = ?", (key,) |
| 1036 | ).fetchone() |
| 1037 | return row["value"] if row else default |
| 1038 | finally: |
| 1039 | conn.close() |
| 1040 | |
| 1041 | |
| 1042 | def set_setting(key: str, value: str): |
| 1043 | """Set a setting value.""" |
| 1044 | init_db() |
| 1045 | conn = _connect() |
| 1046 | try: |
| 1047 | conn.execute( |
| 1048 | """INSERT INTO settings (key, value, updated_at) |
| 1049 | VALUES (?, ?, datetime('now')) |
| 1050 | ON CONFLICT(key) DO UPDATE SET |
| 1051 | value = excluded.value, |
| 1052 | updated_at = datetime('now')""", |
| 1053 | (key, value), |
| 1054 | ) |
| 1055 | conn.commit() |
| 1056 | finally: |
| 1057 | conn.close() |
| 1058 | |
| 1059 | |
| 1060 | # --- Stats --- |
| 1061 | |
| 1062 | |
| 1063 | def get_stats() -> Dict[str, Any]: |
| 1064 | """Get overall database stats.""" |
| 1065 | conn = _connect() |
| 1066 | try: |
| 1067 | topic_count = conn.execute("SELECT COUNT(*) FROM topics WHERE enabled = 1").fetchone()[0] |
| 1068 | finding_count = conn.execute("SELECT COUNT(*) FROM findings").fetchone()[0] |
| 1069 | |
| 1070 | week_ago = (datetime.now(timezone.utc) - timedelta(days=7)).strftime("%Y-%m-%d") |
| 1071 | runs_7d = conn.execute( |
| 1072 | "SELECT COUNT(*) FROM research_runs WHERE run_date >= ?", (week_ago,) |
| 1073 | ).fetchone()[0] |
| 1074 | successful_7d = conn.execute( |
| 1075 | "SELECT COUNT(*) FROM research_runs WHERE run_date >= ? AND status = 'completed'", |
| 1076 | (week_ago,), |
| 1077 | ).fetchone()[0] |
| 1078 | failed_7d = conn.execute( |
| 1079 | "SELECT COUNT(*) FROM research_runs WHERE run_date >= ? AND status = 'failed'", |
| 1080 | (week_ago,), |
| 1081 | ).fetchone()[0] |
| 1082 | cost_7d = conn.execute( |
| 1083 | "SELECT COALESCE(SUM(token_cost), 0) FROM research_runs WHERE run_date >= ?", |
| 1084 | (week_ago,), |
| 1085 | ).fetchone()[0] |
| 1086 | |
| 1087 | # Source breakdown |
| 1088 | sources = {} |
| 1089 | for row in conn.execute( |
| 1090 | "SELECT source, COUNT(*) as cnt FROM findings GROUP BY source" |
| 1091 | ).fetchall(): |
| 1092 | sources[row["source"]] = row["cnt"] |
| 1093 | |
| 1094 | db_path = _get_db_path() |
| 1095 | db_size = db_path.stat().st_size if db_path.exists() else 0 |
| 1096 | |
| 1097 | return { |
| 1098 | "topics_active": topic_count, |
| 1099 | "total_findings": finding_count, |
| 1100 | "db_size_bytes": db_size, |
| 1101 | "runs_7d": runs_7d, |
| 1102 | "successful_7d": successful_7d, |
| 1103 | "failed_7d": failed_7d, |
| 1104 | "cost_7d": cost_7d, |
| 1105 | "sources": sources, |
| 1106 | "daily_budget": get_setting("daily_budget", "5.00"), |
| 1107 | } |
| 1108 | finally: |
| 1109 | conn.close() |
| 1110 | |
| 1111 | |
| 1112 | def get_trending(days: int = 7) -> List[Dict[str, Any]]: |
| 1113 | """Get topics ranked by recent finding activity.""" |
| 1114 | conn = _connect() |
| 1115 | try: |
| 1116 | since = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d") |
| 1117 | rows = conn.execute( |
| 1118 | """SELECT t.name, t.id, |
| 1119 | COUNT(f.id) as new_findings, |
| 1120 | COALESCE(SUM(f.engagement_score), 0) as total_engagement |
| 1121 | FROM topics t |
| 1122 | LEFT JOIN findings f ON f.topic_id = t.id AND f.first_seen >= ? |
| 1123 | WHERE t.enabled = 1 |
| 1124 | GROUP BY t.id |
| 1125 | ORDER BY new_findings DESC""", |
| 1126 | (since,), |
| 1127 | ).fetchall() |
| 1128 | return [dict(r) for r in rows] |
| 1129 | finally: |
| 1130 | conn.close() |
| 1131 | |
| 1132 | |
| 1133 | def finding_from_candidate(candidate: schema.Candidate) -> Dict[str, Any]: |
| 1134 | """Convert a ranked candidate into a persisted finding.""" |
| 1135 | primary_item = schema.candidate_primary_item(candidate) |
| 1136 | corroborating_sources = [ |
| 1137 | source for source in schema.candidate_sources(candidate) |
| 1138 | if source and source != candidate.source |
| 1139 | ] |
| 1140 | summary = candidate.explanation or candidate.snippet or "" |
| 1141 | if corroborating_sources: |
| 1142 | prefix = f"Also seen in: {', '.join(corroborating_sources)}." |
| 1143 | summary = f"{prefix} {summary}".strip() |
| 1144 | body = ( |
| 1145 | primary_item.body |
| 1146 | if primary_item and primary_item.body |
| 1147 | else candidate.snippet or candidate.title |
| 1148 | ) |
| 1149 | author = primary_item.author if primary_item and primary_item.author else "" |
| 1150 | return { |
| 1151 | "source": candidate.source or "unknown", |
| 1152 | "source_url": candidate.url, |
| 1153 | "source_title": candidate.title, |
| 1154 | "author": author, |
| 1155 | "content": body, |
| 1156 | "summary": summary, |
| 1157 | "engagement_score": candidate.engagement or 0, |
| 1158 | "relevance_score": candidate.final_score or candidate.rerank_score or candidate.local_relevance, |
| 1159 | } |
| 1160 | |
| 1161 | |
| 1162 | def findings_from_report( |
| 1163 | report: schema.Report, |
| 1164 | *, |
| 1165 | limit: Optional[int] = None, |
| 1166 | ) -> List[Dict[str, Any]]: |
| 1167 | """Convert report into persisted findings. |
| 1168 | |
| 1169 | Uses ranked candidates (post-rerank) when available for quality scores and explanations. |
| 1170 | Supplements with raw items from items_by_source for HN/PM that didn't rank highly |
| 1171 | but are valuable for watchlist persistence. When ranked_candidates is empty |
| 1172 | (degraded path — rerank failed or was skipped), falls back to supplementing |
| 1173 | all sources from items_by_source so findings aren't silently dropped. |
| 1174 | """ |
| 1175 | findings = [] |
| 1176 | seen_urls = set() |
| 1177 | |
| 1178 | for candidate in report.ranked_candidates: |
| 1179 | findings.append(finding_from_candidate(candidate)) |
| 1180 | seen_urls.add(candidate.url) |
| 1181 | |
| 1182 | supplement_sources = ( |
| 1183 | list(report.items_by_source) |
| 1184 | if not report.ranked_candidates |
| 1185 | else ["hackernews", "polymarket"] |
| 1186 | ) |
| 1187 | for source_name in supplement_sources: |
| 1188 | if source_name not in report.items_by_source: |
| 1189 | continue |
| 1190 | for item in report.items_by_source[source_name]: |
| 1191 | if item.url in seen_urls: |
| 1192 | continue |
| 1193 | findings.append({ |
| 1194 | "source": source_name, |
| 1195 | "source_url": item.url, |
| 1196 | "source_title": item.title, |
| 1197 | "author": item.author or "", |
| 1198 | "content": item.body or "", |
| 1199 | "summary": item.snippet or (item.body[:500] if item.body else ""), |
| 1200 | "engagement_score": item.engagement_score or 0.0, |
| 1201 | "relevance_score": item.local_relevance or 0.5, |
| 1202 | }) |
| 1203 | seen_urls.add(item.url) |
| 1204 | |
| 1205 | return findings[:limit] if limit is not None else findings |
| 1206 | |
| 1207 | |
| 1208 | # --- CLI interface --- |
| 1209 | |
| 1210 | |
| 1211 | def _cli_query(args): |
| 1212 | """Handle CLI query command.""" |
| 1213 | topic = get_topic(args.topic) |
| 1214 | if not topic: |
| 1215 | print(json.dumps({"error": f"Topic not found: {args.topic}"})) |
| 1216 | return |
| 1217 | |
| 1218 | since = None |
| 1219 | if args.since: |
| 1220 | # Parse duration like "7d", "30d". Use UTC to match SQLite's |
| 1221 | # datetime('now') which writes first_seen in UTC. |
| 1222 | days = int(args.since.rstrip("d")) |
| 1223 | since = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d") |
| 1224 | |
| 1225 | findings = get_new_findings(topic["id"], since) |
| 1226 | print(json.dumps({"topic": topic["name"], "findings": findings, "count": len(findings)}, default=str)) |
| 1227 | |
| 1228 | |
| 1229 | def _cli_search(args): |
| 1230 | """Handle CLI search command.""" |
| 1231 | results = search_findings(args.query, limit=args.limit) |
| 1232 | print(json.dumps({"query": args.query, "results": results, "count": len(results)}, default=str)) |
| 1233 | |
| 1234 | |
| 1235 | def _cli_trending(args): |
| 1236 | """Handle CLI trending command.""" |
| 1237 | results = get_trending(args.days) |
| 1238 | print(json.dumps({"trending": results}, default=str)) |
| 1239 | |
| 1240 | |
| 1241 | def _cli_stats(args): |
| 1242 | """Handle CLI stats command.""" |
| 1243 | stats = get_stats() |
| 1244 | print(json.dumps(stats, default=str)) |
| 1245 | |
| 1246 | |
| 1247 | def main(): |
| 1248 | parser = argparse.ArgumentParser(description="Query the last30days research database") |
| 1249 | sub = parser.add_subparsers(dest="command") |
| 1250 | |
| 1251 | # query |
| 1252 | q = sub.add_parser("query", help="Query findings for a topic") |
| 1253 | q.add_argument("topic", help="Topic name") |
| 1254 | q.add_argument("--since", help="Duration like '7d' or '30d'") |
| 1255 | q.set_defaults(func=_cli_query) |
| 1256 | |
| 1257 | # search |
| 1258 | s = sub.add_parser("search", help="Full-text search across findings") |
| 1259 | s.add_argument("query", help="Search query") |
| 1260 | s.add_argument("--limit", type=int, default=20, help="Max results") |
| 1261 | s.set_defaults(func=_cli_search) |
| 1262 | |
| 1263 | # trending |
| 1264 | t = sub.add_parser("trending", help="Show trending topics") |
| 1265 | t.add_argument("--days", type=int, default=7, help="Look back N days") |
| 1266 | t.set_defaults(func=_cli_trending) |
| 1267 | |
| 1268 | # stats |
| 1269 | st = sub.add_parser("stats", help="Show database stats") |
| 1270 | st.set_defaults(func=_cli_stats) |
| 1271 | |
| 1272 | args = parser.parse_args() |
| 1273 | if not args.command: |
| 1274 | parser.print_help() |
| 1275 | sys.exit(1) |
| 1276 | |
| 1277 | # Ensure DB exists |
| 1278 | init_db() |
| 1279 | args.func(args) |
| 1280 | |
| 1281 | |
| 1282 | if __name__ == "__main__": |
| 1283 | main() |
| 1284 |