← back to Paul Conrad Archive
src/conrad/exports.py
191 lines
"""CSV/JSON exports, data/sources.json, REPORT.md and research-gap analysis."""
from __future__ import annotations
import csv
import json
from collections import defaultdict
from pathlib import Path
from . import config, db
from .provenance import flags as provenance_flags
from .rights import PUBLIC_IMAGE_LEVELS, REQUIRES_ARCHIVE
REPO_COLS = [("Huntington", "Huntington Library"), ("Syracuse", "Syracuse University"), ("LOC", "Library of Congress"),
("Ohio State", "Ohio State University"), ("Wichita", "Wichita State University")]
def _w(path: Path, header: list[str], rows) -> int:
n = 0
with path.open("w", newline="") as f:
w = csv.writer(f)
w.writerow(header)
for r in rows:
w.writerow(r)
n += 1
return n
def canonical_rows(conn) -> list[dict]:
"""One row per canonical cartoon (merged_into IS NULL) with every source that points at it."""
rows = [dict(r) for r in conn.execute("SELECT * FROM cartoons WHERE merged_into IS NULL ORDER BY COALESCE(date_start,'9999'), id")]
members = defaultdict(list)
for r in conn.execute("SELECT id, COALESCE(merged_into,id) AS canon FROM cartoons"):
members[r["canon"]].append(r["id"])
srcs = defaultdict(list)
for r in conn.execute("SELECT * FROM cartoon_sources ORDER BY source_id"):
srcs[r["cartoon_id"]].append(dict(r))
subj, ppl = defaultdict(set), defaultdict(set)
for cid, n in conn.execute("SELECT cartoon_id, s.name FROM cartoon_subjects JOIN subjects s ON s.id=subject_id"):
subj[cid].add(n)
for cid, n in conn.execute("SELECT cartoon_id, p.name FROM cartoon_people JOIN people p ON p.id=person_id"):
ppl[cid].add(n)
for r in rows:
ids = members[r["id"]]
r["sources"] = [s for i in ids for s in srcs[i]]
r["subjects"] = sorted({x for i in ids for x in subj[i]})
r["people"] = sorted({x for i in ids for x in ppl[i]})
r["repositories"] = sorted({s["repository"] for s in r["sources"] if s["repository"]})
r["merged_ids"] = [i for i in ids if i != r["id"]]
r["acquisition_methods"] = sorted({s.get("acquisition_method") for s in r["sources"] if s.get("acquisition_method")})
r["provenance_flags"] = provenance_flags(r["acquisition_methods"])
return rows
def export_all(conn) -> dict:
ex = config.EXPORT_DIR
rows = canonical_rows(conn)
counts = {}
counts["cartoons.csv"] = _w(ex / "cartoons.csv",
["id", "canonical_id", "granularity", "title", "date_exact", "date_start", "date_end", "year", "date_is_estimate",
"publication", "repositories", "identifiers", "record_urls", "public_image_links", "people", "subjects",
"rights_text", "notes", "acquisition_methods", "provenance_flags"],
([r["id"], r["canonical_id"], r["granularity"], r["title"], r["date_exact"], r["date_start"], r["date_end"],
r["year"], r["date_is_estimate"], r["publication"], "; ".join(r["repositories"]),
"; ".join(f"{s['source_id']}:{s['identifier']}" for s in r["sources"]),
"; ".join(sorted({s["record_url"] for s in r["sources"] if s["record_url"]})),
"; ".join(sorted({s["image_url"] or s["thumbnail_url"] for s in r["sources"]
if s["access_level"] in PUBLIC_IMAGE_LEVELS and (s["image_url"] or s["thumbnail_url"])})),
"; ".join(r["people"]), "; ".join(r["subjects"]), r["rights_text"], r["notes"],
"; ".join(r["acquisition_methods"]), "; ".join(r["provenance_flags"])] for r in rows))
(ex / "cartoons.json").write_text(json.dumps(rows, ensure_ascii=False, indent=1, default=str))
counts["cartoons.json"] = len(rows)
counts["sources.csv"] = _w(ex / "sources.csv",
["cartoon_id", "canonical_id", "source_id", "repository", "collection_name", "identifier", "box", "folder",
"page", "record_url", "image_url", "thumbnail_url", "local_image", "access_level", "rights_url", "provenance",
"retrieved_at", "acquisition_method"],
conn.execute("""SELECT cs.cartoon_id, c.canonical_id, cs.source_id, cs.repository, cs.collection_name, cs.identifier,
cs.box, cs.folder, cs.page, cs.record_url, cs.image_url, cs.thumbnail_url, cs.local_image,
cs.access_level, cs.rights_url, cs.provenance, cs.retrieved_at, cs.acquisition_method
FROM cartoon_sources cs JOIN cartoons c ON c.id=cs.cartoon_id ORDER BY cs.source_id, cs.identifier"""))
counts["duplicates.csv"] = _w(ex / "duplicates.csv",
["relation", "score", "a_id", "a_canonical", "a_title", "a_date", "b_id", "b_canonical", "b_title", "b_date", "detail",
"action"],
([r[0], r[1], r[2], r[3], r[4], r[5], r[6], r[7], r[8], r[9], r[10],
"auto-linked (merged_into)" if r[0] == "duplicate_of" else "MANUAL REVIEW"]
for r in conn.execute("""SELECT l.relation, l.score, a.id, a.canonical_id, a.title, COALESCE(a.date_exact,a.year),
b.id, b.canonical_id, b.title, COALESCE(b.date_exact,b.year), l.detail
FROM cartoon_links l JOIN cartoons a ON a.id=l.cartoon_id JOIN cartoons b ON b.id=l.related_id
WHERE l.relation IN ('duplicate_of','possible_duplicate') ORDER BY l.relation DESC, l.score DESC""")))
table = per_year(conn)
counts["missing_years.csv"] = _w(ex / "missing_years.csv", ["year", "item_level", "box_range_slots", "folder_records", "status"],
([y, t["item"], t["box_range"], t["folder"],
"NO RECORDS" if not (t["item"] or t["box_range"] or t["folder"]) else
"no item-level records" if not t["item"] else "thin (<5 items)" if t["item"] < 5 else "ok"]
for y, t in table.items() if not t["item"] or t["item"] < 5))
cov = source_coverage(conn)
counts["source_coverage.csv"] = _w(ex / "source_coverage.csv",
["source_id", "name", "repository", "classification", "status", "cartoon_source_rows", "item", "folder",
"box_range", "public_image_links", "requires_archive_access", "last_run_pages", "last_run_errors", "notes"],
([c[k] for k in ["source_id", "name", "repository", "classification", "status", "rows", "item", "folder",
"box_range", "public_image", "archive", "pages", "errors", "notes"]] for c in cov))
gaps = research_gaps(conn)
counts["research_gaps.csv"] = _w(ex / "research_gaps.csv", ["gap_type", "key", "detail"], gaps)
write_sources_json(conn)
return counts
def per_year(conn) -> dict[int, dict]:
t = {y: defaultdict(int) for y in range(config.YEAR_MIN, config.YEAR_MAX + 1)}
repos = defaultdict(set)
for canon, repo in conn.execute("""SELECT COALESCE(k.merged_into,k.id), cs.repository FROM cartoon_sources cs
JOIN cartoons k ON k.id=cs.cartoon_id"""):
repos[canon].add(repo)
for r in conn.execute("""SELECT id, year, granularity, publication, title, date_exact FROM cartoons
WHERE merged_into IS NULL AND year BETWEEN ? AND ?""", (config.YEAR_MIN, config.YEAR_MAX)):
row = t[r["year"]]
row[r["granularity"]] += 1
rs = repos[r["id"]]
for label, repo in REPO_COLS:
if repo in rs:
row[label] += 1
pub = r["publication"] or ""
if pub.startswith("Los Angeles Times") or pub.startswith("LA Times"):
row["LA Times"] += 1
if pub == "Denver Post":
row["Denver Post"] += 1
if "University of Iowa Libraries" in rs:
row["Iowa"] += 1
if not ({repo for _, repo in REPO_COLS} & rs):
row["other"] += 1
if r["granularity"] == "item" and (not r["title"] or not r["date_exact"]):
row["missing_meta"] += 1
return t
def source_coverage(conn) -> list[dict]:
out = []
for s in conn.execute("SELECT * FROM sources ORDER BY id"):
agg = conn.execute("""SELECT COUNT(*) n, SUM(c.granularity='item') i, SUM(c.granularity='folder') f,
SUM(c.granularity='box_range') b, SUM(cs.access_level='online_image') pi,
SUM(cs.access_level='archive_visit') av
FROM cartoon_sources cs JOIN cartoons c ON c.id=cs.cartoon_id WHERE cs.source_id=?""",
(s["id"],)).fetchone()
run = conn.execute("SELECT pages_scanned, errors FROM crawl_runs WHERE source=? ORDER BY id DESC LIMIT 1",
(s["id"],)).fetchone()
out.append({"source_id": s["id"], "name": s["name"], "repository": s["repository"],
"classification": s["classification"], "status": s["status"], "rows": agg["n"] or 0,
"item": agg["i"] or 0, "folder": agg["f"] or 0, "box_range": agg["b"] or 0,
"public_image": agg["pi"] or 0, "archive": agg["av"] or 0,
"pages": run["pages_scanned"] if run else None, "errors": run["errors"] if run else None,
"notes": s["notes"], "url": s["url"], "access_notes": s["access_notes"]})
return out
def write_sources_json(conn) -> None:
cov = source_coverage(conn)
config.SOURCES_JSON.write_text(json.dumps({"generated_at": db.now(), "classification_vocabulary": [
"PUBLIC_API", "PUBLIC_HTML", "PUBLIC_IMAGE", "METADATA_ONLY", "PAYWALLED", "PHYSICAL_ARCHIVE",
"REQUIRES_PERMISSION"], "sources": cov}, indent=1, ensure_ascii=False))
def research_gaps(conn) -> list[tuple]:
gaps = []
t = per_year(conn)
for y, r in t.items():
if r["item"] < 5:
gaps.append(("thin_year", y, f"{r['item']} item-level, {r['box_range']} box-range slots, {r['folder']} folder records"))
n = conn.execute("SELECT COUNT(*) FROM cartoons c JOIN cartoon_sources cs ON cs.cartoon_id=c.id AND cs.source_id='huntington' "
"WHERE c.title IS NULL").fetchone()[0]
gaps.append(("huntington_untitled", "all", f"{n} Huntington CON slots have no title in the public finding aid "
"(internal Huntington database needed)"))
for r in conn.execute("""SELECT cs.box, MIN(c.date_start), MAX(c.date_end), COUNT(*) FROM cartoons c JOIN cartoon_sources cs
ON cs.cartoon_id=c.id AND cs.source_id='huntington' GROUP BY cs.box ORDER BY CAST(cs.box AS INT)"""):
gaps.append(("huntington_box_untitled", f"box {r[0]}", f"{r[3]} slots {r[1]}..{r[2]} without titles"))
for r in conn.execute("""SELECT cs.folder, c.title FROM cartoons c JOIN cartoon_sources cs ON cs.cartoon_id=c.id
AND cs.source_id='seed_syracuse' WHERE c.date_exact IS NULL ORDER BY cs.folder"""):
gaps.append(("syracuse_no_exact_date", r[0], r[1]))
for r in conn.execute("""SELECT c.canonical_id, c.title, c.date_exact, c.year FROM cartoons c
WHERE c.canonical_id LIKE 'loc:%' AND c.granularity='item' AND c.merged_into IS NULL
AND c.id NOT IN (SELECT cartoon_id FROM cartoon_links WHERE relation='contained_in_candidate'
AND detail LIKE 'Huntington%')"""):
gaps.append(("loc_not_in_huntington", r[0], f"{r[1]} ({r[2] or r[3]}) — no Huntington box span match "
"(pre-1967 Denver Post era, bare-year date, or not held there)"))
for lo, hi, label in ((1950, 1964, "denver_post_gap"), (1964, 1993, "latimes_gap"), (1993, 2010, "syndicated_gap")):
for y in range(lo, hi + 1):
r = t[y]
if r["item"] == 0:
gaps.append((label, y, f"no item-level record; {r['box_range']} Huntington slots, {r['folder']} Syracuse folders"))
for r in conn.execute("SELECT id, name, status, notes FROM sources WHERE status IN ('blocked','not_attempted','failed')"):
gaps.append(("source_unmeasured", r[0], f"{r[1]} — {r[2]}: {(r[3] or '')[:200]}"))
return gaps