diff --git a/CHANGELOG.md b/CHANGELOG.md index ee1e501..5218eb1 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 6e6c190..4ffa99d 100644 --- a/README.md +++ b/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 diff --git a/colophon/epub.py b/colophon/epub.py new file mode 100644 index 0000000..87635d7 --- /dev/null +++ b/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)[^>]*>.*?', ' ', 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)]*>(.*?)', 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)]*>(.*?)', 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)]*>', 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)]*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, + } diff --git a/colophon/grimmory.py b/colophon/grimmory.py index 1e5473a..29f59a5 100644 --- a/colophon/grimmory.py +++ b/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: diff --git a/colophon/hardcover.py b/colophon/hardcover.py index eedf5a5..589324b 100644 --- a/colophon/hardcover.py +++ b/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.""" diff --git a/colophon/resolver.py b/colophon/resolver.py index 596d329..0ae7498 100644 --- a/colophon/resolver.py +++ b/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"): diff --git a/tests/test_epub.py b/tests/test_epub.py new file mode 100644 index 0000000..fb74a3a --- /dev/null +++ b/tests/test_epub.py @@ -0,0 +1,98 @@ +"""EPUB inspection — pure / stdlib only, no network. Run: python -m unittest -v""" +import os +import tempfile +import unittest +import zipfile + +from colophon import epub + +CONTAINER = ('' + '') + +COPYRIGHT = ("

Copyright

LOCK IN. Copyright © 2014 by John Scalzi. " + "All rights reserved. ISBN 978-0-7653-7586-5. Published by Tom Doherty " + "Associates, LLC. First Edition: August 2014.

") +CHAPTER = "

It was a bright cold morning and the clock struck thirteen.

" +COVER = "

Cover

" + + +def _opf(identifiers, with_copyright=True): + ids = "".join(identifiers) + items = ['', + ''] + spine = ['', ''] + if with_copyright: + items.insert(1, '') + spine.insert(1, '') + return ('' + ids + + 'Lock InJohn Scalzi' + 'Tor2014-08-26T00:00:00+00:00' + '' + "".join(items) + '' + + "".join(spine) + '') + + +def _write_epub(path, opf, with_copyright=True): + with zipfile.ZipFile(path, "w") as z: + z.writestr("mimetype", "application/epub+zip") + z.writestr("META-INF/container.xml", CONTAINER) + z.writestr("OEBPS/content.opf", opf) + z.writestr("OEBPS/cover.xhtml", COVER) + z.writestr("OEBPS/chap1.xhtml", CHAPTER) + if with_copyright: + z.writestr("OEBPS/copyright.xhtml", COPYRIGHT) + + +class Inspect(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + + def _epub(self, identifiers, with_copyright=True, name="b.epub"): + p = os.path.join(self.dir, name) + _write_epub(p, _opf(identifiers, with_copyright), with_copyright) + return p + + def test_opf_isbns_normalized_and_ordered(self): + r = epub.inspect(self._epub([ + 'urn:isbn:9781466849358', + '978-0-7653-7586-5'])) + self.assertEqual(r["opf_isbns"], ["9781466849358", "9780765375865"]) + + def test_uuid_identifier_is_not_an_isbn(self): + r = epub.inspect(self._epub([ + 'urn:uuid:17ecd1a7-81f1-4703-adb6-0b02375770c4'])) + self.assertEqual(r["opf_isbns"], []) + + def test_isbn10_taken_only_when_marked_or_valid(self): + # marked ISBN-10 (valid checksum 0765375863) is kept + r = epub.inspect(self._epub(['ISBN 0-7653-7586-3'])) + self.assertIn("0765375863", r["opf_isbns"]) + + def test_colophon_text_found(self): + r = epub.inspect(self._epub(['urn:isbn:9781466849358'])) + self.assertIn("All rights reserved", r["colophon_text"]) + self.assertIn("Copyright", r["colophon_text"]) + + def test_metadata_fields(self): + r = epub.inspect(self._epub(['urn:isbn:9781466849358'])) + self.assertEqual(r["opf_title"], "Lock In") + self.assertEqual(r["publisher"], "Tor") + self.assertEqual(r["year"], "2014") + + def test_no_colophon_page_yields_empty_text(self): + r = epub.inspect(self._epub(['urn:isbn:9781466849358'], + with_copyright=False)) + self.assertEqual(r["colophon_text"], "") + + def test_malformed_file_returns_none(self): + p = os.path.join(self.dir, "bad.epub") + with open(p, "w") as f: + f.write("not a zip") + self.assertIsNone(epub.inspect(p)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resolve_escalation.py b/tests/test_resolve_escalation.py new file mode 100644 index 0000000..10c56ba --- /dev/null +++ b/tests/test_resolve_escalation.py @@ -0,0 +1,48 @@ +"""run_resolve EPUB-escalation glue — pure logic, mocked I/O. Run: python -m unittest -v""" +import unittest +from unittest import mock + +from colophon import audit, resolver + +_BOOK = {"book_id": 1, "title": "X", "authors": "Y"} + + +def _propose(conf, source=None, cid="a"): + return {"book_id": 1, "title": "X", "action": "propose", "confidence": conf, + "chosen_id": cid, "chosen_title": "X", "isbn": "9780000000001", + "slug": "x", "source": source} + + +class Escalation(unittest.TestCase): + def test_below_threshold_inspects_and_swaps_in_better(self): + seq = [_propose(0.6), _propose(0.97, source="epub-opf", cid="b")] # first pass, then escalated + with mock.patch.object(audit, "all_books", return_value=[_BOOK]), \ + mock.patch.object(resolver, "_epub_signals", return_value={"opf_isbns": ["9780000000001"]}) as sig, \ + mock.patch.object(resolver, "resolve", side_effect=seq) as res: + out = resolver.run_resolve(book_ids=[1], apply=False, store=None, min_conf=0.9) + sig.assert_called_once_with(1) + self.assertEqual(res.call_count, 2) + self.assertIsNotNone(res.call_args.kwargs.get("file_signals")) # 2nd call carried the signals + self.assertEqual(out["proposals"][0]["source"], "epub-opf") + self.assertEqual(out["proposals"][0]["confidence"], 0.97) + + def test_above_threshold_never_inspects(self): + with mock.patch.object(audit, "all_books", return_value=[_BOOK]), \ + mock.patch.object(resolver, "_epub_signals") as sig, \ + mock.patch.object(resolver, "resolve", side_effect=[_propose(0.95)]): + out = resolver.run_resolve(book_ids=[1], apply=False, store=None, min_conf=0.9) + sig.assert_not_called() + self.assertEqual(out["proposals"][0]["confidence"], 0.95) + + def test_escalation_kept_only_when_strictly_better(self): + seq = [_propose(0.6, cid="a"), _propose(0.5, source="epub-colophon", cid="b")] # worse → discard + with mock.patch.object(audit, "all_books", return_value=[_BOOK]), \ + mock.patch.object(resolver, "_epub_signals", return_value={"colophon_text": "x"}), \ + mock.patch.object(resolver, "resolve", side_effect=seq): + out = resolver.run_resolve(book_ids=[1], apply=False, store=None, min_conf=0.9) + self.assertEqual(out["proposals"][0]["chosen_id"], "a") # kept the original + self.assertIsNone(out["proposals"][0]["source"]) + + +if __name__ == "__main__": + unittest.main()