← back to Paul Conrad Archive
src/conrad/crawlers/loc.py
302 lines
"""Library of Congress Prints & Photographs (deep enumeration, TK-12199 DTD pick A).
robots.txt disallows /search and /pictures/search (Crawl-delay 5), so the crawler does NOT page the
search UI. It uses (a) the documented loc.gov format endpoint /photos/?fa=contributor:...&fo=json, which
robots permits, exhausting its pagination, and (b) per-item JSON (/pictures/item/<id>/?fo=json) for every
known Conrad item id (seed + format endpoint) to verify and enrich each record. Images: metadata links only.
Deep enumeration (TK-12199): every permitted JSON path is robots-checked first and recorded; the format endpoints
(/photos/, /manuscripts/, /books/, /maps/, /collections/cartoon-drawings/) are exhausted with sp= pagination; and
the P&P catalog's sequential control numbers are probed around every known Conrad id through the permitted
/pictures/item/<id>/?fo=json endpoint (neighbour-LCCN probing, bounded by LOC_PROBE_WINDOW / LOC_PROBE_MAX,
Crawl-delay 5 honoured, checkpointed so a restart resumes). /pictures/related (the "neighbors" link) and
/pictures/search stay untouched because robots.txt disallows them.
"""
from __future__ import annotations
import os
import re
from .. import rights
from ..models import CartoonRecord
from ..normalize import GENERIC_SUBJECTS, clean_subject, parse_date, person_from_heading, presidents_in_text
from .base import Blocked, Checkpoint, Crawler, Transient
from .seed_import import _pub_from_call
LOC = "https://www.loc.gov"
FORMAT_QUERIES = [
("photos", {"fa": "contributor:conrad, paul"}),
("photos", {"q": "paul conrad editorial cartoon"}),
("photos", {"q": "conrad, paul, 1924-2010"}),
("photos", {"q": "conrad, paul"}),
("manuscripts", {"fa": "contributor:conrad, paul"}),
("manuscripts", {"q": "paul conrad cartoons"}),
("books", {"fa": "contributor:conrad, paul"}),
("maps", {"fa": "contributor:conrad, paul"}),
("collections/cartoon-drawings", {"fa": "contributor:conrad, paul"}),
("collections/cartoon-drawings", {"q": "conrad"}),
]
KNOWN_GROUPS = {"2010632868": "83 proofs of political cartoons for the Los Angeles Times (group record)"}
# ids surfaced by the seed /pictures/search cache whose creator field is empty but whose notes name Conrad
EXTRA_CANDIDATES = {"2010634752": "seed search cache: [Original editorial cartoon drawings], creator empty"}
ROBOTS_PATHS = ["/photos/", "/manuscripts/", "/books/", "/maps/", "/collections/cartoon-drawings/", "/item/1/",
"/pictures/item/1/", "/search/", "/pictures/search/", "/pictures/related/"]
PROBE_WINDOW = int(os.environ.get("LOC_PROBE_WINDOW", 5))
PROBE_MAX = int(os.environ.get("LOC_PROBE_MAX", 300))
GROUP_TITLE = re.compile(r"^\[?(proofs of|original editorial cartoon drawings|political cartoons by|cartoons\]?$)", re.I)
def _pk_from(url_or_id: str) -> str | None:
m = re.search(r"/(?:item|pictures/item)/([0-9a-z]+)/?", url_or_id or "")
return m[1] if m else None
def item_record(pk: str, d: dict, provenance: str) -> CartoonRecord | None:
it = d.get("item") or {}
creators = " ".join(c.get("title", "") for c in (it.get("creators") or []) if isinstance(c, dict))
creators += " " + " ".join(it.get("contributor_names") or [])
note_txt = " ".join((n.get("note") or "") if isinstance(n, dict) else str(n) for n in (it.get("notes") or []))
if not re.search(r"conrad,\s*paul", creators, re.I) and not re.search(r"creators include:[^.]*paul conrad", note_txt, re.I):
return None
title = it.get("title")
dt = parse_date(it.get("created_published_date") or it.get("created_published") or it.get("date"))
subjects, people = [], []
for s in it.get("subjects") or []:
s = s.get("title") if isinstance(s, dict) else s
if not s:
continue
p = person_from_heading(s)
c = clean_subject(s.replace("--Quotations", ""))
if p:
people.append(p)
elif c and not GENERIC_SUBJECTS.search(c):
subjects.append(c)
summary = it.get("summary") or ""
people += presidents_in_text(" ".join([title or "", summary]))
res = d.get("resources") or []
img = None
if res and isinstance(res[0], dict):
img = res[0].get("image") or res[0].get("large") or res[0].get("medium")
thumb = None
tg = it.get("thumb_gallery")
if isinstance(tg, str) and "notdigitized" not in tg:
thumb = tg
rights_info = it.get("rights_information") or ""
holder = None
m = re.search(r"Copyright\s+(\d{4}),?\s+([^.]+)", rights_info)
if m:
holder = re.sub(r"^The\s+", "", m[2].strip())
medium_txt = str(it.get("medium_brief") or it.get("medium") or "")
media = re.search(r"Media includes:\s*(\d+)\s+([A-Za-z]+)", note_txt) or re.match(r"\s*(\d+)\s+([A-Za-z]+)", medium_txt)
unprocessed = (it.get("call_number") or "").lower().startswith("unprocessed")
devised = (title or "").startswith("[") or "title devised by library staff" in note_txt.lower()
# a lot/group: known group, lot-style title, >1 declared pieces, or an unprocessed lot with a staff-devised title
# (an unprocessed SINGLE print with a transcribed title, e.g. 2023631806, stays an item)
group = pk in KNOWN_GROUPS or bool(GROUP_TITLE.search(title or "")) or bool(media and int(media[1]) > 1) or \
(unprocessed and devised and not (d.get("resources") or []))
pub = holder if holder and re.search(r"times|post", holder, re.I) else _pub_from_call(None, dt["year"])
return CartoonRecord(
canonical_id=f"loc:{pk}", identifier=pk, granularity="folder" if group else "item", title=title,
description=summary or None, date_exact=dt["date_exact"], date_start=dt["date_start"],
date_end=dt["date_end"], year=dt["year"], date_is_estimate=dt["estimate"],
medium=it.get("medium_brief") or (it.get("medium") if isinstance(it.get("medium"), str) else None),
publication=pub, rights_text=rights_info or rights.rights_for("Library of Congress")[0],
copyright_holder=holder, repository="Library of Congress",
collection_name=", ".join(c.get("title", "") for c in (it.get("collections") or []) if isinstance(c, dict))
or "Prints & Photographs Division", folder=it.get("call_number"),
record_url=f"{LOC}/pictures/item/{pk}/", image_url=img, thumbnail_url=thumb,
access_level=rights.ONLINE_IMAGE if (img or thumb) else rights.ONLINE_METADATA,
rights_url="https://www.loc.gov/rr/print/res/rights.html", provenance=provenance,
subjects=subjects, people=sorted(set(people)),
notes=KNOWN_GROUPS.get(pk) or (("group/lot record — not a single cartoon"
+ (f" ({media[1]} {media[2].lower()})" if media else "")
+ ("; unprocessed, no digitized components" if unprocessed else ""))
if group else None),
)
class LOCCrawler(Crawler):
source_id = "loc"
name = "Library of Congress — Prints & Photographs (JSON API)"
repository = "Library of Congress"
url = "https://www.loc.gov/photos/?fa=contributor:conrad,%20paul&fo=json"
classification = "PUBLIC_API"
access_notes = ("Item records online; many Conrad drawings are digitized but flagged 'May be restricted: Copyright "
"... Los Angeles Times' — view at loc.gov (link-out), reproduction needs permission.")
def _robots_audit(self) -> dict[str, bool]:
out = {}
for path in ROBOTS_PATHS:
out[path] = self.http.allowed(LOC + path)
self.notes.append("robots.txt: " + ", ".join(f"{p}={'allow' if ok else 'DISALLOW'}" for p, ok in out.items()))
return out
def crawl(self) -> None:
cp = Checkpoint("loc")
self.books: list[dict] = []
pks: dict[str, str] = {}
allowed = self._robots_audit()
per_query: dict[str, int] = {}
for fmt, params in FORMAT_QUERIES:
if not allowed.get(f"/{fmt}/", True):
self.notes.append(f"skipped /{fmt}/ (robots)")
continue
page = 1
max_pages = 50 if "fa" in params else 3 # keyword queries: stop early, results get irrelevant fast
while page <= max_pages:
try:
data = self.http.get_json(f"{LOC}/{fmt}/", params={**params, "fo": "json", "c": 100, "sp": page})
except (Blocked, Transient) as e:
self.error(f"{LOC}/{fmt}/ {params}", e)
break
self.stats["pages"] += 1
hits = 0
for r in data.get("results") or []:
contrib = " ".join(r.get("contributor") or [])
if fmt == "books" and "conrad, paul" in contrib.lower():
self.books.append(r)
continue
if "conrad, paul" in contrib.lower():
hits += 1
pk = _pk_from(r.get("id") or "")
if pk:
pks.setdefault(pk, f"live:{LOC}/{fmt}/?{params}")
per_query[f"/{fmt}/ {params}"] = per_query.get(f"/{fmt}/ {params}", 0) + hits
if not (data.get("pagination") or {}).get("next") or ("q" in params and not hits):
break
page += 1
self.notes.append("format endpoints (Conrad hits): " + "; ".join(f"{k}={v}" for k, v in per_query.items()))
found_live = set(pks)
# every Conrad pk already known from the seed cache gets verified individually
for (pk,) in self.conn.execute("SELECT identifier FROM cartoon_sources WHERE source_id='seed_loc'"):
pks.setdefault(pk, "live:item-json (seed id verification)")
for pk in KNOWN_GROUPS:
pks.setdefault(pk, "live:item-json (known group record)")
for pk, why in EXTRA_CANDIDATES.items():
pks.setdefault(pk, f"live:item-json ({why})")
done = set(cp.get("done", []))
conrad: set[str] = set()
self.groups: dict[str, dict] = {}
for pk, prov in sorted(pks.items()):
if self._verify(pk, prov):
conrad.add(pk)
done.add(pk)
if len(done) % 10 == 0:
self.conn.commit()
cp.set("done", sorted(done))
self.conn.commit()
cp.set("done", sorted(done))
new_from_probe = self._probe_neighbours(conrad, set(pks), cp)
self.notes.append(f"neighbour-LCCN probing (+/-{PROBE_WINDOW}, cap {PROBE_MAX}): {self.probe_stats}; "
f"new Conrad ids from probing: {sorted(new_from_probe) or 'none'}")
for pk, g in self.groups.items():
self.notes.append(f"group {pk}: {g}")
for b in self.books: # LOC catalog records for Conrad's own books -> verification of the bibliography
t = (b.get("title") or "").strip(" /")
self.conn.execute("INSERT OR IGNORE INTO collections(repository,name,identifier,url,level,notes,source_id) "
"VALUES (?,?,?,?,?,?,?)", ("Library of Congress", t[:300], _pk_from(b.get("id") or ""),
b.get("id"), "book", f"date={b.get('date')}", self.source_id))
self.conn.commit()
self.notes.append(f"LOC book records by Conrad: {len(self.books)}")
self.notes.append(f"format endpoints found {len(found_live)} Conrad ids; verified {len(conrad)} Conrad item JSON "
f"records; /search, /pictures/search, /pictures/related are robots-disallowed (not paged)")
# ------------------------------------------------------------------ helpers
def _verify(self, pk: str, prov: str) -> bool:
url = f"{LOC}/pictures/item/{pk}/"
try:
d = self.http.get_json(url, params={"fo": "json"})
except (Blocked, Transient) as e:
self.error(url, e)
return False
self.stats["pages"] += 1
rec = item_record(pk, d, prov + f" -> {url}?fo=json")
if not rec:
return False
self.save(rec)
if rec.granularity == "folder":
self.groups[pk] = group_components(pk, d)
return True
def _probe_neighbours(self, conrad: set[str], known: set[str], cp: Checkpoint) -> set[str]:
"""Probe sequential P&P control numbers around every Conrad id (permitted item endpoint only).
A hit extends the window from itself; misses (404 / non-Conrad) are checkpointed so restarts skip them."""
probed = dict(cp.get("probed", {})) # pk -> 'conrad' | 'other' | 'missing' | 'error'
self.probe_stats = {"requests": 0, "conrad": 0, "other": 0, "missing": 0, "error": 0, "cached_skip": 0}
queue = sorted(p for p in conrad if p.isdigit() and len(p) >= 8)
new: set[str] = set()
seen = set(known) | set(probed)
while queue and self.probe_stats["requests"] < PROBE_MAX:
base = queue.pop(0)
width = len(base)
for off in [o for k in range(1, PROBE_WINDOW + 1) for o in (k, -k)]:
cand = str(int(base) + off).zfill(width)
if cand in seen:
if probed.get(cand) == "conrad" and cand not in conrad:
conrad.add(cand)
new.add(cand)
self.probe_stats["cached_skip"] += 1 if cand in probed else 0
continue
if self.probe_stats["requests"] >= PROBE_MAX:
break
seen.add(cand)
url = f"{LOC}/pictures/item/{cand}/"
self.probe_stats["requests"] += 1
try:
status, body = self.http.get(url, params={"fo": "json"})
if status != 200:
probed[cand] = "missing"
else:
import json as _json
d = _json.loads(body)
rec = item_record(cand, d, f"live:neighbour-probe of {base} -> {url}?fo=json")
if rec:
self.save(rec)
if rec.granularity == "folder":
self.groups[cand] = group_components(cand, d)
probed[cand] = "conrad"
conrad.add(cand)
new.add(cand)
queue.append(cand) # extend the window from every new hit
else:
probed[cand] = "other"
except (Blocked, Transient, ValueError) as e:
probed[cand] = "error"
self.error(url, e)
self.probe_stats[probed[cand]] += 1
self.stats["pages"] += 1
if self.probe_stats["requests"] % 10 == 0:
self.conn.commit()
cp.set("probed", probed)
self.conn.commit()
cp.set("probed", probed)
from collections import Counter
self.probe_stats["cumulative_probed_ids"] = dict(Counter(probed.values()), total=len(probed))
if queue:
self.notes.append(f"probe budget exhausted with {len(queue)} ids still queued (raise LOC_PROBE_MAX)")
return new
def group_components(pk: str, d: dict) -> dict:
"""Are a group record's components individually addressable through permitted endpoints?"""
it = d.get("item") or {}
res = [r for r in (d.get("resources") or []) if isinstance(r, dict)]
files = sum(len(r.get("files") or []) for r in res)
rel = d.get("related") or {}
notes = " ".join((n.get("note") or "") for n in (it.get("notes") or []) if isinstance(n, dict))
m = re.search(r"Media includes:\s*(\d+)\s+([A-Za-z]+)", notes) or \
re.match(r"\s*(\d+)\s+([A-Za-z]+)", str(it.get("medium_brief") or it.get("medium") or ""))
out = {"call_number": it.get("call_number"), "digitized_resources": len(res), "resource_files": files,
"declared_components": f"{m[1]} {m[2].lower()}" if m else None,
"child_item_links": [x for x in (it.get("related_items") or []) if x],
"neighbors_link_disallowed": bool(rel.get("neighbors"))}
addressable = bool(out["child_item_links"])
out["components_addressable"] = addressable
out["why"] = ("child item records linked from the group" if addressable else
"no child item records, no digitized component files; the only component path is the "
"'neighbors' link under /pictures/related/, which robots.txt disallows")
return out
CRAWLER = LOCCrawler