feat: inspect EPUB colophon to resolve low-confidence mis-seeds

CHANGELOG.md

@@ -6,6 +6,17 @@ All notable changes to this project are documented here. The format is based on
 
 ## [Unreleased]
 
+### Added
+- **EPUB inspection for low-confidence mis-seeds** — when `resolve` lands below the
+  confidence gate (or finds no match), it now inspects the book's own EPUB and
+  re-adjudicates with two extra signals: the OPF `dc:identifier` ISBN (a real ISBN that
+  resolves to a same-author Hardcover book heals deterministically, no LLM —
+  `source=epub-opf`), and the colophon / copyright page text folded into the Haiku prompt
+  (`source=epub-colophon`). Only below-threshold books pay for the file I/O. Gated on the
+  new `COLOPHON_BOOKS_ROOT` (the host path the library mounts from); unset ⇒ feature off.
+  New module `colophon/epub.py` (stdlib-only) plus `hardcover.book_by_isbn` and
+  `grimmory.epub_path` helpers.
+
 ## [0.2.0] — 2026-06-04
 
 ### Added

README.md

@@ -82,6 +82,7 @@ Copy `.env.example` to `.env` and fill it in (or export the variables):
 | `ANTHROPIC_API_KEY` | LLM adjudication (omit to use the `claude` CLI) |
 | `COLOPHON_DB`, `COLOPHON_REPORTS` | where to write the changelog + reports |
 | `COLOPHON_RESOLVE_RETRY_DAYS` | re-query a cached-unresolvable mis-seed after N days (default `0` = never) |
+| `COLOPHON_BOOKS_ROOT` | host path of the library root; set it to let `resolve` inspect a below-threshold book's own EPUB (OPF ISBN + colophon). Unset ⇒ feature off |
 | `SMTP_*`, `OVERSIGHT_TO` | oversight + `maintain --email` summaries |
 
 ## Usage

colophon/epub.py

@@ -0,0 +1,103 @@
+"""EPUB inspection — read-only, standard library only.
+
+Pulls the two identity signals colophon falls back on when title/author resolution
+is weak: the OPF `dc:identifier` ISBN(s) — deterministic, free — and the book's
+colophon / copyright page text, for the Haiku adjudicator (the printed *print* ISBN
+there often matches Hardcover's canonical edition better than the OPF ebook ISBN).
+Never raises for a malformed/DRM'd file: returns None.
+"""
+import os
+import re
+import zipfile
+
+# Keywords that mark a colophon / copyright page (front- or back-matter).
+_COLO = re.compile(r'(?i)\b(isbn|all rights reserved|first published|copyright|©|'
+                   r'published by|a cip|library of congress|e-?book|edition)\b')
+_MAX_COLOPHON = 4000          # ~300 tokens — the marginal cost measured for the resolve call
+_FRONT_BACK = 6              # how many leading/trailing spine items count as matter
+
+
+def _strip(html):
+    html = re.sub(r'(?is)<(script|style)[^>]*>.*?</\1>', ' ', html)
+    return re.sub(r'\s+', ' ', re.sub(r'<[^>]+>', ' ', html)).strip()
+
+
+def _isbn10_ok(s):
+    if len(s) != 10:
+        return False
+    try:
+        total = sum((10 - i) * (10 if c in "Xx" else int(c)) for i, c in enumerate(s))
+    except ValueError:
+        return False
+    return total % 11 == 0
+
+
+def _isbns(opf):
+    """Normalized ISBN-13/10 from every `dc:identifier`, in document order, deduped.
+    A 10-digit value is only taken when the element says ISBN or it checksums — a bare
+    digit run (e.g. a UUID) is not an ISBN."""
+    out = []
+    for raw in re.findall(r'(?is)<dc:identifier[^>]*>(.*?)</dc:identifier>', opf):
+        txt = _strip(raw)
+        d = re.sub(r'[^0-9Xx]', '', txt)
+        if len(d) == 13 and d.isdigit() and d[:3] in ("978", "979"):
+            out.append(d)
+        elif len(d) == 10 and ("isbn" in txt.lower() or _isbn10_ok(d)):
+            out.append(d.upper())
+    return list(dict.fromkeys(out))
+
+
+def _dc(opf, tag):
+    m = re.search(rf'(?is)<dc:{tag}[^>]*>(.*?)</dc:{tag}>', opf)
+    return _strip(m.group(1)) if m else None
+
+
+def inspect(path):
+    """Return {opf_isbns, colophon_text, opf_title, publisher, year} or None.
+
+    `colophon_text` is "" when no page looks like a colophon (e.g. indie EPUBs with
+    only a cover) — the caller should treat empty as "no usable colophon signal"."""
+    try:
+        z = zipfile.ZipFile(path)
+        container = z.read('META-INF/container.xml').decode('utf-8', 'replace')
+        opf_path = re.search(r'full-path="([^"]+)"', container).group(1)
+        opf = z.read(opf_path).decode('utf-8', 'replace')
+    except Exception:
+        return None
+
+    opf_dir = os.path.dirname(opf_path)
+    manifest = {}
+    for item in re.findall(r'(?is)<item\b[^>]*>', opf):
+        mid, href = re.search(r'id="([^"]+)"', item), re.search(r'href="([^"]+)"', item)
+        if mid and href:
+            manifest[mid.group(1)] = href.group(1)
+    spine = re.findall(r'(?is)<itemref\b[^>]*idref="([^"]+)"', opf)
+
+    pages = []
+    for idref in spine:
+        href = manifest.get(idref)
+        if not href:
+            continue
+        zp = href.split('#')[0]
+        zp = (opf_dir + '/' + zp).lstrip('/') if opf_dir else zp
+        try:
+            pages.append(_strip(z.read(zp).decode('utf-8', 'replace')))
+        except KeyError:
+            continue
+
+    best_text, best_score, best_hits = "", -1, 0
+    n = len(pages)
+    for i, t in enumerate(pages):
+        if not t:
+            continue
+        hits = len(_COLO.findall(t[:3000]))
+        score = hits + (2 if (i < _FRONT_BACK or i >= n - _FRONT_BACK) else 0) + (1 if len(t) < 4000 else 0)
+        if score > best_score:
+            best_text, best_score, best_hits = t, score, hits
+    return {
+        "opf_isbns": _isbns(opf),
+        "colophon_text": best_text[:_MAX_COLOPHON] if best_hits else "",
+        "opf_title": _dc(opf, "title"),
+        "publisher": _dc(opf, "publisher"),
+        "year": (_dc(opf, "date") or "")[:4] or None,
+    }

colophon/grimmory.py

@@ -19,6 +19,10 @@
 ADMIN_GROUP = os.environ.get("COLOPHON_ADMIN_GROUP", "admin")
 DB_CONTAINER = os.environ.get("COLOPHON_DB_CONTAINER", "grimmory-db")
 DB_NAME = os.environ.get("COLOPHON_DB_NAME", "grimmory")
+# Host path the grimmory library mounts from (e.g. /mnt/media/.../library). Set this
+# to enable EPUB inspection in resolve; book_file paths are relative to it. Unset =
+# the feature stays off (the resolver simply never inspects files).
+BOOKS_ROOT = os.environ.get("COLOPHON_BOOKS_ROOT")
 
 
 class GrimmoryError(Exception):
@@ -148,6 +152,26 @@ def snapshot(book_id):
     return snap
 
 
+def epub_path(book_id):
+    """Host filesystem path of the book's largest EPUB, or None — read-only.
+
+    Requires BOOKS_ROOT (COLOPHON_BOOKS_ROOT): `book_file` stores paths relative to
+    grimmory's in-container library root, so we re-root them on the host. Returns None
+    when BOOKS_ROOT is unset, the book has no EPUB, or no row matches."""
+    if not BOOKS_ROOT:
+        return None
+    out = _db(
+        "SELECT IFNULL(file_sub_path,''), file_name FROM book_file "
+        f"WHERE book_id={int(book_id)} AND LOWER(file_name) LIKE '%.epub' "
+        "ORDER BY file_size_kb DESC LIMIT 1;"
+    ).strip()
+    if not out:
+        return None
+    parts = out.split("\t")
+    sub, name = (parts[0], parts[1]) if len(parts) == 2 else ("", parts[-1])
+    return os.path.join(BOOKS_ROOT, sub, name) if sub else os.path.join(BOOKS_ROOT, name)
+
+
 def signature(snap):
     """The fields whose change means a refresh has landed."""
     if not snap:

colophon/hardcover.py

@@ -8,6 +8,7 @@
 """
 import json
 import os
+import re
 import subprocess
 import time
 import urllib.error
@@ -90,6 +91,27 @@ def book_by_id(hcid):
     }
 
 
+def _norm_isbn(s):
+    return re.sub(r'[^0-9Xx]', '', s or '').upper()
+
+
+def book_by_isbn(isbn):
+    """Resolve an ISBN-13/10 to a Hardcover book (same dict shape as `book_by_id`),
+    or None. Goes through the allowed Typesense `search` and matches a hit whose
+    `isbns` contains the ISBN — broad Hasura `editions(where:{isbn_13})` filters are
+    403-restricted on this key tier (see `search`)."""
+    want = _norm_isbn(isbn)
+    if not want:
+        return None
+    for hit in search(want, per_page=8):
+        if hit.get("hcid") and want in {_norm_isbn(x) for x in (hit.get("isbns") or [])}:
+            try:
+                return book_by_id(hit["hcid"])
+            except Exception:
+                return None
+    return None
+
+
 def search(query_text, per_page=8):
     """Full-text book search (Hardcover's Typesense `search` — the broad Hasura
     filter is 403-restricted). Returns a list of candidate dicts."""

colophon/resolver.py

@@ -9,7 +9,7 @@
 import os
 import time
 
-from . import anthropic, audit, hardcover, matcher
+from . import anthropic, audit, epub, grimmory, hardcover, matcher
 from .heal import assert_preconditions, heal_book
 
 AUTO_CONF = 0.9          # auto-apply re-seeds at or above this confidence
@@ -68,6 +68,8 @@ def _record_skip(store, book, p):
     "Rules: a candidate with compilation=true is an omnibus/collection — pick it "
     "ONLY if the book itself is that collection, never for a single title; the "
     "author must match; if no candidate is clearly the same work, return NONE. "
+    "The book's own colophon / copyright page may be included — use its printed ISBN, "
+    "publisher and edition as strong evidence, but chosen_id must still be a candidate id. "
     "Never invent an id — chosen_id must be one of the given ids or \"NONE\". Be "
     "conservative: NONE is better than a wrong fix."
 )
@@ -84,9 +86,53 @@ def _record_skip(store, book, p):
 }
 
 
-def resolve(book):
-    """book: {book_id, title, authors, hcid, ...}. Returns a proposal dict."""
+def _resolve_by_isbn(bid, title, authors, isbns):
+    """Deterministic: an OPF ISBN that resolves to a same-author Hardcover book needs
+    no LLM. Heals to the exact edition (the ISBN-13 from the file) when valid, else the
+    book's canonical ISBN. Returns a propose dict (source=epub-opf) or None."""
+    for isbn in isbns:
+        try:
+            cand = hardcover.book_by_isbn(isbn)
+        except Exception:
+            continue
+        if not cand:
+            continue
+        if authors and not matcher._title_match(authors, ", ".join(cand.get("authors") or [])):
+            continue
+        isbn13 = isbn if (len(isbn) == 13 and isbn.isdigit()) else cand.get("isbn")
+        if not isbn13:
+            continue
+        return {"book_id": bid, "title": title, "action": "propose", "source": "epub-opf",
+                "chosen_id": str(cand["hcid"]), "chosen_title": cand.get("title"),
+                "isbn": isbn13, "slug": cand.get("slug"), "confidence": 0.97,
+                "is_set": None, "reason": f"OPF ISBN {isbn} → {cand.get('title')!r}"}
+    return None
+
+
+def _epub_signals(book_id):
+    """Read-only: the book's local EPUB → epub.inspect() dict, or None. Never raises;
+    None when there is no books root, no EPUB, or no usable signal in the file."""
+    try:
+        path = grimmory.epub_path(book_id)
+        if not path or not os.path.exists(path):
+            return None
+        sig = epub.inspect(path)
+        return sig if (sig and (sig.get("opf_isbns") or sig.get("colophon_text"))) else None
+    except Exception:
+        return None
+
+
+def resolve(book, file_signals=None):
+    """book: {book_id, title, authors, hcid, ...}. Returns a proposal dict.
+
+    With file_signals (from epub.inspect): a real OPF ISBN that resolves to a
+    same-author Hardcover book short-circuits the LLM (source=epub-opf); otherwise the
+    colophon text is folded into the adjudication prompt as extra evidence."""
     bid, title, authors = book["book_id"], book["title"], book.get("authors", "")
+    if file_signals:
+        sc = _resolve_by_isbn(bid, title, authors, file_signals.get("opf_isbns") or [])
+        if sc:
+            return sc
     cands = hardcover.search(f"{title} {authors}".strip(), per_page=8)
     catalog = [{
         "id": str(c["hcid"]), "title": c["title"], "subtitle": c.get("subtitle"),
@@ -97,7 +143,9 @@ def resolve(book):
     if not catalog:
         return {"book_id": bid, "title": title, "action": "none", "reason": "no search candidates"}
     valid_ids = {c["id"] for c in catalog}
+    colophon = (file_signals or {}).get("colophon_text") or ""
     user = ("BOOK TO IDENTIFY:\n" + json.dumps({"title": title, "authors": authors}) +
+            ("\n\nBOOK'S OWN COLOPHON / COPYRIGHT PAGE:\n" + colophon if colophon else "") +
             "\n\nCANDIDATES:\n" + json.dumps(catalog, ensure_ascii=False) +
             "\n\nReturn the correct candidate id, or NONE.")
     try:
@@ -119,6 +167,7 @@ def resolve(book):
         "chosen_title": (detail or {}).get("title") or next(c["title"] for c in catalog if c["id"] == chosen),
         "isbn": (detail or {}).get("isbn"), "slug": (detail or {}).get("slug"),
         "confidence": conf, "is_set": d.get("is_set"), "reason": reason,
+        "source": "epub-colophon" if colophon else None,
     }
 
 
@@ -154,6 +203,14 @@ def run_resolve(limit=None, book_ids=None, apply=False, min_conf=AUTO_CONF, g=No
         if i > 0:
             time.sleep(1.0)
         p = resolve(b)
+        # Below the gate (or no match)? Inspect the actual EPUB and re-adjudicate with
+        # its OPF ISBN + colophon as extra signal — the only books that pay for file I/O.
+        if not p.get("source") and (p["action"] == "none" or (p.get("confidence") or 0) < min_conf):
+            sig = _epub_signals(b["book_id"])
+            if sig:
+                p2 = resolve(b, file_signals=sig)
+                if (p2.get("confidence") or 0) > (p.get("confidence") or 0):
+                    p = p2
         if (apply and p["action"] == "propose"
                 and (p.get("confidence") or 0) >= min_conf and p.get("isbn")):
             try:
@@ -187,7 +244,8 @@ def render(result):
          "## Re-seeds\n"]
     for p in sorted(proposed, key=lambda p: -(p.get("confidence") or 0)):
         tag = "✓ APPLIED" if p.get("applied") else ("apply-FAILED" if p.get("apply_error") else "proposed")
-        L.append(f"- `{p['book_id']}` {p['title']!r}  [{tag}]")
+        via = f"  _via {p['source']}_" if p.get("source") else ""
+        L.append(f"- `{p['book_id']}` {p['title']!r}  [{tag}]{via}")
         L.append(f"    → **{p['chosen_title']!r}**  (hcid {p['chosen_id']}, isbn {p.get('isbn')}, "
                  f"conf {p.get('confidence')}{', SET' if p.get('is_set') else ''})")
         if p.get("apply_error"):