← back to Paul Conrad Archive
src/conrad/crawlers/ia_fulltext.py
368 lines
"""Internet Archive FULL-TEXT (OCR) search for printed APPEARANCES of Paul Conrad cartoons — newspapers,
magazines and cartoon anthologies whose scanned pages carry a Conrad credit line
("PAUL CONRAD / Los Angeles Times / © Los Angeles Times Syndicate", "CONRAD © 1971 Los Angeles Times", ...).
How it works (robots-permitted, metadata only):
* archive.org/robots.txt disallows only /control/ and /report/; the public full-text search service
(`/services/search/beta/page_production/?service_backend=fts`) is queried year by year, 1950-2010, for several
query variants, and EVERY result page is walked (no sampling). Each hit is one scanned PAGE
(identifier + sub-file + page_num) with OCR highlight snippets.
* a hit becomes a record ONLY if an OCR snippet carries a Conrad CREDIT-LINE signature (see `classify`) — a short
line naming Conrad next to "Los Angeles Times" / "Times Syndicate" / "Denver Post" / "©". Articles merely
*about* Conrad, other Conrads (Joseph, Conrad Hilton, Conrad de Aenlle, ...) and bare byline noise are rejected
and counted, never saved.
* granularity 'item', but the record is an APPEARANCE of an (unidentified) cartoon on a dated page, not a titled
cartoon: title is a supplied, bracketed description "[Paul Conrad cartoon — <publication>, <date>, p. N]";
the OCR snippet is kept verbatim in notes (flagged "OCR, unverified"). Dedupe folds reprints of the same page.
* IMAGES are never downloaded. For items IA serves openly, the page image URL is taken from IA's own IIIF
presentation manifest (JSON) and stored as image_url on host iiif.archive.org for hotlinking; lending-library
(access-restricted) books get no image URL — their page is only viewable through IA's lending UI.
"""
from __future__ import annotations
import json
import time
import re
from urllib.parse import quote, urlencode
from .. import config, db, rights
from ..models import CartoonRecord
from ..normalize import presidents_in_text
from .base import Blocked, Checkpoint, Crawler, Transient
FTS = "https://archive.org/services/search/beta/page_production/"
YEARS = range(1950, 2011)
# query variants (coordinator/Steve, TK-12199). "Los Angeles Times Syndicate" alone is covered by V2: the FTS index
# is PAGE-level, so '"Conrad" "Los Angeles Times Syndicate"' is exactly "LATS pages whose text also carries Conrad".
VARIANTS = [
("paul_conrad", '"Paul Conrad"', YEARS),
("conrad_lats", '"Conrad" "Los Angeles Times Syndicate"', YEARS),
("conrad_lat", '"Conrad" "Los Angeles Times"', YEARS),
("conrad_denver", '"Conrad" "Denver Post"', range(1950, 1965)),
("conrad_cartoon", '"Conrad cartoon"', YEARS),
]
PAGE_SIZE = 100
MAX_PAGES = 40 # per (variant, year); hitting the cap is logged, never silent
HL = re.compile(r"\{\{\{(.*?)\}\}\}")
OTHER_CONRADS = re.compile(
r"joseph\s+conrad|conrad\s+(?:de\s*a|hilton|burns|black|lister|veidt|n\.? ?hilton|jr|adenauer|bain|dobler|"
r"birdie|richter|aiken|kent|jones)|(?:robert|kent|frank|charles|pete|william|andree|michael|chris|dr\.?|rev\.?|"
r"deacon|the rev)\s+conrad|conrad\s+international|conrad\s+hotel", re.I)
CREDIT_CONTEXT = re.compile(r"los\s*angeles\s*times|l\.\s?a\.\s?times|times\s*syndicate|denver\s*post|syndicate|©|\(c\)", re.I)
PAUL = re.compile(r"paul\s+conrad", re.I)
def _norm_line(s: str) -> str:
return re.sub(r"\s+", " ", HL.sub(r"\1", s)).strip()
AWARD = re.compile(r"prize|award|pulitzer|winner|best\s+cartoon|citation|honorable|finalist|nominat|\b(?:19|20)\d\d\s*[—–-]\s*paul"
r"|cartoons?\s*[—–-]|five cartoons a week|music|copyright entries|;\s*m\s+paul|joined the|work(?:ed)? at|\bmr\.", re.I)
STRONG = re.compile(r"syndicate|©|\(c\)|copyright|reprinted|courtesy|permission", re.I)
DASH_CREDIT = re.compile(r"^[—–-]+\s*(?:paul\s+)?conrad\s*[,.]?\s*(?:the\s+)?(?:los\s*angeles\s*times|denver\s*post)\b", re.I)
# prose credits that mark a REPRINTED cartoon (textbooks / anthologies / magazines)
PROSE_CREDITS = [
re.compile(r"source:\s*paul\s+conrad\s+in\s+the\s+(?:los\s+angeles\s+times|denver\s+post)", re.I),
re.compile(r"(?P<title>[A-Z][^./|]{3,60})\.\s*by\s+paul\s+conrad\.?,?\s*(?:the\s+)?(?:los\s+angeles\s+times|denver\s+post)",
re.I),
re.compile(r"(?:this|the above|the following|original artwork for this)\s+cartoon\s+by\s+paul\s+conrad", re.I),
re.compile(r"paul\s+conrad\s+cartoon\s+reprinted", re.I),
]
QUOTED_CAPTION = re.compile(r"[’”\"!?]\s*$")
SYNDICATE_AD = re.compile(r"syndicate|cartoons a week|editorial cartoons|features", re.I)
def classify_hit(fields: dict, snippets: list[str]) -> tuple[str, str | None]:
"""classify() plus the Editor & Publisher rule: E&P carried syndicate sales ADS for Conrad ('PAUL CONRAD / Five
cartoons a week', LA Times Syndicate) that usually reproduce a sample cartoon — recorded as kind 'syndicate_ad'."""
kind, ev = classify(snippets)
if kind == "credit" or not str(fields.get("identifier") or "").startswith("sim_editor-publisher"):
return kind, ev
for sn in snippets:
lines = sn.split("\n")
for i, raw in enumerate(lines):
line = _norm_line(raw)
if re.fullmatch(r"(?:paul\s+)?conrad[.,]?", line, re.I) and "{{{" in raw:
window = " ".join(_norm_line(x) for x in lines[max(0, i - 2): i + 3])
if SYNDICATE_AD.search(window) and not OTHER_CONRADS.search(window) \
and not re.search(r"prize|award|pulitzer|winner|joined|obituar|died", window, re.I):
return "syndicate_ad", window[:300]
return kind, ev
def classify(snippets: list[str]) -> tuple[str, str | None]:
"""-> (kind, evidence). kind: 'credit' (a printed Conrad cartoon credit), 'other_conrad', 'mention'.
A CREDIT line must (a) carry the highlighted Conrad match, (b) be a short credit/signature line (<= 45 chars, or
<= 90 when the line STARTS with the Conrad credit, e.g. 'PAUL CONRAD Los Angeles Times © LA Times Syndicate'),
(c) not be another Conrad, (d) not sit in an award / biography / roster context (prize lists, 'Five cartoons a
week', 'Mr. Conrad joined ...'), and (e) have STRONG credit evidence within itself and the next two lines
(Syndicate / © / copyright / reprinted / courtesy), or be a dash attribution under a quoted caption
('...HEALTH HAZARD!’ / —PAUL CONRAD, THE DENVER POST'). A bare 'Paul Conrad' in a list of names never counts."""
other = False
for sn in snippets:
flat = _norm_line(sn.replace("\n", " "))
if not OTHER_CONRADS.search(flat):
for rx in PROSE_CREDITS:
m = rx.search(flat)
if m:
return "credit", flat[max(0, m.start() - 60): m.end() + 60][:300]
lines = sn.split("\n")
for i, raw in enumerate(lines):
if not re.search(r"\{\{\{[^}]*conrad[^}]*\}\}\}", raw, re.I):
continue
line = _norm_line(raw)
window = " ".join(_norm_line(x) for x in lines[max(0, i - 1): i + 3])
if OTHER_CONRADS.search(window):
other = True
continue
if AWARD.search(window):
continue
lead = re.match(r"(?:[©@—–-]\s*)?(?:paul\s+)?conrad\b\W*(?:\(?c\)?\s*)?(?:the\s+)?(?:los\s*angeles\s*times|l\.\s?a\."
r"|denver\s*post|©|copyright|courtesy)", line, re.I)
if len(line) > 45 and not (lead and len(line) <= 90):
continue
nxt = " ".join(_norm_line(x) for x in lines[i: i + 3])
prev = _norm_line(lines[i - 1]) if i else ""
if (STRONG.search(nxt) and re.search(r"conrad", line, re.I) and len(re.sub(r"(?i)paul|conrad|\W", "", line)) < 40) \
or (DASH_CREDIT.match(line) and QUOTED_CAPTION.search(prev)):
return "credit", _norm_line(" / ".join(lines[max(0, i - 1): i + 3]))[:300]
return ("other_conrad" if other else "mention"), None
TITLE_RX = [re.compile(r"[“\"]([^”\"]{3,80})[”\"]\s*(?:was|is)\s+the\s+(?:\w+\s+)?title\s+of\s+this\s+cartoon", re.I),
re.compile(r"(?:\d+\s+|^)([A-Z][A-Za-z'’ ,-]{3,60}?)\.\s*By\s+Paul\s+Conrad\.?,?\s*(?:the\s+)?(?:Los\s+Angeles\s+Times|Denver\s+Post)")]
def cartoon_title(evidence: str | None) -> str | None:
"""A cartoon title only when the OCR text itself states one ('"X" was the title of this cartoon', a contents line
'X. By Paul Conrad. Los Angeles Times')."""
for rx in TITLE_RX:
m = rx.search(evidence or "")
if m:
t = m[1].strip(" ,;:")
words = t.split()
if len(words) > 1 and not re.search(r"\d{2,}", t):
return t
return None
MONTHS = {m: i for i, m in enumerate(["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov",
"dec"], 1)}
def issue_date(fields: dict) -> tuple[str | None, int | None]:
"""Best issue date for the scanned page: from the sub-file name ('Aug 11 1998, The Jerusalem Post ...'), the
identifier ('..._1978_11_01'), else the item date. Returns (ISO date or None, year)."""
for s in (fields.get("file_basename") or "", fields.get("identifier") or ""):
m = re.search(r"\b([A-Z][a-z]{2})[a-z]*\.? (\d{1,2}),? (19[4-9]\d|20[01]\d)\b", s)
if m and m[1].lower() in MONTHS:
return f"{m[3]}-{MONTHS[m[1].lower()]:02d}-{int(m[2]):02d}", int(m[3])
m = re.search(r"(19[4-9]\d|20[01]\d)[_-](\d{2})[_-](\d{2})", s)
if m and 1 <= int(m[2]) <= 12 and 1 <= int(m[3]) <= 31:
return f"{m[1]}-{m[2]}-{m[3]}", int(m[1])
d = str(fields.get("date") or "")[:10]
y = fields.get("year")
return None, int(y) if y else (int(d[:4]) if d[:4].isdigit() else None)
def publication(fields: dict) -> str:
t = re.sub(r"\s*,\s*(19|20)\d\d.*$", "", str(fields.get("title") or fields.get("identifier") or "")).strip()
t = re.sub(r"\s+(19|20)\d\d[ _-]\d\d[ _-]\d\d$", "", t)
return t[:120] or fields.get("identifier")
def is_book(fields: dict) -> bool:
cols = set(fields.get("collection") or [])
return bool(cols & {"inlibrary", "printdisabled", "internetarchivebooks", "americana"}) and "newspapers" not in cols
def restricted(fields: dict) -> bool:
return bool(set(fields.get("collection") or []) & {"inlibrary", "printdisabled", "lendinglibrary"})
class IAFullText(Crawler):
source_id = "ia_fulltext"
name = "Internet Archive full-text search — printed Conrad cartoon appearances (newspapers, anthologies)"
repository = "Internet Archive"
url = "https://archive.org/search?query=%22Paul+Conrad%22&sin=TXT"
classification = "PUBLIC_API"
access_notes = ("OCR full-text search (page-level). Records = pages carrying a Conrad credit line. Page images are "
"hotlinked from iiif.archive.org only for openly served items; lending-library books: no image.")
def __init__(self, *a, max_pages: int = MAX_PAGES, years=None, variants=None, **kw):
super().__init__(*a, **kw)
self.max_pages = max_pages
self.years = years
self.variants = variants or VARIANTS
self.ck = Checkpoint("ia_fulltext")
self.counts = {"hits": 0, "pages": 0, "credit": 0, "mention": 0, "other_conrad": 0, "images": 0,
"restricted_no_image": 0, "capped": 0}
self._manifests: dict[str, list | None] = {}
# ------------------------------------------------------------------ search
def fts(self, query: str, page: int) -> dict:
params = {"user_query": query, "hits_per_page": PAGE_SIZE, "page": page, "service_backend": "fts"}
for attempt in range(3): # the FTS service answers a sporadic HTTP 400 that succeeds on retry
try:
return self.http.get_json(FTS, params=params)
except Transient:
if attempt == 2:
raise
time.sleep(5 * (attempt + 1))
def harvest(self) -> dict:
pages: dict[tuple, dict] = {}
for vid, q, yrs in self.variants:
for y in (self.years or yrs):
if y not in yrs:
continue
qy = f"{q} AND year:{y}"
p, seen = 1, 0
while True:
try:
d = self.fts(qy, p)
except (Blocked, Transient, ValueError) as e:
self.error(FTS + "?" + urlencode({"user_query": qy, "page": p}), f"{type(e).__name__}: {e}")
break
self.stats["pages"] += 1
body = (d.get("response") or {}).get("body") or {}
hits = (body.get("hits") or {}).get("hits") or []
total = (body.get("hits") or {}).get("total") or 0
for h in hits:
f = h.get("fields") or {}
key = (f.get("identifier"), f.get("file_basename") or "", f.get("page_num"))
ent = pages.setdefault(key, {"fields": f, "snippets": [], "variants": set()})
ent["snippets"] += (h.get("highlight") or {}).get("text") or []
ent["variants"].add(vid)
seen += len(hits)
self.counts["hits"] += len(hits)
if not hits or seen >= total:
break
if p >= self.max_pages:
self.counts["capped"] += 1
self.notes.append(f"CAPPED {vid} {y}: {seen}/{total} hits walked")
break
p += 1
return pages
# ------------------------------------------------------------------ images
def manifest_images(self, identifier: str) -> list | None:
"""Canvas image URLs from IA's own IIIF presentation manifest (JSON; no image is requested)."""
if identifier in self._manifests:
return self._manifests[identifier]
out = None
try:
m = self.http.get_json(f"https://iiif.archive.org/iiif/3/{quote(identifier, safe='')}/manifest.json")
out = []
for c in m.get("items") or []:
url, label = None, json.dumps(c.get("label") or "")
for ap in c.get("items") or []:
for an in ap.get("items") or []:
b = an.get("body") or {}
url = url or b.get("id")
out.append((url, label))
except (Blocked, Transient, ValueError) as e:
self.error(f"iiif manifest {identifier}", f"{type(e).__name__}: {e}")
self._manifests[identifier] = out
return out
def access_restricted(self, identifier: str) -> bool:
"""IA's own flag (metadata JSON): lending-library / restricted items expose no public page image."""
try:
m = self.http.get_json(f"https://archive.org/metadata/{quote(identifier, safe='')}/metadata")
except (Blocked, Transient, ValueError) as e:
self.error(f"ia metadata {identifier}", f"{type(e).__name__}: {e}")
return True # unknown -> no image claimed
r = (m or {}).get("result") or {}
return str(r.get("access-restricted-item", "")).lower() == "true"
def page_image(self, f: dict) -> str | None:
if restricted(f) or self.access_restricted(f["identifier"]):
return None
canv = self.manifest_images(f["identifier"])
if not canv:
return None
n, base = f.get("page_num"), f.get("file_basename") or ""
if n is None:
return None
# multi-file items: the manifest concatenates every sub-file; pick the canvas whose image path names the
# sub-file and page. Single-file items: canvas index == page_num.
stem = f"{base}_{int(n):04d}.jp2"
for url, _ in canv:
if url and stem and (quote(stem, safe="") in url or stem in url or quote(stem) in url):
return url
if not f.get("result_in_subfile") and int(n) < len(canv):
return canv[int(n)][0]
return None
# ------------------------------------------------------------------ records
def crawl(self) -> None:
pages = self.harvest()
self.counts["pages"] = len(pages)
seen_ids: set[str] = set()
for (ident, base, pn), ent in pages.items():
f = ent["fields"]
kind, evidence = classify_hit(f, ent["snippets"])
self.counts[kind] = self.counts.get(kind, 0) + 1
if kind not in ("credit", "syndicate_ad") or not ident:
continue
iso, year = issue_date(f)
pub = publication(f)
book = is_book(f)
page_url = f"https://archive.org/details/{quote(ident)}" + (f"/{quote(base)}" if f.get("result_in_subfile")
and base else "") + f"/page/n{pn}"
img = self.page_image(f)
if img:
self.counts["images"] += 1
else:
self.counts["restricted_no_image"] += 1
ymatch = re.search(r"(?:©|\(c\)|copyright)\s*(19[5-9]\d|20[01]\d)|times,?\s*(19[5-9]\d|20[01]\d)", evidence or "", re.I)
cy = int(ymatch[1] or ymatch[2]) if ymatch else None
if book:
# an anthology reprint: the cartoon's own year only if the credit line states it
d = dict(date_exact=None, year=cy, date_start=f"{cy}-01-01" if cy else None,
date_end=f"{cy}-12-31" if cy else (f"{year}-12-31" if year else None), date_is_estimate=True)
where = f"reprinted in {pub}" + (f" ({year})" if year else "")
else:
d = dict(date_exact=iso, year=year, date_start=iso or (f"{year}-01-01" if year else None),
date_end=iso or (f"{year}-12-31" if year else None), date_is_estimate=not iso)
where = f"{pub}, {iso or year or 'n.d.'}"
ctitle = cartoon_title(evidence)
cid = f"iafts:{ident}:{base}:{pn}"
if cid in seen_ids:
continue
seen_ids.add(cid)
note = ("SYNDICATE ADVERTISEMENT in Editor & Publisher (normally reproduces a sample Conrad cartoon; unverified)"
if kind == "syndicate_ad" else
"PRINTED APPEARANCE (newspaper/periodical page)" if not book else "PRINTED APPEARANCE (book reprint)")
note += f" found by IA full-text OCR search {sorted(ent['variants'])}; credit line (OCR, unverified): \"{evidence}\""
rec = CartoonRecord(
canonical_id=cid, identifier=f"{ident}/{base}#n{pn}", granularity="item",
title=ctitle or f"[Paul Conrad cartoon — {where}, p. {int(pn) + 1}]",
caption=ctitle, publication=pub if not book else None,
syndicate="Los Angeles Times Syndicate" if re.search(r"syndicate", evidence or "", re.I) else None,
notes=note[:1000], rights_text=rights.COPYRIGHT_NOTE, repository="Internet Archive",
collection_name=pub, page=str(int(pn) + 1), record_url=page_url, image_url=img,
access_level=rights.ONLINE_IMAGE if img else rights.ONLINE_METADATA,
provenance=f"live:archive.org FTS {db.now()[:10]}; {ident}/{base} n{pn}",
acquisition_method="direct_api", subjects=["Editorial cartoons"],
people=presidents_in_text(evidence or ""), **d)
self.save(rec)
self.ck.set("last_run", db.now())
self.ck.set("counts", self.counts)
c = self.counts
self.notes.insert(0, f"FTS hits walked={c['hits']} distinct pages={c['pages']}; credit-line pages saved={c['credit']} + "
f"Editor & Publisher syndicate-ad pages={c.get('syndicate_ad', 0)} "
f"(page image URL recorded={c['images']}, restricted or unresolved/no image={c['restricted_no_image']}); "
f"rejected: about-Conrad mentions={c['mention']}, other Conrads={c['other_conrad']}; "
f"capped query-years={c['capped']}")
if c["credit"] == 0 and c["hits"] == 0:
self.status = "failed"
CRAWLER = IAFullText