634 lines
24 KiB
Python
634 lines
24 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
|
|
from pathlib import Path
|
|
from typing import List, Tuple, Optional, Dict, Set
|
|
from .file import File
|
|
|
|
|
|
class _SafeDict(dict):
|
|
"""dict for str.format_map that leaves unknown fields as an empty string."""
|
|
|
|
def __missing__(self, key): # noqa: ANN001
|
|
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: Optional[List[str]],
|
|
category_roots: Optional[Dict[str, str]],
|
|
) -> Optional[Dict[str, str]]:
|
|
"""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 _link_name(
|
|
self, file_obj: File, tag, templates: Optional[Dict[str, str]]
|
|
) -> str:
|
|
"""Hardlink filename for a tag — a per-category template or the pool name.
|
|
|
|
Applies ``templates[tag.category]`` (e.g. ``"{year} - {title}{ext}"``) to
|
|
the file's ``name_context``; path separators are flattened. Any failure or
|
|
empty result falls back to the pool filename.
|
|
"""
|
|
template = templates.get(tag.category) if templates else None
|
|
if not template:
|
|
return file_obj.filename
|
|
try:
|
|
rendered = template.format_map(_SafeDict(file_obj.name_context()))
|
|
except (ValueError, KeyError, IndexError, AttributeError):
|
|
return file_obj.filename
|
|
rendered = rendered.replace("/", "-").replace("\\", "-").strip()
|
|
return rendered or file_obj.filename
|
|
|
|
def _folder_value(self, tag, transforms: Optional[Dict[str, str]]) -> 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 .csfd import apply_transform
|
|
return apply_transform(tag.name, transforms[tag.category])
|
|
return tag.name
|
|
|
|
def _target_dir(
|
|
self, tag, roots: Optional[Dict[str, str]],
|
|
transforms: Optional[Dict[str, str]] = None,
|
|
) -> Optional[Path]:
|
|
"""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: Optional[Dict[str, str]],
|
|
transforms: Optional[Dict[str, str]] = None,
|
|
reserved_subfolders: Optional[Set[str]] = None,
|
|
) -> Optional[Set[str]]:
|
|
"""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: Optional[List[str]] = None,
|
|
dry_run: bool = False,
|
|
category_roots: Optional[Dict[str, str]] = None,
|
|
category_transforms: Optional[Dict[str, str]] = None,
|
|
category_filename_templates: Optional[Dict[str, str]] = 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
|
|
|
|
for file_obj in files:
|
|
if not file_obj.tags:
|
|
continue
|
|
|
|
for tag in file_obj.tags:
|
|
# Resolve the target dir; None means this category is excluded
|
|
target_dir = self._target_dir(tag, roots, category_transforms)
|
|
if target_dir is None:
|
|
continue
|
|
target_file = target_dir / self._link_name(
|
|
file_obj, tag, category_filename_templates)
|
|
|
|
try:
|
|
if not dry_run:
|
|
# Create directory structure
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Skip if link already exists
|
|
if target_file.exists():
|
|
# Check if it's already a hardlink to the same file
|
|
if self._is_same_file(file_obj.file_path, target_file):
|
|
continue
|
|
else:
|
|
# Different file exists, add suffix
|
|
target_file = self._get_unique_name(target_file)
|
|
|
|
# Create hardlink
|
|
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 _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 / file_obj.filename
|
|
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
|
|
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: Optional[List[str]] = None,
|
|
category_roots: Optional[Dict[str, str]] = None,
|
|
category_transforms: Optional[Dict[str, str]] = None,
|
|
category_filename_templates: Optional[Dict[str, str]] = 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)
|
|
preview = []
|
|
|
|
for file_obj in files:
|
|
if not file_obj.tags:
|
|
continue
|
|
|
|
for tag in file_obj.tags:
|
|
target_dir = self._target_dir(tag, roots, category_transforms)
|
|
if target_dir is None:
|
|
continue
|
|
target_file = target_dir / self._link_name(
|
|
file_obj, tag, category_filename_templates)
|
|
|
|
preview.append((file_obj.file_path, target_file))
|
|
|
|
return preview
|
|
|
|
def find_obsolete_links(
|
|
self,
|
|
files: List[File],
|
|
categories: Optional[List[str]] = None,
|
|
category_roots: Optional[Dict[str, str]] = None,
|
|
category_transforms: Optional[Dict[str, str]] = None,
|
|
category_filename_templates: Optional[Dict[str, str]] = None,
|
|
reserved_subfolders: Optional[Set[str]] = 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, given current files+tags.
|
|
expected_all: set[Path] = set()
|
|
for file_obj in files:
|
|
for tag in file_obj.tags:
|
|
target_dir = self._target_dir(tag, roots, category_transforms)
|
|
if target_dir is None:
|
|
continue
|
|
expected_all.add(target_dir / self._link_name(
|
|
file_obj, tag, category_filename_templates))
|
|
|
|
# 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
|
|
if link_file in expected_all:
|
|
continue
|
|
try:
|
|
source = inode_to_file[link_file.stat().st_ino].file_path
|
|
except (OSError, KeyError):
|
|
source = link_file # orphan (no current pool movie)
|
|
obsolete.append((link_file, source))
|
|
|
|
return obsolete
|
|
|
|
def remove_obsolete_links(
|
|
self,
|
|
files: List[File],
|
|
categories: Optional[List[str]] = None,
|
|
dry_run: bool = False,
|
|
category_roots: Optional[Dict[str, str]] = None,
|
|
category_transforms: Optional[Dict[str, str]] = None,
|
|
category_filename_templates: Optional[Dict[str, str]] = None,
|
|
reserved_subfolders: Optional[Set[str]] = 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: Optional[List[str]] = None,
|
|
dry_run: bool = False,
|
|
category_roots: Optional[Dict[str, str]] = None,
|
|
category_transforms: Optional[Dict[str, str]] = None,
|
|
category_filename_templates: Optional[Dict[str, str]] = None,
|
|
reserved_subfolders: Optional[Set[str]] = 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: Optional[List[str]] = 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
|