← back to Paul Conrad Archive

src/conrad/crawlers/web_images.py

322 lines

"""Public web pages that SHOW Conrad cartoons (TK-12199 image priority): one institutional exhibit and several
secondary copies (dealer, auction, blog, magazine). Each page is fetched through the polite client (robots.txt per
path with wildcard semantics, descriptive UA); every page is registered as its own source with an honest status.

Image URLs are stored as METADATA only (never fetched). Blog / dealer / auction copies are REPRINTS: their rows carry
acquisition_method 'secondary_citation' (dealer/auction/blog) and a note saying so, so dedupe and the viewer prefer the
institutional record whenever the same cartoon exists elsewhere. Dates are recorded only when the page states them."""
from __future__ import annotations

import html as _html
import json
import re
from datetime import datetime, timezone
from urllib.parse import urljoin

from .. import db, rights
from ..models import CartoonRecord
from ..normalize import parse_date, presidents_in_text
from .base import Blocked, Crawler, Transient

SYR = "https://library.syracuse.edu/extsites/cartoonists/conrad.php"
PCG = "https://www.original-political-cartoon.com/cartoon-gallery/artists/conrad-paul-1924-2010/"
INVALUABLE = ["https://www.invaluable.com/artist/paul-conrad-saveb5g89x/sold-at-auction-prices/",
              "https://www.invaluable.com/artist/conrad-paul-francis-67i99nw6g1/sold-at-auction-prices/"]
PHD = "https://pophistorydig.com/topics/paul-conrad-cartoonist/"
TRUTHDIG = "https://www.truthdig.com/author/paul_conrad/"

SOURCES = {
    "syracuse_exhibit": ("Syracuse University Libraries — 'Cartoonists' web exhibit, Paul Conrad page", SYR, "PUBLIC_HTML"),
    "pc_gallery": ("Political Cartoon Gallery (dealer, London) — Conrad originals for sale", PCG, "PUBLIC_HTML"),
    "invaluable": ("Invaluable — auction results for Paul Conrad originals", INVALUABLE[0], "PUBLIC_HTML"),
    "pophistorydig": ("The Pop History Dig — 'Paul Conrad, cartoonist' article (blog reprints)", PHD, "PUBLIC_HTML"),
    "truthdig": ("Truthdig — Paul Conrad cartoons (2000s syndicated)", TRUTHDIG, "PUBLIC_HTML"),
}
MONTHS = "january|february|march|april|may|june|july|august|september|october|november|december"


def _txt(s: str | None) -> str:
    return re.sub(r"\s+", " ", _html.unescape(re.sub(r"<[^>]+>", " ", s or ""))).strip()


def stated_year(text: str) -> tuple[int | None, tuple[str, str] | None]:
    """(year, (start, end)) only when the text itself dates the cartoon; '1970s' -> decade range. Book / exhibit
    years are the caller's problem (it passes cartoon captions only)."""
    ys = sorted({int(y) for y in re.findall(r"\b(19[5-9]\d|200\d|2010)\b(?!s)", text)})
    if len(ys) == 1:
        return ys[0], (f"{ys[0]}-01-01", f"{ys[0]}-12-31")
    m = re.search(r"\b(19[5-9]0)s\b", text)
    if m and not ys:
        d = int(m[1])
        return None, (f"{d}-01-01", f"{d + 9}-12-31")
    return None, None


# ------------------------------------------------------------------------------------------------ parsers (pure)
def parse_syracuse(body: str) -> list[dict]:
    out = []
    for tag in re.findall(r"<img\b[^>]*>", body, re.I):
        alt = re.search(r'\balt="([^"]*)"', tag)
        src = re.search(r'\bsrc="([^"]*)"', tag)
        if alt and src and alt[1].strip().lower().endswith(" cartoon") and "images/" in src[1]:
            out.append(dict(title=_html.unescape(alt[1]).strip()[:-len(" cartoon")].strip(), image=urljoin(SYR, src[1])))
    return out


def parse_pcg_list(body: str) -> list[str]:
    return sorted({urljoin(PCG, u) for u in re.findall(r'href="(/cartoon-gallery/buy/[^"]+/\d+/)"', body)})


def parse_pcg_item(body: str) -> dict:
    t = _txt(re.sub(r"<script.*?</script>|<style.*?</style>", " ", body, flags=re.S))
    img = re.search(r'<img\s[^>]*src="(/media/filer_public_thumbnails/[^"]+)"[^>]*\balt="([^"]*)"', body, re.S)
    def field(name, stop):
        m = re.search(name + r"\s+(.+?)\s+(?:" + stop + r")", t)
        return m[1].strip() if m else None
    pub = field("Publication", "Published|Home|Size|Medium")
    date = field("Published", "Home|Size|Medium|Description")
    alt = _html.unescape(img[2]).strip() if img else ""
    page_title = re.search(r"<title>\s*(.*?)\s*-\s*Cartoon Gallery\s*</title>", body, re.S)
    title = alt if alt and alt != "None" else (_html.unescape(page_title[1]).strip() if page_title else None)
    return dict(title=title, image=urljoin(PCG, img[1]) if img else None,
                publication=pub, published=date, medium=field("Medium", "Publication|Published|Home|Size"),
                size=field("Size", "Medium|Publication|Published"), sold="SOLD" in (img[2] if img else ""))


def _enclosing_object(body: str, pos: int) -> str:
    """The innermost JSON object {...} around position pos (string-aware brace matching)."""
    depth, i = 0, pos
    while i > 0:  # walk back to the unmatched '{'
        i -= 1
        c = body[i]
        if c == "}":
            depth += 1
        elif c == "{":
            if depth == 0:
                break
            depth -= 1
    start, depth, j, in_str, esc = i, 0, i, False, False
    while j < len(body):
        c = body[j]
        if in_str:
            if esc:
                esc = False
            elif c == "\\":
                esc = True
            elif c == '"':
                in_str = False
        elif c == '"':
            in_str = True
        elif c == "{":
            depth += 1
        elif c == "}":
            depth -= 1
            if depth == 0:
                return body[start: j + 1]
        j += 1
    return body[start: pos + 800]


def parse_invaluable(body: str) -> list[dict]:
    out = []
    for m in re.finditer(r'"lotRef"\s*:\s*"([^"]+)"', body):
        seg = _enclosing_object(body, m.start())
        def g(k):
            x = re.search(r'"%s"\s*:\s*("(?:[^"\\]|\\.)*"|[\d.]+)' % k, seg)
            if not x:
                return None
            try:
                return json.loads(x[1]) if x[1].startswith('"') else x[1]
            except ValueError:
                return x[1].strip('"')
        title = _html.unescape(g("lotTitle") or "")
        if not re.search(r"paul\s+conrad|conrad,?\s+paul\s+(?:f|1924)", title, re.I) or re.search(r"15\d\d|16\d\d", title):
            continue
        # cartoons only: an original drawing / political cartoon lot; prints, lithographs and portraits are not
        if not re.search(r"political cartoon|cartoon|\bink\b|1924-2010\)\s*[\"“]", title, re.I) or re.search(r"lithograph|signed print|portrait", title, re.I):
            continue
        ts = g("dateTimeUTCUnix")
        named = re.sub(r"^.*?(?:1924\s*-\s*2010\)?,?|\(political cartoon\)\s*paul conrad)\s*", "", title, flags=re.I)
        named = re.split(r",\s*(?:ink|pen|original)\b", named, flags=re.I)[0].strip(" .,'\"\\")
        out.append(dict(ref=m[1], title=title, named=named if len(named) > 3 else None, house=g("houseName"), lot=g("lotNumber"), photo=g("photoPath"),
                        sold=datetime.fromtimestamp(int(float(ts)), timezone.utc).date().isoformat() if ts else None))
    uniq = {d["ref"]: d for d in out}
    return list(uniq.values())


def parse_pophistorydig(body: str) -> list[dict]:
    out = []
    for tag in re.findall(r"<img\b[^>]*>", body, re.I):
        alt = re.search(r'\balt="([^"]*)"', tag)
        src = re.search(r'\bsrc="([^"]*)"', tag)
        if not (alt and src):
            continue
        a = _html.unescape(alt[1]).strip()
        if "conrad" not in a.lower() or re.search(r"working on a drawing|at his desk|book|photo of", a, re.I):
            continue
        if not re.search(r"cartoon|caricature|caption|send up|cast|depict|slaying|conrad['’]s", a, re.I):
            continue
        out.append(dict(alt=a, image=urljoin(PHD, src[1])))
    return out


def parse_og(body: str) -> dict:
    def meta(p):
        m = re.search(r'<meta[^>]+(?:property|name)="%s"[^>]+content="([^"]*)"' % re.escape(p), body)
        return _html.unescape(m[1]).strip() if m else None
    return dict(title=meta("og:title"), image=meta("og:image"), published=meta("article:published_time"),
                desc=meta("og:description"))


# ------------------------------------------------------------------------------------------------ crawler
class WebImages(Crawler):
    source_id = "web_images"
    name = "Public web pages showing Conrad cartoons (exhibit, dealer, auction, blog, magazine)"
    repository = None
    url = SYR
    classification = "PUBLIC_HTML"
    access_notes = "Each sub-source is registered separately; images are URL metadata only; copies flagged as reprints."

    def __init__(self, *a, **kw):
        super().__init__(*a, **kw)
        self.per: dict[str, int] = {}
        self.imgs: dict[str, int] = {}
        self.status_of: dict[str, tuple[str, str]] = {}

    def fetch(self, sid: str, url: str) -> str | None:
        try:
            st, body = self.http.get(url)
            self.stats["pages"] += 1
        except Blocked as e:
            self.status_of[sid] = ("blocked", str(e))
            self.error(url, f"BLOCKED: {e}")
            return None
        except Transient as e:
            self.status_of.setdefault(sid, ("failed", str(e)))
            self.error(url, e)
            return None
        if st != 200:
            self.status_of.setdefault(sid, ("failed", f"HTTP {st} at {url}"))
            return None
        return body

    def put(self, sid: str, rec: CartoonRecord) -> None:
        self.stats["seen"] += 1
        st = db.save_record(self.conn, rec, sid)
        if st in ("added", "updated"):
            self.stats[st] += 1
        self.per[sid] = self.per.get(sid, 0) + 1
        if rec.image_url:
            self.imgs[sid] = self.imgs.get(sid, 0) + 1

    def base(self, sid, cid, ident, title, url, image, method, note, **kw) -> CartoonRecord:
        text = " ".join(filter(None, [title, kw.get("description")]))
        return CartoonRecord(canonical_id=cid, identifier=ident, granularity="item", title=title, record_url=url,
                             image_url=image, access_level=rights.ONLINE_IMAGE if image else rights.ONLINE_METADATA,
                             rights_text=rights.COPYRIGHT_NOTE, notes=note[:1000], acquisition_method=method,
                             provenance=f"live:{sid} {db.now()[:10]}", people=presidents_in_text(text),
                             subjects=["Editorial cartoons"], **kw)

    # --- sub-sources
    def syracuse(self):
        b = self.fetch("syracuse_exhibit", SYR)
        for i, it in enumerate(parse_syracuse(b or ""), 1):
            self.put("syracuse_exhibit", self.base(
                "syracuse_exhibit", f"syrx:{re.sub(r'[^0-9a-z]+', '-', it['image'].rsplit('/', 1)[-1].lower())}",
                it["image"].rsplit("/", 1)[-1], it["title"], SYR, it["image"], "direct_html",
                "Syracuse University Libraries 'Cartoonists' web exhibit (Special Collections); title from the exhibit's "
                "image alt text; undated on the page", repository="Syracuse University Libraries",
                collection_name="Cartoonists web exhibit"))

    def pcg(self):
        b = self.fetch("pc_gallery", PCG)
        for u in parse_pcg_list(b or ""):
            d = self.fetch("pc_gallery", u)
            if not d:
                continue
            it = parse_pcg_item(d)
            if not it["title"]:
                continue
            title = re.sub(r"^SOLD\s+", "", it["title"]).strip()
            pd = parse_date(it["published"]) if it["published"] else {"date_exact": None, "date_start": None,
                                                                      "date_end": None, "year": None}
            caveat = ""
            if it["publication"] and "los angeles" in it["publication"].lower() and pd["year"] and pd["year"] < 1964:
                caveat = " | CAVEAT: dealer states the LA Times before 1964, when Conrad was at the Denver Post"
            self.put("pc_gallery", self.base(
                "pc_gallery", f"pcg:{u.rstrip('/').rsplit('/', 1)[-1]}", u, title, u, it["image"], "secondary_citation",
                f"DEALER COPY (Political Cartoon Gallery, London) of an original drawing; publication/date as stated by "
                f"the dealer: {it['publication'] or '?'} / {it['published'] or '?'}; medium {it['medium'] or '?'}, size "
                f"{it['size'] or '?'}{caveat}", repository="Political Cartoon Gallery (dealer)",
                collection_name="dealer stock", publication=it["publication"], medium=it["medium"],
                dimensions=it["size"], date_exact=pd["date_exact"], date_start=pd["date_start"],
                date_end=pd["date_end"], year=pd["year"], date_is_estimate=not pd["date_exact"]))

    def invaluable(self):
        for page in INVALUABLE:
            b = self.fetch("invaluable", page)
            for it in parse_invaluable(b or ""):
                img = f"https://image.invaluable.com/housePhotos/{it['photo']}" if it.get("photo") else None
                self.put("invaluable", self.base(
                    "invaluable", f"inv:{it['ref']}", it["ref"],
                    it["named"] or f"[Paul Conrad original cartoon — {it['house'] or 'auction'} lot {it['lot'] or '?'}, "
                                   f"sold {it['sold'] or '?'}]", page, img, "secondary_citation",
                    f"AUCTION RECORD (Invaluable) lot title '{it['title']}'"
                    f"{'' if it['named'] else '; the lot title names no cartoon'}; "
                    f"sale date {it['sold']} is NOT the cartoon's date", repository=it["house"] or "auction house",
                    collection_name="Invaluable auction results"))

    def pophistorydig(self):
        b = self.fetch("pophistorydig", PHD)
        for it in parse_pophistorydig(b or ""):
            y, rng = stated_year(it["alt"])
            title = it["alt"] if len(it["alt"]) <= 160 else it["alt"][:157] + "..."
            self.put("pophistorydig", self.base(
                "pophistorydig", f"phd:{it['image'].rsplit('/', 1)[-1].lower()}", it["image"], title, PHD,
                it["image"], "secondary_citation",
                f"BLOG REPRINT (The Pop History Dig); title = the blog's image caption; "
                f"{'year stated in caption' if y else ('decade stated in caption' if rng else 'undated in source')}",
                repository="The Pop History Dig (blog)", collection_name="pophistorydig.com", description=it["alt"],
                year=y, date_start=rng[0] if rng else None, date_end=rng[1] if rng else None, date_is_estimate=True))

    def truthdig(self):
        b = self.fetch("truthdig", TRUTHDIG)
        for u in sorted(set(re.findall(r'href="(https://www\.truthdig\.com/cartoons/[^"]+)"', b or ""))):
            d = self.fetch("truthdig", u)
            if not d:
                continue
            og = parse_og(d)
            if not og["title"]:
                continue
            title = re.sub(r"\s*[-|–]\s*Truthdig\s*$", "", og["title"]).strip()
            iso = (og["published"] or "")[:10] or None
            self.put("truthdig", self.base(
                "truthdig", f"td:{u.rstrip('/').rsplit('/', 1)[-1]}", u, title, u, og["image"], "secondary_citation",
                "Truthdig web publication of a syndicated Conrad cartoon (Tribune Media Services); date = Truthdig's "
                "posting date, not necessarily first publication", repository="Truthdig", collection_name="Truthdig cartoons",
                publication="Truthdig", date_exact=None, date_start=iso, date_end=iso,
                year=int(iso[:4]) if iso else None, date_is_estimate=True))

    def crawl(self) -> None:
        for sid, (name, url, cls) in SOURCES.items():
            db.upsert_source(self.conn, sid, name, url=url, classification=cls, status="running")
        for fn in (self.syracuse, self.pcg, self.invaluable, self.pophistorydig, self.truthdig):
            try:
                fn()
            except (Blocked, Transient) as e:  # pragma: no cover - fetch() already catches these
                self.error(fn.__name__, e)
        summary = []
        for sid, (name, url, cls) in SOURCES.items():
            n, i = self.per.get(sid, 0), self.imgs.get(sid, 0)
            st, why = self.status_of.get(sid, ("worked" if n else "failed", ""))
            if n and st != "blocked":
                st = "worked"
            db.upsert_source(self.conn, sid, name, url=url, classification="REQUIRES_PERMISSION" if st == "blocked" else cls,
                             status=st, notes=f"records={n}; image URLs recorded={i}" + (f"; {why}" if why else ""))
            summary.append(f"{sid}: {n} rec / {i} img ({st})")
        self.conn.commit()
        self.notes.append("; ".join(summary))


CRAWLER = WebImages