diff --git a/archive.py b/archive.py new file mode 100644 index 0000000..e9beafd --- /dev/null +++ b/archive.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Oracle soft-cap archival. + +Bounds live `entries` growth by moving old / excess rows into +`entries_archive` (preserving data — soft cap, not hard delete). +Two triggers: + --days N : move entries not updated in N days (default 30) + --cap N : if live entries exceed N, archive oldest beyond the cap (default 5000) +Default is a real run (data moves). Use --dry-run to report only. +""" + +import argparse +import os +import sqlite3 +import time + +DB = os.path.join(os.path.dirname(__file__), "oracle.db") + +ARCHIVE_SCHEMA = """ +CREATE TABLE IF NOT EXISTS entries_archive ( + id INTEGER PRIMARY KEY, + source TEXT, source_id TEXT, url TEXT, title TEXT, + extracted_text TEXT, summary TEXT, category_tags TEXT, + signal_score REAL, raw_metadata TEXT, + first_seen TEXT, last_updated TEXT, + archived_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) +); +""" + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--days", type=int, default=30, help="Archive entries not updated in N days") + ap.add_argument("--cap", type=int, default=5000, help="Soft cap on live entries; archive oldest beyond this") + ap.add_argument("--dry-run", action="store_true", help="Report only, make no changes") + args = ap.parse_args() + + if not os.path.exists(DB): + print("No oracle.db — nothing to archive") + return + + conn = sqlite3.connect(DB) + conn.execute(ARCHIVE_SCHEMA) + + cutoff = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - args.days * 86400)) + + cur = conn.cursor() + cur.execute("SELECT id, last_updated FROM entries") + rows = cur.fetchall() + n_total = len(rows) + + old_ids = [r[0] for r in rows if (r[1] or "") < cutoff] + beyond = max(0, n_total - args.cap) + if beyond > 0: + ordered = sorted(rows, key=lambda r: r[1] or "")[:beyond] + cap_ids = [r[0] for r in ordered] + else: + cap_ids = [] + + move_ids = sorted(set(old_ids) | set(cap_ids)) + + if not move_ids: + print(f"Archive check: {n_total} live entries, none older than {args.days}d " + f"or beyond cap {args.cap}. Nothing to archive.") + conn.close() + return + + print(f"Archive check: {n_total} live entries -> would archive {len(move_ids)} " + f"(old={len(old_ids)}, cap={len(cap_ids)}).") + + if args.dry_run: + print("DRY RUN — no changes made.") + conn.close() + return + + q = ",".join("?" * len(move_ids)) + conn.execute( + f"""INSERT OR REPLACE INTO entries_archive + (id, source, source_id, url, title, extracted_text, summary, + category_tags, signal_score, raw_metadata, first_seen, last_updated) + SELECT id, source, source_id, url, title, extracted_text, summary, + category_tags, signal_score, raw_metadata, first_seen, last_updated + FROM entries WHERE id IN ({q})""", + move_ids, + ) + conn.execute(f"DELETE FROM entries WHERE id IN ({q})", move_ids) + conn.commit() + conn.close() + print(f"Archived {len(move_ids)} entries (live now {n_total - len(move_ids)}). " + f"Preserved in entries_archive.") + + +if __name__ == "__main__": + main() diff --git a/oracle-pipeline.sh b/oracle-pipeline.sh new file mode 100755 index 0000000..8d2b383 --- /dev/null +++ b/oracle-pipeline.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +# +# Oracle daily pipeline — cron entry point (no_agent; stdout is delivered verbatim). +# Runs: fetch+store (pipeline.py) -> summarize (summarize.py) -> soft-cap archive (archive.py). +# run_log captures per-source failure + zero-fetch degradation for visibility. +# +set -u + +ORACLE_DIR="/home/vpsadmin/oracle" +LOG_DIR="$ORACLE_DIR/logs" +TS="$(date -u +%Y%m%d-%H%M%S)" +LOG="$LOG_DIR/cron_run_${TS}.log" + +mkdir -p "$LOG_DIR" +cd "$ORACLE_DIR" || { echo "FATAL: cannot cd $ORACLE_DIR"; exit 1; } + +{ + echo "=== Oracle pipeline run: $(date -u) ===" + python3 pipeline.py --limit 20 + echo + echo "=== Summarization Engine ===" + python3 summarize.py + echo + echo "=== Soft-cap archive (dry-safe default: 30d / 5000 cap) ===" + python3 archive.py --days 30 --cap 5000 + echo + echo "=== run_log tail (failure visibility) ===" + python3 -c " +import sqlite3 +c = sqlite3.connect('oracle.db') +for r in c.execute('SELECT id,run_time,total_fetched,total_stored,sources_failed,notes FROM run_log ORDER BY id DESC LIMIT 1'): + print(' run', r[0], '|', r[1], '| fetched', r[2], '| stored', r[3], '| failed', r[4], '| notes:', r[5]) +print(' live entries:', c.execute('SELECT COUNT(*) FROM entries').fetchone()[0]) +print(' archived entries:', c.execute('SELECT COUNT(*) FROM entries_archive').fetchone()[0]) +" + echo "=== Done ===" +} 2>&1 | tee "$LOG" diff --git a/pipeline.py b/pipeline.py index 05ecbaa..8ac7432 100644 --- a/pipeline.py +++ b/pipeline.py @@ -231,7 +231,14 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo # Record run log (failure visibility + growth control) — before conn.close() ok = [s for s, st in source_stats.items() if not st.get("error")] failed = [s for s, st in source_stats.items() if st.get("error")] - notes = "; ".join(f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")) or "all sources ok" + # Zero-fetch (e.g. Reddit fully rate-limited) raises no exception but + # is still a degraded run — record it so run_log can tell + # "intermittent vs consistently-broken" apart over time. + zero = [s for s, st in source_stats.items() if st.get("fetched", 0) == 0 and not st.get("error")] + notes_parts = [f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")] + if zero: + notes_parts.append(f"no-fetch (degraded): {', '.join(zero)}") + notes = "; ".join(notes_parts) or "all sources ok" try: conn.execute(""" INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes)