759 lines
29 KiB
Python
759 lines
29 KiB
Python
"""
|
|
Hardlink Manager for Curator
|
|
|
|
Creates directory structure based on file tags and creates hardlinks
|
|
to organize files without duplicating them on disk.
|
|
|
|
Example:
|
|
A file with tags "žánr/Komedie", "žánr/Akční", "rok/1988" will create:
|
|
|
|
output/
|
|
├── žánr/
|
|
│ ├── Komedie/
|
|
│ │ └── film.mkv (hardlink)
|
|
│ └── Akční/
|
|
│ └── film.mkv (hardlink)
|
|
└── rok/
|
|
└── 1988/
|
|
└── film.mkv (hardlink)
|
|
"""
|
|
|
|
import os
|
|
import random
|
|
import shutil
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
from .file import File
|
|
from .naming import is_valid_year, parse_pool_stem
|
|
|
|
|
|
class _SafeDict(dict):
|
|
"""dict for str.format_map that leaves unknown fields as an empty string."""
|
|
|
|
def __missing__(self, key):
|
|
return ""
|
|
|
|
|
|
class HardlinkManager:
|
|
"""Manager for creating hardlink-based directory structures from tagged files.
|
|
|
|
The output layout is driven by a *category → root folder* mapping
|
|
(``category_roots``). Each tag is placed at
|
|
``output/<root>/<tag_name>/<file>``; an empty root means the tag's own
|
|
folders sit directly at the output root (e.g. genre folders next to the
|
|
"Dle roku" / "Dle země původu" folders). The legacy ``categories`` list
|
|
(folder == category name) is still accepted and treated as the identity
|
|
mapping ``{cat: cat}``.
|
|
"""
|
|
|
|
def __init__(self, output_dir: Path):
|
|
"""
|
|
Initialize HardlinkManager.
|
|
|
|
Args:
|
|
output_dir: Base directory where the tag-based structure will be created
|
|
"""
|
|
self.output_dir = Path(output_dir)
|
|
self.created_links: list[Path] = []
|
|
self.errors: list[tuple[Path, str]] = []
|
|
|
|
def _resolve_roots(
|
|
self,
|
|
categories: list[str] | None,
|
|
category_roots: dict[str, str] | None,
|
|
) -> dict[str, str] | None:
|
|
"""Normalize the two filter styles into a category → root-folder map.
|
|
|
|
``None`` means "all categories", folder == category name.
|
|
"""
|
|
if category_roots is not None:
|
|
return dict(category_roots)
|
|
if categories is not None:
|
|
return {cat: cat for cat in categories}
|
|
return None
|
|
|
|
def _clean_title(self, file_obj: File) -> str:
|
|
"""The movie's clean title (no trailing ``(YYYY)``), path-safe."""
|
|
ctx = file_obj.name_context()
|
|
title = str(ctx.get("title") or file_obj.file_path.stem or "").strip()
|
|
parsed = parse_pool_stem(title)
|
|
if parsed is not None:
|
|
title = parsed[0]
|
|
title = title.replace("/", "-").replace("\\", "-").strip()
|
|
return title or file_obj.file_path.stem
|
|
|
|
def _default_link_name(self, file_obj: File) -> str:
|
|
"""Filmotéka hardlink name: the clean title + extension (no year)."""
|
|
return f"{self._clean_title(file_obj)}{file_obj.file_path.suffix}"
|
|
|
|
def _year_link_name(self, file_obj: File) -> str | None:
|
|
"""``Title (YYYY).ext`` used to disambiguate a name collision, or None."""
|
|
year = str(file_obj.name_context().get("year") or "").strip()
|
|
if not is_valid_year(year):
|
|
return None
|
|
return f"{self._clean_title(file_obj)} ({year}){file_obj.file_path.suffix}"
|
|
|
|
def _render_template(self, file_obj: File, template: str) -> str:
|
|
"""Render a per-category filename template; fall back to the clean name."""
|
|
try:
|
|
rendered = template.format_map(_SafeDict(file_obj.name_context()))
|
|
except ValueError, KeyError, IndexError, AttributeError:
|
|
return self._default_link_name(file_obj)
|
|
rendered = rendered.replace("/", "-").replace("\\", "-").strip()
|
|
return rendered or self._default_link_name(file_obj)
|
|
|
|
def _plan_links(
|
|
self,
|
|
files: list[File],
|
|
roots: dict[str, str] | None,
|
|
transforms: dict[str, str] | None,
|
|
templates: dict[str, str] | None,
|
|
) -> list[tuple[File, Path]]:
|
|
"""Resolve the tag tree to a collision-free ``(file, link_path)`` plan.
|
|
|
|
The default hardlink name is the movie's **clean title** (no year). When
|
|
two *different* movies would land on the same name in the same folder,
|
|
the colliding ones fall back to the year-qualified ``Title (YYYY).ext`` to
|
|
tell them apart; a residual clash (same title *and* year) gets a stable
|
|
numeric suffix so every path is unique. A per-category
|
|
``filename_template`` overrides the default inside that category's folders.
|
|
|
|
Both link creation and obsolete detection consume this one plan, so they
|
|
can never disagree on a name (which is what caused hardlink churn before).
|
|
"""
|
|
# (file_obj, target_dir, name, is_templated), one per file+folder
|
|
entries: list[list] = []
|
|
seen_fd: set[tuple[str, str]] = set()
|
|
for file_obj in files:
|
|
for tag in file_obj.tags:
|
|
target_dir = self._target_dir(tag, roots, transforms)
|
|
if target_dir is None:
|
|
continue
|
|
key = (str(file_obj.file_path), str(target_dir))
|
|
if key in seen_fd: # one link per movie per folder
|
|
continue
|
|
seen_fd.add(key)
|
|
template = templates.get(tag.category) if templates else None
|
|
if template:
|
|
entries.append(
|
|
[
|
|
file_obj,
|
|
target_dir,
|
|
self._render_template(file_obj, template),
|
|
True,
|
|
]
|
|
)
|
|
else:
|
|
entries.append(
|
|
[file_obj, target_dir, self._default_link_name(file_obj), False]
|
|
)
|
|
|
|
# Disambiguate default-named collisions (different movies) by their year.
|
|
groups: dict[tuple[Path, str], set[str]] = defaultdict(set)
|
|
for file_obj, target_dir, name, templated in entries:
|
|
if not templated:
|
|
groups[(target_dir, name)].add(str(file_obj.file_path))
|
|
for entry in entries:
|
|
file_obj, target_dir, name, templated = entry
|
|
if not templated and len(groups[(target_dir, name)]) > 1:
|
|
year_name = self._year_link_name(file_obj)
|
|
if year_name:
|
|
entry[2] = year_name
|
|
|
|
# Final pass: guarantee unique paths (stable numeric suffix on any tie).
|
|
by_path: dict[tuple[Path, str], list[list]] = defaultdict(list)
|
|
for entry in entries:
|
|
by_path[(entry[1], entry[2])].append(entry)
|
|
plan: list[tuple[File, Path]] = []
|
|
for (target_dir, name), group in by_path.items():
|
|
if len(group) == 1:
|
|
plan.append((group[0][0], target_dir / name))
|
|
continue
|
|
ordered = sorted(group, key=lambda e: str(e[0].file_path))
|
|
stem, suffix = Path(name).stem, Path(name).suffix
|
|
for i, entry in enumerate(ordered):
|
|
fname = name if i == 0 else f"{stem}_{i + 1}{suffix}"
|
|
plan.append((entry[0], target_dir / fname))
|
|
return plan
|
|
|
|
def _folder_value(self, tag, transforms: dict[str, str] | None) -> str:
|
|
"""Folder name for a tag — its value run through the category transform.
|
|
|
|
The tag keeps its exact value; the grouping (e.g. rating → ten-point
|
|
band) only happens here, when naming the folder.
|
|
"""
|
|
if transforms and tag.category in transforms:
|
|
from .transforms import apply_transform
|
|
|
|
return apply_transform(tag.name, transforms[tag.category])
|
|
return tag.name
|
|
|
|
def _target_dir(
|
|
self,
|
|
tag,
|
|
roots: dict[str, str] | None,
|
|
transforms: dict[str, str] | None = None,
|
|
) -> Path | None:
|
|
"""Output directory for a tag, or None if its category is excluded."""
|
|
if roots is None:
|
|
folder = tag.category
|
|
elif tag.category in roots:
|
|
folder = roots[tag.category]
|
|
else:
|
|
return None
|
|
base = self.output_dir / folder if folder else self.output_dir
|
|
return base / self._folder_value(tag, transforms)
|
|
|
|
def _managed_top_dirs(
|
|
self,
|
|
files: list[File],
|
|
roots: dict[str, str] | None,
|
|
transforms: dict[str, str] | None = None,
|
|
reserved_subfolders: set[str] | None = None,
|
|
) -> set[str] | None:
|
|
"""Top-level output folders owned by the tag tree (None = all of them).
|
|
|
|
A category with a non-empty root owns that root folder. A category placed
|
|
at the output root (empty root, e.g. genres) owns its own folders at the
|
|
root — and, so a genre whose last movie dropped the tag still gets its
|
|
now-stale folder cleaned, *every* root-level folder is treated as owned
|
|
except the grouping roots and any ``reserved_subfolders`` (copy-as-is
|
|
mirrors, "Tipy dne", …). The obsolete check still only removes links to
|
|
current pool files, so reserved/foreign folders are doubly protected.
|
|
"""
|
|
if roots is None:
|
|
return None
|
|
grouping_roots = {folder for folder in roots.values() if folder}
|
|
has_root_level = any(not folder for folder in roots.values())
|
|
tops: set[str] = set(grouping_roots)
|
|
|
|
# genre folders currently present on the movies
|
|
for cat, folder in roots.items():
|
|
if not folder:
|
|
for file_obj in files:
|
|
for tag in file_obj.tags:
|
|
if tag.category == cat:
|
|
tops.add(self._folder_value(tag, transforms))
|
|
|
|
# plus any other root-level folder (catches genres that lost their last
|
|
# movie), minus grouping roots and reserved (mirrors / Tipy dne / …)
|
|
if has_root_level and self.output_dir.exists():
|
|
reserved = set(reserved_subfolders or set())
|
|
for entry in self.output_dir.iterdir():
|
|
if (
|
|
entry.is_dir()
|
|
and entry.name not in grouping_roots
|
|
and entry.name not in reserved
|
|
):
|
|
tops.add(entry.name)
|
|
return tops
|
|
|
|
def create_structure_for_files(
|
|
self,
|
|
files: list[File],
|
|
categories: list[str] | None = None,
|
|
dry_run: bool = False,
|
|
category_roots: dict[str, str] | None = None,
|
|
category_transforms: dict[str, str] | None = None,
|
|
category_filename_templates: dict[str, str] | None = None,
|
|
) -> tuple[int, int]:
|
|
"""
|
|
Create hardlink structure for given files based on their tags.
|
|
|
|
Args:
|
|
files: List of File objects to process
|
|
categories: Optional list of categories to include (None = all)
|
|
dry_run: If True, only simulate without creating actual links
|
|
category_roots: Optional category → root-folder map (see class doc);
|
|
overrides ``categories`` when given.
|
|
category_transforms: Optional category → transform-name map applied to
|
|
the tag value when naming its folder (e.g. rating → decade band).
|
|
category_filename_templates: Optional category → hardlink-name template
|
|
applied only inside that category's folders.
|
|
|
|
Returns:
|
|
Tuple of (successful_links, failed_links)
|
|
"""
|
|
self.created_links = []
|
|
self.errors = []
|
|
|
|
roots = self._resolve_roots(categories, category_roots)
|
|
success_count = 0
|
|
fail_count = 0
|
|
|
|
plan = self._plan_links(
|
|
files, roots, category_transforms, category_filename_templates
|
|
)
|
|
for file_obj, target_file in plan:
|
|
try:
|
|
if not dry_run:
|
|
target_file.parent.mkdir(parents=True, exist_ok=True)
|
|
if target_file.exists():
|
|
# Already the right link → nothing to do (not re-counted).
|
|
if self._is_same_file(file_obj.file_path, target_file):
|
|
continue
|
|
# A wrong occupant sits on the planned path; the plan is
|
|
# authoritative, so replace it.
|
|
target_file.unlink()
|
|
|
|
os.link(file_obj.file_path, target_file)
|
|
|
|
self.created_links.append(target_file)
|
|
success_count += 1
|
|
|
|
except OSError as e:
|
|
self.errors.append((file_obj.file_path, str(e)))
|
|
fail_count += 1
|
|
|
|
return success_count, fail_count
|
|
|
|
def mirror_as_is(
|
|
self, source_dir: Path, subfolder: str | None = None, dry_run: bool = False
|
|
) -> tuple[int, int]:
|
|
"""Mirror a "copy-as-is" folder 1:1 into the output as a hardlinked clone.
|
|
|
|
Recreates the exact directory hierarchy of ``source_dir`` under
|
|
``output_dir/subfolder`` (or directly under ``output_dir`` when
|
|
``subfolder`` is None) and hardlinks every file. Curator metadata files
|
|
(``.!tag`` / ``.!ftag`` / ``.!gtag`` / ``.!index``) are skipped.
|
|
|
|
Used for Seriály: the pool structure is the source of truth and is cloned
|
|
verbatim instead of being rebuilt from tags.
|
|
|
|
Returns:
|
|
Tuple of (successful_links, failed_links)
|
|
"""
|
|
source_dir = Path(source_dir)
|
|
if not source_dir.is_dir():
|
|
return 0, 0
|
|
|
|
base = self.output_dir / subfolder if subfolder else self.output_dir
|
|
success_count = 0
|
|
fail_count = 0
|
|
|
|
for src_file in source_dir.rglob("*"):
|
|
if not src_file.is_file():
|
|
continue
|
|
if src_file.name.endswith((".!tag", ".!ftag", ".!gtag", ".!index")):
|
|
continue
|
|
|
|
target_file = base / src_file.relative_to(source_dir)
|
|
try:
|
|
if not dry_run:
|
|
target_file.parent.mkdir(parents=True, exist_ok=True)
|
|
if target_file.exists():
|
|
if self._is_same_file(src_file, target_file):
|
|
success_count += 1
|
|
continue
|
|
target_file.unlink()
|
|
os.link(src_file, target_file)
|
|
self.created_links.append(target_file)
|
|
success_count += 1
|
|
except OSError as e:
|
|
self.errors.append((src_file, str(e)))
|
|
fail_count += 1
|
|
|
|
return success_count, fail_count
|
|
|
|
def remove_mirror(self, name: str) -> bool:
|
|
"""Remove ``output_dir/name`` entirely (file or folder tree).
|
|
|
|
Used to drop a whole mirrored library (or one entry of the shared
|
|
mixed-library-series bucket) when it falls out of the miniDLNA
|
|
generation's scope. Mirrors are pure hardlinks, so deleting the folder
|
|
loses nothing — it's fully regenerable from the pool. ``name`` may
|
|
contain "/" to address one entry inside a bucket folder.
|
|
|
|
Returns whether anything was actually removed.
|
|
"""
|
|
target = self.output_dir / name
|
|
if not target.exists():
|
|
return False
|
|
if target.is_dir():
|
|
shutil.rmtree(target)
|
|
else:
|
|
target.unlink()
|
|
return True
|
|
|
|
def _fill_special_folder(
|
|
self, chosen: list[File], subfolder: str, dry_run: bool
|
|
) -> tuple[int, int]:
|
|
"""Empty ``subfolder`` and hardlink ``chosen`` files into it."""
|
|
base = self.output_dir / subfolder
|
|
|
|
if not dry_run and base.exists():
|
|
for child in base.iterdir():
|
|
if child.is_file() or child.is_symlink():
|
|
try:
|
|
child.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
created = 0
|
|
fail = 0
|
|
for file_obj in chosen:
|
|
target = base / self._default_link_name(file_obj)
|
|
try:
|
|
if not dry_run:
|
|
base.mkdir(parents=True, exist_ok=True)
|
|
if target.exists():
|
|
if self._is_same_file(file_obj.file_path, target):
|
|
created += 1
|
|
continue
|
|
# Collision with another movie: prefer the year-qualified
|
|
# name, else a numeric suffix.
|
|
year_name = self._year_link_name(file_obj)
|
|
if year_name and not (base / year_name).exists():
|
|
target = base / year_name
|
|
else:
|
|
target = self._get_unique_name(target)
|
|
os.link(file_obj.file_path, target)
|
|
self.created_links.append(target)
|
|
created += 1
|
|
except OSError as e:
|
|
self.errors.append((file_obj.file_path, str(e)))
|
|
fail += 1
|
|
return created, fail
|
|
|
|
def generate_recently_added(
|
|
self,
|
|
files: list[File],
|
|
count: int = 10,
|
|
subfolder: str = "- Nově přidané",
|
|
dry_run: bool = False,
|
|
) -> tuple[int, int]:
|
|
"""(Re)fill ``subfolder`` with the ``count`` most recently added movies.
|
|
|
|
Ordering is by ``File.added_timestamp()`` (import time, ctime fallback),
|
|
newest first. Returns (created_links, failed_links).
|
|
"""
|
|
chosen = sorted(files, key=lambda f: f.added_timestamp(), reverse=True)[:count]
|
|
return self._fill_special_folder(chosen, subfolder, dry_run)
|
|
|
|
def generate_random_tips(
|
|
self,
|
|
files: list[File],
|
|
count: int = 10,
|
|
subfolder: str = "Tipy dne",
|
|
dry_run: bool = False,
|
|
) -> tuple[int, int]:
|
|
"""(Re)fill ``subfolder`` with ``count`` random files as hardlinks.
|
|
|
|
The folder is emptied first so each run picks a fresh random selection.
|
|
``files`` should be the pool movies only (copy-as-is mirrors excluded).
|
|
|
|
Returns:
|
|
Tuple of (created_links, failed_links)
|
|
"""
|
|
chosen = random.sample(files, min(count, len(files)))
|
|
return self._fill_special_folder(chosen, subfolder, dry_run)
|
|
|
|
def _is_same_file(self, path1: Path, path2: Path) -> bool:
|
|
"""Check if two paths point to the same file (same inode)."""
|
|
try:
|
|
return path1.stat().st_ino == path2.stat().st_ino
|
|
except OSError:
|
|
return False
|
|
|
|
def _get_unique_name(self, path: Path) -> Path:
|
|
"""Get a unique filename by adding a numeric suffix."""
|
|
stem = path.stem
|
|
suffix = path.suffix
|
|
parent = path.parent
|
|
counter = 1
|
|
|
|
while True:
|
|
new_name = f"{stem}_{counter}{suffix}"
|
|
new_path = parent / new_name
|
|
if not new_path.exists():
|
|
return new_path
|
|
counter += 1
|
|
|
|
def remove_created_links(self) -> int:
|
|
"""
|
|
Remove all hardlinks created by the last operation.
|
|
|
|
Returns:
|
|
Number of links removed
|
|
"""
|
|
removed = 0
|
|
for link_path in self.created_links:
|
|
try:
|
|
if link_path.exists() and link_path.is_file():
|
|
link_path.unlink()
|
|
removed += 1
|
|
|
|
# Try to remove empty parent directories
|
|
self._remove_empty_parents(link_path.parent)
|
|
except OSError:
|
|
pass
|
|
|
|
self.created_links = []
|
|
return removed
|
|
|
|
def _remove_empty_parents(self, path: Path) -> None:
|
|
"""Remove empty parent directories up to output_dir."""
|
|
try:
|
|
while path != self.output_dir and path.is_dir():
|
|
if any(path.iterdir()):
|
|
break # Directory not empty
|
|
path.rmdir()
|
|
path = path.parent
|
|
except OSError:
|
|
pass
|
|
|
|
def get_preview(
|
|
self,
|
|
files: list[File],
|
|
categories: list[str] | None = None,
|
|
category_roots: dict[str, str] | None = None,
|
|
category_transforms: dict[str, str] | None = None,
|
|
category_filename_templates: dict[str, str] | None = None,
|
|
) -> list[tuple[Path, Path]]:
|
|
"""
|
|
Get a preview of what links would be created.
|
|
|
|
Args:
|
|
files: List of File objects
|
|
categories: Optional list of categories to include
|
|
category_roots: Optional category → root-folder map (overrides
|
|
``categories`` when given).
|
|
category_transforms: Optional category → transform-name map for folders.
|
|
category_filename_templates: Optional category → hardlink-name template.
|
|
|
|
Returns:
|
|
List of tuples (source_path, target_path)
|
|
"""
|
|
roots = self._resolve_roots(categories, category_roots)
|
|
plan = self._plan_links(
|
|
files, roots, category_transforms, category_filename_templates
|
|
)
|
|
return [(file_obj.file_path, target_file) for file_obj, target_file in plan]
|
|
|
|
def find_obsolete_links(
|
|
self,
|
|
files: list[File],
|
|
categories: list[str] | None = None,
|
|
category_roots: dict[str, str] | None = None,
|
|
category_transforms: dict[str, str] | None = None,
|
|
category_filename_templates: dict[str, str] | None = None,
|
|
reserved_subfolders: set[str] | None = None,
|
|
) -> list[tuple[Path, Path]]:
|
|
"""
|
|
Find hardlinks in the output directory that no longer match file tags.
|
|
|
|
Scans the managed parts of the output directory for hardlinks that point
|
|
to source files but whose path no longer matches the file's current tags.
|
|
Only the tag-tree's own top-level folders are scanned, so copy-as-is
|
|
mirrors (e.g. Seriály) are left untouched.
|
|
|
|
Args:
|
|
files: List of File objects (source files)
|
|
categories: Optional list of categories to check (None = all)
|
|
category_roots: Optional category → root-folder map (overrides
|
|
``categories`` when given).
|
|
category_transforms: Optional category → transform-name map for folders.
|
|
|
|
Returns:
|
|
List of tuples (link_path, source_path) for obsolete links
|
|
"""
|
|
obsolete: list[tuple[Path, Path]] = []
|
|
|
|
if not self.output_dir.exists():
|
|
return obsolete
|
|
|
|
roots = self._resolve_roots(categories, category_roots)
|
|
|
|
# Build a map of source file inodes to File objects
|
|
inode_to_file: dict[int, File] = {}
|
|
for file_obj in files:
|
|
try:
|
|
inode = file_obj.file_path.stat().st_ino
|
|
inode_to_file[inode] = file_obj
|
|
except OSError:
|
|
continue
|
|
|
|
# Every link the tag tree *should* contain, mapped to the inode(s) it may
|
|
# point at — taken from the very same plan that creates the links, so the
|
|
# two never disagree on a name. Matching by inode too — not just the path
|
|
# — is what catches an **orphan that squats the expected name**: a link
|
|
# whose name is right but which points to an old inode (movie
|
|
# re-imported/replaced) is still obsolete, while the correctly created
|
|
# link is *not* wrongly swept.
|
|
expected: dict[Path, set[int]] = {}
|
|
for file_obj, path in self._plan_links(
|
|
files, roots, category_transforms, category_filename_templates
|
|
):
|
|
try:
|
|
expected.setdefault(path, set()).add(file_obj.file_path.stat().st_ino)
|
|
except OSError:
|
|
continue
|
|
|
|
# Scan only the tag-tree's own top-level folders (skip copy-as-is mirrors).
|
|
# Inside them, any file that isn't an expected link is obsolete — whether
|
|
# a tag was removed, the movie was replaced (orphan inode), or the folder
|
|
# itself is a leftover from a renamed layout.
|
|
top_dirs = self._managed_top_dirs(
|
|
files, roots, category_transforms, reserved_subfolders
|
|
)
|
|
for top in self.output_dir.iterdir():
|
|
if not top.is_dir():
|
|
continue
|
|
if top_dirs is not None and top.name not in top_dirs:
|
|
continue
|
|
|
|
for link_file in top.rglob("*"):
|
|
if not link_file.is_file():
|
|
continue
|
|
try:
|
|
link_inode = link_file.stat().st_ino
|
|
except OSError:
|
|
continue
|
|
# Valid only when both the path *and* the inode match a current
|
|
# movie; a name-only match (orphan squatting the name) is stale.
|
|
if link_inode in expected.get(link_file, ()):
|
|
continue
|
|
source = inode_to_file.get(link_inode)
|
|
obsolete.append((link_file, source.file_path if source else link_file))
|
|
|
|
return obsolete
|
|
|
|
def remove_obsolete_links(
|
|
self,
|
|
files: list[File],
|
|
categories: list[str] | None = None,
|
|
dry_run: bool = False,
|
|
category_roots: dict[str, str] | None = None,
|
|
category_transforms: dict[str, str] | None = None,
|
|
category_filename_templates: dict[str, str] | None = None,
|
|
reserved_subfolders: set[str] | None = None,
|
|
) -> tuple[int, list[Path]]:
|
|
"""
|
|
Remove hardlinks that no longer match file tags.
|
|
|
|
Args:
|
|
files: List of File objects
|
|
categories: Optional list of categories to check
|
|
dry_run: If True, only return what would be removed
|
|
category_roots: Optional category → root-folder map (overrides
|
|
``categories`` when given).
|
|
category_transforms: Optional category → transform-name map for folders.
|
|
category_filename_templates: Optional category → hardlink-name template.
|
|
reserved_subfolders: Output top-level folders to never touch (mirrors,
|
|
"Tipy dne", …).
|
|
|
|
Returns:
|
|
Tuple of (removed_count, list_of_removed_paths)
|
|
"""
|
|
obsolete = self.find_obsolete_links(
|
|
files,
|
|
categories,
|
|
category_roots,
|
|
category_transforms,
|
|
category_filename_templates,
|
|
reserved_subfolders,
|
|
)
|
|
removed_paths = []
|
|
|
|
if dry_run:
|
|
return len(obsolete), [link for link, _ in obsolete]
|
|
|
|
for link_path, _ in obsolete:
|
|
try:
|
|
link_path.unlink()
|
|
removed_paths.append(link_path)
|
|
|
|
# Try to remove empty parent directories
|
|
self._remove_empty_parents(link_path.parent)
|
|
except OSError:
|
|
pass
|
|
|
|
return len(removed_paths), removed_paths
|
|
|
|
def sync_structure(
|
|
self,
|
|
files: list[File],
|
|
categories: list[str] | None = None,
|
|
dry_run: bool = False,
|
|
category_roots: dict[str, str] | None = None,
|
|
category_transforms: dict[str, str] | None = None,
|
|
category_filename_templates: dict[str, str] | None = None,
|
|
reserved_subfolders: set[str] | None = None,
|
|
) -> tuple[int, int, int, int]:
|
|
"""
|
|
Synchronize hardlink structure with current file tags.
|
|
|
|
This will:
|
|
1. Remove hardlinks for removed tags
|
|
2. Create new hardlinks for new tags
|
|
|
|
Args:
|
|
files: List of File objects
|
|
categories: Optional list of categories to sync
|
|
dry_run: If True, only simulate
|
|
category_roots: Optional category → root-folder map (overrides
|
|
``categories`` when given).
|
|
category_transforms: Optional category → transform-name map for folders.
|
|
category_filename_templates: Optional category → hardlink-name template.
|
|
reserved_subfolders: Output top-level folders to never touch (mirrors,
|
|
"Tipy dne", …).
|
|
|
|
Returns:
|
|
Tuple of (created, create_failed, removed, remove_failed)
|
|
"""
|
|
# First find how many obsolete links there are
|
|
obsolete_count = len(
|
|
self.find_obsolete_links(
|
|
files,
|
|
categories,
|
|
category_roots,
|
|
category_transforms,
|
|
category_filename_templates,
|
|
reserved_subfolders,
|
|
)
|
|
)
|
|
|
|
# Remove obsolete links
|
|
removed, _removed_paths = self.remove_obsolete_links(
|
|
files,
|
|
categories,
|
|
dry_run,
|
|
category_roots,
|
|
category_transforms,
|
|
category_filename_templates,
|
|
reserved_subfolders,
|
|
)
|
|
remove_failed = obsolete_count - removed if not dry_run else 0
|
|
|
|
# Then create new links
|
|
created, create_failed = self.create_structure_for_files(
|
|
files,
|
|
categories,
|
|
dry_run,
|
|
category_roots,
|
|
category_transforms,
|
|
category_filename_templates,
|
|
)
|
|
|
|
return created, create_failed, removed, remove_failed
|
|
|
|
|
|
def create_hardlink_structure(
|
|
files: list[File], output_dir: Path, categories: list[str] | None = None
|
|
) -> tuple[int, int, list[tuple[Path, str]]]:
|
|
"""
|
|
Convenience function to create hardlink structure.
|
|
|
|
Args:
|
|
files: List of File objects to process
|
|
output_dir: Base directory for output
|
|
categories: Optional list of categories to include
|
|
|
|
Returns:
|
|
Tuple of (successful_count, failed_count, errors_list)
|
|
"""
|
|
manager = HardlinkManager(output_dir)
|
|
success, fail = manager.create_structure_for_files(files, categories)
|
|
return success, fail, manager.errors
|