Daily Art backend: curated Met pool, daily cached pick, /api/art (prototype)
The engine for the /art room (design-independent; deploy held for Codex review).
- goodnews/art.py: harvest a curated pool of public-domain HIGHLIGHT artworks from the
Met (isHighlight+isPublicDomain+hasImages -> masterworks, never potsherds; CC0). Daily
deterministic pick from the least-recently-shown (no soon-repeats, same for everyone),
fetch metadata + download the image to OUR cache (data/art_cache) so the homepage never
waits on or hotlinks the museum. Bulletproof: bad object/image falls through candidates;
a failed day keeps the last piece (room never empty). Injectable HTTP for tests.
- Schema: art_pool + daily_art. /api/art/today (edge-cacheable) + /api/art/image/{id}
(served from cache, immutable). CLI `art [--harvest] [--force]` + a non-fatal cycle step.
- Tests (5, mocked HTTP) + verified live against the Met: harvested 1641 works,
picked/cached "Repose" by John White Alexander. 371 tests green.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+187
@@ -0,0 +1,187 @@
|
||||
"""Daily Art — the /art room. One gorgeous public-domain masterwork a day, picked from
|
||||
a curated pool of museum highlights and cached to OUR origin (image + metadata), so the
|
||||
homepage never waits on, nor hotlinks, the museum.
|
||||
|
||||
Source: The Met Collection API (no key; public-domain works are CC0 — free, unrestricted,
|
||||
caching encouraged). Curation = isHighlight + isPublicDomain + hasImages, so the pool is
|
||||
masterworks, never potsherds. Built to be bulletproof: a failed pick falls through to the
|
||||
next candidate, and a failed day keeps yesterday's piece — the room is never empty.
|
||||
|
||||
Network calls go through module-level _http_* helpers so tests can monkeypatch them.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
from .localtime import local_today
|
||||
|
||||
MET_BASE = "https://collectionapi.metmuseum.org/public/collection/v1"
|
||||
# Broad, visual, museum-grade terms. Each is filtered to public-domain highlights with
|
||||
# images, then deduped — a diverse pool of a few thousand masterworks.
|
||||
HARVEST_QUERIES = ("painting", "portrait", "landscape", "still life", "flowers",
|
||||
"sculpture", "drawing", "garden", "river", "sunset")
|
||||
_UA = {"User-Agent": "upbeatBytes/1.0 (+https://upbeatbytes.com)"}
|
||||
_PICK_ATTEMPTS = 8 # candidates to try before giving up for the day
|
||||
_NO_REPEAT_POOL = 40 # pick the daily piece from the N least-recently-shown
|
||||
_MIN_IMAGE_BYTES = 3000 # smaller than this = not a real image
|
||||
|
||||
|
||||
def _http_json(url: str, timeout: int = 20) -> dict:
|
||||
req = urllib.request.Request(url, headers=_UA)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
return json.loads(r.read().decode("utf-8"))
|
||||
|
||||
|
||||
def _http_bytes(url: str, timeout: int = 30) -> tuple[bytes, str]:
|
||||
req = urllib.request.Request(url, headers=_UA)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
return r.read(), (r.headers.get("Content-Type") or "")
|
||||
|
||||
|
||||
def cache_dir() -> Path:
|
||||
"""Where cached images live — beside the DB, so the host cycle writes and the API
|
||||
container reads the same mounted volume."""
|
||||
override = os.environ.get("GOODNEWS_ART_CACHE")
|
||||
if override:
|
||||
d = Path(override)
|
||||
else:
|
||||
db = Path(os.environ.get("GOODNEWS_DB", "data/goodnews.sqlite3"))
|
||||
d = db.parent / "art_cache"
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
return d
|
||||
|
||||
|
||||
# --- harvest: build the curated pool of object IDs -------------------------------
|
||||
|
||||
def harvest_pool(conn: sqlite3.Connection, queries=HARVEST_QUERIES, source: str = "met") -> dict:
|
||||
"""Query the Met for public-domain highlight images across broad art terms; store the
|
||||
deduped object IDs. Cheap: each search returns all matching IDs in one call. Per-query
|
||||
failure is non-fatal."""
|
||||
found, errors = set(), 0
|
||||
for q in queries:
|
||||
url = (f"{MET_BASE}/search?isHighlight=true&hasImages=true&isPublicDomain=true"
|
||||
f"&q={urllib.request.quote(q)}")
|
||||
try:
|
||||
data = _http_json(url)
|
||||
for oid in (data.get("objectIDs") or []):
|
||||
if isinstance(oid, int):
|
||||
found.add(oid)
|
||||
except Exception: # noqa: BLE001 — non-fatal per query
|
||||
errors += 1
|
||||
before = conn.execute("SELECT COUNT(*) FROM art_pool WHERE source=?", (source,)).fetchone()[0]
|
||||
conn.executemany(
|
||||
"INSERT OR IGNORE INTO art_pool (source, object_id) VALUES (?, ?)",
|
||||
[(source, oid) for oid in found],
|
||||
)
|
||||
conn.commit()
|
||||
after = conn.execute("SELECT COUNT(*) FROM art_pool WHERE source=?", (source,)).fetchone()[0]
|
||||
return {"queried": len(queries), "errors": errors, "found": len(found),
|
||||
"added": after - before, "pool": after}
|
||||
|
||||
|
||||
# --- daily pick: choose, fetch, cache --------------------------------------------
|
||||
|
||||
def _object(object_id: int) -> dict:
|
||||
return _http_json(f"{MET_BASE}/objects/{object_id}")
|
||||
|
||||
|
||||
def _download_image(obj: dict, object_id: int) -> str | None:
|
||||
"""Download the web-large (then full) image to our cache; return the filename or None."""
|
||||
for key in ("primaryImageSmall", "primaryImage"):
|
||||
url = obj.get(key)
|
||||
if not url:
|
||||
continue
|
||||
try:
|
||||
data, ctype = _http_bytes(url)
|
||||
except Exception: # noqa: BLE001
|
||||
continue
|
||||
if not ctype.startswith("image/") or len(data) < _MIN_IMAGE_BYTES:
|
||||
continue
|
||||
ext = ".png" if "png" in ctype else ".jpg"
|
||||
fname = f"{object_id}{ext}"
|
||||
try:
|
||||
(cache_dir() / fname).write_bytes(data)
|
||||
except OSError:
|
||||
return None
|
||||
return fname
|
||||
return None
|
||||
|
||||
|
||||
def _candidates(conn: sqlite3.Connection, art_date: str, source: str) -> list[int]:
|
||||
"""The N least-recently-shown pool IDs, rotated deterministically by the date so the
|
||||
same piece shows for everyone that day and pieces don't repeat soon."""
|
||||
rows = conn.execute(
|
||||
"SELECT object_id FROM art_pool WHERE source=? ORDER BY shown_at IS NOT NULL, shown_at, object_id LIMIT ?",
|
||||
(source, _NO_REPEAT_POOL),
|
||||
).fetchall()
|
||||
ids = [r[0] for r in rows]
|
||||
if not ids:
|
||||
return ids
|
||||
seed = int(hashlib.sha256(art_date.encode()).hexdigest(), 16) % len(ids)
|
||||
return ids[seed:] + ids[:seed] # rotate so the daily choice is stable but varies
|
||||
|
||||
|
||||
def pick_daily(conn: sqlite3.Connection, art_date: str | None = None, source: str = "met",
|
||||
force: bool = False) -> dict | None:
|
||||
"""Pick + cache the day's art. Idempotent (skips if today's already done unless force).
|
||||
Tries successive candidates so a bad object/image never breaks the day; returns the
|
||||
stored row, or None if nothing could be fetched (caller keeps the prior day's piece)."""
|
||||
art_date = art_date or local_today()
|
||||
existing = conn.execute("SELECT * FROM daily_art WHERE art_date=?", (art_date,)).fetchone()
|
||||
if existing and not force:
|
||||
return dict(existing)
|
||||
for oid in _candidates(conn, art_date, source):
|
||||
try:
|
||||
obj = _object(oid)
|
||||
except Exception: # noqa: BLE001
|
||||
continue
|
||||
if not obj.get("isPublicDomain"):
|
||||
continue
|
||||
fname = _download_image(obj, oid)
|
||||
if not fname:
|
||||
continue
|
||||
conn.execute(
|
||||
"INSERT INTO daily_art (art_date, source, object_id, title, artist, date_text, medium, "
|
||||
"department, credit, source_url, image_file) VALUES (?,?,?,?,?,?,?,?,?,?,?) "
|
||||
"ON CONFLICT(art_date) DO UPDATE SET object_id=excluded.object_id, title=excluded.title, "
|
||||
"artist=excluded.artist, date_text=excluded.date_text, medium=excluded.medium, "
|
||||
"department=excluded.department, credit=excluded.credit, source_url=excluded.source_url, "
|
||||
"image_file=excluded.image_file",
|
||||
(art_date, source, oid, obj.get("title") or "Untitled",
|
||||
obj.get("artistDisplayName") or None, obj.get("objectDate") or None,
|
||||
obj.get("medium") or None, obj.get("department") or None,
|
||||
obj.get("creditLine") or None, obj.get("objectURL") or None, fname),
|
||||
)
|
||||
conn.execute("UPDATE art_pool SET shown_at=? WHERE source=? AND object_id=?",
|
||||
(art_date, source, oid))
|
||||
conn.commit()
|
||||
return dict(conn.execute("SELECT * FROM daily_art WHERE art_date=?", (art_date,)).fetchone())
|
||||
return None # nothing fetched today — get_today falls back to the latest piece
|
||||
|
||||
|
||||
def get_today(conn: sqlite3.Connection, art_date: str | None = None) -> dict | None:
|
||||
"""Today's piece if present, else the most recent one cached (room is never empty)."""
|
||||
if art_date:
|
||||
row = conn.execute("SELECT * FROM daily_art WHERE art_date=?", (art_date,)).fetchone()
|
||||
if row:
|
||||
return dict(row)
|
||||
row = conn.execute("SELECT * FROM daily_art ORDER BY art_date DESC LIMIT 1").fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def run_daily(conn: sqlite3.Connection, source: str = "met") -> dict:
|
||||
"""Cycle entry point: ensure the pool exists, then ensure today has a piece. Bounded
|
||||
and non-fatal — safe to call every cycle (it no-ops once the day is picked)."""
|
||||
pool = conn.execute("SELECT COUNT(*) FROM art_pool WHERE source=?", (source,)).fetchone()[0]
|
||||
harvested = None
|
||||
if pool == 0:
|
||||
harvested = harvest_pool(conn, source=source)
|
||||
picked = pick_daily(conn, source=source)
|
||||
return {"pool": conn.execute("SELECT COUNT(*) FROM art_pool WHERE source=?", (source,)).fetchone()[0],
|
||||
"harvested": harvested, "picked_object": picked.get("object_id") if picked else None}
|
||||
Reference in New Issue
Block a user