ForexBot2/scripts/migrate_signal_log_news.py

#!/usr/bin/env python3
"""
migrate_signal_log_news.py — move inline news blobs into news_snapshots.

Legacy rows store the full RSS headline list (~33 KB) and calendar events
(~1.2 KB) inline on EVERY scan row; the same blob repeats ~5× because the
news caches outlive a candle. This script interns each distinct blob once
(content hash), stamps `headlines_hash` / `events_hash` on the row and
NULLs the inline columns. Readers use the `signals_with_news` view or
`signal_log.read_news`.

Idempotent and resumable (batched transactions). Run --vacuum with BOTH
live writers (telegram bot, dashboard publisher) stopped — VACUUM needs
exclusive access and ~1× the final size of free disk.

    python scripts/migrate_signal_log_news.py --dry-run
    python scripts/migrate_signal_log_news.py
    python scripts/migrate_signal_log_news.py --vacuum      # bots stopped
"""
import argparse
import hashlib
import os
import sqlite3
import sys
from datetime import datetime, timezone

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import app_config  # noqa: E402
from signal_log import SignalLogger  # noqa: E402


def _hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def migrate(con: sqlite3.Connection, batch: int = 500) -> dict:
    """Intern legacy inline blobs; returns counters. Caller owns the
    connection (schema must already exist — open a SignalLogger once)."""
    stats = {"rows": 0, "headlines_interned": 0, "events_interned": 0}
    now = datetime.now(timezone.utc).isoformat()
    while True:
        rows = con.execute(
            "SELECT id, headlines_json, events_json FROM signals"
            " WHERE (headlines_json IS NOT NULL AND headlines_hash IS NULL)"
            "    OR (events_json IS NOT NULL AND events_hash IS NULL)"
            " ORDER BY id LIMIT ?", (batch,)).fetchall()
        if not rows:
            break
        con.execute("BEGIN IMMEDIATE")
        try:
            for rowid, headlines, events in rows:
                sets, params = [], []
                if headlines is not None:
                    h = _hash(headlines)
                    cur = con.execute(
                        "INSERT OR IGNORE INTO news_snapshots"
                        " (hash, kind, body, first_seen, last_seen)"
                        " VALUES (?, 'headlines', ?, ?, ?)",
                        (h, headlines, now, now))
                    stats["headlines_interned"] += cur.rowcount
                    sets += ["headlines_hash=?", "headlines_json=NULL"]
                    params.append(h)
                if events is not None:
                    h = _hash(events)
                    cur = con.execute(
                        "INSERT OR IGNORE INTO news_snapshots"
                        " (hash, kind, body, first_seen, last_seen)"
                        " VALUES (?, 'events', ?, ?, ?)",
                        (h, events, now, now))
                    stats["events_interned"] += cur.rowcount
                    sets += ["events_hash=?", "events_json=NULL"]
                    params.append(h)
                params.append(rowid)
                con.execute(
                    f"UPDATE signals SET {', '.join(sets)} WHERE id=?",
                    params)
                stats["rows"] += 1
            con.commit()
        except Exception:
            con.rollback()
            raise
    return stats


def prune_orphans(con: sqlite3.Connection) -> int:
    cur = con.execute(
        "DELETE FROM news_snapshots WHERE hash NOT IN"
        " (SELECT headlines_hash FROM signals WHERE headlines_hash IS NOT NULL"
        "  UNION SELECT events_hash FROM signals WHERE events_hash IS NOT NULL)")
    con.commit()
    return cur.rowcount


def prune_news_older_than(con: sqlite3.Connection, days: int) -> int:
    """Drop the news reference (not the row) from rows older than N days,
    then delete snapshots nothing references any more."""
    # logged_at is ISO8601 with a 'T' and a +00:00 offset; SQLite's
    # datetime() returns 'YYYY-MM-DD HH:MM:SS'. Comparing them as raw
    # strings put every row from the cutoff CALENDAR DAY on the wrong
    # side ('T' > ' '), so a full day never pruned. Normalise to the
    # same shape first — all logged_at values are UTC (see _iso()).
    cur = con.execute(
        "UPDATE signals SET headlines_hash=NULL, events_hash=NULL"
        " WHERE substr(replace(logged_at, 'T', ' '), 1, 19)"
        " < datetime('now', ?)", (f"-{int(days)} days",))
    con.commit()
    prune_orphans(con)
    return cur.rowcount


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[1])
    ap.add_argument("--db", default=None)
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--batch", type=int, default=500)
    ap.add_argument("--prune-news-older-than", type=int, default=None,
                    metavar="DAYS")
    ap.add_argument("--prune-orphans", action="store_true")
    ap.add_argument("--vacuum", action="store_true",
                    help="checkpoint + VACUUM (all writers must be stopped)")
    ap.add_argument("--no-backup", action="store_true")
    args = ap.parse_args(argv)
    path = args.db or app_config.signal_log_path()
    if not os.path.exists(path):
        print(f"no database at {path}")
        return 1
    size0 = os.path.getsize(path)
    if args.dry_run:
        con = sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True)
        n = con.execute(
            "SELECT COUNT(*) FROM signals WHERE headlines_json IS NOT NULL"
            " OR events_json IS NOT NULL").fetchone()[0]
        d = con.execute(
            "SELECT COUNT(DISTINCT headlines_json) FROM signals"
            " WHERE headlines_json IS NOT NULL").fetchone()[0]
        print(f"{n} rows carry inline news blobs ({d} distinct headline "
              f"blobs); file size {size0 / 1e6:.1f} MB")
        con.close()
        return 0
    if not args.no_backup:
        backup = f"{path}.pre-news-{datetime.now(timezone.utc):%Y%m%d}"
        if not os.path.exists(backup):
            src = sqlite3.connect(path)
            dst = sqlite3.connect(backup)
            src.backup(dst)
            dst.close()
            src.close()
            print(f"backup written: {backup}")
    lg = SignalLogger(path)          # applies the additive schema
    lg.close()
    con = sqlite3.connect(path)
    con.execute("PRAGMA busy_timeout=10000")
    try:
        print("migrate:", migrate(con, batch=args.batch))
        if args.prune_news_older_than is not None:
            print("rows de-referenced:",
                  prune_news_older_than(con, args.prune_news_older_than))
        if args.prune_orphans:
            print("orphan snapshots removed:", prune_orphans(con))
        if args.vacuum:
            try:
                con.execute("PRAGMA wal_checkpoint(TRUNCATE)")
                con.execute("VACUUM")
            except sqlite3.OperationalError as exc:
                print(f"VACUUM failed ({exc}) — stop the bot and dashboard "
                      "publisher first, then re-run with --vacuum")
                return 2
    finally:
        con.close()
    print(f"size {size0 / 1e6:.1f} MB -> {os.path.getsize(path) / 1e6:.1f} MB"
          + ("" if args.vacuum else " (run --vacuum to reclaim space)"))
    return 0


if __name__ == "__main__":
    sys.exit(main())