From fab17129e2e0fe88a610518071cada2218f4c377 Mon Sep 17 00:00:00 2001 From: dank Date: Wed, 4 Nov 2026 14:45:29 -0800 Subject: [PATCH] feat(media): dedupe + relayout campaign tools for Audiobooks and eBooks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 204 ++++++++++++++++ scripts/media-ab-reorg.py | 420 +++++++++++++++++++++++++++++++++ scripts/media-ab-verifydups.py | 89 +++++++ scripts/media-ebook-dedupe.py | 239 +++++++++++++++++++ 4 files changed, 952 insertions(+) create mode 100755 scripts/abs-post-reorg.py create mode 100755 scripts/media-ab-reorg.py create mode 100755 scripts/media-ab-verifydups.py create mode 100755 scripts/media-ebook-dedupe.py diff --git a/scripts/abs-post-reorg.py b/scripts/abs-post-reorg.py new file mode 100755 index 0000000..0eb24db --- /dev/null +++ b/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 ("srchostdstarray" 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() diff --git a/scripts/media-ab-reorg.py b/scripts/media-ab-reorg.py new file mode 100755 index 0000000..da40efb --- /dev/null +++ b/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() diff --git a/scripts/media-ab-verifydups.py b/scripts/media-ab-verifydups.py new file mode 100755 index 0000000..dbd5367 --- /dev/null +++ b/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 size 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() diff --git a/scripts/media-ebook-dedupe.py b/scripts/media-ebook-dedupe.py new file mode 100755 index 0000000..3277ddd --- /dev/null +++ b/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()