#!/usr/bin/env python3 """ Seed logs/newsdata.db from the headline blobs still on disk. ============================================================== The durable headline corpus (``news_archive.py``) only started collecting when its hook shipped. Everything before that survives only inside signal logs, in two different shapes: * ``logs/signals.db`` — modern schema, headlines interned into ``news_snapshots`` (kind ``headlines``) and referenced by hash. * an older orphan DB (default ``~/signals.db``) — pre-interning, blobs inline in ``signals.headlines_json``, short-form item keys with no ``description``, ``link``, ``publisher`` or ``score_by_currency``. Both are opened READ-ONLY. The script is idempotent: re-running it adds nothing new and never downgrades a score. Provenance is RECOMPUTED, not trusted ------------------------------------- Neither source records which scorer produced its ``score`` — that flag only exists from this change onwards. So each headline is re-scored with the live keyword scorer and compared: * stored score differs from the keyword score -> ``llm`` (DeepSeek wrote it) * exact match on a non-zero score -> ``keyword`` * both zero -> ``unknown`` (indistinguishable) Rows are stamped ``origin='backfill'`` because they come from the post-filter pool: ``get_fundamental_context`` drops headlines inside ±0.1 before they ever reach a signal row, so genuinely neutral headlines are under-represented in this history. Rows collected live by the hook in ``news_engine`` do not have that bias and are stamped ``'live'``. Usage ----- venv/bin/python scripts/backfill_news_archive.py --dry-run venv/bin/python scripts/backfill_news_archive.py venv/bin/python scripts/backfill_news_archive.py --legacy-db /path/to/old.db """ from __future__ import annotations import argparse import json import os import sqlite3 import sys from collections import Counter from typing import Any, Dict, Iterator, Tuple sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import app_config # noqa: E402 import news_archive # noqa: E402 from config import NewsConfig # noqa: E402 from news_engine import NewsEngine # noqa: E402 def _open_ro(path: str) -> sqlite3.Connection: """Read-only, but NOT immutable=1 — that flag is blind to WAL rows.""" return sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True) def _has_table(con: sqlite3.Connection, name: str) -> bool: row = con.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)).fetchone() return row is not None def iter_blobs(path: str) -> Iterator[Tuple[str, str]]: """Yield (headlines_json, seen_at) from either DB shape.""" con = _open_ro(path) try: if _has_table(con, "news_snapshots"): yield from con.execute( "SELECT body, first_seen FROM news_snapshots" " WHERE kind='headlines' ORDER BY first_seen") cols = {r[1] for r in con.execute("PRAGMA table_info(signals)")} if "headlines_json" in cols: order = "logged_at" if "logged_at" in cols else "id" yield from con.execute( "SELECT headlines_json, " + ("logged_at" if "logged_at" in cols else "''") + " FROM signals WHERE headlines_json IS NOT NULL" f" ORDER BY {order}") finally: con.close() class _Row: """Duck-typed stand-in for NewsSentiment, so news_archive.build_row flattens a stored dict exactly the way it flattens a live headline.""" __slots__ = ("headline", "source", "timestamp", "currencies_mentioned", "sentiment_score", "keywords", "sentiment_by_currency", "link", "description", "publisher", "score_source") def __init__(self, item: Dict[str, Any], score_source: str): self.headline = item.get("headline") self.source = item.get("source") self.timestamp = item.get("timestamp") self.currencies_mentioned = item.get("currencies") or [] self.sentiment_score = item.get("score") self.keywords = item.get("keywords") or [] self.sentiment_by_currency = item.get("score_by_currency") or {} self.link = item.get("link") self.description = item.get("description") self.publisher = item.get("publisher") self.score_source = score_source def classify(engine: NewsEngine, item: Dict[str, Any]) -> str: """Recover which scorer produced this stored score.""" text = item.get("headline") if not isinstance(text, str) or not text.strip(): return "unknown" try: stored = float(item.get("score") or 0.0) except (TypeError, ValueError): return "unknown" keyword = engine._analyze_headline_sentiment(text) if abs(stored - keyword) > 1e-9: return "llm" if abs(stored) < 1e-9: return "unknown" # both zero — the two scorers are indistinguishable return "keyword" def collect(paths, engine: NewsEngine): """Newest wins per headline: later blobs carry upgraded LLM scores.""" best: Dict[str, Tuple[str, Dict[str, Any], str]] = {} stats = Counter() for path in paths: if not os.path.exists(path): print(f" skip (not found): {path}") continue blobs = 0 for body, seen_at in iter_blobs(path): try: items = json.loads(body) except Exception: stats["bad_blob"] += 1 continue blobs += 1 for item in items: if not isinstance(item, dict): continue text = item.get("headline") if not isinstance(text, str) or not text.strip(): continue source = classify(engine, item) prior = best.get(text) if prior is None or ( news_archive._SOURCE_RANK.get(source, 0) >= news_archive._SOURCE_RANK.get(prior[0], 0)): best[text] = (source, item, seen_at or "") stats["items"] += 1 print(f" {path}: {blobs} blobs") for source, _, _ in best.values(): stats[source] += 1 return best, stats def main() -> int: default_legacy = os.path.join(os.path.expanduser("~"), "signals.db") ap = argparse.ArgumentParser(description=__doc__.split("\n")[1]) ap.add_argument("--db", default=app_config.signal_log_path(), help="modern signal-log DB (default: config path)") ap.add_argument("--legacy-db", default=default_legacy, help=f"older inline-blob DB (default: {default_legacy})") ap.add_argument("--out", default=app_config.news_archive_path(), help="archive DB to write (default: config path)") ap.add_argument("--dry-run", action="store_true", help="count only, write nothing") args = ap.parse_args() engine = NewsEngine(NewsConfig()) print("Reading headline blobs (read-only):") best, stats = collect([args.db, args.legacy_db], engine) print(f"\n headline items scanned : {stats['items']}") print(f" distinct headlines : {len(best)}") print(f" llm (DeepSeek) : {stats['llm']}") print(f" keyword : {stats['keyword']}") print(f" unknown (both zero) : {stats['unknown']}") if stats["bad_blob"]: print(f" unparseable blobs : {stats['bad_blob']}") if args.dry_run: print("\n--dry-run: nothing written.") return 0 if not best: print("\nNothing to write.") return 0 rows = [] for text, (source, item, seen_at) in best.items(): row = news_archive.build_row( _Row(item, source), teacher_model=None, # not recorded by either source teacher_prompt_sha1=None, origin="backfill", first_seen=seen_at or None, last_seen=seen_at or None) if row is not None: rows.append(row) archive = news_archive.NewsArchive(args.out) try: written = archive.store(rows) finally: archive.close() print(f"\nWrote {written} rows to {args.out}") con = _open_ro(args.out) try: print("\nArchive now holds:") for source, count in con.execute( "SELECT score_source, COUNT(*) FROM headlines" " GROUP BY score_source ORDER BY COUNT(*) DESC"): print(f" {source:8s} {count}") finally: con.close() return 0 if __name__ == "__main__": raise SystemExit(main())