← back to Paul Conrad Archive
src/conrad/web/app.py
363 lines
"""Local research viewer (FastAPI). Binds 127.0.0.1:8787 only.
IMAGE RULE (Steve, TK-12199): images are shown ONLY where the holding institution publishes a public image, and only by
HOTLINKING the institution's own URL (see imagehost.py allowlist). Nothing is ever downloaded, proxied, cached or stored;
a Content-Security-Policy img-src allowlist makes the browser refuse every other image host. All other records show
"View at <repository>" link-outs.
"""
from __future__ import annotations
import base64
import html
import os
import secrets
from pathlib import Path
from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse, Response
from fastapi.staticfiles import StaticFiles
from .. import db
from .. import provenance as prov
from . import imagehost
STATIC = Path(__file__).parent / "static"
class AuthConfigError(RuntimeError):
pass
def auth_settings() -> tuple[str, str] | None:
"""Basic-auth credentials from env. CONRAD_REQUIRE_AUTH=1 without both creds -> refuse to start (fail closed)."""
user, pw = os.environ.get("CONRAD_BASIC_USER", ""), os.environ.get("CONRAD_BASIC_PASS", "")
if user and pw:
return user, pw
if os.environ.get("CONRAD_REQUIRE_AUTH", "") in ("1", "true", "yes"):
raise AuthConfigError("CONRAD_REQUIRE_AUTH=1 but CONRAD_BASIC_USER / CONRAD_BASIC_PASS are not both set")
return None
def create_app() -> FastAPI:
creds = auth_settings()
try: # make sure the denormalised search index exists (cheap no-op when already built)
_c = db.connect()
db.init_db(_c)
if not _c.execute("SELECT 1 FROM cartoon_index LIMIT 1").fetchone() \
or _c.execute("SELECT 1 FROM cartoon_index WHERE search_text IS NULL LIMIT 1").fetchone():
db.refresh_index(_c)
except Exception: # noqa: BLE001 — a read-only / missing DB must not stop the app from starting
pass
application = FastAPI(title="Paul Conrad Master Archive", docs_url=None, redoc_url=None, openapi_url=None)
@application.middleware("http")
async def guard(request: Request, call_next):
# robots.txt stays readable so crawlers can see Disallow: / (everything else is behind auth when enabled)
if creds and request.url.path != "/robots.txt":
ok = False
hdr = request.headers.get("authorization", "")
if hdr.lower().startswith("basic "):
try:
u, _, pw = base64.b64decode(hdr[6:]).decode("utf-8").partition(":")
ok = secrets.compare_digest(u, creds[0]) and secrets.compare_digest(pw, creds[1])
except Exception: # noqa: BLE001
ok = False
if not ok:
resp = Response("Authentication required", status_code=401,
headers={"WWW-Authenticate": 'Basic realm="conrad-archive", charset="UTF-8"'})
resp.headers["X-Robots-Tag"] = "noindex, nofollow"
return resp
resp = await call_next(request)
resp.headers["X-Robots-Tag"] = "noindex, nofollow"
resp.headers["Referrer-Policy"] = "no-referrer"
resp.headers["Content-Security-Policy"] = imagehost.CSP
return resp
application.mount("/static", StaticFiles(directory=STATIC), name="static")
application.include_router(router)
return application
from fastapi import APIRouter # noqa: E402
router = APIRouter()
app = None # set at the bottom of the module (after routes are registered)
SORTS = {
"date_desc": "COALESCE(c.date_start,'0000') DESC, c.id DESC",
"date_asc": "COALESCE(c.date_start,'9999') ASC, c.id",
"title": "CASE WHEN c.title IS NULL THEN 1 ELSE 0 END, LOWER(c.title), c.id",
"repository": "ix.repos, COALESCE(c.date_start,'9999'), c.id",
"year": "c.year IS NULL, c.year, c.date_start, c.id",
}
def conn():
return db.connect()
_IMG_SOURCES_SQL = """SELECT COALESCE(k.merged_into, k.id) AS cid, x.source_id, x.repository, x.collection_name,
x.record_url, x.rights_url, x.image_url, x.thumbnail_url
FROM cartoon_sources x JOIN cartoons k ON k.id = x.cartoon_id
WHERE (x.image_url IS NOT NULL OR x.thumbnail_url IS NOT NULL) {extra}
ORDER BY cid, x.source_id LIKE 'seed%', x.source_id"""
def display_images(c, ids=None) -> dict:
"""{canonical cartoon id: display_image dict} for records whose holding institution publishes an image on an
allowlisted host. ids=None -> every record (used by the has_image filter)."""
extra, args = "", []
if ids is not None:
ids = list(ids)
if not ids:
return {}
extra = "AND COALESCE(k.merged_into, k.id) IN (" + ",".join("?" * len(ids)) + ")"
args = ids
grouped: dict = {}
for r in c.execute(_IMG_SOURCES_SQL.format(extra=extra), args):
grouped.setdefault(r["cid"], []).append(r)
out = {}
for cid, rows in grouped.items():
di = imagehost.pick(rows)
if di:
out[cid] = di
return out
def _where(q, year, year_from, year_to, person, president, subject, publication, repository, collection, granularity,
has_image=False, c=None):
w, a = ["c.merged_into IS NULL"], []
if has_image:
ids = sorted(display_images(c or conn()))
w.append("c.id IN (" + (",".join(str(int(i)) for i in ids) or "NULL") + ")")
if granularity and granularity != "all":
w.append("c.granularity = ?"); a.append(granularity)
if q:
# one LIKE over the denormalised blob (title/caption/description/publication/people/subjects/
# repositories/collection names), built by db.refresh_index — keeps q fast and covers holding-repository text
w.append("c.id IN (SELECT id FROM cartoon_index WHERE search_text LIKE ?)")
a.append(f"%{' '.join(q.lower().split())}%")
if year:
w.append("c.year = ?"); a.append(year)
if year_from:
w.append("c.year >= ?"); a.append(year_from)
if year_to:
w.append("c.year <= ?"); a.append(year_to)
if person:
w.append("c.id IN (SELECT cp.cartoon_id FROM cartoon_people cp JOIN people p ON p.id=cp.person_id WHERE p.name = ?)"); a.append(person)
if president:
w.append("c.id IN (SELECT cp.cartoon_id FROM cartoon_people cp JOIN people p ON p.id=cp.person_id WHERE p.is_president=1 AND p.name = ?)"); a.append(president)
if subject:
w.append("c.id IN (SELECT cs.cartoon_id FROM cartoon_subjects cs JOIN subjects s ON s.id=cs.subject_id WHERE s.name = ?)"); a.append(subject)
if publication:
w.append("c.publication = ?"); a.append(publication)
if repository:
w.append("c.id IN (SELECT COALESCE(k.merged_into,k.id) FROM cartoon_sources x JOIN cartoons k ON k.id=x.cartoon_id WHERE x.repository = ?)"); a.append(repository)
if collection:
w.append("c.id IN (SELECT COALESCE(k.merged_into,k.id) FROM cartoon_sources x JOIN cartoons k ON k.id=x.cartoon_id WHERE x.collection_name = ?)"); a.append(collection)
return " AND ".join(w), a
@router.get("/", response_class=HTMLResponse)
def index():
return FileResponse(STATIC / "index.html")
@router.get("/api/search")
def search(q: str | None = None, year: int | None = None, year_from: int | None = None, year_to: int | None = None,
person: str | None = None, president: str | None = None, subject: str | None = None,
publication: str | None = None, repository: str | None = None, collection: str | None = None,
granularity: str = "item", has_image: bool = False, sort: str = "date_desc",
limit: int = Query(60, le=500), offset: int = 0):
c = conn()
where, args = _where(q, year, year_from, year_to, person, president, subject, publication, repository, collection,
granularity, has_image, c)
total = c.execute(f"SELECT COUNT(*) FROM cartoons c WHERE {where}", args).fetchone()[0]
rows = c.execute(f"""
SELECT c.id, c.canonical_id, c.granularity, c.title, c.caption, c.date_exact, c.date_start, c.date_end, c.year,
c.date_is_estimate, c.publication, c.created_at, ix.repos, ix.n_sources, ix.people, ix.record_url
FROM cartoons c LEFT JOIN cartoon_index ix ON ix.id = c.id
WHERE {where} ORDER BY {SORTS.get(sort, SORTS['date_desc'])} LIMIT ? OFFSET ?""",
[*args, limit, offset]).fetchall()
results = [dict(r) for r in rows]
methods = prov.methods_by_canonical(c, [r["id"] for r in results])
imgs = display_images(c, [r["id"] for r in results])
for r in results:
r["display_image"] = imgs.get(r["id"])
r["acquisition_methods"] = methods.get(r["id"], [])
r["provenance_flags"] = prov.flags(r["acquisition_methods"])
return {"total": total, "offset": offset, "results": results}
@router.get("/api/facets")
def facets(granularity: str = "item"):
c = conn()
g = "" if granularity == "all" else "AND c.granularity = ?"
ga = [] if granularity == "all" else [granularity]
def top(sql, n=300):
return [dict(r) for r in c.execute(sql + f" LIMIT {n}", ga).fetchall()]
return {
"timeline": top(f"SELECT c.year AS k, COUNT(*) AS n FROM cartoons c WHERE c.merged_into IS NULL AND c.year IS NOT NULL {g} GROUP BY c.year ORDER BY c.year", 200),
"presidents": top(f"SELECT p.name AS k, COUNT(*) AS n FROM cartoon_people cp JOIN people p ON p.id=cp.person_id JOIN cartoons c ON c.id=cp.cartoon_id WHERE p.is_president=1 AND c.merged_into IS NULL {g} GROUP BY p.name ORDER BY n DESC"),
"people": top(f"SELECT p.name AS k, COUNT(*) AS n FROM cartoon_people cp JOIN people p ON p.id=cp.person_id JOIN cartoons c ON c.id=cp.cartoon_id WHERE c.merged_into IS NULL {g} GROUP BY p.name ORDER BY n DESC, p.name"),
"subjects": top(f"SELECT s.name AS k, COUNT(*) AS n FROM cartoon_subjects cs JOIN subjects s ON s.id=cs.subject_id JOIN cartoons c ON c.id=cs.cartoon_id WHERE c.merged_into IS NULL {g} GROUP BY s.name ORDER BY n DESC, s.name"),
"publications": top(f"SELECT c.publication AS k, COUNT(*) AS n FROM cartoons c WHERE c.merged_into IS NULL AND c.publication IS NOT NULL {g} GROUP BY c.publication ORDER BY n DESC"),
"repositories": top(f"SELECT x.repository AS k, COUNT(DISTINCT COALESCE(c.merged_into,c.id)) AS n FROM cartoon_sources x JOIN cartoons c ON c.id=x.cartoon_id WHERE 1=1 {g} GROUP BY x.repository ORDER BY n DESC"),
"collections": top(f"SELECT x.collection_name AS k, COUNT(DISTINCT COALESCE(c.merged_into,c.id)) AS n FROM cartoon_sources x JOIN cartoons c ON c.id=x.cartoon_id WHERE x.collection_name IS NOT NULL {g} GROUP BY x.collection_name ORDER BY n DESC"),
}
@router.get("/api/stats")
def stats():
c = conn()
return {r["granularity"]: r["n"] for r in c.execute(
"SELECT granularity, COUNT(*) n FROM cartoons WHERE merged_into IS NULL GROUP BY granularity")}
def _detail(cid: int) -> dict:
c = conn()
row = c.execute("SELECT * FROM cartoons WHERE id=?", (cid,)).fetchone()
if not row:
raise HTTPException(404, "not found")
if row["merged_into"]:
row = c.execute("SELECT * FROM cartoons WHERE id=?", (row["merged_into"],)).fetchone()
cid = row["id"]
ids = [r[0] for r in c.execute("SELECT id FROM cartoons WHERE id=? OR merged_into=?", (cid, cid))]
ph = ",".join("?" * len(ids))
d = dict(row)
# raw image_url / thumbnail_url are NOT sent to the browser; only the allowlisted display_image (hotlink) is
d["sources"] = [dict(r) for r in c.execute(
f"""SELECT source_id, repository, collection_name, identifier, box, folder, page, record_url, access_level,
rights_url, provenance, acquisition_method, retrieved_at, (image_url IS NOT NULL OR thumbnail_url IS NOT NULL) AS has_online_image
FROM cartoon_sources WHERE cartoon_id IN ({ph}) ORDER BY source_id LIKE 'seed%', source_id""", ids)]
d["people"] = [r[0] for r in c.execute(f"SELECT DISTINCT p.name FROM cartoon_people cp JOIN people p ON p.id=cp.person_id WHERE cp.cartoon_id IN ({ph}) ORDER BY 1", ids)]
d["subjects"] = [r[0] for r in c.execute(f"SELECT DISTINCT s.name FROM cartoon_subjects cs JOIN subjects s ON s.id=cs.subject_id WHERE cs.cartoon_id IN ({ph}) ORDER BY 1", ids)]
d["links"] = [dict(r) for r in c.execute(
f"""SELECT l.relation, l.score, l.detail, o.id AS other_id, o.title AS other_title, o.canonical_id AS other_canonical
FROM cartoon_links l JOIN cartoons o ON o.id = CASE WHEN l.cartoon_id IN ({ph}) THEN l.related_id ELSE l.cartoon_id END
WHERE l.cartoon_id IN ({ph}) OR l.related_id IN ({ph})""", ids * 3)]
d["merged_ids"] = [i for i in ids if i != cid]
d["display_image"] = display_images(c, [cid]).get(cid)
d["acquisition_methods"] = sorted({s["acquisition_method"] for s in d["sources"] if s["acquisition_method"]})
d["provenance_flags"] = prov.flags(d["acquisition_methods"])
return d
@router.get("/api/cartoon/{cid}")
def cartoon_json(cid: int):
return _detail(cid)
def _e(x) -> str:
return html.escape("" if x is None else str(x))
def _source_row(s: dict) -> str:
link = "—"
if s["record_url"]:
link = ('<a rel="noopener noreferrer" target="_blank" href="' + _e(s["record_url"]) + '">View at '
+ _e(s["repository"]) + " ↗</a>")
if s["rights_url"]:
link += ' · <a rel="noopener noreferrer" target="_blank" href="' + _e(s["rights_url"]) + '">rights</a>'
online = ' <span class="badge">image online at repository</span>' if s["has_online_image"] else ""
cells = [_e(s["repository"]), _e(s["collection_name"]), _e(s["identifier"]), _e(s["box"]), _e(s["folder"]),
'<span class="badge">' + _e(s["access_level"]) + "</span>" + online
+ ' <span class="badge' + (' warn' if s.get("acquisition_method") == "seed_via_reader_bypass" else "") + '">'
+ _e(s.get("acquisition_method")) + "</span>", link,
'<span class="muted small">' + _e(s["provenance"]) + "</span>"]
return "<tr>" + "".join("<td>" + c + "</td>" for c in cells) + "</tr>"
def _link_row(lk: dict) -> str:
return ("<li>" + _e(lk["relation"]) + ': <a href="/cartoon/' + str(lk["other_id"]) + '">'
+ _e(lk["other_title"] or lk["other_canonical"]) + '</a> <span class="muted">' + _e(lk["detail"] or "")
+ "</span></li>")
GRANULARITY_LABEL = {"item": "Item-level record", "folder": "Folder-level record (not a single cartoon)",
"box_range": "Box-range slot (range-level; date interpolated within the box span)"}
@router.get("/cartoon/{cid}", response_class=HTMLResponse)
def cartoon_page(cid: int):
d = _detail(cid)
date = d["date_exact"] or ((d["date_start"] + " – " + d["date_end"]) if d["date_start"] else "undated")
est = ' <span class="badge warn">estimated date</span>' if d["date_is_estimate"] else ""
caption = ("<blockquote>" + _e(d["caption"]) + "</blockquote>") if d["caption"] else ""
desc = ("<p>" + _e(d["description"]) + "</p>") if d["description"] else ""
people = ", ".join('<a href="/?person=' + _e(p) + '">' + _e(p) + "</a>" for p in d["people"]) or "—"
subjects = ", ".join('<a href="/?subject=' + _e(x) + '">' + _e(x) + "</a>" for x in d["subjects"]) or "—"
holder = (" (copyright: " + _e(d["copyright_holder"]) + ")") if d["copyright_holder"] else ""
links = "".join(_link_row(lk) for lk in d["links"][:50]) or '<li class="muted">none</li>'
rows = "".join(_source_row(s) for s in d["sources"])
pflags = d["provenance_flags"]
pbadge = ""
if prov.FLAG_BYPASS in pflags:
txt = prov.BADGE_TEXT if prov.FLAG_BYPASS_UNVERIFIED in pflags else \
"acquired via third-party reader past a bot challenge — also verified by a direct fetch"
pbadge += ' <span class="badge warn provenance-flag" data-flag="' + prov.FLAG_BYPASS + '">' + _e(txt) + "</span>"
if prov.FLAG_SECONDARY_ONLY in pflags:
pbadge += (' <span class="badge warn provenance-flag" data-flag="' + prov.FLAG_SECONDARY_ONLY
+ '">known only from a secondary citation — not seen in a holding repository</span>')
title = _e(d["title"] or "[untitled / not cataloged]")
figure = ""
di = d["display_image"]
if di and imagehost.is_allowed(di["url"]):
rights = (' · <a rel="noopener noreferrer" target="_blank" href="' + _e(di["rights_url"]) + '">rights</a>'
if di["rights_url"] else "")
rec = (' · <a rel="noopener noreferrer" target="_blank" href="' + _e(di["record_url"]) + '">View at '
+ _e(di["repository"]) + " ↗</a>") if di["record_url"] else ""
figure = ('<figure class="hero"><img src="' + _e(di["url"]) + '" alt="' + title
+ '" loading="lazy" referrerpolicy="no-referrer" decoding="async">'
+ '<figcaption class="credit small">Image: ' + _e(di["credit"])
+ " (displayed from the institution’s site; not copied here)" + rec + rights
+ "</figcaption></figure>")
body = PAGE.format(figure=figure,
head_title=_e(d["title"] or d["canonical_id"]), title=title, gran=_e(GRANULARITY_LABEL[d["granularity"]]),
date=_e(date), est=est + pbadge, pub=_e(d["publication"] or "publication unknown"), canonical=_e(d["canonical_id"]),
caption=caption, desc=desc, people=people, subjects=subjects, medium=_e(d["medium"] or "—"),
rights=_e(d["rights_text"] or "—") + holder, notes=_e(d["notes"] or "—"), n_sources=len(d["sources"]),
rows=rows, links=links, created=_e(d["created_at"]), updated=_e(d["updated_at"]))
return HTMLResponse(body)
PAGE = """<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"><meta name="robots" content="noindex,nofollow">
<title>{head_title} — Conrad Archive</title><link rel="stylesheet" href="/static/style.css"></head>
<body><header class="top"><a href="/" class="brand">Paul Conrad Master Archive</a>
<span class="muted">images only where the holding institution publishes one (shown from its site, never copied)</span></header>
<main class="detail"><p><a href="/">← back to search</a></p>
<h1>{title}</h1>
{figure}
<p class="meta"><span class="badge">{gran}</span> <b>{date}</b>{est} · {pub} · <span class="muted">{canonical}</span></p>
{caption}{desc}
<dl><dt>People</dt><dd>{people}</dd><dt>Subjects</dt><dd>{subjects}</dd><dt>Medium</dt><dd>{medium}</dd>
<dt>Rights</dt><dd>{rights}</dd><dt>Notes</dt><dd>{notes}</dd></dl>
<h2>Where to see it ({n_sources} source records)</h2>
<div class="tablewrap"><table><thead><tr><th>Repository</th><th>Collection</th><th>Identifier</th><th>Box</th>
<th>Folder</th><th>Access</th><th>Link-out</th><th>Provenance</th></tr></thead><tbody>{rows}</tbody></table></div>
<p class="muted small">Images are never copied or stored here. Where the holding institution publishes one it is displayed straight from the institution's site; otherwise follow the link-out.</p>
<h2>Related records</h2><ul>{links}</ul>
<p class="muted small">record created <span title="{created}">{created}</span> · updated {updated}</p>
</main></body></html>"""
@router.get("/robots.txt", response_class=PlainTextResponse)
def robots():
return "User-agent: *\nDisallow: /\n"
app = create_app()
def main() -> None:
import uvicorn
host = os.environ.get("CONRAD_HOST", "127.0.0.1")
port = int(os.environ.get("CONRAD_PORT", "8787"))
uvicorn.run(app, host=host, port=port, proxy_headers=True, forwarded_allow_ips="127.0.0.1")
if __name__ == "__main__":
main()