← back to Paul Conrad Archive
src/conrad/crawlers/discovery.py
206 lines
"""Discovery pass: find and classify other repositories holding Conrad material.
Uses only free, robots-permitted APIs: DPLA (key from env/secrets), Smithsonian Open Access (api.data.gov key,
if present). Hosts that refuse automated access (Calisphere, HathiTrust catalog, OSU PastPerfect, Empire ADC)
are probed once and classified. Presidential libraries / WorldCat are listed as UNVERIFIED leads only.
Every discovered repository is written to data/sources.json by scripts/report.py.
"""
from __future__ import annotations
import re
from .. import config, db, rights
from ..models import CartoonRecord
from ..normalize import parse_date, presidents_in_text
from .base import Blocked, Crawler, Transient
CONRAD_RE = re.compile(r"conrad,\s*paul(?:,?\s*1924)?|paul conrad", re.I)
CARTOON_RE = re.compile(r"cartoon|caricature|drawing|editorial|los angeles times|denver post", re.I)
# Name-authority forms that pin THIS Paul Conrad (b. 27 Jun 1924): the NPG/Smithsonian form carries the birth date,
# LC form the life dates. A record whose creator matches one of these is kept without the cartoon-word test
# (NPG's Time-cover originals describe the Time donation, not the medium).
AUTHORITY_RE = re.compile(r"paul conrad, born 27 jun 1924|conrad,\s*paul,\s*1924", re.I)
PROBES = [ # (source_id, name, url, classification-if-blocked, note)
("calisphere", "Calisphere (UC Libraries)", "https://calisphere.org/search/?q=%22paul+conrad%22", "REQUIRES_PERMISSION",
"AWS WAF JS challenge for non-browser clients"),
("hathitrust", "HathiTrust catalog", "https://catalog.hathitrust.org/Search/Home?lookfor=%22Conrad%2C+Paul%2C+1924-2010%22",
"REQUIRES_PERMISSION", "robots.txt unreadable (403); treated as disallowed"),
("empire_adc", "Empire Archival Discovery Cooperative", "https://empireadc.org/search/catalog/nsyu_2642890",
"REQUIRES_PERMISSION", "AWS WAF JS challenge"),
("loc_lcib", "LOC Information Bulletin (Oct 1999) — Conrad gift article", "https://www.loc.gov/loc/lcib/9910/conrad.html",
"PUBLIC_HTML", "context article about Conrad's gift of drawings to LOC"),
]
LEADS = [ # listed, NOT verified by this crawler (pulitzer + nixon_library are probed by secondary_citations)
("worldcat", "WorldCat", "https://search.worldcat.org/search?q=au%3AConrad%2C+Paul%2C+1924-2010", "METADATA_ONLY",
"No keyless API; bibliographic lead only (not crawled)."),
("lbj_library", "LBJ Presidential Library", "https://www.lbjlibrary.org/", "PHYSICAL_ARCHIVE", "UNVERIFIED lead."),
("reagan_library", "Ronald Reagan Presidential Library", "https://www.reaganlibrary.gov/", "PHYSICAL_ARCHIVE",
"UNVERIFIED lead."),
]
class Discovery(Crawler):
source_id = "discovery"
name = "Discovery pass (DPLA, Smithsonian, probes)"
repository = "multiple"
url = "https://api.dp.la/v2/items"
classification = "PUBLIC_API"
def crawl(self) -> None:
self.refs: list[str] = []
# clear earlier non-cartoon rows written by a previous version of this pass (records are re-derived here)
self.dpla()
self.smithsonian()
for sid, name, url, cls, note in PROBES:
status = "worked"
try:
st, _ = self.http.get(url)
self.stats["pages"] += 1
if st != 200:
status = "failed"
self.error(url, f"HTTP {st}")
res_note = f"HTTP {st}"
except (Blocked, Transient) as e:
status, res_note = "blocked", str(e)
self.error(url, f"BLOCKED: {e}")
db.upsert_source(self.conn, sid, name, url=url, classification=cls if status == "blocked" else "PUBLIC_HTML",
status=status, notes=f"{note}; probe: {res_note}")
if self.refs:
self.notes.append("non-cartoon references: " + " || ".join(self.refs[:20]))
for sid, name, url, cls, note in LEADS:
db.upsert_source(self.conn, sid, name, url=url, classification=cls, status="not_attempted", notes=note)
self.conn.commit()
# ------------------------------------------------------------------ DPLA
def dpla(self) -> None:
key = config.secret("DPLA_API_KEY")
if not key:
db.upsert_source(self.conn, "dpla", "DPLA", url=self.url, classification="PUBLIC_API", status="not_attempted",
notes="no DPLA_API_KEY available")
return
db.upsert_source(self.conn, "dpla", "Digital Public Library of America (API)", url="https://dp.la/",
classification="PUBLIC_API", status="running")
providers: dict[str, int] = {}
kept = 0
queries = [{"q": '"paul conrad" cartoon'}, {"q": '"conrad, paul"'}, {"q": '"paul conrad" los angeles times'},
{"sourceResource.creator": '"Paul Conrad, born 27 Jun 1924"'}, {"sourceResource.creator": '"Conrad, Paul"'}]
for qp in queries:
q = " ".join(f"{k}={v}" for k, v in qp.items())
for page in range(1, 6):
try:
d = self.http.get_json("https://api.dp.la/v2/items", params={**qp, "page_size": 100, "page": page},
secret_params={"api_key": key})
except (Blocked, Transient) as e:
self.error("https://api.dp.la/v2/items", e)
break
self.stats["pages"] += 1
docs = d.get("docs") or []
for doc in docs:
sr = doc.get("sourceResource") or {}
creators = " ".join(sr.get("creator") or []) if isinstance(sr.get("creator"), list) else str(sr.get("creator") or "")
title = " ".join(sr.get("title") or []) if isinstance(sr.get("title"), list) else str(sr.get("title") or "")
subj = " ".join(s.get("name", "") for s in sr.get("subject") or [] if isinstance(s, dict))
blob = " ".join([creators, title, subj, " ".join(sr.get("description") or [] if isinstance(sr.get("description"), list) else [str(sr.get("description") or "")])])
if not ((CONRAD_RE.search(creators) and CARTOON_RE.search(blob)) or AUTHORITY_RE.search(creators)):
continue
types = sr.get("type") or []
types = types if isinstance(types, list) else [types]
if types and not ({"image", "physical object"} & {str(t).lower() for t in types}):
self.refs.append(f"DPLA non-image ref ({','.join(map(str, types))}): {title[:90]}")
continue
prov = (doc.get("dataProvider") or {})
prov = prov.get("name") if isinstance(prov, dict) else str(prov)
providers[prov] = providers.get(prov, 0) + 1
dt = parse_date((sr.get("date") or [{}])[0].get("displayDate") if isinstance(sr.get("date"), list) else None)
time_cover = "time magazine donated" in blob.lower() and "cover art" in blob.lower()
rec = CartoonRecord(
canonical_id=f"dpla:{doc['id']}", identifier=doc["id"], granularity="item", title=title or None,
publication="Time" if time_cover else None,
notes="original Time cover art (NPG Time Collection, donated by Time 1978)" if time_cover else None,
description=blob[:1000], date_exact=dt["date_exact"], date_start=dt["date_start"],
date_end=dt["date_end"], year=dt["year"], repository=prov, collection_name="via DPLA",
record_url=doc.get("isShownAt"), thumbnail_url=doc.get("object"),
access_level=rights.ONLINE_IMAGE if doc.get("object") else rights.ONLINE_METADATA,
rights_text=" ".join(sr.get("rights") or []) if isinstance(sr.get("rights"), list) else (sr.get("rights") or rights.COPYRIGHT_NOTE),
provenance=f"live:DPLA q={q}", people=presidents_in_text(title),
)
self.stats["seen"] += 1
st = db.save_record(self.conn, rec, "dpla")
if st in ("added", "updated"):
self.stats[st] += 1
kept += 1
if len(docs) < 100:
break
db.upsert_source(self.conn, "dpla", "Digital Public Library of America (API)", url="https://dp.la/",
classification="PUBLIC_API", status="worked",
notes=f"kept {kept} Conrad-cartoon records; providers={providers}")
self.conn.commit()
# ------------------------------------------------------------ Smithsonian
def smithsonian(self) -> None:
key = config.secret("SI_API_KEY")
if not key:
db.upsert_source(self.conn, "smithsonian", "Smithsonian Open Access", url="https://api.si.edu/",
classification="PUBLIC_API", status="not_attempted", notes="no SI_API_KEY")
return
db.upsert_source(self.conn, "smithsonian", "Smithsonian Open Access API", url="https://api.si.edu/openaccess/",
classification="PUBLIC_API", status="running")
units: dict[str, int] = {}
kept = 0
for q in ('"paul conrad"', 'conrad cartoon'):
try:
d = self.http.get_json("https://api.si.edu/openaccess/api/v1.0/search",
params={"q": q, "rows": 100}, secret_params={"api_key": key})
except (Blocked, Transient) as e:
self.error("https://api.si.edu/openaccess/api/v1.0/search", e)
continue
self.stats["pages"] += 1
for row in (d.get("response") or {}).get("rows") or []:
c = row.get("content") or {}
blob = str(c)[:20000]
if not (CONRAD_RE.search(blob) and CARTOON_RE.search(blob)):
continue
if re.search(r"conrad,\s*(?!paul)", row.get("title", ""), re.I):
continue
unit = row.get("unitCode") or "SI"
units[unit] = units.get(unit, 0) + 1
if unit.startswith("SIL"): # Smithsonian Libraries: books / artist files, not cartoon objects
t = row.get("title") or ""
if re.match(r"pro and conrad", t, re.I):
self.conn.execute("UPDATE books SET verified=1, notes=COALESCE(notes,'') || ? WHERE title='Pro and Conrad'",
(f" | held by Smithsonian Libraries ({(c.get('descriptiveNonRepeating') or {}).get('record_link') or row.get('id')})",))
self.refs.append(f"SI {unit} library record: {t[:90]}")
continue
dnr = c.get("descriptiveNonRepeating") or {}
title = row.get("title")
is_folder = bool(re.search(r"\[folder\]", title or "", re.I))
ind = (c.get("indexedStructured") or {})
dates = ind.get("date") or []
dt = parse_date(dates[0] if dates else None)
media = ((dnr.get("online_media") or {}).get("media") or [{}])[0]
rec = CartoonRecord(
canonical_id=f"si:{row.get('id')}", identifier=row.get("id"),
granularity="folder" if is_folder else "item", title=title, date_start=dt["date_start"],
date_end=dt["date_end"], date_exact=dt["date_exact"], year=dt["year"],
repository=f"Smithsonian ({dnr.get('data_source') or unit})", collection_name=unit,
record_url=dnr.get("record_link") or dnr.get("guid"),
thumbnail_url=media.get("thumbnail"), image_url=media.get("content"),
access_level=rights.ONLINE_IMAGE if media.get("thumbnail") else rights.ONLINE_METADATA,
rights_text=((c.get("freetext") or {}).get("creditLine") or [{}])[0].get("content") or rights.COPYRIGHT_NOTE,
provenance=f"live:api.si.edu q={q}",
notes="artist/library file folder, not a cartoon" if is_folder else None,
people=presidents_in_text(title),
)
self.stats["seen"] += 1
st = db.save_record(self.conn, rec, "smithsonian")
if st in ("added", "updated"):
self.stats[st] += 1
kept += 1
db.upsert_source(self.conn, "smithsonian", "Smithsonian Open Access API", url="https://api.si.edu/openaccess/",
classification="PUBLIC_API", status="worked", notes=f"kept {kept} records; units={units}")
self.conn.commit()
CRAWLER = Discovery