107 lines
4.4 KiB
Python
107 lines
4.4 KiB
Python
"""db.py — SQLite storage for ani-web: users, history, searches, events.
|
|
|
|
One file per data dir: <DATA>/aniweb.db (+ -wal/-shm while running).
|
|
Fresh connection per call: sqlite3 connections can't cross threads without
|
|
a lock, the threadpool runs the sync endpoints, and at this scale the
|
|
reconnect cost is zero. WAL: SQLite serializes writers itself, so the
|
|
lost-update races of the JSON read-modify-write era are gone.
|
|
|
|
One-time import: a table that is empty while its legacy JSON still exists
|
|
gets absorbed (users.json, history/*.json, searches/*.json); the files are
|
|
deletable once the tables are populated (the empty-table guard makes a
|
|
second import impossible). Legacy ghost rows (same episode stored with int
|
|
AND str ep_id) are deduped on import: newest t per (user, slug, ep).
|
|
"""
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from pathlib import Path
|
|
|
|
DATA = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data"))
|
|
DB = DATA / "aniweb.db"
|
|
EVENT_RETENTION = 90 * 86400 # ponytail: pruned at boot, not by a cron
|
|
|
|
|
|
def connect() -> sqlite3.Connection:
|
|
c = sqlite3.connect(DB, timeout=10)
|
|
c.execute("PRAGMA journal_mode=WAL")
|
|
return c
|
|
|
|
|
|
def init() -> None:
|
|
"""Schema + one-time JSON import + event pruning. Idempotent."""
|
|
DATA.mkdir(exist_ok=True)
|
|
c = connect()
|
|
c.executescript("""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
name TEXT PRIMARY KEY, salt TEXT NOT NULL, hash TEXT NOT NULL,
|
|
iter INTEGER, created REAL);
|
|
CREATE TABLE IF NOT EXISTS history (
|
|
user TEXT NOT NULL, slug_id TEXT NOT NULL, ep_id INTEGER NOT NULL,
|
|
t REAL, payload TEXT NOT NULL,
|
|
PRIMARY KEY (user, slug_id, ep_id));
|
|
CREATE TABLE IF NOT EXISTS searches (
|
|
user TEXT NOT NULL, q TEXT NOT NULL, t REAL,
|
|
PRIMARY KEY (user, q COLLATE NOCASE));
|
|
CREATE TABLE IF NOT EXISTS events (
|
|
id INTEGER PRIMARY KEY, ts REAL, user TEXT, type TEXT, payload TEXT);
|
|
CREATE INDEX IF NOT EXISTS ix_events_user_ts ON events (user, ts);
|
|
""")
|
|
_import_json(c)
|
|
c.execute("DELETE FROM events WHERE ts < ?", (time.time() - EVENT_RETENTION,))
|
|
c.commit()
|
|
c.close()
|
|
|
|
|
|
def _import_json(c: sqlite3.Connection) -> None:
|
|
def empty(table: str) -> bool:
|
|
return c.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0
|
|
|
|
if empty("users") and (DATA / "users.json").exists():
|
|
for name, d in json.loads((DATA / "users.json").read_text()).items():
|
|
c.execute("INSERT INTO users VALUES (?,?,?,?,?)",
|
|
(name, d["salt"], d["hash"], d.get("iter"), d.get("created", 0.0)))
|
|
|
|
if empty("history") and (DATA / "history").is_dir():
|
|
best = {}
|
|
for f in (DATA / "history").glob("*.json"):
|
|
try:
|
|
rows = json.loads(f.read_text())
|
|
except (json.JSONDecodeError, OSError):
|
|
continue
|
|
for e in rows:
|
|
try:
|
|
k = (f.stem, str(e.get("slug_id")), int(e["ep_id"]))
|
|
except (KeyError, TypeError, ValueError):
|
|
continue
|
|
if e.get("t", 0) >= best.get(k, {"t": 0})["t"]:
|
|
best[k] = e
|
|
for (user, slug, ep), e in best.items():
|
|
e = dict(e, ep_id=ep) # normalize ghost-row int/str
|
|
c.execute("INSERT INTO history VALUES (?,?,?,?,?)",
|
|
(user, slug, ep, e.get("t", 0.0), json.dumps(e)))
|
|
|
|
if empty("searches") and (DATA / "searches").is_dir():
|
|
# no per-item timestamps in the JSON: file mtime, oldest-last order
|
|
for f in (DATA / "searches").glob("*.json"):
|
|
try:
|
|
rows = json.loads(f.read_text())
|
|
except (json.JSONDecodeError, OSError):
|
|
continue
|
|
t0, seen = f.stat().st_mtime, set()
|
|
for i, q in enumerate(rows):
|
|
q = str(q)
|
|
if q.lower() in seen:
|
|
continue
|
|
seen.add(q.lower())
|
|
c.execute("INSERT OR REPLACE INTO searches VALUES (?,?,?)",
|
|
(f.stem, q, t0 - i))
|
|
|
|
|
|
def event(c: sqlite3.Connection, user: str, type: str, payload: dict) -> None:
|
|
"""Append one metric event (caller commits). Watch-time is derivable
|
|
from the 'position' stream: consecutive pos deltas per (user, ep)."""
|
|
c.execute("INSERT INTO events (ts, user, type, payload) VALUES (?,?,?,?)",
|
|
(time.time(), user, type, json.dumps(payload)))
|