Phase 5: cron entry script, soft-cap archive, run_log zero-fetch degradation
- oracle-pipeline.sh: single cron entry point (pipeline -> summarize -> archive) - archive.py: soft-cap archival to entries_archive (preserve, not delete) - pipeline.py: record zero-fetch (rate-limited) runs as degraded in run_log.notes Verified: full script runs exit 0, run_log captures per-source status.
This commit is contained in:
+94
@@ -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()
|
||||||
Executable
+37
@@ -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"
|
||||||
+8
-1
@@ -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()
|
# Record run log (failure visibility + growth control) — before conn.close()
|
||||||
ok = [s for s, st in source_stats.items() if not st.get("error")]
|
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")]
|
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:
|
try:
|
||||||
conn.execute("""
|
conn.execute("""
|
||||||
INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes)
|
INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes)
|
||||||
|
|||||||
Reference in New Issue
Block a user