diff --git a/CHANGELOG.md b/CHANGELOG.md index 16d216c..ee1e501 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,25 @@ All notable changes to this project are documented here. The format is based on ## [Unreleased] +## [0.2.0] — 2026-06-04 + +### Added +- **`maintain` command** — runs backfill + resolve in a single process (each phase + guarded so one failing doesn't skip the other), writes a one-page summary, and with + `--email` always sends it as a daily heartbeat — including on an aborted run. Exits + non-zero on a phase failure (so a supervising timer is marked failed and retries); an + SMTP failure does not change the exit code. +- **Resolve skip-list** — mis-seeds that resolve to no-match or a below-threshold match + are remembered (`resolve_skip` table) and not re-queried on later runs, cutting the + repeated Hardcover + LLM calls for permanently-unresolvable books. The skip key is the + normalized title+author, so a book is re-queried automatically once its metadata + changes. `resolve --force` re-checks everything; `resolve --clear-skips` forgets the + list; `COLOPHON_RESOLVE_RETRY_DAYS` (default `0` = never) sets an optional re-check TTL. + +### Changed +- `resolve` reports now show the cached-unresolvable set, surfacing standing + below-threshold matches with a one-line manual-apply command. + ## [0.1.0] — 2026-06-03 Initial public release. @@ -26,5 +45,6 @@ Initial public release. per-run limits, and a circuit-breaker. - Standard-library-only; configurable entirely via environment variables. -[Unreleased]: https://github.com/vidaks/colophon/compare/v0.1.0...HEAD +[Unreleased]: https://github.com/vidaks/colophon/compare/v0.2.0...HEAD +[0.2.0]: https://github.com/vidaks/colophon/releases/tag/v0.2.0 [0.1.0]: https://github.com/vidaks/colophon/releases/tag/v0.1.0 diff --git a/README.md b/README.md index 17b897a..6e6c190 100644 --- a/README.md +++ b/README.md @@ -81,7 +81,8 @@ Copy `.env.example` to `.env` and fill it in (or export the variables): | `COLOPHON_HARDCOVER_KEY` | Hardcover API key (or `COLOPHON_HARDCOVER_QUERY_CMD`) | | `ANTHROPIC_API_KEY` | LLM adjudication (omit to use the `claude` CLI) | | `COLOPHON_DB`, `COLOPHON_REPORTS` | where to write the changelog + reports | -| `SMTP_*`, `OVERSIGHT_TO` | weekly oversight email (only sent when flagged) | +| `COLOPHON_RESOLVE_RETRY_DAYS` | re-query a cached-unresolvable mis-seed after N days (default `0` = never) | +| `SMTP_*`, `OVERSIGHT_TO` | oversight + `maintain --email` summaries | ## Usage @@ -90,7 +91,9 @@ python -m colophon.cli precheck # assert the files-never-touched python -m colophon.cli backfill # survey + propose (dry-run) python -m colophon.cli backfill --apply # heal broken ISBNs python -m colophon.cli resolve --apply # LLM-resolve mis-identified books (>= 0.9 conf) +python -m colophon.cli resolve --force # re-query the cached-unresolvable mis-seeds python -m colophon.cli series-audit # series-number report (read-only) +python -m colophon.cli maintain --apply --email # backfill + resolve in one run + summary email python -m colophon.cli oversight --days 7 # weekly changelog review + verdict python -m colophon.cli log # the change history python -m colophon.cli revert --apply # undo a metadata run diff --git a/colophon/cli.py b/colophon/cli.py index 9688a40..de9bbac 100644 --- a/colophon/cli.py +++ b/colophon/cli.py @@ -140,11 +140,15 @@ def cmd_audit(args, g, store): def cmd_resolve(args, g, store): from . import anthropic from .resolver import render, run_resolve + if args.clear_skips: + n = store.skip_clear() + print(f"cleared {n} cached-unresolvable entr{'y' if n == 1 else 'ies'}") + return 0 if not anthropic.have_key(): print("(no ANTHROPIC_API_KEY — using the `claude` CLI; production needs the vaulted key)") try: - res = run_resolve(limit=args.limit, book_ids=args.book or None, - apply=args.apply, min_conf=args.min_conf, g=g, store=store) + res = run_resolve(limit=args.limit, book_ids=args.book or None, apply=args.apply, + min_conf=args.min_conf, g=g, store=store, force=args.force) except PreconditionError as e: print(e) return 2 @@ -194,6 +198,35 @@ def cmd_oversight(args, g, store): return 0 +def cmd_maintain(args, g, store): + """Run the nightly sweep (backfill + resolve) and report. With --email, always + send the summary (a daily heartbeat) — even on an aborted run. Exits non-zero + when a phase failed so the systemd unit fails + the next timer fire retries; an + SMTP failure does NOT change the exit code (the heal work still happened).""" + from . import maintain as M + res = body = None + try: + res = M.run_maintain(g, store, limit=args.limit, min_conf=args.min_conf, + apply=args.apply, force=args.force) + body = M.render_summary(res) + reports_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, "reports")) + os.makedirs(reports_dir, exist_ok=True) + path = os.path.join(reports_dir, f"maintain-{time.strftime('%Y%m%dT%H%M%S')}.md") + with open(path, "w") as f: + f.write(body) + print(body) + print(f"(report written to {path})") + finally: + if args.email: + from . import oversight + subject = M.subject(res) if res else "Colophon daily [CRASH] — see host journal" + mail = body or ("Colophon maintain crashed before producing a summary.\n" + "Check `journalctl -u colophon.service` on the host.\n") + ok, detail = oversight.send_email(subject, mail) + print(f"email: {detail}") + return 0 if (res and res["ok"]) else 1 + + def main(argv=None): p = argparse.ArgumentParser(prog="colophon", description="autonomous library metadata healer (Phase 0)") sub = p.add_subparsers(dest="cmd", required=True) @@ -219,19 +252,27 @@ def main(argv=None): rs.add_argument("--limit", type=int, default=None) rs.add_argument("--apply", action="store_true", help="auto-apply re-seeds at conf >= --min-conf") rs.add_argument("--min-conf", type=float, default=0.9) + rs.add_argument("--force", action="store_true", help="re-query books on the cached-unresolvable skip-list") + rs.add_argument("--clear-skips", action="store_true", help="forget all cached-unresolvable entries, then exit") sn = sub.add_parser("series-audit", help="Phase 3 — series-numbering audit (read-only unless --apply)") sn.add_argument("--limit", type=int, default=None) sn.add_argument("--apply", action="store_true", help="heal clean number mismatches (reuses the heal path)") ov = sub.add_parser("oversight", help="Phase 4b — weekly changelog oversight + verdict (emails only if flagged)") ov.add_argument("--days", type=int, default=7) ov.add_argument("--email", action="store_true", help="email the digest only when flagged (DRIFT/REVIEW)") + mt = sub.add_parser("maintain", help="nightly sweep: backfill + resolve in one run, with a summary (dry-run unless --apply)") + mt.add_argument("--limit", type=int, default=20, help="max books for the backfill survey") + mt.add_argument("--min-conf", type=float, default=0.9, help="auto-apply gate for resolve re-seeds") + mt.add_argument("--apply", action="store_true", help="actually write (default: dry-run)") + mt.add_argument("--force", action="store_true", help="re-query cached-unresolvable mis-seeds this run") + mt.add_argument("--email", action="store_true", help="always email the summary (a daily heartbeat)") args = p.parse_args(argv) g, store = Grimmory(), Store() fn = {"precheck": cmd_precheck, "heal": cmd_heal, "log": cmd_log, "revert": cmd_revert, "backfill": cmd_backfill, "audit": cmd_audit, "resolve": cmd_resolve, "series-audit": cmd_series_audit, - "oversight": cmd_oversight}[args.cmd] + "oversight": cmd_oversight, "maintain": cmd_maintain}[args.cmd] try: return fn(args, g, store) except (GrimmoryError, PreconditionError) as e: diff --git a/colophon/maintain.py b/colophon/maintain.py new file mode 100644 index 0000000..b147dfd --- /dev/null +++ b/colophon/maintain.py @@ -0,0 +1,114 @@ +"""Daily maintenance run — backfill + resolve in one process, with a summary. + +Runs the two write phases the timer fires nightly, each guarded so a failure in +one does not skip the other, then composes a tight summary (counts + what changed ++ standing manual items + a one-line health verdict). The CLI emails it — always, +including on an aborted run — so the summary doubles as a heartbeat. The process +still exits non-zero when a phase failed, so the systemd unit is marked failed and +the next timer fire retries. + +This module never raises for a phase or precondition failure: it captures the +failure into the result so the caller can always render + send a report. +""" +import time + +from . import backfill +from .heal import assert_preconditions +from .resolver import run_resolve + + +def run_maintain(g, store, limit=20, min_conf=0.9, apply=False, force=False): + res = {"ts": time.strftime("%Y-%m-%d %H:%M"), "apply": apply, "ok": True, + "aborted": False, "backfill": None, "resolve": None, "errors": []} + if apply: + try: + assert_preconditions(g) + except Exception as e: # noqa: BLE001 — precondition/network failure: abort cleanly, still report + res["ok"], res["aborted"] = False, True + res["errors"].append(f"preconditions: {str(e)[:200]}") + return res + try: + res["backfill"] = backfill.run(g, store, limit=limit, apply=apply) + if res["backfill"]["errors"] or res["backfill"]["aborted"]: + res["ok"] = False + except Exception as e: # noqa: BLE001 + res["ok"] = False + res["errors"].append(f"backfill: {str(e)[:200]}") + try: + res["resolve"] = run_resolve(apply=apply, min_conf=min_conf, g=g, store=store, force=force) + props = res["resolve"]["proposals"] + if any(p.get("apply_error") for p in props) or any(p.get("aborted") for p in props): + res["ok"] = False + except Exception as e: # noqa: BLE001 + res["ok"] = False + res["errors"].append(f"resolve: {str(e)[:200]}") + return res + + +def verdict(res): + if res["aborted"]: + return "ABORTED" + return "OK" if res["ok"] else "ERRORS" + + +def _healed_count(res): + bf = (res["backfill"] or {}).get("healed", 0) + rs = sum(1 for p in (res["resolve"] or {}).get("proposals", []) if p.get("applied")) + return bf + rs + + +def subject(res): + n = _healed_count(res) + mode = "" if res["apply"] else " [dry-run]" + return f"Colophon daily [{verdict(res)}]{mode} — {n} healed ({res['ts']})" + + +def render_summary(res): + L = [f"Colophon daily maintenance — {res['ts']}", + f"STATUS: {verdict(res)}" + ("" if res["apply"] else " (dry-run — nothing written)"), ""] + if res.get("errors"): + L.append("Errors:") + L += [f" - {e}" for e in res["errors"]] + L.append("") + + bf = res["backfill"] + if bf: + heals = [p for p in bf["proposals"] if p["action"] == "heal"] + flags = [p for p in bf["proposals"] if p["action"].startswith("review")] + L.append(f"Backfill (broken ISBN → canonical): {len(bf['proposals'])} surveyed · " + f"{bf['healed']} healed · {len(flags)} flagged · {bf['errors']} errors" + + (" ABORTED (circuit-breaker)" if bf["aborted"] else "")) + for p in heals: + L.append(f" + book {p['book_id']} → hcid {p.get('hcid')} {p.get('hc_title')!r}") + else: + L.append("Backfill: did not run") + + rs = res["resolve"] + if rs: + props = rs["proposals"] + applied = [p for p in props if p.get("applied")] + proposed = [p for p in props if p["action"] == "propose" and not p.get("applied")] + none = [p for p in props if p["action"] == "none"] + err = [p for p in props if p["action"] == "error"] + skipped = rs.get("skipped", []) + L.append(f"Resolve (mis-seed → correct identity): {len(props)} queried · " + f"{len(applied)} auto-healed · {len(proposed)} below-threshold · " + f"{len(none)} no-match · {len(err)} error · {len(skipped)} cached-skipped") + for p in applied: + L.append(f" + book {p['book_id']} {p['title']!r} → {p.get('chosen_title')!r} " + f"(conf {p.get('confidence')})") + # New + standing below-threshold matches are the actionable items (human nudge). + standing = [(p["book_id"], p.get("title"), p.get("chosen_title"), p.get("confidence")) + for p in proposed] + standing += [(s["book_id"], s.get("title"), s.get("chosen_title"), s.get("conf")) + for s in skipped if s.get("action") == "propose-below"] + if standing: + L.append(" Below-threshold matches (apply manually if right — " + "`colophon resolve --book --apply --min-conf 0`):") + for bid, title, chosen, conf in standing: + L.append(f" ? book {bid} {title!r} → {chosen!r} (conf {conf})") + else: + L.append("Resolve: did not run") + + L += ["", "Full reports under reports/ on the host."] + return "\n".join(L) + "\n" diff --git a/colophon/resolver.py b/colophon/resolver.py index 0743e04..596d329 100644 --- a/colophon/resolver.py +++ b/colophon/resolver.py @@ -6,13 +6,58 @@ ids). Proposes; writes nothing. Auto-apply is a later, agreement-gated stage. """ import json +import os import time -from . import anthropic, audit, hardcover +from . import anthropic, audit, hardcover, matcher from .heal import assert_preconditions, heal_book AUTO_CONF = 0.9 # auto-apply re-seeds at or above this confidence ABORT_ERRORS = 3 # circuit-breaker +# Days after which a cached-unresolvable book is re-queried anyway. 0 = never +# (the default): once placed on the skip-list a book is only re-tried when its +# title/author changes (fingerprint miss) or on an explicit `resolve --force`. +RETRY_DAYS = int(os.environ.get("COLOPHON_RESOLVE_RETRY_DAYS", "0") or "0") + + +def _fingerprint(book): + """The normalized title+author the resolve query is built from. Stable across + processes (plain text, not the per-process-salted builtin hash()).""" + return matcher._norm(book.get("title", "")) + " :: " + matcher._norm(book.get("authors", "")) + + +def _expired(last_seen, days=RETRY_DAYS): + if days <= 0: + return False + cutoff = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(time.time() - days * 86400)) + return (last_seen or "") < cutoff + + +def _cached_skip(book, skips, days=RETRY_DAYS): + """Return the live skip row for this book, or None if it should be re-queried + (no entry / title-or-author changed / entry expired).""" + row = skips.get(book["book_id"]) + if row and row["fingerprint"] == _fingerprint(book) and not _expired(row["last_seen"], days): + return row + return None + + +def _record_skip(store, book, p): + """After an apply-mode resolve, keep the skip-list current for this book.""" + if p.get("applied"): + store.skip_clear(book["book_id"]) # resolved — no longer unresolvable + return + if p.get("apply_error") or p["action"] == "error": + return # transient failure — retry next run + fp = _fingerprint(book) + if p["action"] == "none": + store.skip_put(book["book_id"], book.get("title"), fp, "none", + reason=p.get("reason"), conf=p.get("confidence")) + elif p["action"] == "propose": # matched, but below the auto-apply gate + store.skip_put(book["book_id"], book.get("title"), fp, "propose-below", + reason=p.get("reason"), conf=p.get("confidence"), + chosen_id=p.get("chosen_id"), chosen_title=p.get("chosen_title"), + isbn=p.get("isbn")) SYSTEM = ( "You identify the correct Hardcover book for a library entry whose current " @@ -77,17 +122,32 @@ def resolve(book): } -def run_resolve(limit=None, book_ids=None, apply=False, min_conf=AUTO_CONF, g=None, store=None): +def run_resolve(limit=None, book_ids=None, apply=False, min_conf=AUTO_CONF, g=None, + store=None, force=False): """Resolve the given books (or every audit-flagged mis-seed). With apply=True, heal the proposals at confidence >= min_conf via the real heal engine (set the - resolved ISBN + id + locks → refresh); the rest stay propose-only.""" + resolved ISBN + id + locks → refresh); the rest stay propose-only. + + Books already on the skip-list (resolved to no-match / below-threshold on a prior + run, title+author unchanged) are filtered out and returned under `skipped` instead + of being re-queried. `force` ignores the skip-list; an explicit `book_ids` is always + treated as a deliberate re-check (skip-list bypassed for those).""" if apply: assert_preconditions(g) + explicit = bool(book_ids) if book_ids: allb = {b["book_id"]: b for b in audit.all_books()} books = [allb[i] for i in book_ids if i in allb] else: books = [b for b, _ in audit.run_audit(limit=limit)["categories"].get("review-misseed", [])] + skipped = [] + if store and not explicit and not force: + skips = store.skip_map() + active = [] + for b in books: + row = _cached_skip(b, skips) + (skipped if row else active).append(row or b) + books = active run_id = (store.new_run_id() + "-resolve") if (apply and store) else None proposals, errors = [], 0 for i, b in enumerate(books): @@ -102,13 +162,14 @@ def run_resolve(limit=None, book_ids=None, apply=False, min_conf=AUTO_CONF, g=No except Exception as e: p["applied"], p["apply_error"] = False, str(e)[:100] errors += 1 - if errors >= ABORT_ERRORS: - p["aborted"] = True - proposals.append(p) - break + if apply and store: + _record_skip(store, b, p) proposals.append(p) + if p.get("apply_error") and errors >= ABORT_ERRORS: + p["aborted"] = True + break return {"count": len(proposals), "proposals": proposals, "run_id": run_id, - "applied": sum(1 for p in proposals if p.get("applied"))} + "applied": sum(1 for p in proposals if p.get("applied")), "skipped": skipped} def render(result): @@ -117,10 +178,12 @@ def render(result): none = [p for p in props if p["action"] == "none"] err = [p for p in props if p["action"] == "error"] applied = [p for p in proposed if p.get("applied")] + skipped = result.get("skipped", []) L = [f"# Colophon mis-seed resolution — {time.strftime('%Y-%m-%d %H:%M')}", f"\n**{len(props)}** mis-seeds · **{len(applied)}** auto-applied · " f"**{len(proposed) - len(applied)}** proposed (below threshold) · " - f"**{len(none)}** no-match · **{len(err)}** error\n", + f"**{len(none)}** no-match · **{len(err)}** error · " + f"**{len(skipped)}** cached-skipped\n", "## 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") @@ -139,4 +202,18 @@ def render(result): L.append("\n## Errors\n") for p in err: L.append(f"- `{p['book_id']}` {p['title']!r} — {p.get('reason', '')}") + if skipped: + below = [s for s in skipped if s.get("action") == "propose-below"] + L.append(f"\n## Cached unresolvable — not re-queried ({len(skipped)})\n") + if below: + L.append("Standing proposals (a match was found but below the auto-apply " + "gate — apply manually if correct):\n") + for s in sorted(below, key=lambda s: -(s.get("conf") or 0)): + L.append(f"- `{s['book_id']}` {s.get('title')!r} → **{s.get('chosen_title')!r}** " + f"(hcid {s.get('chosen_id')}, isbn {s.get('isbn')}, conf {s.get('conf')})") + L.append(f" apply: `colophon resolve --book {s['book_id']} --apply --min-conf 0`") + n_none = len(skipped) - len(below) + if n_none: + L.append(f"\n{n_none} cached no-match. `colophon resolve --force` re-checks every " + "skipped book; `colophon resolve --clear-skips` forgets them.") return "\n".join(L) + "\n" diff --git a/colophon/store.py b/colophon/store.py index 51b4ca8..4d38343 100644 --- a/colophon/store.py +++ b/colophon/store.py @@ -32,6 +32,28 @@ def __init__(self, path=DB_PATH): )""" ) c.execute("CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value TEXT)") + # Mis-seeds the resolver could not place (no match, or a match below the + # auto-apply threshold). Keyed by book; `fingerprint` is the normalized + # title+author the resolve query was built from — if the book's title or + # author later changes, the fingerprint no longer matches and the book is + # re-queried. This is what stops the nightly sweep re-asking Hardcover + + # Haiku about the same unresolvable books every run. + c.execute( + """CREATE TABLE IF NOT EXISTS resolve_skip( + book_id INTEGER PRIMARY KEY, + title TEXT, + fingerprint TEXT NOT NULL, + action TEXT NOT NULL, + reason TEXT, + conf REAL, + chosen_id TEXT, + chosen_title TEXT, + isbn TEXT, + first_seen TEXT NOT NULL, + last_seen TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 1 + )""" + ) def epoch(self): """Re-audit epoch. Settled (locked) books are only re-opened when this is @@ -77,6 +99,37 @@ def recent(self, n=20): return [dict(r) for r in c.execute( "SELECT * FROM changes ORDER BY id DESC LIMIT ?", (int(n),))] + # --- resolve skip-list (don't re-query the unresolvable) --- + + def skip_map(self): + """All skip entries as {book_id: row-dict}.""" + with self._conn() as c: + return {r["book_id"]: dict(r) for r in c.execute("SELECT * FROM resolve_skip")} + + def skip_put(self, book_id, title, fingerprint, action, reason=None, conf=None, + chosen_id=None, chosen_title=None, isbn=None): + now = time.strftime("%Y-%m-%dT%H:%M:%S") + with self._conn() as c: + c.execute( + "INSERT INTO resolve_skip(book_id,title,fingerprint,action,reason,conf," + "chosen_id,chosen_title,isbn,first_seen,last_seen,attempts) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?,1) " + "ON CONFLICT(book_id) DO UPDATE SET title=excluded.title, " + "fingerprint=excluded.fingerprint, action=excluded.action, " + "reason=excluded.reason, conf=excluded.conf, chosen_id=excluded.chosen_id, " + "chosen_title=excluded.chosen_title, isbn=excluded.isbn, " + "last_seen=excluded.last_seen, attempts=resolve_skip.attempts+1", + (int(book_id), title, fingerprint, action, reason, conf, chosen_id, + chosen_title, isbn, now, now), + ) + + def skip_clear(self, book_id=None): + """Drop one skip entry (book_id given) or all of them. Returns rows removed.""" + with self._conn() as c: + if book_id is None: + return c.execute("DELETE FROM resolve_skip").rowcount + return c.execute("DELETE FROM resolve_skip WHERE book_id=?", (int(book_id),)).rowcount + @staticmethod def loads(row, field): v = row.get(field) diff --git a/pyproject.toml b/pyproject.toml index f214eee..5818057 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "colophon" -version = "0.1.0" +version = "0.2.0" description = "Autonomous library-metadata healer for Booklore-family book servers" readme = "README.md" requires-python = ">=3.9" diff --git a/tests/test_resolve_skip.py b/tests/test_resolve_skip.py new file mode 100644 index 0000000..ae45dfb --- /dev/null +++ b/tests/test_resolve_skip.py @@ -0,0 +1,126 @@ +"""Skip-list logic — pure / SQLite only, no network. Run: python -m unittest -v""" +import os +import tempfile +import unittest + +from colophon import resolver +from colophon.store import Store + + +def _book(book_id=1, title="The Wool Trilogy", authors="Hugh Howey"): + return {"book_id": book_id, "title": title, "authors": authors} + + +class Fingerprint(unittest.TestCase): + def test_stable_and_normalized(self): + a = resolver._fingerprint(_book(title="The Wool Trilogy", authors="Hugh Howey")) + b = resolver._fingerprint(_book(title=" the WOOL trilogy ", authors="hugh howey")) + self.assertEqual(a, b) # case/whitespace/punctuation-insensitive + + def test_changes_with_title_or_author(self): + base = resolver._fingerprint(_book()) + self.assertNotEqual(base, resolver._fingerprint(_book(title="Wool Omnibus"))) + self.assertNotEqual(base, resolver._fingerprint(_book(authors="Someone Else"))) + + def test_no_builtin_hash(self): + # A normalized string, never the per-process-salted builtin hash(). + fp = resolver._fingerprint(_book()) + self.assertIsInstance(fp, str) + self.assertIn("wool", fp) + + +class Expiry(unittest.TestCase): + def test_zero_days_never_expires(self): + self.assertFalse(resolver._expired("1970-01-01T00:00:00", days=0)) + + def test_old_entry_expires_with_ttl(self): + self.assertTrue(resolver._expired("1970-01-01T00:00:00", days=30)) + now = __import__("time").strftime("%Y-%m-%dT%H:%M:%S") + self.assertFalse(resolver._expired(now, days=30)) + + +class StoreRoundtrip(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + self.store = Store(path=os.path.join(self.dir, "t.db")) + + def test_put_get_clear(self): + self.store.skip_put(7, "Dune Saga", "fp-1", "none", reason="no candidates") + m = self.store.skip_map() + self.assertIn(7, m) + self.assertEqual(m[7]["action"], "none") + self.assertEqual(m[7]["attempts"], 1) + self.assertEqual(self.store.skip_clear(7), 1) + self.assertEqual(self.store.skip_map(), {}) + + def test_upsert_increments_attempts_keeps_first_seen(self): + self.store.skip_put(7, "Dune Saga", "fp-1", "none") + first = self.store.skip_map()[7]["first_seen"] + self.store.skip_put(7, "Dune Saga", "fp-1", "none") + row = self.store.skip_map()[7] + self.assertEqual(row["attempts"], 2) + self.assertEqual(row["first_seen"], first) + + def test_clear_all(self): + self.store.skip_put(1, "a", "f1", "none") + self.store.skip_put(2, "b", "f2", "none") + self.assertEqual(self.store.skip_clear(), 2) + + +class CachedSkipFilter(unittest.TestCase): + def test_matches_only_on_same_fingerprint(self): + b = _book(book_id=5) + skips = {5: {"fingerprint": resolver._fingerprint(b), + "last_seen": "2999-01-01T00:00:00"}} + self.assertIsNotNone(resolver._cached_skip(b, skips, days=0)) + + def test_no_skip_when_title_changed(self): + b = _book(book_id=5, title="Original") + skips = {5: {"fingerprint": resolver._fingerprint(b), + "last_seen": "2999-01-01T00:00:00"}} + moved = _book(book_id=5, title="Corrected Title") + self.assertIsNone(resolver._cached_skip(moved, skips, days=0)) + + def test_no_skip_when_absent(self): + self.assertIsNone(resolver._cached_skip(_book(book_id=9), {}, days=0)) + + +class RecordSkip(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + self.store = Store(path=os.path.join(self.dir, "t.db")) + + def test_none_is_recorded(self): + b = _book(book_id=10) + resolver._record_skip(self.store, b, {"action": "none", "reason": "no candidates"}) + row = self.store.skip_map()[10] + self.assertEqual(row["action"], "none") + self.assertEqual(row["fingerprint"], resolver._fingerprint(b)) + + def test_below_threshold_propose_keeps_match_detail(self): + b = _book(book_id=11) + resolver._record_skip(self.store, b, { + "action": "propose", "confidence": 0.85, "chosen_id": "430000", + "chosen_title": "The Phoenix Project", "isbn": "9780988262577"}) + row = self.store.skip_map()[11] + self.assertEqual(row["action"], "propose-below") + self.assertEqual(row["chosen_id"], "430000") + self.assertEqual(row["isbn"], "9780988262577") + + def test_applied_clears_existing_skip(self): + b = _book(book_id=12) + resolver._record_skip(self.store, b, {"action": "none"}) + self.assertIn(12, self.store.skip_map()) + resolver._record_skip(self.store, b, {"action": "propose", "applied": True}) + self.assertNotIn(12, self.store.skip_map()) + + def test_transient_errors_not_recorded(self): + b = _book(book_id=13) + resolver._record_skip(self.store, b, {"action": "error", "reason": "Haiku timeout"}) + resolver._record_skip(self.store, _book(book_id=14), + {"action": "propose", "apply_error": "grimmory 500"}) + self.assertEqual(self.store.skip_map(), {}) + + +if __name__ == "__main__": + unittest.main()