dank
fab17129e2
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.
421 lines
17 KiB
Python
Executable File
421 lines
17 KiB
Python
Executable File
#!/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()
|