← back to Paul Conrad Archive
src/conrad/crawlers/seed_import.py
235 lines
"""Seed importer: ingests the READ-ONLY research corpus built under TK-12179
(~/Projects/paul-conrad-cartoons/research). Every record carries provenance 'seed:<file>'.
Nothing is ever written to the seed directory."""
from __future__ import annotations
import json
import re
from pathlib import Path
from .. import config, rights
from ..models import CartoonRecord
from ..normalize import (GENERIC_SUBJECTS, clean_subject, parse_date, person_from_heading, presidents_in_text)
from .base import Crawler
OAC_URL = "https://oac.cdlib.org/findaid/ark:/13030/c8z03dxd/"
WSU_OBJ = "https://archivesspace.wichita.edu/repositories/3/archival_objects/"
SYR_URL = "https://library.syracuse.edu/digital/guides/c/conrad_p.htm"
def read_json(p: Path):
return json.loads(p.read_text())
def huntington_slot_records(slots: list[dict], provenance: str) -> list[CartoonRecord]:
"""One box_range record per CON folder number. Dates are the BOX span (range-level, not item-level)."""
out = []
txt, url = rights.rights_for("Huntington Library")
for s in slots:
con = s["con_number"]
est = s.get("date_est")
out.append(CartoonRecord(
canonical_id=f"hunt:CON-{con}", identifier=f"CON {con}", granularity="box_range",
title=None, date_start=s["date_start"], date_end=s["date_end"],
year=int(est[:4]) if est else int(s["date_start"][:4]), date_is_estimate=True,
description=f"Original drawing slot CON {con} in Box {s['box']} ({s['series']}); box span "
f"{s['date_start']} to {s['date_end']}. Title not cataloged in the public finding aid.",
notes=f"range-level slot; interpolated date_est={est} (linear within box span, NOT a cataloged date)",
publication="Los Angeles Times" if s["date_start"] >= "1964" and s["date_end"] <= "1993-12-31" else None,
medium="Original drawing", rights_text=txt, rights_url=url,
repository="Huntington Library", collection_name="Paul Conrad Papers (mssCON 1-12360)",
box=str(s["box"]), folder=f"CON {con}", record_url=s.get("source_url") or OAC_URL,
access_level=rights.ARCHIVE_VISIT, provenance=provenance,
))
return out
def loc_record(r: dict, provenance: str) -> CartoonRecord | None:
creator = (r.get("creator") or "")
if not re.search(r"conrad,\s*paul", creator, re.I):
return None
pk = str(r.get("pk") or r.get("id"))
d = parse_date(r.get("created_published_date"))
subj_raw = r.get("subjects") or []
subjects, people = [], []
for s in subj_raw:
c = clean_subject(s.replace("--Quotations", ""))
if not c or GENERIC_SUBJECTS.search(c):
continue
p = person_from_heading(s)
(people if p else subjects).append(p or c)
people += presidents_in_text(r.get("title"))
img = r.get("image") or {}
digitized = img.get("full") and "notdigitized" not in (img.get("full") or "")
txt, rurl = rights.rights_for("Library of Congress")
return CartoonRecord(
canonical_id=f"loc:{pk}", identifier=pk, granularity="item", title=r.get("title"),
date_exact=d["date_exact"], date_start=d["date_start"], date_end=d["date_end"], year=d["year"],
date_is_estimate=d["estimate"], medium=r.get("medium_brief") or r.get("medium"),
publication=_pub_from_call(r.get("call_number"), d["year"]),
rights_text=txt, rights_url=rurl, repository="Library of Congress",
collection_name="Prints & Photographs Division", folder=r.get("call_number"),
record_url=(r.get("links") or {}).get("item") or f"https://www.loc.gov/pictures/item/{pk}/",
image_url=img.get("full") if digitized else None, thumbnail_url=img.get("thumb") if digitized else None,
access_level=rights.ONLINE_IMAGE if digitized else rights.ONLINE_METADATA,
provenance=provenance, subjects=subjects, people=sorted(set(people)),
)
def _pub_from_call(call: str | None, year: int | None) -> str | None:
if year and 1950 <= year <= 1963:
return "Denver Post"
if year and 1964 <= year <= 1993:
return "Los Angeles Times"
if year and 1994 <= year <= 2010:
return "Los Angeles Times Syndicate / Tribune Media Services"
return None
def parse_wsu_md(md: str, obj_id: str) -> dict | None:
h = (re.search(r"^Title:\s*(.+?)\s*\|\s*Wichita State University", md, re.M) or [None, ""])[1]
m = re.match(r"^(.*),\s*(\d{4}(?:-\d{2}(?:-\d{2})?)?|undated)$", h)
title = (m[1] if m else h).strip()
date = m[2] if m and m[2] != "undated" else None
scope = re.search(r"## Scope and Contents\s+([\s\S]*?)\n##", md)
box = re.search(r"File\s*[—-]\s*Box:\s*(\w+)", md)
series = re.search(r"\[(Series \d+ -- [^\]]+?)\]\(https://archivesspace", md)
return {"id": obj_id, "title": title, "date": date, "scope": re.sub(r"\s+", " ", scope[1]).strip() if scope else None,
"box": box[1] if box else None, "series": series[1] if series else None}
def wsu_record(it: dict, provenance: str) -> CartoonRecord:
d = parse_date(it["date"])
untitled = re.fullmatch(r"untitled", it["title"] or "", re.I) is not None
title = None if untitled else it["title"].strip('"').strip()
subjects = [s.strip() for s in re.split(r",\s*", (it.get("scope") or "").rstrip(".")) if s.strip()]
people = presidents_in_text(" ".join([it["title"] or "", it.get("scope") or ""]))
txt, rurl = rights.rights_for("Wichita State University")
return CartoonRecord(
canonical_id=f"wsu:{it['id']}", identifier=it["id"], granularity="item", title=title,
caption=it["title"] if (it["title"] or "").startswith('"') else None,
description=it.get("scope"), date_exact=d["date_exact"], date_start=d["date_start"], date_end=d["date_end"],
year=d["year"], publication=_pub_from_call(None, d["year"]), rights_text=txt, rights_url=rurl,
repository="Wichita State University", collection_name="Cartoon Collection of Paul Conrad (MS 90-18)",
box=it.get("box"), record_url=WSU_OBJ + it["id"], access_level=rights.ARCHIVE_VISIT,
provenance=provenance, subjects=subjects, people=people,
notes=None if it.get("series") else "title/date from a sibling page's collection-organization listing",
)
def syracuse_folder_records(syr: dict, provenance: str) -> list[CartoonRecord]:
heads_by_folder: dict[str, list[str]] = {}
for h in syr["headings"]:
for f in h["folders"]:
heads_by_folder.setdefault(f, []).append(h["heading"])
txt, rurl = rights.rights_for("Syracuse University")
out = []
for f in syr["folders"]:
label = f["label"]
d = parse_date(label)
heads = heads_by_folder.get(f["folder"], [])
people = sorted({p for p in (person_from_heading(h) for h in heads) if p})
subjects = sorted({h for h in heads if not person_from_heading(h)})
out.append(CartoonRecord(
canonical_id=f"syr:{f['folder']}", identifier=f["folder"], granularity="folder",
title=f"Cartoons, {label} (folder {f['folder']})", date_start=d["date_start"], date_end=d["date_end"],
year=d["year"] if d["year"] else None,
description=f"Syracuse folder {f['folder']}: original cartoons dated {label}. Folder-level record — "
f"the finding aid indexes {len(heads)} subject/name headings to this folder but does not "
f"list individual cartoons.",
publication="Los Angeles Times" if (d["year"] or 0) >= 1964 else None,
rights_text=txt, rights_url=rurl, repository="Syracuse University",
collection_name="Paul Conrad Cartoons, [1944?]-1986", box=f["folder"].split("F")[0],
folder=f["folder"], record_url=f"{SYR_URL}#{f['folder']}", access_level=rights.ARCHIVE_VISIT,
provenance=provenance, subjects=subjects, people=people,
))
return out
def catalog_record(r: dict, provenance: str) -> CartoonRecord | None:
if r.get("record_type") != "documented" or r["id"].startswith("pulitzer-"):
return None # year-of-work awards / collection summaries are not individual cartoons
d = parse_date(r.get("date"))
return CartoonRecord(
canonical_id=f"cat:{r['id']}", identifier=r["id"], granularity="item", title=r["title"],
description=r.get("subject"), date_exact=d["date_exact"], date_start=d["date_start"], date_end=d["date_end"],
year=d["year"], date_is_estimate=d["estimate"], publication=r.get("publication"),
rights_text=rights.COPYRIGHT_NOTE, repository="Secondary citation",
collection_name="Documented in published secondary sources", record_url=r.get("citation_url"),
access_level=rights.ONLINE_METADATA, provenance=provenance,
people=presidents_in_text(" ".join([r.get("title") or "", r.get("subject") or ""])),
subjects=list(r.get("topics") or []),
notes="secondary-source citation (obituary/encyclopedia/LOC page); image intentionally not carried over",
)
class SeedImport(Crawler):
source_id = "seed"
name = "Seed corpus (TK-12179, read-only)"
repository = "multiple"
url = str(config.SEED_RESEARCH)
classification = "METADATA_ONLY"
access_notes = "Local cached metadata built by paul-conrad-cartoons/scripts/build-corpus.mjs; read-only."
SUBSOURCES = {
"seed_huntington": ("Seed: Huntington EAD slots (corpus.json)", "Huntington Library", OAC_URL, "PHYSICAL_ARCHIVE"),
"seed_loc": ("Seed: LOC P&P search cache (raw/loc-page-1.json)", "Library of Congress",
"https://www.loc.gov/pictures/search/?q=Conrad%2C%20Paul", "PUBLIC_API"),
"seed_wichita": ("Seed: Wichita State ArchivesSpace cache (raw/wsu/*.md via r.jina.ai)",
"Wichita State University", "https://archivesspace.wichita.edu/repositories/3/resources/166",
"PHYSICAL_ARCHIVE"),
"seed_syracuse": ("Seed: Syracuse finding-aid cache (raw/syr.html, syr.md)", "Syracuse University", SYR_URL,
"PHYSICAL_ARCHIVE"),
"seed_catalog": ("Seed: hand-built secondary-source catalog (data/cartoons.json)", "Secondary citation",
str(config.SEED_ROOT / "data/cartoons.json"), "METADATA_ONLY"),
}
def crawl(self) -> None:
R = config.SEED_RESEARCH
from .. import db
for sid, (name, repo, url, cls) in self.SUBSOURCES.items():
db.upsert_source(self.conn, sid, name, repository=repo, url=url, classification=cls, status="seed_only",
access_notes="seed cache; provenance recorded per row")
corpus = read_json(R / "corpus.json")
self.stats["pages"] += 1
self._save_all(huntington_slot_records(corpus["huntington"], "seed:research/corpus.json#huntington"),
"seed_huntington")
raw = read_json(R / "raw/loc-page-1.json")
self.stats["pages"] += 1
self._save_all([x for x in (loc_record(r, "seed:research/raw/loc-page-1.json") for r in raw["results"]) if x],
"seed_loc")
items: dict[str, dict] = {}
for p in sorted((R / "raw/wsu").glob("*.md")):
md = p.read_text()
self.stats["pages"] += 1
it = parse_wsu_md(md, p.stem)
if it:
items[p.stem] = it
# sibling listing in the "Collection organization" sidebar
for m in re.finditer(r"\[([^\]\[]*), (\d{4}-\d{2}-\d{2}|undated)\]\(" + re.escape(WSU_OBJ) + r"(\d+)\)", md):
oid = m[3]
if oid not in items and m[2] != "undated":
items[oid] = {"id": oid, "title": m[1], "date": m[2], "scope": None, "box": None, "series": None}
self._save_all([wsu_record(it, "seed:research/raw/wsu/" + (k + ".md" if it.get("series") else "*.md sidebar"))
for k, it in sorted(items.items())], "seed_wichita")
self._save_all(syracuse_folder_records(corpus["syracuse"], "seed:research/corpus.json#syracuse"),
"seed_syracuse")
catp = config.SEED_ROOT / "data/cartoons.json"
if catp.exists():
cat = read_json(catp)
self._save_all([x for x in (catalog_record(r, "seed:data/cartoons.json") for r in cat["records"]) if x],
"seed_catalog")
self.notes.append(f"huntington_repairs={corpus.get('huntington_repairs')}")
def _save_all(self, recs, source_id):
from .. import db
for rec in recs:
self.stats["seen"] += 1
st = db.save_record(self.conn, rec, source_id)
if st in ("added", "updated"):
self.stats[st] += 1
self.conn.commit()
CRAWLER = SeedImport