← back to Paul Conrad Archive

src/conrad/crawlers/huntington.py

202 lines

"""Huntington Library — Paul Conrad Papers (mssCON 1-12360).

Live-parses the public EAD finding aid (the same XML OAC renders) for EVERY series / subseries / box /
item component. Original-drawing boxes carry CON number ranges; each CON number becomes a box_range
slot (date = box span; NOT item-level). Non-drawing series (correspondence, book files, tear sheets…)
are inventoried into `collections` for context, never counted as cartoons.
"""
from __future__ import annotations

import re
import xml.etree.ElementTree as ET
from datetime import date

from bs4 import BeautifulSoup

from .. import rights
from ..models import CartoonRecord
from .base import Blocked, Crawler, Transient

EAD_URL = "https://cinco-prd.s3.amazonaws.com/media/ead/conrad.xml"
OAC_URL = "https://oac.cdlib.org/findaid/ark:/13030/c8z03dxd/"
CATALOG_URL = "https://catalog.huntington.org/record=b1768021"
VERSO_URL = "https://www.huntington.org/verso/volunteering-decipher-paul-conrad"
COLL = "Paul Conrad Papers (mssCON 1-12360)"

MONTHS = {m: i for i, m in enumerate(["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"], 1)}


def _mon(w):
    return MONTHS.get((w or "").lower()[:3])


def _last(y, m):
    import calendar
    return calendar.monthrange(y, m)[1]


def parse_span(span: str):
    """'1970, Mar- Aug' | '1969, Sept- 1970, Feb' | '1967- 1968' | '1995, May 22- July' -> (start, end)."""
    parts = [p.strip() for p in re.split(r"\s*[-–—]\s*", span) if p.strip()]
    if not parts:
        return None
    L, R = parts[0], (parts[1] if len(parts) > 1 else "")
    lm = re.match(r"^(\d{4})(?:,\s*([A-Za-z]+)\.?(?:\s+(\d{1,2}))?)?", L)
    if not lm:
        return None
    y1, m1, d1 = int(lm[1]), _mon(lm[2]) or 1, int(lm[3]) if lm[3] else 1
    y2 = y1
    ry = re.match(r"^(\d{4})(?:,\s*([A-Za-z]+)\.?(?:\s+(\d{1,2}))?)?", R)
    rm = re.match(r"^([A-Za-z]+)\.?(?:\s+(\d{1,2}))?", R)
    if ry:
        y2 = int(ry[1]); m2 = _mon(ry[2]) or 12; d2 = int(ry[3]) if ry[3] else _last(y2, m2)
    elif rm and _mon(rm[1]):
        m2 = _mon(rm[1]); d2 = int(rm[2]) if rm[2] else _last(y2, m2)
    elif lm[2]:
        m2 = m1; d2 = d1 if lm[3] else _last(y1, m1)
    else:
        m2, d2 = 12, 31
    return date(y1, m1, d1).isoformat(), date(y2, m2, d2).isoformat()


def _txt(el) -> str:
    return re.sub(r"\s+", " ", "".join(el.itertext())).strip() if el is not None else ""


def parse_ead(xml: str) -> dict:
    root = ET.fromstring(xml.encode("utf-8"))
    arch = root.find("archdesc")
    did = arch.find("did")
    coll = {
        "title": _txt(did.find("unittitle")), "unitid": _txt(did.find("unitid")),
        "dates": _txt(did.find("unitdate")),
        "extent": "; ".join(_txt(e) for e in did.iter("extent")) or _txt(did.find("physdesc")),
        "access": _txt(arch.find("accessrestrict")), "use": _txt(arch.find("userestrict")),
        "abstract": _txt(did.find("abstract")),
    }
    comps = []

    def walk(el, path):
        for c in el:
            if not re.fullmatch(r"c0\d|c", c.tag):
                continue
            cd = c.find("did")
            title = _txt(cd.find("unittitle")) if cd is not None else ""
            unitid = _txt(cd.find("unitid")) if cd is not None else ""
            unitdate = _txt(cd.find("unitdate")) if cd is not None else ""
            conts = {(k.get("type") or "").lower(): _txt(k) for k in (cd.findall("container") if cd is not None else [])}
            comps.append({"level": c.get("level"), "tag": c.tag, "title": title, "unitid": unitid, "unitdate": unitdate,
                          "box": conts.get("box"), "folder": conts.get("folder"), "path": path})
            walk(c, path + [title.rstrip(".")])
    walk(arch.find("dsc"), [])
    return {"collection": coll, "components": comps}


def drawing_boxes(comps: list[dict]) -> list[dict]:
    boxes = []
    for c in comps:
        sub = c["path"][1] if len(c["path"]) > 1 else ""
        if not sub.startswith("Original Drawings") or not c["title"].startswith("Drawings,"):
            continue
        src = c["title"] if re.search(r"CON\s*\d", c["title"]) else c["unitid"]
        cm = re.search(r"CON\s*(\d+)\s*[-–]\s*(\d+)", src)
        span_text = re.sub(r"\.?\s*CON[\s\S]*$", "", c["title"].replace("Drawings,", "", 1)).strip().rstrip(".")
        sp = parse_span(span_text)
        boxes.append({"series": " > ".join(c["path"][:2]), "subseries": sub, "box": c["box"] or "", "title": c["title"],
                      "span": span_text, "con_start": int(cm[1]) if cm else None, "con_end": int(cm[2]) if cm else None,
                      "date_start": sp[0] if sp else None, "date_end": sp[1] if sp else None})
    # repair transcription typos where an end number overruns the next box start (e.g. 'CON 4890- 49152')
    repairs = []
    for i, b in enumerate(boxes):
        nxt = next((x for x in boxes[i + 1:] if x["subseries"] == b["subseries"] and x["con_start"]), None)
        if b["con_start"] and b["con_end"] and nxt and b["con_end"] >= nxt["con_start"]:
            repairs.append(f"box {b['box']}: {b['con_start']}-{b['con_end']} -> {b['con_start']}-{nxt['con_start'] - 1}")
            b["con_end"] = nxt["con_start"] - 1
    return boxes, repairs


def slot_records(boxes: list[dict]) -> list[CartoonRecord]:
    from datetime import date as D
    txt, rurl = rights.rights_for("Huntington Library")
    out = []
    for b in boxes:
        if not (b["con_start"] and b["con_end"] and b["date_start"]):
            continue
        n = b["con_end"] - b["con_start"] + 1
        t0, t1 = D.fromisoformat(b["date_start"]).toordinal(), D.fromisoformat(b["date_end"]).toordinal()
        for k in range(n):
            con = b["con_start"] + k
            est = D.fromordinal(round(t0 + (t1 - t0) * (0 if n == 1 else k / (n - 1)))).isoformat()
            out.append(CartoonRecord(
                canonical_id=f"hunt:CON-{con}", identifier=f"CON {con}", granularity="box_range",
                date_start=b["date_start"], date_end=b["date_end"], year=int(est[:4]), date_is_estimate=True,
                description=f"Original drawing slot CON {con} in Box {b['box']} ({b['series']}); box span "
                            f"'{b['span']}'. 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 b["date_start"] >= "1964" and b["date_end"] <= "1993-12-31" else None,
                medium="Original drawing", rights_text=txt, rights_url=rurl, repository="Huntington Library",
                collection_name=COLL, box=b["box"], folder=f"CON {con}", record_url=OAC_URL,
                access_level=rights.ARCHIVE_VISIT, provenance=f"live:{EAD_URL}",
            ))
    return out


class Huntington(Crawler):
    source_id = "huntington"
    name = "Huntington Library — Paul Conrad Papers (EAD finding aid)"
    repository = "Huntington Library"
    url = EAD_URL
    classification = "PHYSICAL_ARCHIVE"
    access_notes = ("Originals viewable in the Huntington reading room (reader credentials required). Public EAD "
                    "lists box-level CON ranges only; the Huntington's internal item database is not public.")

    def crawl(self) -> None:
        status, xml = self.http.get(EAD_URL)
        self.stats["pages"] += 1
        if status != 200:
            raise Transient(f"EAD HTTP {status}")
        ead = parse_ead(xml)
        c = ead["collection"]
        self.conn.execute(
            """INSERT OR REPLACE INTO collections(repository,name,identifier,url,extent,date_range,level,notes,source_id)
               VALUES (?,?,?,?,?,?,?,?,?)""",
            ("Huntington Library", c["title"] or COLL, c["unitid"], OAC_URL, c["extent"], c["dates"], "collection",
             f"ACCESS: {c['access'][:600]} | USE: {c['use'][:600]}", self.source_id))
        n_boxes = 0
        for comp in ead["components"]:
            if comp["level"] in ("box", "Box") or (comp["box"] and comp["level"] not in ("item",)):
                n_boxes += 1
                self.conn.execute(
                    """INSERT OR REPLACE INTO collections(repository,name,identifier,url,extent,date_range,level,notes,source_id)
                       VALUES (?,?,?,?,?,?,?,?,?)""",
                    ("Huntington Library", " > ".join(comp["path"] + [comp["title"]])[:400], f"Box {comp['box']}",
                     OAC_URL, None, comp["unitdate"], "box", comp["unitid"], self.source_id))
            if len(comp["path"]) > 1 and comp["path"][1].startswith("Book Files") and comp["level"] == "item":
                self.conn.execute("INSERT OR IGNORE INTO collections(repository,name,identifier,url,level,notes,source_id)"
                                  " VALUES (?,?,?,?,?,?,?)", ("Huntington Library", comp["title"][:400], comp["unitid"],
                                                              OAC_URL, "book_file", "Huntington Book Files series", self.source_id))
        boxes, repairs = drawing_boxes(ead["components"])
        for rec in slot_records(boxes):
            self.save(rec)
        self.conn.commit()
        self.notes.append(f"EAD: {len(ead['components'])} components, {n_boxes} boxes, {len(boxes)} original-drawing "
                          f"boxes; typo repairs: {repairs}")
        # collection-level catalog record + volunteer article (context only)
        for url in (CATALOG_URL, VERSO_URL):
            try:
                st, html = self.http.get(url)
                self.stats["pages"] += 1
                if st == 200:
                    soup = BeautifulSoup(html, "html.parser")
                    text = re.sub(r"\s+", " ", soup.get_text(" "))
                    m = re.search(r"(\d[\d,]*)\s+(?:original\s+)?(?:cartoons|drawings)", text, re.I)
                    self.notes.append(f"{url}: ok" + (f" (mentions '{m.group(0)}')" if m else ""))
                else:
                    self.error(url, f"HTTP {st}")
            except (Blocked, Transient) as e:
                self.error(url, e)
                self.notes.append(f"{url}: {e}")


CRAWLER = Huntington