200 lines
8.3 KiB
Python
200 lines
8.3 KiB
Python
import shutil
|
|
from pathlib import Path
|
|
|
|
from .. import config
|
|
from ..naming import render_template, sanitize
|
|
from . import tagger
|
|
|
|
|
|
def _cleanup_sources(src_dir: Path, paths: list[Path]) -> None:
|
|
"""Remove emptied source folders, deepest first. Folders that still hold
|
|
files and the download dir itself are left alone."""
|
|
download_root = Path(config.DOWNLOAD_DIR).resolve()
|
|
candidates = {src_dir}
|
|
for p in paths:
|
|
d = p.parent
|
|
while d == src_dir or src_dir in d.parents:
|
|
candidates.add(d)
|
|
if d == src_dir:
|
|
break
|
|
d = d.parent
|
|
for d in sorted(candidates, key=lambda p: len(p.parts), reverse=True):
|
|
if d.resolve() == download_root:
|
|
continue # merged items can point at the download dir itself
|
|
if d.is_dir() and not any(f.is_file() for f in d.rglob("*")):
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
|
|
def import_item(item_path: str, files: list[str], is_dir: bool, request, library) -> str:
|
|
"""Move scanned files into the library, renamed per the library's templates.
|
|
Returns the destination path (folder for multi-file audiobooks, else the file)."""
|
|
root = Path(library.root_path)
|
|
folder = root / render_template(library.folder_template, request)
|
|
base = render_template(library.file_template, request)
|
|
|
|
paths = [Path(f) for f in files]
|
|
if is_dir and len(paths) > 1:
|
|
width = max(2, len(str(len(paths))))
|
|
targets = [
|
|
folder / sanitize(f"{base} - Part {i:0{width}d}{src.suffix.lower()}")
|
|
for i, src in enumerate(paths, 1)
|
|
]
|
|
# check every target first: otherwise a second title with the same
|
|
# rendered name would silently overwrite the first one
|
|
taken = [d for d in targets if d.exists()]
|
|
if taken:
|
|
raise FileExistsError(
|
|
f"Destination already exists: {taken[0]}"
|
|
+ (f" (und {len(taken) - 1} weitere)" if len(taken) > 1 else "")
|
|
+ " — trägt ein anderer Titel nach diesem Namensschema denselben Namen?"
|
|
)
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
for i, (src, dest) in enumerate(zip(paths, targets), 1):
|
|
shutil.move(str(src), dest)
|
|
if request.media_type == "audiobook":
|
|
tagger.tag_audio(dest, request, track=i, total=len(paths))
|
|
_cleanup_sources(Path(item_path), paths)
|
|
return str(folder)
|
|
|
|
src = paths[0]
|
|
dest = folder / sanitize(f"{base}{src.suffix.lower()}")
|
|
if dest.exists():
|
|
raise FileExistsError(f"Destination already exists: {dest}")
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(src), dest)
|
|
if request.media_type == "audiobook":
|
|
tagger.tag_audio(dest, request)
|
|
if is_dir:
|
|
# a folder holding a single audiobook file is emptied by the move too
|
|
_cleanup_sources(Path(item_path), paths)
|
|
return str(dest)
|
|
|
|
|
|
def append_to_import(files: list[str], request, library) -> str:
|
|
"""Add files to an already imported audiobook. Numbering continues after
|
|
what is there, and all tracks are re-tagged with the new total."""
|
|
dest_dir = Path(request.imported_path or "")
|
|
if not dest_dir.is_dir():
|
|
raise FileNotFoundError(
|
|
f"Zielordner existiert nicht (mehr): {request.imported_path}"
|
|
)
|
|
existing = sorted(
|
|
f for f in dest_dir.iterdir()
|
|
if f.is_file() and f.suffix.lower() in config.AUDIOBOOK_EXTENSIONS
|
|
)
|
|
paths = [Path(f) for f in files]
|
|
total = len(existing) + len(paths)
|
|
width = max(2, len(str(total)))
|
|
base = render_template(library.file_template, request)
|
|
|
|
for i, src in enumerate(paths, len(existing) + 1):
|
|
dest = dest_dir / sanitize(f"{base} - Part {i:0{width}d}{src.suffix.lower()}")
|
|
if dest.exists():
|
|
raise FileExistsError(f"Destination already exists: {dest}")
|
|
shutil.move(str(src), dest)
|
|
if request.media_type == "audiobook":
|
|
tagger.tag_audio(dest, request, track=i, total=total)
|
|
if request.media_type == "audiobook":
|
|
for i, f in enumerate(existing, 1):
|
|
tagger.tag_audio(f, request, track=i, total=total)
|
|
for src_dir in {p.parent for p in paths}:
|
|
_cleanup_sources(src_dir, paths)
|
|
return str(dest_dir)
|
|
|
|
|
|
def relocate(request, library) -> str:
|
|
"""Move an imported audiobook to where the templates say it belongs now.
|
|
Extras move along, emptied folders are removed."""
|
|
src = Path(request.imported_path or "")
|
|
if not src.exists():
|
|
raise FileNotFoundError(f"Pfad existiert nicht (mehr): {request.imported_path}")
|
|
|
|
root = Path(library.root_path)
|
|
folder = root / render_template(library.folder_template, request)
|
|
base = render_template(library.file_template, request)
|
|
|
|
if src.is_file():
|
|
dest = folder / sanitize(f"{base}{src.suffix.lower()}")
|
|
if dest.resolve() == src.resolve():
|
|
return str(src)
|
|
if dest.exists():
|
|
raise FileExistsError(f"Destination already exists: {dest}")
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(src), dest)
|
|
if request.media_type == "audiobook":
|
|
tagger.tag_audio(dest, request)
|
|
_remove_if_empty(src.parent, root)
|
|
return str(dest)
|
|
|
|
# one rename keeps the folder's inode, so Audiobookshelf sees a move and keeps
|
|
# the item; filling a new folder file by file makes it add a new item halfway
|
|
# and cache whatever cover it saw at that moment
|
|
moved_from = None
|
|
if not folder.exists():
|
|
folder.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
src.rename(folder)
|
|
moved_from, src = src, folder
|
|
except OSError:
|
|
pass # another filesystem, or a target inside the source: file by file
|
|
|
|
audio = sorted(
|
|
(f for f in src.iterdir()
|
|
if f.is_file() and f.suffix.lower() in config.AUDIOBOOK_EXTENSIONS),
|
|
key=lambda f: f.name,
|
|
)
|
|
extras = [f for f in src.iterdir() if f.is_file() and f not in audio]
|
|
same_folder = folder.resolve() == src.resolve() if folder.exists() else False
|
|
if not same_folder and folder.exists() and any(folder.iterdir()):
|
|
raise FileExistsError(f"Destination already exists: {folder}")
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
|
|
width = max(2, len(str(len(audio))))
|
|
targets = [
|
|
folder / sanitize(
|
|
f"{base} - Part {i:0{width}d}{f.suffix.lower()}" if len(audio) > 1
|
|
else f"{base}{f.suffix.lower()}"
|
|
)
|
|
for i, f in enumerate(audio, 1)
|
|
]
|
|
# renaming in place may aim at a name that is already taken: fine when the
|
|
# occupant is one of the files being moved, never when it is a stranger
|
|
sources = {f.resolve() for f in audio}
|
|
taken = [d for d in targets if d.exists() and d.resolve() not in sources]
|
|
if taken or len(set(targets)) != len(targets):
|
|
raise FileExistsError(
|
|
f"Destination already exists: {taken[0] if taken else targets[0]}"
|
|
" — trägt ein anderer Titel nach diesem Namensschema denselben Namen?"
|
|
)
|
|
current = list(audio)
|
|
for i, dest in enumerate(targets):
|
|
if current[i].resolve() != dest.resolve():
|
|
if dest.exists():
|
|
# a file further down the list still sits in this slot
|
|
j = next(k for k, c in enumerate(current) if c.resolve() == dest.resolve())
|
|
stashed = dest.with_name(dest.name + ".wordarr-tmp")
|
|
shutil.move(str(current[j]), stashed)
|
|
current[j] = stashed
|
|
shutil.move(str(current[i]), dest)
|
|
current[i] = dest
|
|
if request.media_type == "audiobook":
|
|
tagger.tag_audio(dest, request, track=i + 1, total=len(audio))
|
|
for extra in extras:
|
|
target = folder / extra.name
|
|
if target.resolve() != extra.resolve() and not target.exists():
|
|
shutil.move(str(extra), target)
|
|
if not same_folder:
|
|
_remove_if_empty(src, root)
|
|
if moved_from:
|
|
_remove_if_empty(moved_from.parent, root)
|
|
return str(folder)
|
|
|
|
|
|
def _remove_if_empty(folder: Path, root: Path) -> None:
|
|
"""Drop the folder and its emptied parents, never touching the library root."""
|
|
while folder.is_dir() and folder.resolve() != root.resolve() and root in folder.parents:
|
|
if any(folder.iterdir()):
|
|
return
|
|
folder.rmdir()
|
|
folder = folder.parent
|