feat(media): dedupe + relayout campaign tools for Audiobooks and eBooks

media-ab-reorg.py: planner for the ABS layout {Author}/{Series}/{Seq -
Year - Title {Narrator}} — claims one book folder per DB item (tracks
of duplicate recordings must not interleave), gates any book folder
that would nest above another (ABS's swallow defect), routes comic and
ebook-only items to intake, holds dedup losers under the campaign dir.
media-ab-verifydups.py: full-blake2b confirmation of size+head/tail
collision survivors. media-ebook-dedupe.py: dedupe the eBooks pile
against the Calibre library (identity key title-core+first-surname,
OPF metadata authority) and emit the calibredb ingest list.
abs-post-reorg.py: server-side ABS driver (cover export, DB-row wipe —
never hard=1 — scan, status) with credentials out of argv.

scripts/abs-post-reorg.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""abs-post-reorg.py — drive the Audiobookshelf side of the 2026-11-04 relayout.
+
+Runs ON Dankserver (the ABS host). Credentials come from the machine-local
+file ~/.config/secrets/audiobookshelf and never enter argv; requests go
+through urllib so neither the password nor the bearer token is world-readable
+in /proc. The delete arm is the DB-row arm ONLY — `hard=1` is fs.remove() of
+the item folder and must never reach this file (see the Audiobookshelf vault
+page; after the move the covering paths are whole books).
+
+Subcommands
+  covers  --list LIST.tsv        copy each ABS /metadata cover onto the book
+                                 folder ("srchost<TAB>dstarray" per line);
+                                 run BEFORE the mover so the cover rides the
+                                 book's move, or AFTER with dst already final
+  wipe    [--apply]              delete every library item DB row (progress
+                                 included — the campaign accepted losing the
+                                 11 progress rows); refuses without --apply
+  scan    [--wait]               trigger the library scan; --wait polls
+                                 isFetchingLibrary then prints the item total
+  status                         item total + isMissing census
+
+Paging law (measured twice): page on `page` (0-based) + `limit`; `offset` is
+ignored outright. Count unique ids; stop on `total`.
+"""
+import argparse
+import json
+import os
+import shutil
+import sys
+import time
+import urllib.error
+import urllib.request
+
+BASE = "http://127.0.0.1:8000/audiobookshelf"
+SECRET = os.environ.get("ABS_SECRET", "/home/dank/.config/secrets/audiobookshelf")
+
+
+def cred():
+    kv = {}
+    with open(SECRET) as fh:
+        for line in fh:
+            if "=" in line:
+                k, _, v = line.strip().partition("=")
+                kv[k] = v
+    return kv.get("ABS_ADMIN_USER", "frank"), kv.get("ABS_ADMIN_PASSWORD")
+
+
+def call(method, path, body=None, token=None, base=BASE):
+    data = json.dumps(body).encode() if body is not None else None
+    req = urllib.request.Request(base + path, data=data, method=method)
+    req.add_header("Content-Type", "application/json")
+    if token:
+        req.add_header("Authorization", f"Bearer {token}")
+    try:
+        with urllib.request.urlopen(req, timeout=30) as r:
+            raw = r.read().decode()
+            try:
+                return r.status, json.loads(raw)
+            except json.JSONDecodeError:
+                return r.status, raw.strip()
+    except urllib.error.HTTPError as e:
+        return e.code, e.read().decode()[:200]
+
+
+def login():
+    user, password = cred()
+    if not password:
+        sys.exit(f"refused: {SECRET} carries no ABS_ADMIN_PASSWORD")
+    code, body = call("POST", "/login", {"username": user, "password": password})
+    if code != 200:
+        sys.exit(f"login failed {code}: {body}")
+    return body["user"]["token"]      # the token is inside user, not top level
+
+
+def library_id(token, name="Audiobooks"):
+    code, libs = call("GET", "/api/libraries", token=token)
+    if code != 200:
+        sys.exit(f"libraries failed {code}: {libs}")
+    for lib in libs.get("libraries", []):
+        if lib["name"] == name:
+            return lib["id"]
+    sys.exit(f"no library named {name}")
+
+
+def all_item_ids(token, lid):
+    ids, page = [], 0
+    code, one = call("GET", f"/api/libraries/{lid}/items?limit=1", token=token)
+    total = one.get("total", -1) if code == 200 else -1
+    while True:
+        code, body = call("GET", f"/api/libraries/{lid}/items?page={page}&limit=500",
+                          token=token)
+        if code != 200:
+            sys.exit(f"items page {page} failed {code}")
+        results = body.get("results", [])
+        if not results:
+            break
+        ids += [it["id"] for it in results]
+        page += 1
+        if total >= 0 and page * 500 >= total:
+            break
+    return list(dict.fromkeys(ids)), total
+
+
+def cmd_covers(args):
+    copied = missing = exists = 0
+    for line in open(args.list):
+        src, _, dst = line.rstrip("\n").partition("\t")
+        if not dst:
+            continue
+        if os.path.exists(dst):
+            exists += 1
+            continue
+        if not os.path.exists(src):
+            missing += 1
+            continue
+        os.makedirs(os.path.dirname(dst), exist_ok=True)
+        shutil.copy2(src, dst)
+        copied += 1
+    print(f"covers copied: {copied}  already there: {exists}  missing source: {missing}")
+
+
+def cmd_wipe(args, token, lid):
+    ids, total = all_item_ids(token, lid)
+    print(f"library {lid}: {len(ids)} items (total={total})")
+    if not args.apply:
+        print("dry run: nothing deleted (add --apply)")
+        return
+    gone = 0
+    for i, iid in enumerate(ids):
+        code, _ = call("DELETE", f"/api/items/{iid}", token=token)
+        if code == 200:
+            gone += 1
+        elif i < 3:
+            print(f"delete {iid}: {code}")
+        if i % 500 == 499:
+            print(f"  {i + 1}/{len(ids)} deleted")
+    print(f"deleted {gone}/{len(ids)} item rows")
+
+
+def cmd_scan(args, token, lid):
+    code, body = call("POST", f"/api/libraries/{lid}/scan", token=token)
+    print(f"scan: {code} {body if code != 200 else 'accepted'}")
+    if not args.wait:
+        return
+    last = -1
+    while True:
+        time.sleep(30)
+        code, lib = call("GET", f"/api/libraries/{lid}", token=token)
+        code2, one = call("GET", f"/api/libraries/{lid}/items?limit=1", token=token)
+        total = one.get("total", -1) if code2 == 200 else -1
+        fetching = lib.get("isFetchingLibrary") if code == 200 else None
+        print(f"  fetching={fetching} items={total}")
+        if fetching is False and total == last and total >= 0:
+            break
+        last = total
+    print(f"scan settled at {total} items")
+
+
+def cmd_status(token, lid):
+    missing = 0
+    seen, page = set(), 0
+    while True:
+        code, body = call("GET", f"/api/libraries/{lid}/items?page={page}&limit=500",
+                          token=token)
+        results = body.get("results", []) if code == 200 else []
+        if not results:
+            break
+        for it in results:
+            if it["id"] not in seen:
+                seen.add(it["id"])
+                if it.get("isMissing"):
+                    missing += 1
+        page += 1
+    code, lib = call("GET", f"/api/libraries/{lid}", token=token)
+    print(f"items: {len(seen)}  isMissing: {missing}  "
+          f"pending tasks: {lib.get('pendingTasks') if code == 200 else '?'}")
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    c = sub.add_parser("covers")
+    c.add_argument("--list", required=True)
+    w = sub.add_parser("wipe")
+    w.add_argument("--apply", action="store_true")
+    s = sub.add_parser("scan")
+    s.add_argument("--wait", action="store_true")
+    sub.add_parser("status")
+    args = ap.parse_args()
+    token = login()
+    lid = library_id(token)
+    if args.cmd == "covers":
+        cmd_covers(args)
+    elif args.cmd == "wipe":
+        cmd_wipe(args, token, lid)
+    elif args.cmd == "scan":
+        cmd_scan(args, token, lid)
+    else:
+        cmd_status(token, lid)
+
+
+if __name__ == "__main__":
+    main()

scripts/deploy-remote.sh

@@ -323,11 +323,13 @@ copy_alloy_config() {
     # The repo config carries the token GRAFANA_HOST_LABEL; the installed copy
     # carries the host's name, so Loki streams say WHICH machine's journal a
     # line came from without a second config file per host.
-    # *_CONF_ROOT are test seams: the bats cases execute these copiers for
+    # *_CONF_ROOT and ALLOY_CONFIG_SRC are test seams: the bats cases execute
-    # real against a sandbox with a stubbed run_privileged.
+    # these copiers for real against a sandbox with a stubbed run_privileged;
+    # ALLOY_CONFIG_SRC lets the refusal case feed a deliberately broken config.
     local root="${ALLOY_CONF_ROOT:-/opt/alloy}" tmp
+    local src="${ALLOY_CONFIG_SRC:-shared/alloy/config.alloy}"
     tmp="$(mktemp)"
-    sed "s/GRAFANA_HOST_LABEL/$TARGET/g" shared/alloy/config.alloy > "$tmp"
+    sed "s/GRAFANA_HOST_LABEL/$TARGET/g" "$src" > "$tmp"
     if ! grep -q "host = \"$TARGET\"," "$tmp" || grep -q "GRAFANA_HOST_LABEL" "$tmp"; then
         echo "alloy config: host-label substitution failed" >&2
         rm -f "$tmp"

scripts/media-ab-reorg.py

@@ -0,0 +1,420 @@
+#!/usr/bin/env python3
+"""media-ab-reorg.py — plan the Audiobooks dedupe + ABS-structure relayout.
+
+2026-11-04 campaign. Turns the Audiobooks library into the layout Audiobookshelf
+documents (https://audiobookshelf.org/docs/documentation/libraries/book-library/
+directory-structure/):
+
+    {Author}/[{Series}/]{seq} - {year} - {Title} {Narrator}/…files…
+
+The authority for every book's author/series/title/year/narrator is the ABS
+database dump (--items, JSON lines: [relPath, isMissing, title, authors, series,
+seq, year, narratorsJSON, coverPath, ebookFlag]); files are attached to their
+book by longest-prefix match on the item's relPath, which is how ABS's own
+grouping works. Folder names are then re-emitted so ABS's folder parser reads
+the same metadata back off disk.
+
+Three outputs, all consumable by scripts/media-apply.py (run everything with
+--root /mnt/entropy; hold/intake/park live outside the library on purpose):
+
+  --plan        mkdir/keep/move rows: dedupe losers -> --hold, ebook-only items
+                and comics -> --intake (the eBooks pile feeds Calibre next),
+                unowned media -> --park, everything else -> the new layout.
+  --covers      rows "srchost \\t dstarray": covers ABS keeps in /metadata get a
+                copy beside the book (cp, it crosses filesystems); run this
+                BEFORE apply so the cover lands inside the moved folder.
+  --exceptions  every judgement call: name collisions, byte-dups that pair two
+                different books, collapsed items, stray files.
+
+Dedupe law: only CONFIRMED byte-identities (--dups from
+media-ab-verifydups.py) remove a file. One copy survives per book: inside a
+folder the cleanest-named file wins; across folders the item whose root matches
+its author wins. Equal bytes under two different book identities keep both and
+raise an exception — a trailer embedded in two books is not a duplicate.
+
+Rerunnable: a second run against a moved library emits keep rows, not moves.
+"""
+import argparse
+import json
+import os
+import re
+from collections import defaultdict
+
+AUDIO = {"m4b", "mp3", "m4a", "flac", "opus", "ogg", "oga", "mp4", "aac", "wma",
+         "aiff", "aif", "wav", "webm", "webma", "mka", "awb", "caf", "mpg",
+         "mpg2", "mpeg"}
+EBOOK = {"epub", "pdf", "mobi", "azw3", "cbr", "cbz", "fb2"}
+COMIC = {"cbr", "cbz"}
+MEDIA = AUDIO | {"epub", "pdf", "mobi", "azw3", "cbr", "cbz"}
+PLACEHOLDER_AUTHOR = {"unknown", "unknown artist", "various", "various artists",
+                      "various authors", "n/a", ""}
+BAD_CHARS = re.compile(r'[<>:"/\\|?*]')
+DISC_DIR = re.compile(r"^(disc|cd|disk)\s*\d", re.I)
+KEEP_HINT = re.compile(r"\s*\((?:2|3|copy|backup|duplicate)\)\s*$", re.I)
+META_HOST = "/home/dank/Containers/audiobookshelf"
+
+
+def clean(name):
+    name = BAD_CHARS.sub(" ", name or "")
+    name = re.sub(r"\s+", " ", name).strip().strip(".")
+    return name or "Untitled"
+
+
+def clean_author(name):
+    """DB author string -> folder name: one author, no 'zz' sort artifact."""
+    name = re.sub(r"^zz", "", (name or "").strip())
+    return clean(name)
+
+
+def first_author(authors):
+    parts = [clean_author(p) for p in (authors or "").split("|")]
+    parts = [p for p in parts if p]
+    return parts[0] if parts else "Unknown"
+
+
+def parse_narrators(blob):
+    try:
+        v = json.loads(blob) if blob else []
+    except (TypeError, json.JSONDecodeError):
+        v = []
+    if isinstance(v, str):
+        v = [v]
+    return [str(x).strip() for x in (v or []) if str(x).strip()]
+
+
+def core_title(title, author):
+    """Title as the book folder should show it: release prefixes stripped."""
+    t = (title or "Untitled").strip()
+    a = (author or "").strip()
+    if a and t.lower().startswith(a.lower() + " -"):
+        t = t[len(a):].lstrip(" -")
+    return t.strip() or "Untitled"
+
+
+def bookname(seq, year, title, narrators):
+    seq = (seq or "").strip()
+    seq = seq if re.fullmatch(r"\d+(\.\d+)?", seq) else ""
+    year = (year or "").strip()
+    year = year if re.fullmatch(r"(1[5-9]|20)\d\d", year) else ""
+    t = title
+    if seq:
+        # drop a sequence prefix already carried inside the title
+        t = re.sub(rf"^{re.escape(seq)}\s*[\-.]\s+", "", t)
+    if year:
+        t = re.sub(rf"^{year}\s*[\-.]\s+", "", t)
+        t = re.sub(rf"\s*\({year}\)\s*$", "", t)
+    name = t
+    if seq and year:
+        name = f"{seq} - {year} - {name}"
+    elif seq:
+        name = f"{seq} - {name}"
+    elif year:
+        name = f"{year} - {name}"
+    if len(narrators) == 1:
+        name = f"{name} {{{clean(narrators[0])}}}"
+    return clean(name)
+
+
+class Campaign:
+    def __init__(self, args):
+        self.root = args.root.rstrip("/")
+        self.lib = self.root + "/Audiobooks"
+        self.hold = args.hold
+        self.intake = args.intake
+        self.park = args.park
+        self.items = {}          # relPath -> item dict (healthy only)
+        self.stale = []          # isMissing=1 rows
+        self.moved = {}          # src -> (dst, kind)
+        self.claims = {}         # dst -> src (collision guard)
+        self.loser_files = set()
+        self.loser_items = set()
+        self.exc = defaultdict(list)
+
+    # ------------------------------------------------------------- inputs
+    def load_items(self, path):
+        for line in open(path):
+            r = json.loads(line)
+            rel = r[0]
+            item = {"rel": rel, "title": r[2] or "", "authors": r[3] or "",
+                    "series": r[4] or "", "seq": r[5] or "", "year": r[6] or "",
+                    "narr": parse_narrators(r[7]), "cover": r[8] or "",
+                    "ebook": r[9]}
+            if r[1]:
+                self.stale.append(rel)
+                continue
+            self.items[rel] = item
+
+    def load_files(self, path):
+        files = []
+        for line in open(path):
+            sz, ext, p = line.rstrip("\n").split("\t", 2)
+            if p.startswith(self.lib + "/"):
+                files.append((int(sz), ext.lstrip(".").lower(), p))
+        return files
+
+    def owner(self, d):
+        """The healthy DB item whose folder is d or an ancestor of d."""
+        rel = os.path.relpath(d, self.lib)
+        while rel and rel != ".":
+            if rel in self.items:
+                return rel
+            rel = os.path.dirname(rel)
+        return None
+
+    # ------------------------------------------------------------- dedupe
+    def identity(self, item_rel):
+        if item_rel is None:
+            return None
+        it = self.items[item_rel]
+        return (first_author(it["authors"]).lower(),
+                core_title(it["title"], first_author(it["authors"])).lower(),
+                (it["year"] or "").strip())
+
+    def root_score(self, item_rel):
+        it = self.items[item_rel]
+        top = item_rel.split("/")[0]
+        s = 0
+        if top.lower() == first_author(it["authors"]).lower():
+            s += 2
+        if not top.startswith("_") and "various" not in top.lower():
+            s += 1
+        return (s, -len(item_rel))
+
+    def dedupe(self, files, dups_path):
+        by_hash = defaultdict(list)
+        if dups_path:
+            for line in open(dups_path):
+                h, sz, p = line.rstrip("\n").split("\t", 2)
+                if p.startswith(self.lib + "/"):
+                    by_hash[h].append(p)
+        owner_of = {}
+        dirs = {os.path.dirname(p) for _, _, p in files}
+        for d in dirs:
+            owner_of[d] = self.owner(d)
+        files_by_owner = defaultdict(list)
+        for _, _, p in files:
+            files_by_owner[owner_of[os.path.dirname(p)]].append(p)
+
+        for h, paths in by_hash.items():
+            owners = {owner_of.get(os.path.dirname(p)) for p in paths}
+            owners = {o for o in owners if o is not None}
+            per_item = defaultdict(list)
+            for p in paths:
+                per_item[owner_of.get(os.path.dirname(p))].append(p)
+            if owners:
+                ids = {o: self.identity(o) for o in owners}
+                # group items by identity; loose (None) files follow their own rule
+                same_as_loser = []
+                keepers = []
+                byid = defaultdict(list)
+                for o in owners:
+                    byid[ids[o]].append(o)
+                for ident, os_ in byid.items():
+                    os_.sort(key=self.root_score, reverse=True)
+                    keepers.append(os_[0])
+                    same_as_loser.extend(os_[1:])
+                keepers = set(keepers)
+                loser_items = set(same_as_loser)
+            else:
+                loser_items = set()
+                keepers = set()
+            for o, ps in per_item.items():
+                if o is None:
+                    continue  # loose files decided in placement
+                if o in loser_items:
+                    self.loser_files.update(ps)
+                    continue
+                if len(ps) > 1 and o in keepers:
+                    # redundant copies inside the surviving folder
+                    keeper = min(ps, key=lambda p: (
+                        bool(KEEP_HINT.search(os.path.splitext(p)[0])),
+                        len(p), p))
+                    self.loser_files.update(p for p in ps if p != keeper)
+            for o in loser_items:
+                self.loser_items.add(o)
+        # collapsed items: every media file lost
+        for o, ps in files_by_owner.items():
+            if o is None:
+                continue
+            media = [p for p in ps if os.path.splitext(p)[1].lstrip(".").lower()
+                     in MEDIA]
+            if media and all(p in self.loser_files for p in media):
+                self.loser_items.add(o)
+                self.exc["collapsed items (all files were byte-duplicates)"].append(o)
+
+    # ---------------------------------------------------------- placement
+    def target(self, item):
+        author = first_author(item["authors"])
+        series = clean(item["series"]) if item["series"] else ""
+        title = core_title(item["title"], author)
+        name = bookname(item["seq"], item["year"], title, item["narr"])
+        parts = [self.lib, clean(author)]
+        if series:
+            parts.append(clean(series))
+        parts.append(name)
+        return os.path.join(*parts)
+
+    def claim(self, dst, src, why):
+        """One destination, one source; collisions rename with a marker."""
+        if dst not in self.claims:
+            self.claims[dst] = src
+            return dst
+        stem, ext = os.path.splitext(dst)
+        n = 2
+        while f"{stem} ({n}){ext}" in self.claims:
+            n += 1
+        dst2 = f"{stem} ({n}){ext}"
+        self.claims[dst2] = src
+        self.exc[why].append(f"{src} -> renamed to {dst2}")
+        return dst2
+
+    def place(self, files):
+        audio_by_item = defaultdict(int)
+        for _, ext, p in files:
+            o = self.owner(os.path.dirname(p))
+            if o and ext in AUDIO:
+                audio_by_item[o] += 1
+        cover_dirs = {}
+        # Claim one BOOK FOLDER per item: two items for the same title (the
+        # _unsorted era filed some books 3 times) must not merge their tracks
+        # into one item — the loser gets a variant folder instead.
+        for rel in sorted(self.items, key=lambda r: self.root_score(r),
+                          reverse=True):
+            it = self.items[rel]
+            t = self.target(it)
+            if t in self.claims:
+                n = 2
+                while f"{t} (variant {n})" in self.claims:
+                    n += 1
+                t = f"{t} (variant {n})"
+                self.exc["same title filed by two DB items"].append(f"{rel} -> {t}")
+            self.claims[t] = rel
+            cover_dirs[rel] = t
+            if (it["cover"].startswith("/metadata") and audio_by_item[rel]
+                    and rel not in self.loser_items):
+                src = META_HOST + it["cover"]
+                self.covers.append((src, os.path.join(t,
+                                                      "cover" + os.path.splitext(src)[1].lower())))
+        # Containment gate: an item folder holding media files must never sit
+        # above another item folder — ABS's grouping would swallow the child
+        # (the docs' PROBLEM example). Rename the ANCESTOR book out of the way.
+        for _ in range(3):
+            moved = False
+            folders = sorted(set(cover_dirs.values()))
+            for anc in folders:
+                for d in folders:
+                    if d != anc and d.startswith(anc + os.sep):
+                        n = 1
+                        cand = f"{anc} (book)" if n == 1 else f"{anc} (book {n})"
+                        while cand in self.claims:
+                            n += 1
+                            cand = f"{anc} (book {n})"
+                        self.exc["book folder nested above another book"].append(
+                            f"{anc} -> {cand}")
+                        self.claims[cand] = self.claims.pop(anc)
+                        for r, t in cover_dirs.items():
+                            if t == anc:
+                                cover_dirs[r] = cand
+                        self.covers = [(s, d.replace(anc + os.sep, cand + os.sep, 1))
+                                       for s, d in self.covers]
+                        moved = True
+                        break
+                if moved:
+                    break
+            if not moved:
+                break
+        for _, ext, p in files:
+            if p in self.loser_files:
+                d = os.path.join(self.hold, os.path.relpath(p, self.root))
+                self.move(p, d, "dup")
+                continue
+            d = os.path.dirname(p)
+            o = self.owner(d)
+            if o is None:
+                rel = os.path.relpath(p, self.lib)
+                if ext in MEDIA:
+                    self.move(p, os.path.join(self.park, rel), "park")
+                else:
+                    self.move(p, os.path.join(self.park, rel), "park-junk")
+                continue
+            it = self.items[o]
+            if ext in COMIC:
+                key = clean(f"{first_author(it['authors'])} - {core_title(it['title'], '')}")
+                dst = os.path.join(self.intake, "_comics", key, os.path.basename(p))
+                self.move(p, self.claim(dst, p, "intake comic collisions"), "comic")
+                continue
+            if audio_by_item[o] == 0:  # ebook-only item belongs to the ebook stack
+                key = f"{first_author(it['authors'])} - {bookname(it['seq'], it['year'], core_title(it['title'], first_author(it['authors'])), it['narr'])}"
+                dst = os.path.join(self.intake, key, os.path.relpath(p, os.path.join(self.lib, o)))
+                self.move(p, self.claim(dst, p, "intake ebook collisions"), "ebook-only")
+                continue
+            rel_in = os.path.relpath(p, os.path.join(self.lib, o))
+            dst = os.path.join(cover_dirs[o], rel_in)
+            if dst == p:
+                self.kept.append(p)
+                self.claims.setdefault(dst, p)
+                continue
+            self.move(p, self.claim(dst, p, "target collisions"), "move")
+
+    def move(self, src, dst, kind):
+        self.moves.append((src, dst, kind))
+
+    def write(self, plan_path, exc_path):
+        dirs = {os.path.dirname(dst) for _, dst, _ in self.moves}
+        dirs |= {os.path.dirname(d) for _, d in self.covers}
+        all_dirs = set()
+        for d in dirs:
+            while d and d != self.root:
+                all_dirs.add(d)
+                d = os.path.dirname(d)
+        with open(plan_path, "w") as fh:
+            fh.write("action\tsrc\tdst\tkind\n")
+            for d in sorted(all_dirs):
+                fh.write(f"mkdir\t\t{d}\tcampaign\n")
+            for p in self.kept:
+                fh.write(f"keep\t{p}\t\tkeep\n")
+            for src, dst, kind in sorted(self.moves):
+                fh.write(f"move\t{src}\t{dst}\t{kind}\n")
+        if self.exc:
+            with open(exc_path, "w") as fh:
+                for why in sorted(self.exc):
+                    fh.write(f"# {why}\n")
+                    for line in sorted(set(self.exc[why])):
+                        fh.write(line + "\n")
+                    fh.write("\n")
+        mv = [m for m in self.moves if m[2] != "dup"]
+        print(f"plan: {len(all_dirs)} folders, {len(self.moves)} moves "
+              f"({len(mv)} relayout, {len(self.moves)-len(mv)} dup-hold), "
+              f"{len(self.kept)} keep, {len(self.covers)} cover-copies")
+
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--root", default="/mnt/entropy")
+    ap.add_argument("--enum", required=True)
+    ap.add_argument("--items", required=True)
+    ap.add_argument("--dups", default=None)
+    ap.add_argument("--hold", default="/mnt/entropy/.media-campaign/dup-hold-20261104")
+    ap.add_argument("--intake", default="/mnt/entropy/eBooks/_from-audiobooks")
+    ap.add_argument("--park", default="/mnt/entropy/NeedsProcessing/_from-audiobooks-20261104")
+    ap.add_argument("--plan", required=True)
+    ap.add_argument("--exceptions", required=True)
+    ap.add_argument("--covers", required=True)
+    args = ap.parse_args()
+
+    c = Campaign(args)
+    c.moves, c.kept, c.covers = [], [], []
+    c.load_items(args.items)
+    files = c.load_files(args.enum)
+    c.dedupe(files, args.dups)
+    c.place(files)
+    c.write(args.plan, args.exceptions)
+    with open(args.covers, "w") as fh:
+        for src, dst in sorted(set(c.covers)):
+            fh.write(f"{src}\t{dst}\n")
+    print(f"covers: {len(set(c.covers))}  stale DB rows (delete after scan): {len(c.stale)}")
+
+
+if __name__ == "__main__":
+    main()

scripts/media-ab-verifydups.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""media-ab-verifydups.py — confirm byte-duplicates among size-group survivors.
+
+Stage 2 of the 2026-11-04 Audiobooks dedupe. Stage 1 (partial blake2b of the
+first+last 4 MiB, done by the campaign's hashcand.py) over-claims: two files
+of equal size and equal head+tail are CANDIDATES, not copies. This tool takes
+the candidate files from --partial (rows: size, hash, path), groups them by
+(size, partial-hash), full-hashes only members of groups with 2+ members, and
+writes verdicts: one line per file whose full hash ties it to at least one
+other file.
+
+Output rows: full-hash <TAB> size <TAB> path   (sorted by hash) — a hash that
+appears once is not written at all; singletons are not duplicates.
+
+Read-only. Never moves, never deletes. Run it after the partial pass finishes
+(pgrep it first; two hashers on one output file have raced before).
+"""
+import argparse
+import hashlib
+import os
+from collections import defaultdict
+
+
+def full_hash(path, chunk=1 << 20):
+    h = hashlib.blake2b(digest_size=32)
+    with open(path, "rb") as fh:
+        for block in iter(lambda: fh.read(chunk), b""):
+            h.update(block)
+    return h.hexdigest()
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--partial", required=True, help="partial-hashes.tsv")
+    ap.add_argument("--out", required=True, help="full-hash verdicts TSV")
+    ap.add_argument("--progress-every", type=int, default=50)
+    args = ap.parse_args()
+
+    groups = defaultdict(list)
+    with open(args.partial) as fh:
+        for line in fh:
+            f = line.rstrip("\n").split("\t")
+            if len(f) != 3 or f[0] == "ERR":
+                continue
+            groups[(int(f[0]), f[1])].append(f[2])
+    survivors = sorted({p for g in groups.values() if len(g) > 1 for p in g})
+    print(f"survivor files to full-hash: {len(survivors)}")
+
+    done = set()
+    if os.path.exists(args.out):
+        with open(args.out) as fh:
+            for line in fh:
+                done.add(line.rstrip("\n").split("\t", 2)[2])
+    full = {}
+    for line in open(args.out) if os.path.exists(args.out) else []:
+        h, s, p = line.rstrip("\n").split("\t", 2)
+        full[p] = h
+
+    todo = [p for p in survivors if p not in done]
+    out = open(args.out, "a")
+    for i, p in enumerate(todo):
+        try:
+            full[p] = full_hash(p)
+        except OSError as e:
+            print(f"SKIP {p}: {e}")
+            continue
+        if i % args.progress_every == args.progress_every - 1:
+            out.flush()
+    out.close()
+
+    # Rewrite the whole verdict file keeping only TRUE duplicates (2+ fulls).
+    by_hash = defaultdict(list)
+    for p in survivors:
+        if p in full:
+            by_hash[full[p]].append(p)
+    with open(args.out, "w") as out:
+        for h, ps in sorted(by_hash.items()):
+            if len(ps) > 1:
+                for p in sorted(ps):
+                    out.write(f"{h}\t{os.path.getsize(p)}\t{p}\n")
+    dups = {h: ps for h, ps in by_hash.items() if len(ps) > 1}
+    n = sum(len(ps) - 1 for ps in dups.values())
+    waste = sum((len(ps) - 1) * os.path.getsize(ps[0]) for ps in dups.values())
+    print(f"true dup sets: {len(dups)}  redundant files: {n}  "
+          f"waste: {waste / 2**30:.1f} GiB")
+
+
+if __name__ == "__main__":
+    main()

scripts/media-ebook-dedupe.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+"""media-ebook-dedupe.py — dedupe the eBooks pile against the Calibre library.
+
+2026-11-04 campaign, ebook side. The Calibre library (/mnt/entropy/CalibreBooks,
+6,289 books) is DB-owned and healthy; /mnt/entropy/eBooks is a raw pile whose
+best books already live in the library, several of them twice in the pile.
+Audiobooks' orphaned ebooks/comics land in the pile first (media-ab-reorg's
+intake), so this campaign runs after the audiobook move.
+
+Inputs
+  --meta      ebook-meta.tsv: JSON lines {p, sz, title, authors, ident} for
+              epub (embedded OPF) and mobi/azw3 (Palm header title); the
+              producer is the campaign's opfcensus.py.
+  --enum      enum TSV (size, ext, path) for every tree.
+  --calibre   calibre-books.tsv: JSON lines [id, title, authors, series, path,
+              pubdate, formats] dumped from metadata.db.
+  --dups      full blake2b verdicts (hash, size, path) — only confirmed
+              byte-identities remove a file. Optional: without it, identity
+              rules alone run and every hash-dependent pair stays a KEEP.
+
+Outputs
+  --plan      media-apply plan (--root /mnt/entropy): hash-duplicates of a
+              file already in CalibreBooks -> HOLD; byte-identical files inside
+              the pile keep the best format of one book once; non-book junk
+              (mp4/mp3/images) -> PARK.
+  --ingest    one JSON line per surviving BOOK (a set of format files of one
+              identity): {paths, title, authors, source}. The ingest runner
+              feeds these to calibredb inside the calibre-web image.
+  --exceptions every identity pair the file-level evidence could not settle.
+
+Match law: identity is (core title, first author surname). Calibre's
+"Dinniman| Matt" and an OPF "Matt Dinniman" share surname "dinniman". Title
+core drops the leading article and all non-alphanumerics. Identity alone NEVER
+deletes against the pile — it only deletes when the same identity sits in the
+Calibre library with the same format (the pile copy adds nothing), or when a
+byte-hash confirms the pair. Format preference when the pile holds one book in
+several copies: epub > azw3 > mobi > pdf > fb2 > lit > rtf > txt.
+"""
+import argparse
+import json
+import os
+import re
+import unicodedata
+from collections import defaultdict
+
+BOOK_EXT = {"epub", "azw3", "mobi", "pdf", "fb2", "lit", "rtf", "txt", "cbr", "cbz"}
+FORMAT_RANK = {"epub": 0, "azw3": 1, "mobi": 2, "pdf": 3, "fb2": 4,
+               "cbr": 5, "cbz": 5, "lit": 6, "rtf": 7, "txt": 8}
+ARTICLE = re.compile(r"^(the|a|an)\s+", re.I)
+
+
+def ascii_fold(s):
+    return "".join(c for c in unicodedata.normalize("NFKD", s or "")
+                   if not unicodedata.combining(c))
+
+
+def title_core(title):
+    t = ascii_fold(title).lower()
+    t = ARTICLE.sub("", t)
+    t = re.sub(r"\(.*?\)", " ", t)          # (1998), (Unabridged) — noise
+    t = re.sub(r"[^a-z0-9 ]", " ", t)
+    t = re.sub(r"\b(1[5-9]|20)\d\d\b", " ", t)  # years
+    return " ".join(t.split())[:120]
+
+
+def surname(author):
+    a = ascii_fold(author).strip()
+    a = re.split(r"[;,]", a)[0] if "," in a else " ".join(a.split()[-1:])
+    return re.sub(r"[^a-z]", "", a.lower())
+
+
+def first_author(authors):
+    parts = [p.strip() for p in re.split(r"[;&]", authors or "") if p.strip()]
+    return parts[0] if parts else ""
+
+
+def stem_title(path):
+    s = os.path.splitext(os.path.basename(path))[0]
+    return ascii_fold(s)
+
+
+class Campaign:
+    def __init__(self, args):
+        self.args = args
+        self.root = args.root.rstrip("/")
+        self.pile = self.root + "/eBooks"
+        self.lib = self.root + "/CalibreBooks"
+        self.hold = args.hold
+        self.park = args.park
+        self.meta = {}           # path -> meta dict
+        self.pile_files = []     # (sz, ext, path)
+        self.cal_by_path = {}    # file path under CalibreBooks -> calibre row
+        self.cal_identity = {}   # (tcore, surname) -> calibre row
+        self.drop = {}           # path -> reason
+        self.exc = defaultdict(list)
+        self.moves = []
+        self.dirs_pile_book = defaultdict(list)  # book dir (pile) -> files
+
+    # ------------------------------------------------------------- inputs
+    def load(self):
+        for line in open(self.args.meta):
+            r = json.loads(line)
+            self.meta[r["p"]] = r
+        by_hash = defaultdict(list)
+        if self.args.dups:
+            for line in open(self.args.dups):
+                h, sz, p = line.rstrip("\n").split("\t", 2)
+                by_hash[h].append(p)
+        self.dup_sets = [ps for ps in by_hash.values() if len(ps) > 1]
+        for line in open(self.args.enum):
+            sz, ext, p = line.rstrip("\n").split("\t", 2)
+            ext = ext.lstrip(".").lower()
+            if p.startswith(self.pile + "/") and p not in self.park:
+                self.pile_files.append((int(sz), ext, p))
+        for line in open(self.args.calibre):
+            row = json.loads(line)
+            self.cal_identity[(title_core(row[1]), surname(first_author(row[2])))] = row
+
+    def identity(self, p):
+        m = self.meta.get(p)
+        if m and m.get("title"):
+            t = title_core(m["title"])
+            a = surname(first_author(m.get("authors", "")))
+            if t:
+                return (t, a)
+        return (title_core(stem_title(p)), "")
+
+    # ------------------------------------------------------------- rules
+    def run(self):
+        pile_paths = {p for _, _, p in self.pile_files}
+        # R1/R2: confirmed byte-duplicates
+        for ps in self.dup_sets:
+            in_lib = [p for p in ps if p.startswith(self.lib + "/")]
+            in_pile = [p for p in ps if p.startswith(self.pile + "/")]
+            if in_lib:
+                for p in in_pile:
+                    self.drop[p] = "hash-duplicate of CalibreBooks file"
+            if len(in_pile) > 1:
+                keep = min(in_pile, key=lambda p: (FORMAT_RANK.get(
+                    os.path.splitext(p)[1].lstrip(".").lower(), 9), len(p), p))
+                for p in in_pile:
+                    if p != keep:
+                        self.drop[p] = "hash-duplicate inside the pile"
+        # R3/R4: identity against the library (same format only deletes)
+        cal_files = defaultdict(set)   # identity -> formats in the library
+        for key, row in self.cal_identity.items():
+            cal_files[key] = set(f for f in row[6].split("; ") if f)
+        for sz, ext, p in self.pile_files:
+            if p in self.drop or ext not in BOOK_EXT:
+                continue
+            key = self.identity(p)
+            if key[0] and key in self.cal_identity:
+                if ext in cal_files[key]:
+                    self.drop[p] = f"same identity+format already in Calibre ({key[0][:40]})"
+                else:
+                    self.exc["pile holds a format the library lacks (add-format on ingest)"].append(
+                        f"{p}\t{key[0][:50]}\t{self.cal_identity[key][0]}")
+        # R5: surviving files grouped by (identity, parent-ish book dir)
+        books = defaultdict(list)
+        for sz, ext, p in self.pile_files:
+            if p in self.drop:
+                continue
+            if ext not in BOOK_EXT:
+                self.moves.append((p, os.path.join(self.park, os.path.relpath(p, self.root)),
+                                   "non-book"))
+                continue
+            m = self.meta.get(p, {})
+            if m.get("err") and ext in ("epub",):
+                self.exc["unreadable container (park for a human)"].append(p)
+                self.moves.append((p, os.path.join(self.park, os.path.relpath(p, self.root)),
+                                   "broken-book"))
+                continue
+            key = self.identity(p)
+            books[key].append(p)
+        for key, ps in books.items():
+            first = min(ps, key=lambda p: (FORMAT_RANK.get(
+                os.path.splitext(p)[1].lstrip(".").lower(), 9), len(p), p))
+            m = self.meta.get(first, {})
+            self.ingest = {
+                "key": list(key),
+                "paths": sorted(ps),
+                "title": m.get("title") or stem_title(first),
+                "authors": m.get("authors") or "",
+                "ident": m.get("ident") or "",
+            }
+            self.ingest_list.append(self.ingest)
+        for p, why in self.drop.items():
+            self.moves.append((p, os.path.join(self.hold, os.path.relpath(p, self.root)), "dup"))
+
+    def write(self):
+        dirs = {os.path.dirname(d) for _, d, _ in self.moves}
+        all_dirs = set()
+        for d in dirs:
+            while d and d != self.root:
+                all_dirs.add(d)
+                d = os.path.dirname(d)
+        with open(self.args.plan, "w") as fh:
+            fh.write("action\tsrc\tdst\tkind\n")
+            for d in sorted(all_dirs):
+                fh.write(f"mkdir\t\t{d}\tcampaign\n")
+            for src, dst, kind in sorted(self.moves):
+                fh.write(f"move\t{src}\t{dst}\t{kind}\n")
+        with open(self.args.ingest, "w") as fh:
+            for b in self.ingest_list:
+                fh.write(json.dumps(b, ensure_ascii=False) + "\n")
+        with open(self.args.exceptions, "w") as fh:
+            for why in sorted(self.exc):
+                fh.write(f"# {why}\n")
+                for line in sorted(set(self.exc[why])):
+                    fh.write(line + "\n")
+                fh.write("\n")
+        ndrop = sum(1 for _, _, k in self.moves if k == "dup")
+        print(f"pile books to ingest: {len(self.ingest_list)}  "
+              f"files dropped as duplicates: {ndrop}  "
+              f"non-book/broken moved out: {sum(1 for _, _, k in self.moves if k in ('non-book', 'broken-book'))}")
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--root", default="/mnt/entropy")
+    ap.add_argument("--meta", required=True)
+    ap.add_argument("--enum", required=True)
+    ap.add_argument("--calibre", required=True)
+    ap.add_argument("--dups", default=None)
+    ap.add_argument("--hold", default="/mnt/entropy/.media-campaign/dup-hold-ebooks-20261104")
+    ap.add_argument("--park", default="/mnt/entropy/NeedsProcessing/_from-ebooks-20261104")
+    ap.add_argument("--plan", required=True)
+    ap.add_argument("--ingest", required=True)
+    ap.add_argument("--exceptions", required=True)
+    args = ap.parse_args()
+    c = Campaign(args)
+    c.ingest_list = []
+    c.load()
+    c.run()
+    c.write()
+
+
+if __name__ == "__main__":
+    main()

tests/observability_quadlet.bats

@@ -799,14 +799,13 @@ assert not any('container_network' in e or 'container_fs_' in e for e in names),
         skip "franksdank/alloy image not present on this box"
     fi
     local sandbox="$BATS_TEST_TMPDIR/opt" cfg="$BATS_TEST_TMPDIR/broken.alloy"
-    sed 's/GRAFANA_HOST_LABEL/dankclaw/g' shared/alloy/config.alloy > "$cfg"
+    sed 's/GRAFANA_HOST_LABEL/dankclaw/g' "$REPO_ROOT/shared/alloy/config.alloy" > "$cfg"
     printf 'bogus_attr = "x"\n' >> "$cfg"
     ! (
         cd "$REPO_ROOT"
         eval "$(sed -n '/^copy_alloy_config()/,/^}/p' scripts/deploy-remote.sh)"
         run_privileged() { "$@"; }
-        cp() { if [ "$1" = "$cfg" ]; then return 1; fi; command cp "$@"; }
+        export TARGET=dankclaw ALLOY_CONF_ROOT="$sandbox" ALLOY_CONFIG_SRC="$cfg"
-        export TARGET=dankclaw ALLOY_CONF_ROOT="$sandbox"
         copy_alloy_config
     )
     [ ! -f "$sandbox/config/config.alloy" ]
@@ -820,7 +819,7 @@ assert not any('container_network' in e or 'container_fs_' in e for e in names),
         skip "franksdank/alloy image not present on this box"
     fi
     local cfg="$BATS_TEST_TMPDIR/rendered.alloy"
-    sed 's/GRAFANA_HOST_LABEL/dankclaw/g' shared/alloy/config.alloy > "$cfg"
+    sed 's/GRAFANA_HOST_LABEL/dankclaw/g' "$REPO_ROOT/shared/alloy/config.alloy" > "$cfg"
     podman run --rm -v "$cfg:/c.alloy:ro" --entrypoint /usr/bin/alloy \
         localhost/franksdank/alloy:latest validate /c.alloy
 }