diff --git a/__pycache__/apply_abs_mock_report.cpython-311.pyc b/__pycache__/apply_abs_mock_report.cpython-311.pyc new file mode 100644 index 0000000..39554e7 Binary files /dev/null and b/__pycache__/apply_abs_mock_report.cpython-311.pyc differ diff --git a/__pycache__/generate_abs_mock_report.cpython-311.pyc b/__pycache__/generate_abs_mock_report.cpython-311.pyc index c4f69ec..b30796c 100644 Binary files a/__pycache__/generate_abs_mock_report.cpython-311.pyc and b/__pycache__/generate_abs_mock_report.cpython-311.pyc differ diff --git a/apply_abs_mock_report.py b/apply_abs_mock_report.py new file mode 100644 index 0000000..d16f0d0 --- /dev/null +++ b/apply_abs_mock_report.py @@ -0,0 +1,278 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import argparse +import csv +import shutil +import sys +from dataclasses import dataclass +from pathlib import Path, PurePosixPath + + +REQUIRED_COLUMNS = {"current_path", "proposed_abs_path", "status"} + + +class ReportError(ValueError): + pass + + +@dataclass(frozen=True) +class ReportRow: + line_number: int + source_path: Path + target_rel_path: PurePosixPath + status: str + + +@dataclass(frozen=True) +class TransferOperation: + line_number: int + source_path: Path + target_rel_path: PurePosixPath + target_path: Path + + +def parse_report_row(raw_row: dict[str, str], line_number: int) -> ReportRow: + source_raw = (raw_row.get("current_path") or "").strip() + target_raw = (raw_row.get("proposed_abs_path") or "").strip() + status = (raw_row.get("status") or "").strip() + + if not source_raw: + raise ReportError(f"Line {line_number}: missing current_path") + if not target_raw: + raise ReportError(f"Line {line_number}: missing proposed_abs_path") + + target_rel_path = PurePosixPath(target_raw) + if target_rel_path.is_absolute(): + raise ReportError(f"Line {line_number}: proposed_abs_path must stay relative: {target_raw}") + if not target_rel_path.parts: + raise ReportError(f"Line {line_number}: proposed_abs_path is empty") + if any(part in {"", ".", ".."} for part in target_rel_path.parts): + raise ReportError( + f"Line {line_number}: proposed_abs_path contains an unsafe path component: {target_raw}" + ) + + return ReportRow( + line_number=line_number, + source_path=Path(source_raw).expanduser(), + target_rel_path=target_rel_path, + status=status, + ) + + +def load_report(path: Path) -> list[ReportRow]: + with path.open(encoding="utf-8", newline="") as handle: + reader = csv.DictReader(handle, delimiter="\t") + fieldnames = set(reader.fieldnames or []) + missing = REQUIRED_COLUMNS - fieldnames + if missing: + missing_fields = ", ".join(sorted(missing)) + raise ReportError(f"Report is missing required columns: {missing_fields}") + + rows: list[ReportRow] = [] + for line_number, raw_row in enumerate(reader, start=2): + if not any((value or "").strip() for value in raw_row.values()): + continue + rows.append(parse_report_row(raw_row, line_number)) + return rows + + +def is_same_or_child(path: Path, other: Path) -> bool: + try: + path.relative_to(other) + except ValueError: + return False + return True + + +def validate_non_overlapping_paths( + entries: list[tuple[Path, str]], + *, + kind: str, +) -> None: + for index, (path, description) in enumerate(entries): + for other_path, other_description in entries[index + 1 :]: + if is_same_or_child(path, other_path) or is_same_or_child(other_path, path): + raise ReportError( + f"{kind} paths overlap: {description} <-> {other_description}" + ) + + +def build_execution_plan( + rows: list[ReportRow], + destination_root: Path, + *, + selected_status: str, +) -> tuple[list[TransferOperation], int]: + selected_rows = rows if selected_status == "all" else [ + row for row in rows if row.status == selected_status + ] + skipped_rows = len(rows) - len(selected_rows) + + resolved_destination_root = destination_root.resolve() + operations: list[TransferOperation] = [] + seen_sources: dict[Path, int] = {} + seen_targets: dict[PurePosixPath, int] = {} + + for row in selected_rows: + source_path = row.source_path.resolve() + if not source_path.exists(): + raise ReportError( + f"Line {row.line_number}: source path does not exist: {row.source_path}" + ) + if not source_path.is_dir(): + raise ReportError( + f"Line {row.line_number}: source path is not a directory: {row.source_path}" + ) + + target_path = resolved_destination_root.joinpath(*row.target_rel_path.parts) + if target_path.exists(): + raise ReportError( + f"Line {row.line_number}: destination already exists: {target_path}" + ) + if is_same_or_child(target_path, source_path): + raise ReportError( + f"Line {row.line_number}: destination cannot point inside the source tree: {target_path}" + ) + + previous_source_line = seen_sources.get(source_path) + if previous_source_line is not None: + raise ReportError( + f"Line {row.line_number}: duplicate source path already used on line {previous_source_line}: {source_path}" + ) + previous_target_line = seen_targets.get(row.target_rel_path) + if previous_target_line is not None: + raise ReportError( + f"Line {row.line_number}: duplicate proposed_abs_path already used on line {previous_target_line}: " + f"{row.target_rel_path.as_posix()}" + ) + + seen_sources[source_path] = row.line_number + seen_targets[row.target_rel_path] = row.line_number + operations.append( + TransferOperation( + line_number=row.line_number, + source_path=source_path, + target_rel_path=row.target_rel_path, + target_path=target_path, + ) + ) + + validate_non_overlapping_paths( + [(operation.source_path, f"line {operation.line_number}: {operation.source_path}") for operation in operations], + kind="Source", + ) + validate_non_overlapping_paths( + [(operation.target_path, f"line {operation.line_number}: {operation.target_path}") for operation in operations], + kind="Destination", + ) + + return operations, skipped_rows + + +def execute_plan( + operations: list[TransferOperation], + *, + mode: str, + dry_run: bool, + verbose: bool, +) -> None: + for operation in operations: + if dry_run or verbose: + print( + f"{'plan' if dry_run else mode}\t{operation.source_path}\t{operation.target_path}" + ) + if dry_run: + continue + + operation.target_path.parent.mkdir(parents=True, exist_ok=True) + if mode == "copy": + shutil.copytree(operation.source_path, operation.target_path) + else: + shutil.move(str(operation.source_path), str(operation.target_path)) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Copy or move audiobook folders into the directory structure proposed by a mock ABS report." + ) + parser.add_argument( + "--report", + default="reports/audiobookshelf_mock_report.tsv", + help="TSV report produced by generate_abs_mock_report.py", + ) + parser.add_argument( + "--destination-root", + "--dest-root", + "--destination", + required=True, + help="Root directory where the new structure should be created", + ) + parser.add_argument( + "--mode", + choices=("copy", "move"), + default="copy", + help="Whether to copy or move each source directory", + ) + parser.add_argument( + "--status", + choices=("ready", "review", "all"), + default="ready", + help="Which report rows should be applied", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Validate the report and print the planned operations without changing files", + ) + parser.add_argument( + "--verbose", + action="store_true", + help="Print each completed operation", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + report_path = Path(args.report).expanduser().resolve() + destination_root = Path(args.destination_root).expanduser().resolve() + + try: + if not report_path.exists(): + raise ReportError(f"Report does not exist: {report_path}") + if destination_root.exists() and not destination_root.is_dir(): + raise ReportError(f"Destination root is not a directory: {destination_root}") + + rows = load_report(report_path) + operations, skipped_rows = build_execution_plan( + rows, + destination_root, + selected_status=args.status, + ) + execute_plan( + operations, + mode=args.mode, + dry_run=args.dry_run, + verbose=args.verbose, + ) + except (OSError, ReportError) as error: + print(error, file=sys.stderr) + return 1 + + print(f"report\t{report_path}") + print(f"destination_root\t{destination_root}") + print(f"mode\t{args.mode}") + print(f"selected_status\t{args.status}") + print(f"rows_total\t{len(rows)}") + print(f"rows_selected\t{len(operations)}") + print(f"rows_skipped\t{skipped_rows}") + print(f"dry_run\t{'yes' if args.dry_run else 'no'}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/reports/audiobookshelf_mock_report_audiobooki_nowe.ods b/reports/audiobookshelf_mock_report_audiobooki_nowe.ods new file mode 100644 index 0000000..e3e4b1f Binary files /dev/null and b/reports/audiobookshelf_mock_report_audiobooki_nowe.ods differ diff --git a/reports/audiobookshelf_mock_report_audiobooki_nowe.tsv b/reports/audiobookshelf_mock_report_audiobooki_nowe.tsv index 75c87ae..4603854 100644 --- a/reports/audiobookshelf_mock_report_audiobooki_nowe.tsv +++ b/reports/audiobookshelf_mock_report_audiobooki_nowe.tsv @@ -2,9 +2,9 @@ verification_status verification_source verification_note status current_path au unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Abercrombie Joe - Cykl The Devils (tom 1) Diabły (Audiobooki2.pl) 73 Diabły (1).mp3 Abmercombie Joe high folder Abercrombie Joe - Cykl The Devils (tom 1) Diabły (Audiobooki2 pl) path Abmercombie Joe/Abercrombie Joe - Cykl The Devils (tom 1) Diably (Audiobooki2 pl) unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Abercrombie Joe - Czerwona kraina (Audiobooki2.pl) 42 Czerwona kraina (01).mp3 Abmercombie Joe high folder Abercrombie Joe - Czerwona kraina (Audiobooki2 pl) path Abmercombie Joe/Abercrombie Joe - Czerwona kraina (Audiobooki2 pl) unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Abercrombie Joe - Zemsta najlepiej smakuje na zimno (Audiobooki2.pl) 71 Zemsta najlepiej smakuje na zimno 01.mp3 Abmercombie Joe high folder Abercrombie Joe - Zemsta najlepiej smakuje na zimno (Audiobooki2 pl) path Abmercombie Joe/Abercrombie Joe - Zemsta najlepiej smakuje na zimno (Audiobooki2 pl) -unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Cykl Pierwsze Prawo/Joe Abercrombie - Cykl Pierwsze prawo (tom 2) Nim zawisna (Audiobooki2.pl) 73 01 Nim zawisną .mp3 Abmercombie Joe high folder Joe Abercrombie - Cykl Pierwsze prawo (tom 2) Nim zawisna (Audiobooki2 pl) path Abmercombie Joe/Joe Abercrombie - Cykl Pierwsze prawo (tom 2) Nim zawisna (Audiobooki2 pl) ignored grouping folder 'Cykl Pierwsze Prawo' -unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Cykl Pierwsze Prawo/Joe Abercrombie - Cykl Pierwsze prawo (tom 3) Ostateczny argument (Audiobooki2.pl) 86 Ostateczny argument (01).mp3 Abmercombie Joe high folder Joe Abercrombie - Cykl Pierwsze prawo (tom 3) Ostateczny argument (Audiobooki2 pl) path Abmercombie Joe/Joe Abercrombie - Cykl Pierwsze prawo (tom 3) Ostateczny argument (Audiobooki2 pl) ignored grouping folder 'Cykl Pierwsze Prawo' -unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Cykl Pierwsze Prawo/Joe Abercrombie - Cykl Pierwsze prawo Tom 01 Ostrze czyta Filip Kosior 128kbps 45 Ostrze (1).mp3 Abmercombie Joe high folder Joe Abercrombie - Cykl Pierwsze prawo Tom 01 Ostrze path Filip Kosior 128kbps Abmercombie Joe/Joe Abercrombie - Cykl Pierwsze prawo Tom 01 Ostrze {Filip Kosior 128kbps} ignored grouping folder 'Cykl Pierwsze Prawo'; narrator inferred from folder name +unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Cykl Pierwsze Prawo/Joe Abercrombie - Cykl Pierwsze prawo (tom 2) Nim zawisna (Audiobooki2.pl) 73 01 Nim zawisną .mp3 Abmercombie Joe high folder Abercrombie Cykl Pierwsze prawo tom 2 Nim zawisna Audiobooki2 pl path Abmercombie Joe/Abercrombie Cykl Pierwsze prawo tom 2 Nim zawisna Audiobooki2 pl ignored grouping folder 'Cykl Pierwsze Prawo' +unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Cykl Pierwsze Prawo/Joe Abercrombie - Cykl Pierwsze prawo (tom 3) Ostateczny argument (Audiobooki2.pl) 86 Ostateczny argument (01).mp3 Abmercombie Joe high folder Abercrombie Cykl Pierwsze prawo tom 3 Ostateczny argument Audiobooki2 pl path Abmercombie Joe/Abercrombie Cykl Pierwsze prawo tom 3 Ostateczny argument Audiobooki2 pl ignored grouping folder 'Cykl Pierwsze Prawo' +unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Abmercombie Joe/Cykl Pierwsze Prawo/Joe Abercrombie - Cykl Pierwsze prawo Tom 01 Ostrze czyta Filip Kosior 128kbps 45 Ostrze (1).mp3 Abmercombie Joe high folder Abercrombie Cykl Pierwsze prawo Tom 01 Ostrze path Filip Kosior 128kbps Abmercombie Joe/Abercrombie Cykl Pierwsze prawo Tom 01 Ostrze {Filip Kosior 128kbps} ignored grouping folder 'Cykl Pierwsze Prawo'; narrator inferred from folder name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Adler-Olsen/Jussi Adler-Olsen - Departament Q tom 1-8 (Audiobooki.pl)/Adler-Olsen Jussi - Departament Q 01. Kobieta w Klatce 15 01. Departament Q Tom 1 - Kobieta w Klatce - Prolog.mp3 Jussi Adler-Olsen medium mixed-folder Departament Q 01 Kobieta w Klatce path Jussi Adler-Olsen/Departament Q/Vol. 01 - Kobieta w Klatce author inferred from a weak path signal; series normalized from folder context; sequence inferred from folder context; author came from nested folder 'Jussi Adler - Olsen - Departament Q tom 1 - 8 (Audiobooki pl)' unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Adler-Olsen/Jussi Adler-Olsen - Departament Q tom 1-8 (Audiobooki.pl)/Adler-Olsen Jussi - Departament Q 02. Zabójcy Bażantów 15 01. Departament Q Tom 2 - Zabójcy Bażantów - Prolog.mp3 Jussi Adler-Olsen medium mixed-folder Departament Q 02 Zabójcy Bażantów path Jussi Adler-Olsen/Departament Q/Vol. 02 - Zabojcy Bazantow author inferred from a weak path signal; series normalized from folder context; sequence inferred from folder context; author came from nested folder 'Jussi Adler - Olsen - Departament Q tom 1 - 8 (Audiobooki pl)' unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Adler-Olsen/Jussi Adler-Olsen - Departament Q tom 1-8 (Audiobooki.pl)/Adler-Olsen Jussi - Departament Q 03. Wybawienie 18 01. Departament Q Tom 3 - Wybawienie - Prolog.mp3 Jussi Adler-Olsen medium mixed-folder Departament Q 03 Wybawienie path Jussi Adler-Olsen/Departament Q/Vol. 03 - Wybawienie author inferred from a weak path signal; series normalized from folder context; sequence inferred from folder context; author came from nested folder 'Jussi Adler - Olsen - Departament Q tom 1 - 8 (Audiobooki pl)' @@ -16,7 +16,7 @@ unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Brandon Sanderson unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Brandon Sanderson - Legion (czyta T.Sobczak) 96kbps 32 01.Legion.mp3 Brandon Sanderson medium mixed-folder Legion 96kbps path T Sobczak Brandon Sanderson/Legion 96kbps {T Sobczak} author inferred from a weak path signal; narrator inferred from folder name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Brandon Sanderson - Słoneczny mąż (czyta D.Odija) 107 kbps 19 01.mp3 Brandon Sanderson medium mixed-folder Słoneczny mąż 107 kbps path D Odija Brandon Sanderson/Sloneczny maz 107 kbps {D Odija} author inferred from a weak path signal; narrator inferred from folder name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Brandon Sanderson - Warkocz ze Szmaragdowego Morza (czyta M. Kowalik) 128kbps 67 01.mp3 Brandon Sanderson medium mixed-folder Warkocz ze Szmaragdowego Morza 128kbps path M Kowalik Brandon Sanderson/Warkocz ze Szmaragdowego Morza 128kbps {M Kowalik} author inferred from a weak path signal; narrator inferred from folder name -unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Chmielarz Wojciech/Wojciech Chmielarz - Prosta sprawa (2020) czyta Przemysław Bluszcz [audiobook PL] (Audiobooki.pl) 53 01.mp3 Wojciech Chmielarz high folder Prosta sprawa (2020) path Przemysław Bluszcz [audiobook PL] (Audiobooki pl) Wojciech Chmielarz/Prosta sprawa (2020) {Przemyslaw Bluszcz [audiobook PL] (Audiobooki pl)} narrator inferred from folder name; author order normalized from current folder/file name +unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Chmielarz Wojciech/Wojciech Chmielarz - Prosta sprawa (2020) czyta Przemysław Bluszcz [audiobook PL] (Audiobooki.pl) 52 01.mp3 Wojciech Chmielarz high folder Prosta sprawa (2020) path Przemysław Bluszcz [audiobook PL] (Audiobooki pl) Wojciech Chmielarz/Prosta sprawa (2020) {Przemyslaw Bluszcz [audiobook PL] (Audiobooki pl)} narrator inferred from folder name; author order normalized from current folder/file name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Chmielarz Wojciech/Wojciech Chmielarz - Wampir czyta Mateusz Znaniecki 160kbps 80 00 Wampir.mp3 Wojciech Chmielarz high folder Wampir path Mateusz Znaniecki 160kbps Wojciech Chmielarz/Wampir {Mateusz Znaniecki 160kbps} narrator inferred from folder name; author order normalized from current folder/file name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Chmielarz Wojciech/Wojciech Chmielarz - Zombie czyta Grzegorz Przybył 96kbps 15 01.mp3 Wojciech Chmielarz high folder Zombie path Grzegorz Przybył 96kbps Wojciech Chmielarz/Zombie {Grzegorz Przybyl 96kbps} narrator inferred from folder name; author order normalized from current folder/file name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Chmielarz Wojciech/Ćwiek Jakub i Chmielarz Wojciech - Skowyt 8 1.mp3 Chmielarz Wojciech high folder Ćwiek Jakub i Chmielarz Wojciech - Skowyt path Chmielarz Wojciech/Cwiek Jakub i Chmielarz Wojciech - Skowyt @@ -35,8 +35,8 @@ unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Coben_Bolitar/Cob unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Dan Simons/Dan Simmons - cykl Hyperion (tom 2) Upadek Hyperiona (Audiobooki2.pl)/Dan Simmons - cykl Hyperion (tom 2) Upadek Hyperiona 46 01 Upadek Hyperiona.mp3 Dan Simons high folder Dan Simmons - cykl Hyperion (tom 2) Upadek Hyperiona path Dan Simons/Dan Simmons - cykl Hyperion (tom 2) Upadek Hyperiona ignored grouping folder 'Dan Simmons - cykl Hyperion (tom 2) Upadek Hyperiona (Audiobooki2 pl)' unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Dan Simons/Dan Simmons - cykl Hyperion (tom 3) Endymion (Audiobooki2.pl) 106 001 Endymion.mp3 Dan Simons high folder Dan Simmons - cykl Hyperion (tom 3) Endymion (Audiobooki2 pl) path Dan Simons/Dan Simmons - cykl Hyperion (tom 3) Endymion (Audiobooki2 pl) unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Dan Simons/Dan Simmons - cykl Hyperion (tom 4) Triumf Endymiona (Audiobooki2.pl)/Dan Simmons - cykl Hyperion (tom 4) Triumf Endymiona 148 001 Triumf Endymiona.mp3 Dan Simons high folder Dan Simmons - cykl Hyperion (tom 4) Triumf Endymiona path Dan Simons/Dan Simmons - cykl Hyperion (tom 4) Triumf Endymiona ignored grouping folder 'Dan Simmons - cykl Hyperion (tom 4) Triumf Endymiona (Audiobooki2 pl)' -unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Glukhowsky Dmitry/Metro 2033/Dmitry Glukhovsky - Metro 2033 czyta Krzysztof Gosztyła (Audiobooki.pl) 44 01. Metro 2033.mp3 Glukhowsky Dmitry high folder Dmitry Glukhovsky Metro 2033 path Krzysztof Gosztyła (Audiobooki pl) Glukhowsky Dmitry/Dmitry Glukhovsky/Metro 2033 {Krzysztof Gosztyla (Audiobooki pl)} ignored grouping folder 'Metro 2033'; narrator inferred from folder name; series normalized from folder context -unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Glukhowsky Dmitry/Metro 2034/Dmitry Glukhovsky - Metro 2034 czyta Krzysztof Gosztyła [audiobook PL] (Audiobooki.pl) 40 01 Metro 2034.mp3 Glukhowsky Dmitry high folder Dmitry Glukhovsky Metro 2034 path Krzysztof Gosztyła [audiobook PL] (Audiobooki pl) Glukhowsky Dmitry/Dmitry Glukhovsky/Metro 2034 {Krzysztof Gosztyla [audiobook PL] (Audiobooki pl)} ignored grouping folder 'Metro 2034'; narrator inferred from folder name; series normalized from folder context +unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Glukhowsky Dmitry/Metro 2033/Dmitry Glukhovsky - Metro 2033 czyta Krzysztof Gosztyła (Audiobooki.pl) 44 01. Metro 2033.mp3 Glukhowsky Dmitry high folder Glukhovsky Metro 2033 path Krzysztof Gosztyła Audiobooki pl Glukhowsky Dmitry/Glukhovsky/Metro 2033 {Krzysztof Gosztyla Audiobooki pl} ignored grouping folder 'Metro 2033'; narrator inferred from folder name; series normalized from folder context +unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Glukhowsky Dmitry/Metro 2034/Dmitry Glukhovsky - Metro 2034 czyta Krzysztof Gosztyła [audiobook PL] (Audiobooki.pl) 40 01 Metro 2034.mp3 Glukhowsky Dmitry high folder Glukhovsky Metro 2034 path Krzysztof Gosztyła audiobook PL Audiobooki pl Glukhowsky Dmitry/Glukhovsky/Metro 2034 {Krzysztof Gosztyla audiobook PL Audiobooki pl} ignored grouping folder 'Metro 2034'; narrator inferred from folder name; series normalized from folder context unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Glukhowsky Dmitry/Metro 2035/Dimitry Glukhovsky - Metro 2035 czyta Krzysztof Gosztyła (Audiobooki2.pl) 54 (eds-pl) Dmitry_Glukhovsky-METRO 2035 (01).mp3 Glukhowsky Dmitry high folder Dimitry Glukhovsky - Metro 2035 path Krzysztof Gosztyła (Audiobooki2 pl) Glukhowsky Dmitry/Dimitry Glukhovsky - Metro 2035 {Krzysztof Gosztyla (Audiobooki2 pl)} ignored grouping folder 'Metro 2035'; narrator inferred from folder name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Kuzminska Malgorzata Michal/Cykl Anna Serfin Sebastian Strzygon/T1-Sleboda 13 Śleboda (1).mp3 Kuzminska Malgorzata Michal high folder 1 Sleboda path Kuzminska Malgorzata Michal/Vol. 1 - Sleboda ignored grouping folder 'Cykl Anna Serfin Sebastian Strzygon'; sequence inferred from folder name unverified ready /mnt/nextcloudExtDS/Ksiazki/Audiobooki-Nowe/Kuzminska Malgorzata Michal/Cykl Anna Serfin Sebastian Strzygon/T2-Pionek 17 01 Pionek.mp3 Kuzminska Malgorzata Michal high folder 2 Pionek path Kuzminska Malgorzata Michal/Vol. 2 - Pionek ignored grouping folder 'Cykl Anna Serfin Sebastian Strzygon'; sequence inferred from folder name diff --git a/tests/__pycache__/test_apply_abs_mock_report.cpython-311.pyc b/tests/__pycache__/test_apply_abs_mock_report.cpython-311.pyc new file mode 100644 index 0000000..efc7775 Binary files /dev/null and b/tests/__pycache__/test_apply_abs_mock_report.cpython-311.pyc differ diff --git a/tests/__pycache__/test_generate_abs_mock_report.cpython-311.pyc b/tests/__pycache__/test_generate_abs_mock_report.cpython-311.pyc index 5451c3d..d3ffe2a 100644 Binary files a/tests/__pycache__/test_generate_abs_mock_report.cpython-311.pyc and b/tests/__pycache__/test_generate_abs_mock_report.cpython-311.pyc differ diff --git a/tests/test_apply_abs_mock_report.py b/tests/test_apply_abs_mock_report.py new file mode 100644 index 0000000..8e42ccb --- /dev/null +++ b/tests/test_apply_abs_mock_report.py @@ -0,0 +1,230 @@ +import csv +import io +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import apply_abs_mock_report as apply_report + + +class ApplyAbsMockReportTests(unittest.TestCase): + def write_report(self, path: Path, rows: list[dict[str, str]]) -> None: + fieldnames = [ + "verification_status", + "verification_source", + "verification_note", + "status", + "current_path", + "audio_file_count", + "sample_audio_file", + "author", + "author_confidence", + "author_source", + "series", + "sequence", + "publish_year", + "title", + "title_source", + "narrator", + "proposed_abs_path", + "notes", + ] + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fieldnames, delimiter="\t") + writer.writeheader() + writer.writerows(rows) + + def create_book(self, root: Path, relative_path: str) -> Path: + book_root = root / relative_path + (book_root / "Disc 1").mkdir(parents=True) + (book_root / "Disc 1" / "01.mp3").write_bytes(b"audio") + (book_root / "cover.jpg").write_bytes(b"cover") + return book_root + + def test_copy_mode_recreates_report_structure(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + source_root = tmp / "source" + destination_root = tmp / "destination" + report_path = tmp / "report.tsv" + book_root = self.create_book(source_root, "Old Author/Old Book") + + self.write_report( + report_path, + [ + { + "status": "ready", + "current_path": str(book_root), + "proposed_abs_path": "New Author/New Series/Vol. 01 - New Book", + } + ], + ) + + exit_code = apply_report.main( + [ + "--report", + str(report_path), + "--destination-root", + str(destination_root), + "--mode", + "copy", + ] + ) + + self.assertEqual(exit_code, 0) + copied_root = destination_root / "New Author" / "New Series" / "Vol. 01 - New Book" + self.assertTrue((copied_root / "Disc 1" / "01.mp3").exists()) + self.assertTrue((copied_root / "cover.jpg").exists()) + self.assertTrue((book_root / "Disc 1" / "01.mp3").exists()) + + def test_move_mode_removes_source_tree(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + source_root = tmp / "source" + destination_root = tmp / "destination" + report_path = tmp / "report.tsv" + book_root = self.create_book(source_root, "Old Author/Old Book") + + self.write_report( + report_path, + [ + { + "status": "ready", + "current_path": str(book_root), + "proposed_abs_path": "Author/Book", + } + ], + ) + + exit_code = apply_report.main( + [ + "--report", + str(report_path), + "--destination-root", + str(destination_root), + "--mode", + "move", + ] + ) + + self.assertEqual(exit_code, 0) + moved_root = destination_root / "Author" / "Book" + self.assertTrue((moved_root / "Disc 1" / "01.mp3").exists()) + self.assertFalse(book_root.exists()) + + def test_dry_run_prints_plan_without_creating_destination(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + source_root = tmp / "source" + destination_root = tmp / "destination" + report_path = tmp / "report.tsv" + book_root = self.create_book(source_root, "Old Author/Old Book") + + self.write_report( + report_path, + [ + { + "status": "ready", + "current_path": str(book_root), + "proposed_abs_path": "Author/Book", + } + ], + ) + + stdout = io.StringIO() + with mock.patch("sys.stdout", stdout): + exit_code = apply_report.main( + [ + "--report", + str(report_path), + "--destination-root", + str(destination_root), + "--dry-run", + ] + ) + + self.assertEqual(exit_code, 0) + self.assertIn("plan\t", stdout.getvalue()) + self.assertFalse(destination_root.exists()) + + def test_duplicate_targets_fail_validation(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + source_root = tmp / "source" + report_path = tmp / "report.tsv" + first_book = self.create_book(source_root, "Author A/Book One") + second_book = self.create_book(source_root, "Author B/Book Two") + + self.write_report( + report_path, + [ + { + "status": "ready", + "current_path": str(first_book), + "proposed_abs_path": "Author/Shared", + }, + { + "status": "ready", + "current_path": str(second_book), + "proposed_abs_path": "Author/Shared", + }, + ], + ) + + stderr = io.StringIO() + with mock.patch("sys.stderr", stderr): + exit_code = apply_report.main( + [ + "--report", + str(report_path), + "--destination-root", + str(tmp / "destination"), + ] + ) + + self.assertEqual(exit_code, 1) + self.assertIn("duplicate proposed_abs_path", stderr.getvalue()) + + def test_default_status_only_applies_ready_rows(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + source_root = tmp / "source" + destination_root = tmp / "destination" + report_path = tmp / "report.tsv" + ready_book = self.create_book(source_root, "Ready Author/Ready Book") + review_book = self.create_book(source_root, "Review Author/Review Book") + + self.write_report( + report_path, + [ + { + "status": "ready", + "current_path": str(ready_book), + "proposed_abs_path": "Ready Author/Ready Book", + }, + { + "status": "review", + "current_path": str(review_book), + "proposed_abs_path": "Review Author/Review Book", + }, + ], + ) + + exit_code = apply_report.main( + [ + "--report", + str(report_path), + "--destination-root", + str(destination_root), + ] + ) + + self.assertEqual(exit_code, 0) + self.assertTrue((destination_root / "Ready Author" / "Ready Book").exists()) + self.assertFalse((destination_root / "Review Author" / "Review Book").exists()) + self.assertTrue(review_book.exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_generate_abs_mock_report.py b/tests/test_generate_abs_mock_report.py index 2769a8a..ef9fd0d 100644 --- a/tests/test_generate_abs_mock_report.py +++ b/tests/test_generate_abs_mock_report.py @@ -83,6 +83,69 @@ class InferBookTests(unittest.TestCase): self.assertEqual(row["sequence"], "05") self.assertEqual(row["title"], "Pogrzebany") + def test_build_proposed_path_omits_author_from_subfolders(self) -> None: + proposed_path = report.build_proposed_abs_path( + "Jussi Adler-Olsen", + "Adler-Olsen Jussi - Departament Q", + "04", + "", + "Kartoteka 64 - Jussi Adler-Olsen", + "", + ) + + self.assertEqual( + proposed_path, + "Jussi Adler-Olsen/Departament Q/Vol. 04 - Kartoteka 64", + ) + + def test_build_proposed_path_omits_author_only_series_folder(self) -> None: + proposed_path = report.build_proposed_abs_path( + "Jeffrey Archer", + "Archer", + "", + "", + "Kain I Abel", + "", + ) + + self.assertEqual(proposed_path, "Jeffrey Archer/Kain I Abel") + + def test_strips_nearly_matching_author_prefix_from_title(self) -> None: + row = self.infer_row( + "Abmercombie Joe/Abercrombie Joe - Czerwona kraina (Audiobooki2.pl)", + ["Czerwona kraina (01).mp3"], + ) + + self.assertEqual(row["author"], "Abmercombie Joe") + self.assertEqual(row["title"], "Czerwona kraina (Audiobooki2 pl)") + self.assertEqual( + row["proposed_abs_path"], + "Abmercombie Joe/Czerwona kraina (Audiobooki2 pl)", + ) + + def test_strips_nearly_matching_author_prefix_from_nested_title(self) -> None: + row = self.infer_row( + "Dan Simons/Dan Simmons - cykl Hyperion (tom 3) Endymion (Audiobooki2.pl)", + ["001 Endymion.mp3"], + ) + + self.assertEqual(row["author"], "Dan Simons") + self.assertEqual(row["title"], "cykl Hyperion (tom 3) Endymion (Audiobooki2 pl)") + self.assertEqual( + row["proposed_abs_path"], + "Dan Simons/cykl Hyperion (tom 3) Endymion (Audiobooki2 pl)", + ) + + def test_strips_multi_author_prefix_when_current_author_is_in_root(self) -> None: + row = self.infer_row( + "Chmielarz Wojciech/Ćwiek Jakub i Chmielarz Wojciech - Skowyt", + ["1.mp3"], + ) + + self.assertEqual(row["author"], "Chmielarz Wojciech") + self.assertEqual(row["title"], "Skowyt") + self.assertEqual(row["proposed_abs_path"], "Chmielarz Wojciech/Skowyt") + if __name__ == "__main__": unittest.main()