← back to Paul Conrad Archive
src/conrad/db.py
277 lines
"""SQLite schema + write helpers. Records are never deleted; duplicates are linked via merged_into."""
from __future__ import annotations
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from . import config
SCHEMA = """
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS sources (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
repository TEXT,
url TEXT,
classification TEXT CHECK (classification IN ('PUBLIC_API','PUBLIC_HTML','PUBLIC_IMAGE','METADATA_ONLY',
'PAYWALLED','PHYSICAL_ARCHIVE','REQUIRES_PERMISSION')),
status TEXT, -- worked | blocked | seed_only | not_attempted | partial
access_notes TEXT,
notes TEXT,
last_checked TEXT
);
CREATE TABLE IF NOT EXISTS cartoons (
id INTEGER PRIMARY KEY,
canonical_id TEXT NOT NULL UNIQUE,
creator TEXT NOT NULL DEFAULT 'Conrad, Paul, 1924-2010',
granularity TEXT NOT NULL CHECK (granularity IN ('item','folder','box_range')),
date_exact TEXT, date_start TEXT, date_end TEXT, year INTEGER,
date_is_estimate INTEGER NOT NULL DEFAULT 0,
title TEXT, caption TEXT, description TEXT, publication TEXT, syndicate TEXT,
medium TEXT, dimensions TEXT, signed_name TEXT, rights_text TEXT, copyright_holder TEXT,
notes TEXT,
merged_into INTEGER REFERENCES cartoons(id), -- NULL = canonical; else points at the canonical cartoon
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_cartoons_year ON cartoons(year);
CREATE INDEX IF NOT EXISTS ix_cartoons_gran ON cartoons(granularity);
CREATE TABLE IF NOT EXISTS cartoon_sources (
id INTEGER PRIMARY KEY,
cartoon_id INTEGER NOT NULL REFERENCES cartoons(id),
source_id TEXT NOT NULL REFERENCES sources(id),
repository TEXT, collection_name TEXT, identifier TEXT NOT NULL,
box TEXT, folder TEXT, page TEXT,
record_url TEXT, image_url TEXT, thumbnail_url TEXT,
local_image TEXT CHECK (local_image IS NULL), -- copyright rule: images are NEVER stored locally
access_level TEXT, rights_url TEXT, provenance TEXT, retrieved_at TEXT,
acquisition_method TEXT CHECK (acquisition_method IS NULL OR acquisition_method IN
('direct_api','direct_html','seed_direct','seed_via_reader_bypass','secondary_citation')),
UNIQUE (source_id, identifier)
);
CREATE INDEX IF NOT EXISTS ix_cs_cartoon ON cartoon_sources(cartoon_id);
CREATE TABLE IF NOT EXISTS people (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, is_president INTEGER NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS cartoon_people (cartoon_id INTEGER REFERENCES cartoons(id), person_id INTEGER REFERENCES people(id),
PRIMARY KEY (cartoon_id, person_id));
CREATE TABLE IF NOT EXISTS subjects (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE);
CREATE TABLE IF NOT EXISTS cartoon_subjects (cartoon_id INTEGER REFERENCES cartoons(id), subject_id INTEGER REFERENCES subjects(id),
PRIMARY KEY (cartoon_id, subject_id));
CREATE TABLE IF NOT EXISTS publications (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, date_start TEXT, date_end TEXT,
classification TEXT, notes TEXT);
CREATE TABLE IF NOT EXISTS collections (
id INTEGER PRIMARY KEY, repository TEXT NOT NULL, name TEXT NOT NULL, identifier TEXT, url TEXT,
extent TEXT, date_range TEXT, level TEXT, notes TEXT, source_id TEXT,
UNIQUE (repository, name, identifier));
CREATE TABLE IF NOT EXISTS books (
id INTEGER PRIMARY KEY, title TEXT NOT NULL UNIQUE, year INTEGER, publisher TEXT, isbn TEXT, oclc TEXT,
ia_identifier TEXT, ia_access TEXT, hathi_url TEXT, record_url TEXT, notes TEXT, verified INTEGER NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS book_cartoon_refs (book_id INTEGER REFERENCES books(id), cartoon_id INTEGER REFERENCES cartoons(id),
page TEXT, note TEXT, PRIMARY KEY (book_id, cartoon_id));
CREATE TABLE IF NOT EXISTS cartoon_links (
cartoon_id INTEGER NOT NULL REFERENCES cartoons(id), related_id INTEGER NOT NULL REFERENCES cartoons(id),
relation TEXT NOT NULL CHECK (relation IN ('duplicate_of','possible_duplicate','contained_in_candidate')),
score REAL, detail TEXT, PRIMARY KEY (cartoon_id, related_id, relation));
CREATE INDEX IF NOT EXISTS ix_cartoons_merged ON cartoons(merged_into);
-- denormalised search index for the viewer (rebuilt by refresh_index after dedupe)
CREATE TABLE IF NOT EXISTS cartoon_index (
id INTEGER PRIMARY KEY REFERENCES cartoons(id), repos TEXT, n_sources INTEGER, people TEXT, record_url TEXT,
search_text TEXT); -- lowercased title/caption/description/publication/people/subjects/repositories/collections
CREATE TABLE IF NOT EXISTS crawl_runs (
id INTEGER PRIMARY KEY, source TEXT NOT NULL, started TEXT NOT NULL, completed TEXT,
pages_scanned INTEGER DEFAULT 0, records_seen INTEGER DEFAULT 0, records_added INTEGER DEFAULT 0,
records_updated INTEGER DEFAULT 0, errors INTEGER DEFAULT 0, status TEXT, notes TEXT);
CREATE TABLE IF NOT EXISTS errors (
id INTEGER PRIMARY KEY, run_id INTEGER REFERENCES crawl_runs(id), source TEXT, url TEXT, error TEXT, at TEXT);
"""
def now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def connect(path: Path | str | None = None) -> sqlite3.Connection:
conn = sqlite3.connect(str(path or config.DB_PATH), timeout=60, check_same_thread=False)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
return conn
def init_db(conn: sqlite3.Connection) -> None:
conn.executescript(SCHEMA)
migrate(conn)
conn.commit()
def migrate(conn: sqlite3.Connection) -> None:
"""Idempotent in-place migrations for databases created by an older SCHEMA."""
cols = {r[1] for r in conn.execute("PRAGMA table_info(cartoon_sources)")}
if "acquisition_method" not in cols:
conn.execute("""ALTER TABLE cartoon_sources ADD COLUMN acquisition_method TEXT CHECK (acquisition_method IS NULL
OR acquisition_method IN ('direct_api','direct_html','seed_direct','seed_via_reader_bypass',
'secondary_citation'))""")
conn.execute("CREATE INDEX IF NOT EXISTS ix_cs_method ON cartoon_sources(acquisition_method)")
if "search_text" not in {r[1] for r in conn.execute("PRAGMA table_info(cartoon_index)")}:
conn.execute("ALTER TABLE cartoon_index ADD COLUMN search_text TEXT")
from .provenance import backfill
backfill(conn) # only rows still NULL
# ---------------------------------------------------------------- sources
def upsert_source(conn, id: str, name: str, *, repository=None, url=None, classification=None,
status=None, access_notes=None, notes=None) -> None:
conn.execute(
"""INSERT INTO sources (id,name,repository,url,classification,status,access_notes,notes,last_checked)
VALUES (?,?,?,?,?,?,?,?,?)
ON CONFLICT(id) DO UPDATE SET name=excluded.name,
repository=COALESCE(excluded.repository,repository), url=COALESCE(excluded.url,url),
classification=COALESCE(excluded.classification,classification),
status=COALESCE(excluded.status,status), access_notes=COALESCE(excluded.access_notes,access_notes),
notes=COALESCE(excluded.notes,notes), last_checked=excluded.last_checked""",
(id, name, repository, url, classification, status, access_notes, notes, now()),
)
# ---------------------------------------------------------------- cartoons
CARTOON_FIELDS = ["granularity", "date_exact", "date_start", "date_end", "year", "date_is_estimate", "title",
"caption", "description", "publication", "syndicate", "medium", "dimensions", "signed_name",
"rights_text", "copyright_holder", "notes"]
SOURCE_FIELDS = ["repository", "collection_name", "box", "folder", "page", "record_url", "image_url",
"thumbnail_url", "access_level", "rights_url", "provenance", "acquisition_method"]
def _id_for(conn, table: str, name: str) -> int:
row = conn.execute(f"SELECT id FROM {table} WHERE name=?", (name,)).fetchone()
if row:
return row[0]
return conn.execute(f"INSERT INTO {table}(name) VALUES (?)", (name,)).lastrowid
def save_record(conn, rec, source_id: str) -> str:
"""Insert or update one cartoon + its cartoon_sources row. Returns 'added' | 'updated' | 'unchanged'."""
from .normalize import is_president # local import avoids a cycle
ts = now()
row = conn.execute("SELECT * FROM cartoons WHERE canonical_id=?", (rec.canonical_id,)).fetchone()
vals = {f: getattr(rec, f) for f in CARTOON_FIELDS}
vals["date_is_estimate"] = int(bool(vals["date_is_estimate"]))
if row is None:
cols = ["canonical_id", *CARTOON_FIELDS, "created_at", "updated_at"]
cid = conn.execute(
f"INSERT INTO cartoons ({','.join(cols)}) VALUES ({','.join('?' * len(cols))})",
[rec.canonical_id, *vals.values(), ts, ts],
).lastrowid
status = "added"
else:
cid = row["id"]
# only fill/overwrite with non-empty new values (never blank out richer data)
changes = {k: v for k, v in vals.items() if v not in (None, "") and row[k] != v}
if changes:
sets = ",".join(f"{k}=?" for k in changes)
conn.execute(f"UPDATE cartoons SET {sets}, updated_at=? WHERE id=?", [*changes.values(), ts, cid])
status = "updated"
else:
status = "unchanged"
# source row
svals = {f: getattr(rec, f) for f in SOURCE_FIELDS}
if not svals["acquisition_method"]:
from .provenance import derive
svals["acquisition_method"] = derive(source_id, rec.provenance)
ex = conn.execute("SELECT id FROM cartoon_sources WHERE source_id=? AND identifier=?",
(source_id, rec.identifier)).fetchone()
if ex:
conn.execute(f"UPDATE cartoon_sources SET cartoon_id=?, {','.join(f'{k}=?' for k in svals)}, retrieved_at=? WHERE id=?",
[cid, *svals.values(), ts, ex[0]])
else:
cols = ["cartoon_id", "source_id", "identifier", *SOURCE_FIELDS, "retrieved_at"]
conn.execute(f"INSERT INTO cartoon_sources ({','.join(cols)}) VALUES ({','.join('?' * len(cols))})",
[cid, source_id, rec.identifier, *svals.values(), ts])
for s in rec.subjects or []:
s = s.strip()
if s:
conn.execute("INSERT OR IGNORE INTO cartoon_subjects VALUES (?,?)", (cid, _id_for(conn, "subjects", s)))
for p in rec.people or []:
p = p.strip()
if p:
pid = _id_for(conn, "people", p)
if is_president(p):
conn.execute("UPDATE people SET is_president=1 WHERE id=?", (pid,))
conn.execute("INSERT OR IGNORE INTO cartoon_people VALUES (?,?)", (cid, pid))
if rec.publication:
conn.execute("INSERT OR IGNORE INTO publications(name) VALUES (?)", (rec.publication,))
return status
# ---------------------------------------------------------------- crawl runs / errors
def start_run(conn, source: str) -> int:
rid = conn.execute("INSERT INTO crawl_runs(source,started,status) VALUES (?,?,?)", (source, now(), "running")).lastrowid
conn.commit()
return rid
def finish_run(conn, run_id: int, stats: dict, status: str, notes: str | None = None) -> None:
conn.execute(
"""UPDATE crawl_runs SET completed=?, pages_scanned=?, records_seen=?, records_added=?, records_updated=?,
errors=?, status=?, notes=? WHERE id=?""",
(now(), stats.get("pages", 0), stats.get("seen", 0), stats.get("added", 0), stats.get("updated", 0),
stats.get("errors", 0), status, notes, run_id),
)
conn.commit()
def log_error(conn, run_id: int | None, source: str, url: str | None, error: str) -> None:
conn.execute("INSERT INTO errors(run_id,source,url,error,at) VALUES (?,?,?,?,?)", (run_id, source, url, error[:2000], now()))
conn.commit()
def dump_json(obj) -> str:
return json.dumps(obj, ensure_ascii=False, indent=2, default=str)
def refresh_index(conn) -> int:
"""Rebuild cartoon_index: one row per canonical cartoon with aggregated repositories / people / first link-out,
plus a lowercased search_text blob (title, caption, description, publication, people, subjects, repositories,
collection names — across the canonical AND every record merged into it) that the viewer's q= box matches."""
from collections import defaultdict
repos, n, url = defaultdict(list), defaultdict(int), {}
text: dict[int, list[str]] = defaultdict(list)
for canon, repo, coll, rurl, sid in conn.execute(
"""SELECT COALESCE(k.merged_into,k.id), x.repository, x.collection_name, x.record_url, x.source_id
FROM cartoon_sources x JOIN cartoons k ON k.id=x.cartoon_id ORDER BY x.source_id LIKE 'seed%', x.source_id"""):
n[canon] += 1
if repo and repo not in repos[canon]:
repos[canon].append(repo)
text[canon] += [repo or "", coll or ""]
if rurl and canon not in url:
url[canon] = rurl
for canon, *vals in conn.execute(
"SELECT COALESCE(merged_into,id), title, caption, description, publication FROM cartoons"):
text[canon] += [v or "" for v in vals]
people = defaultdict(list)
for cid, canon, name in conn.execute(
"""SELECT cp.cartoon_id, COALESCE(k.merged_into,k.id), p.name FROM cartoon_people cp
JOIN people p ON p.id=cp.person_id JOIN cartoons k ON k.id=cp.cartoon_id"""):
people[cid].append(name)
text[canon].append(name)
for canon, name in conn.execute(
"""SELECT COALESCE(k.merged_into,k.id), s.name FROM cartoon_subjects cs
JOIN subjects s ON s.id=cs.subject_id JOIN cartoons k ON k.id=cs.cartoon_id"""):
text[canon].append(name)
ids = [r[0] for r in conn.execute("SELECT id FROM cartoons WHERE merged_into IS NULL")]
def blob(i: int) -> str:
seen, out = set(), []
for t in text.get(i, []):
t = " ".join(t.lower().split())
if t and t not in seen:
seen.add(t)
out.append(t)
return " | ".join(out)
conn.execute("DELETE FROM cartoon_index")
conn.executemany("INSERT INTO cartoon_index (id, repos, n_sources, people, record_url, search_text) VALUES (?,?,?,?,?,?)",
[(i, ",".join(repos[i]) or None, n[i], "; ".join(people[i]) or None, url.get(i), blob(i)) for i in ids])
conn.commit()
return len(ids)