349 lines
12 KiB
Python
349 lines
12 KiB
Python
"""SQLite persistence.
|
|
|
|
Two jobs beyond bookkeeping:
|
|
|
|
1. Remember the first time we ever saw a version, so the Monday digest can
|
|
say "this dropped 3 days ago" and so risk scoring can reward letting a
|
|
release age.
|
|
2. Remember what has already been reported, so the digest highlights genuinely
|
|
new things instead of re-listing the same pending update every week.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from .models import Evidence, Finding, RunSummary, Verdict, VersionCheck, iso, utcnow
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS runs (
|
|
run_id TEXT PRIMARY KEY,
|
|
started TEXT NOT NULL,
|
|
finished TEXT,
|
|
mode TEXT,
|
|
items_checked INTEGER DEFAULT 0,
|
|
updates_found INTEGER DEFAULT 0,
|
|
errors INTEGER DEFAULT 0,
|
|
new_since_last INTEGER DEFAULT 0,
|
|
http_requests INTEGER DEFAULT 0,
|
|
cache_hits INTEGER DEFAULT 0
|
|
);
|
|
|
|
-- One row per (item, version) ever observed. first_seen is the anchor for
|
|
-- "how long has this been out in the wild".
|
|
CREATE TABLE IF NOT EXISTS version_sightings (
|
|
item_id TEXT NOT NULL,
|
|
version TEXT NOT NULL,
|
|
first_seen TEXT NOT NULL,
|
|
last_seen TEXT NOT NULL,
|
|
released TEXT,
|
|
release_url TEXT,
|
|
notes TEXT,
|
|
PRIMARY KEY (item_id, version)
|
|
);
|
|
|
|
-- Latest snapshot per item, rewritten each scan.
|
|
CREATE TABLE IF NOT EXISTS findings (
|
|
item_id TEXT PRIMARY KEY,
|
|
run_id TEXT,
|
|
installed TEXT,
|
|
latest TEXT,
|
|
comparison TEXT,
|
|
bump TEXT,
|
|
check_status TEXT,
|
|
check_detail TEXT,
|
|
released TEXT,
|
|
verdict_json TEXT,
|
|
evidence_json TEXT,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
-- What the last digest told the team about, so we can diff.
|
|
CREATE TABLE IF NOT EXISTS reported (
|
|
item_id TEXT NOT NULL,
|
|
version TEXT NOT NULL,
|
|
reported_at TEXT NOT NULL,
|
|
channel TEXT NOT NULL,
|
|
PRIMARY KEY (item_id, version, channel)
|
|
);
|
|
|
|
-- Human decisions: "we installed this", "we are holding this until X".
|
|
CREATE TABLE IF NOT EXISTS decisions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
item_id TEXT NOT NULL,
|
|
version TEXT,
|
|
decision TEXT NOT NULL, -- installed | holding | skipped
|
|
note TEXT,
|
|
decided_at TEXT NOT NULL,
|
|
decided_by TEXT
|
|
);
|
|
|
|
-- Cache for expensive Claude web-search version lookups, so mid-week scans
|
|
-- reuse Monday's answer instead of paying for it again.
|
|
CREATE TABLE IF NOT EXISTS lookup_cache (
|
|
item_id TEXT PRIMARY KEY,
|
|
version TEXT,
|
|
released TEXT,
|
|
source_url TEXT,
|
|
confidence TEXT,
|
|
notes TEXT,
|
|
looked_up_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_sight_item ON version_sightings(item_id);
|
|
CREATE INDEX IF NOT EXISTS idx_dec_item ON decisions(item_id);
|
|
"""
|
|
|
|
|
|
def _parse_dt(text: str | None) -> datetime | None:
|
|
if not text:
|
|
return None
|
|
try:
|
|
dt = datetime.fromisoformat(text)
|
|
except ValueError:
|
|
return None
|
|
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
class Store:
|
|
def __init__(self, path: Path):
|
|
self.path = Path(path)
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._conn = sqlite3.connect(str(self.path))
|
|
self._conn.row_factory = sqlite3.Row
|
|
self._conn.executescript(SCHEMA)
|
|
self._conn.commit()
|
|
|
|
def close(self) -> None:
|
|
self._conn.close()
|
|
|
|
@contextmanager
|
|
def _tx(self):
|
|
try:
|
|
yield self._conn
|
|
self._conn.commit()
|
|
except Exception:
|
|
self._conn.rollback()
|
|
raise
|
|
|
|
# -- runs --------------------------------------------------------------
|
|
|
|
def start_run(self, run: RunSummary) -> None:
|
|
with self._tx() as c:
|
|
c.execute(
|
|
"INSERT OR REPLACE INTO runs (run_id, started, mode) VALUES (?,?,?)",
|
|
(run.run_id, iso(run.started), run.mode),
|
|
)
|
|
|
|
def finish_run(self, run: RunSummary) -> None:
|
|
with self._tx() as c:
|
|
c.execute(
|
|
"""UPDATE runs SET finished=?, items_checked=?, updates_found=?,
|
|
errors=?, new_since_last=?, http_requests=?, cache_hits=?
|
|
WHERE run_id=?""",
|
|
(
|
|
iso(run.finished or utcnow()), run.items_checked, run.updates_found,
|
|
run.errors, run.new_since_last, run.http_requests, run.cache_hits,
|
|
run.run_id,
|
|
),
|
|
)
|
|
|
|
def last_run(self, mode: str | None = None) -> RunSummary | None:
|
|
q = "SELECT * FROM runs WHERE finished IS NOT NULL"
|
|
args: tuple = ()
|
|
if mode:
|
|
q += " AND mode=?"
|
|
args = (mode,)
|
|
q += " ORDER BY finished DESC LIMIT 1"
|
|
row = self._conn.execute(q, args).fetchone()
|
|
if not row:
|
|
return None
|
|
return RunSummary(
|
|
run_id=row["run_id"],
|
|
started=_parse_dt(row["started"]) or utcnow(),
|
|
finished=_parse_dt(row["finished"]),
|
|
items_checked=row["items_checked"] or 0,
|
|
updates_found=row["updates_found"] or 0,
|
|
errors=row["errors"] or 0,
|
|
new_since_last=row["new_since_last"] or 0,
|
|
mode=row["mode"] or "scan",
|
|
)
|
|
|
|
def recent_runs(self, limit: int = 10) -> list[dict]:
|
|
rows = self._conn.execute(
|
|
"SELECT * FROM runs ORDER BY started DESC LIMIT ?", (limit,)
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
# -- version sightings -------------------------------------------------
|
|
|
|
def record_sighting(self, check: VersionCheck) -> bool:
|
|
"""Record that ``check.latest`` exists. Returns True if this is the
|
|
first time we have ever seen this version for this item."""
|
|
if not check.latest:
|
|
return False
|
|
now = iso(utcnow())
|
|
row = self._conn.execute(
|
|
"SELECT first_seen FROM version_sightings WHERE item_id=? AND version=?",
|
|
(check.item_id, check.latest),
|
|
).fetchone()
|
|
with self._tx() as c:
|
|
if row:
|
|
c.execute(
|
|
"UPDATE version_sightings SET last_seen=? WHERE item_id=? AND version=?",
|
|
(now, check.item_id, check.latest),
|
|
)
|
|
return False
|
|
c.execute(
|
|
"""INSERT INTO version_sightings
|
|
(item_id, version, first_seen, last_seen, released, release_url, notes)
|
|
VALUES (?,?,?,?,?,?,?)""",
|
|
(
|
|
check.item_id, check.latest, now, now,
|
|
iso(check.released), check.release_url, check.release_notes,
|
|
),
|
|
)
|
|
return True
|
|
|
|
def first_seen(self, item_id: str, version: str) -> datetime | None:
|
|
row = self._conn.execute(
|
|
"SELECT first_seen FROM version_sightings WHERE item_id=? AND version=?",
|
|
(item_id, version),
|
|
).fetchone()
|
|
return _parse_dt(row["first_seen"]) if row else None
|
|
|
|
def sighting_count(self, item_id: str) -> int:
|
|
row = self._conn.execute(
|
|
"SELECT COUNT(*) AS n FROM version_sightings WHERE item_id=?", (item_id,)
|
|
).fetchone()
|
|
return row["n"] if row else 0
|
|
|
|
# -- findings ----------------------------------------------------------
|
|
|
|
def save_finding(self, run_id: str, f: Finding) -> None:
|
|
with self._tx() as c:
|
|
c.execute(
|
|
"""INSERT OR REPLACE INTO findings
|
|
(item_id, run_id, installed, latest, comparison, bump,
|
|
check_status, check_detail, released, verdict_json,
|
|
evidence_json, updated_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
|
|
(
|
|
f.item.id, run_id, f.item.installed, f.check.latest,
|
|
f.comparison, f.bump, f.check.status, f.check.detail,
|
|
iso(f.check.released),
|
|
json.dumps(f.verdict.to_dict()) if f.verdict else None,
|
|
json.dumps([e.to_dict() for e in f.evidence]),
|
|
iso(utcnow()),
|
|
),
|
|
)
|
|
|
|
def load_finding_row(self, item_id: str) -> dict | None:
|
|
row = self._conn.execute(
|
|
"SELECT * FROM findings WHERE item_id=?", (item_id,)
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def all_finding_rows(self) -> list[dict]:
|
|
rows = self._conn.execute("SELECT * FROM findings").fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
# -- digest bookkeeping ------------------------------------------------
|
|
|
|
def was_reported(self, item_id: str, version: str, channel: str = "email") -> bool:
|
|
row = self._conn.execute(
|
|
"SELECT 1 FROM reported WHERE item_id=? AND version=? AND channel=?",
|
|
(item_id, version, channel),
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
def mark_reported(self, item_id: str, version: str, channel: str = "email") -> None:
|
|
with self._tx() as c:
|
|
c.execute(
|
|
"""INSERT OR REPLACE INTO reported (item_id, version, reported_at, channel)
|
|
VALUES (?,?,?,?)""",
|
|
(item_id, version, iso(utcnow()), channel),
|
|
)
|
|
|
|
# -- decisions ---------------------------------------------------------
|
|
|
|
def add_decision(self, item_id: str, decision: str, version: str | None = None,
|
|
note: str | None = None, by: str | None = None) -> None:
|
|
with self._tx() as c:
|
|
c.execute(
|
|
"""INSERT INTO decisions (item_id, version, decision, note, decided_at, decided_by)
|
|
VALUES (?,?,?,?,?,?)""",
|
|
(item_id, version, decision, note, iso(utcnow()), by),
|
|
)
|
|
|
|
def latest_decision(self, item_id: str) -> dict | None:
|
|
row = self._conn.execute(
|
|
"SELECT * FROM decisions WHERE item_id=? ORDER BY decided_at DESC LIMIT 1",
|
|
(item_id,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def decisions_for(self, item_id: str) -> list[dict]:
|
|
rows = self._conn.execute(
|
|
"SELECT * FROM decisions WHERE item_id=? ORDER BY decided_at DESC", (item_id,)
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
# -- web-search lookup cache -------------------------------------------
|
|
|
|
def get_lookup(self, item_id: str, max_age_days: float) -> dict | None:
|
|
row = self._conn.execute(
|
|
"SELECT * FROM lookup_cache WHERE item_id=?", (item_id,)
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
when = _parse_dt(row["looked_up_at"])
|
|
if not when:
|
|
return None
|
|
age = (utcnow() - when).total_seconds() / 86400.0
|
|
if age > max_age_days:
|
|
return None
|
|
out = dict(row)
|
|
out["age_days"] = age
|
|
return out
|
|
|
|
def put_lookup(self, item_id: str, version: str | None, released: str | None,
|
|
source_url: str | None, confidence: str | None,
|
|
notes: str | None) -> None:
|
|
with self._tx() as c:
|
|
c.execute(
|
|
"""INSERT OR REPLACE INTO lookup_cache
|
|
(item_id, version, released, source_url, confidence, notes,
|
|
looked_up_at)
|
|
VALUES (?,?,?,?,?,?,?)""",
|
|
(item_id, version, released, source_url, confidence, notes,
|
|
iso(utcnow())),
|
|
)
|
|
|
|
|
|
def rehydrate_verdict(blob: str | None) -> Verdict | None:
|
|
if not blob:
|
|
return None
|
|
try:
|
|
return Verdict(**json.loads(blob))
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def rehydrate_evidence(blob: str | None) -> list[Evidence]:
|
|
if not blob:
|
|
return []
|
|
out = []
|
|
try:
|
|
for d in json.loads(blob):
|
|
d = dict(d)
|
|
d["published"] = _parse_dt(d.get("published"))
|
|
out.append(Evidence(**d))
|
|
except (ValueError, TypeError):
|
|
return []
|
|
return out
|