Extract stage: vision title + edition-cue extraction, offline-tested
Per-photo raw results cached under data/extract_raw/ (gitignored) so re-runs are free, --only re-extracts a single photo, and titles.json is rebuilt with dedupe that keeps conflicting-edition sightings separate. HEIC converts via pillow-heif; images downscale to <=1568px long edge; model JSON parsed defensively (code fences, surrounding prose). Vision callable is injectable — tests use a local fake; the real one uses the anthropic SDK (approved) with the model from config.toml. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Eric Wagoner
co-authored by
Claude Fable 5
Claude Fable 5
parent
96a1e09956
commit
b7a1ef8549
+4
-1
@@ -35,7 +35,10 @@ def extract(
|
||||
config: ConfigOpt = None,
|
||||
) -> None:
|
||||
"""Stage 1: extract game titles + edition cues from shelf photos."""
|
||||
_not_implemented("extract", 3)
|
||||
from bggpipe.extract import run_extract
|
||||
|
||||
cfg = load_config(config)
|
||||
run_extract(cfg, only=only)
|
||||
|
||||
|
||||
@app.command()
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
"""Stage 1 — extract game titles + edition cues from shelf photos.
|
||||
|
||||
Each photo is sent to Claude vision once and the raw result is cached under
|
||||
data/extract_raw/<photo>.json (making re-runs free and --only targeted).
|
||||
titles.json is rebuilt from all raw files on every run, deduping identical
|
||||
titles across photos unless their edition cues conflict — conflicting cues
|
||||
mean two different editions on the shelf, which stay separate entries.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
|
||||
from bggpipe.config import Config
|
||||
from bggpipe.normalize import normalize_title
|
||||
|
||||
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".heic"}
|
||||
MAX_LONG_EDGE = 1568 # Anthropic vision sweet spot (spec)
|
||||
JPEG_QUALITY = 85
|
||||
_CONFIDENCE_ORDER = {"low": 0, "medium": 1, "high": 2}
|
||||
_CODE_FENCE = re.compile(r"^```(?:json)?\s*|\s*```$", re.MULTILINE)
|
||||
|
||||
# vision(image_b64, media_type) -> model's raw text response
|
||||
VisionFn = Callable[[str, str], str]
|
||||
|
||||
VISION_PROMPT = """\
|
||||
You are cataloging a photo of board game shelves.
|
||||
|
||||
List every board game or card game title visible (spines and face-out boxes),
|
||||
transcribed exactly as printed. Exclude books, card sleeves, storage boxes,
|
||||
and anything that is not a board/card game.
|
||||
|
||||
If the SAME title appears more than once with visibly different boxes, report
|
||||
each as a separate entry — the owner has multiple editions of some games.
|
||||
|
||||
Respond with ONLY a JSON array, no prose, where each element is:
|
||||
{
|
||||
"title_raw": "title as printed on the box",
|
||||
"confidence": "high" | "medium" | "low",
|
||||
"publisher_hint": "publisher name or logo if legible, else null",
|
||||
"edition_hint": "edition wording if visible ('2nd Edition', 'Deluxe',
|
||||
'Big Box', anniversary marks), else null",
|
||||
"year_hint": copyright/print year as an integer if legible, else null,
|
||||
"language_hint": "language of the box text if determinable, else null",
|
||||
"art_notes": "distinctive box-art notes (colorway, artwork style) that
|
||||
could identify the edition, else null"
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
def prepare_image(path: Path) -> tuple[str, str]:
|
||||
"""Load a photo (JPEG/PNG/HEIC), downscale to <=1568px long edge, return
|
||||
(base64 JPEG, media_type). HEIC converts transparently via pillow-heif."""
|
||||
from PIL import Image
|
||||
from pillow_heif import register_heif_opener
|
||||
|
||||
register_heif_opener()
|
||||
with Image.open(path) as img:
|
||||
img = img.convert("RGB")
|
||||
long_edge = max(img.size)
|
||||
if long_edge > MAX_LONG_EDGE:
|
||||
scale = MAX_LONG_EDGE / long_edge
|
||||
img = img.resize(
|
||||
(round(img.width * scale), round(img.height * scale)),
|
||||
Image.LANCZOS,
|
||||
)
|
||||
buffer = io.BytesIO()
|
||||
img.save(buffer, format="JPEG", quality=JPEG_QUALITY)
|
||||
return base64.standard_b64encode(buffer.getvalue()).decode(), "image/jpeg"
|
||||
|
||||
|
||||
def parse_vision_json(text: str) -> list[dict]:
|
||||
"""Parse the model's JSON defensively: strip code fences, find the array."""
|
||||
cleaned = _CODE_FENCE.sub("", text).strip()
|
||||
if not cleaned.startswith("["):
|
||||
start, end = cleaned.find("["), cleaned.rfind("]")
|
||||
if start == -1 or end == -1:
|
||||
raise ValueError(f"no JSON array in vision response: {text[:200]!r}")
|
||||
cleaned = cleaned[start : end + 1]
|
||||
entries = json.loads(cleaned)
|
||||
if not isinstance(entries, list):
|
||||
raise ValueError("vision response is not a JSON array")
|
||||
return [e for e in entries if isinstance(e, dict) and e.get("title_raw")]
|
||||
|
||||
|
||||
def default_vision(model: str) -> VisionFn:
|
||||
"""The real Anthropic-API-backed vision callable (needs ANTHROPIC_API_KEY)."""
|
||||
import anthropic
|
||||
|
||||
client = anthropic.Anthropic()
|
||||
|
||||
def vision(image_b64: str, media_type: str) -> str:
|
||||
response = client.messages.create(
|
||||
model=model,
|
||||
max_tokens=4000,
|
||||
messages=[
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "image",
|
||||
"source": {
|
||||
"type": "base64",
|
||||
"media_type": media_type,
|
||||
"data": image_b64,
|
||||
},
|
||||
},
|
||||
{"type": "text", "text": VISION_PROMPT},
|
||||
],
|
||||
}
|
||||
],
|
||||
)
|
||||
return next(b.text for b in response.content if b.type == "text")
|
||||
|
||||
return vision
|
||||
|
||||
|
||||
def extract_photo(photo: Path, vision: VisionFn) -> list[dict]:
|
||||
image_b64, media_type = prepare_image(photo)
|
||||
raw_entries = parse_vision_json(vision(image_b64, media_type))
|
||||
entries = []
|
||||
for raw in raw_entries:
|
||||
year = raw.get("year_hint")
|
||||
entries.append(
|
||||
{
|
||||
"title_raw": str(raw["title_raw"]).strip(),
|
||||
"confidence": raw.get("confidence") or "medium",
|
||||
"publisher_hint": raw.get("publisher_hint") or "",
|
||||
"edition_hint": raw.get("edition_hint") or "",
|
||||
"year_hint": int(year)
|
||||
if isinstance(year, int | str) and str(year).isdigit()
|
||||
else None,
|
||||
"language_hint": raw.get("language_hint") or "",
|
||||
"art_notes": raw.get("art_notes") or "",
|
||||
"source_photos": [photo.name],
|
||||
}
|
||||
)
|
||||
return entries
|
||||
|
||||
|
||||
def _cues_conflict(a: dict, b: dict) -> bool:
|
||||
"""Two sightings conflict if any edition cue is set on both and differs —
|
||||
that means visibly different boxes, i.e. separate editions (spec)."""
|
||||
for key in ("publisher_hint", "edition_hint", "language_hint"):
|
||||
va, vb = a.get(key) or "", b.get(key) or ""
|
||||
if va and vb and normalize_title(va) != normalize_title(vb):
|
||||
return True
|
||||
ya, yb = a.get("year_hint"), b.get("year_hint")
|
||||
return bool(ya and yb and ya != yb)
|
||||
|
||||
|
||||
def _merge(a: dict, b: dict) -> dict:
|
||||
merged = dict(a)
|
||||
merged["source_photos"] = sorted(set(a["source_photos"]) | set(b["source_photos"]))
|
||||
if _CONFIDENCE_ORDER.get(b["confidence"], 1) > _CONFIDENCE_ORDER.get(
|
||||
a["confidence"], 1
|
||||
):
|
||||
merged["confidence"] = b["confidence"]
|
||||
for key in (
|
||||
"publisher_hint",
|
||||
"edition_hint",
|
||||
"year_hint",
|
||||
"language_hint",
|
||||
"art_notes",
|
||||
):
|
||||
merged[key] = merged.get(key) or b.get(key)
|
||||
return merged
|
||||
|
||||
|
||||
def dedupe_entries(entries: list[dict]) -> list[dict]:
|
||||
"""Collapse same-normalized-title sightings unless their cues conflict."""
|
||||
result: list[dict] = []
|
||||
for entry in entries:
|
||||
entry = {**entry, "title_normalized": normalize_title(entry["title_raw"])}
|
||||
for existing in result:
|
||||
if existing["title_normalized"] == entry[
|
||||
"title_normalized"
|
||||
] and not _cues_conflict(existing, entry):
|
||||
existing.update(_merge(existing, entry))
|
||||
break
|
||||
else:
|
||||
result.append(entry)
|
||||
return result
|
||||
|
||||
|
||||
def rebuild_titles(raw_dir: Path, titles_path: Path) -> list[dict]:
|
||||
entries: list[dict] = []
|
||||
for raw_file in sorted(raw_dir.glob("*.json")):
|
||||
entries.extend(json.loads(raw_file.read_text()))
|
||||
deduped = dedupe_entries(entries)
|
||||
titles_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
titles_path.write_text(json.dumps(deduped, indent=2, ensure_ascii=False) + "\n")
|
||||
return deduped
|
||||
|
||||
|
||||
def run_extract(
|
||||
cfg: Config, *, only: str | None = None, vision: VisionFn | None = None
|
||||
) -> list[dict]:
|
||||
photos = (
|
||||
sorted(
|
||||
p for p in cfg.photos_dir.iterdir() if p.suffix.lower() in IMAGE_EXTENSIONS
|
||||
)
|
||||
if cfg.photos_dir.is_dir()
|
||||
else []
|
||||
)
|
||||
if only:
|
||||
photos = [p for p in photos if p.name == only]
|
||||
if not photos:
|
||||
raise FileNotFoundError(
|
||||
f"--only {only!r}: no such photo in {cfg.photos_dir}"
|
||||
)
|
||||
if not photos:
|
||||
typer.echo(
|
||||
f"No photos found in {cfg.photos_dir}/ — add JPEG/PNG/HEIC files first."
|
||||
)
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
raw_dir = cfg.data_dir / "extract_raw"
|
||||
raw_dir.mkdir(parents=True, exist_ok=True)
|
||||
vision = vision or default_vision(cfg.model)
|
||||
|
||||
for photo in photos:
|
||||
raw_path = raw_dir / f"{photo.name}.json"
|
||||
if raw_path.exists() and not only:
|
||||
typer.echo(f" {photo.name}: already extracted, skipping")
|
||||
continue
|
||||
entries = extract_photo(photo, vision)
|
||||
raw_path.write_text(json.dumps(entries, indent=2, ensure_ascii=False) + "\n")
|
||||
typer.echo(f" {photo.name}: {len(entries)} title(s)")
|
||||
|
||||
deduped = rebuild_titles(raw_dir, cfg.titles_path)
|
||||
typer.echo(f"Wrote {len(deduped)} unique title(s) to {cfg.titles_path}.")
|
||||
return deduped
|
||||
Reference in New Issue
Block a user