Honza/Curator
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Migrate from Poetry and pyenv to uv

Browse Source
This commit is contained in:
Honza
2026-10-12 15:48:26 +02:00
parent c399a6aed3
commit c69f5f9166
93 changed files with 13068 additions and 2401 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/hardlink_manager.py
+128 -95
View File
@@ -17,12 +17,13 @@ Example:
└── 1988/
└── film.mkv (hardlink)
"""
import os
import random
import shutil
from collections import defaultdict
from pathlib import Path
from typing import List, Tuple, Optional, Dict, Set
from .file import File
from .naming import is_valid_year, parse_pool_stem
@@ -30,7 +31,7 @@ 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): # noqa: ANN001
def __missing__(self, key):
return ""
@@ -54,14 +55,14 @@ class HardlinkManager:
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]] = []
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]]:
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.
@@ -86,7 +87,7 @@ class HardlinkManager:
"""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) -> Optional[str]:
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):
@@ -97,18 +98,18 @@ class HardlinkManager:
"""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):
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: Optional[Dict[str, str]],
transforms: Optional[Dict[str, str]],
templates: Optional[Dict[str, str]],
) -> List[Tuple[File, Path]]:
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
@@ -122,7 +123,7 @@ class HardlinkManager:
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] = []
entries: list[list] = []
seen_fd: set[tuple[str, str]] = set()
for file_obj in files:
for tag in file_obj.tags:
@@ -136,10 +137,17 @@ class HardlinkManager:
template = templates.get(tag.category) if templates else None
if template:
entries.append(
[file_obj, target_dir, self._render_template(file_obj, template), True])
[
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])
[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)
@@ -154,10 +162,10 @@ class HardlinkManager:
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)
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]] = []
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))
@@ -169,7 +177,7 @@ class HardlinkManager:
plan.append((entry[0], target_dir / fname))
return plan
def _folder_value(self, tag, transforms: Optional[Dict[str, str]]) -> str:
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
@@ -177,13 +185,16 @@ class HardlinkManager:
"""
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: Optional[Dict[str, str]],
transforms: Optional[Dict[str, str]] = None,
) -> Optional[Path]:
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
@@ -195,10 +206,12 @@ class HardlinkManager:
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]]:
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
@@ -213,7 +226,7 @@ class HardlinkManager:
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)
tops: set[str] = set(grouping_roots)
# genre folders currently present on the movies
for cat, folder in roots.items():
@@ -228,20 +241,23 @@ class HardlinkManager:
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):
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,
files: list[File],
categories: list[str] | None = 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]:
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.
@@ -267,7 +283,8 @@ class HardlinkManager:
fail_count = 0
plan = self._plan_links(
files, roots, category_transforms, category_filename_templates)
files, roots, category_transforms, category_filename_templates
)
for file_obj, target_file in plan:
try:
if not dry_run:
@@ -292,11 +309,8 @@ class HardlinkManager:
return success_count, fail_count
def mirror_as_is(
self,
source_dir: Path,
subfolder: str | None = None,
dry_run: bool = False
) -> Tuple[int, int]:
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
@@ -363,8 +377,8 @@ class HardlinkManager:
return True
def _fill_special_folder(
self, chosen: List[File], subfolder: str, dry_run: bool
) -> Tuple[int, int]:
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
@@ -404,11 +418,11 @@ class HardlinkManager:
def generate_recently_added(
self,
files: List[File],
files: list[File],
count: int = 10,
subfolder: str = "- Nově přidané",
dry_run: bool = False,
) -> Tuple[int, int]:
) -> tuple[int, int]:
"""(Re)fill ``subfolder`` with the ``count`` most recently added movies.
Ordering is by ``File.added_timestamp()`` (import time, ctime fallback),
@@ -419,11 +433,11 @@ class HardlinkManager:
def generate_random_tips(
self,
files: List[File],
files: list[File],
count: int = 10,
subfolder: str = "Tipy dne",
dry_run: bool = False,
) -> Tuple[int, int]:
) -> 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.
@@ -491,12 +505,12 @@ class HardlinkManager:
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]]:
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.
@@ -513,18 +527,19 @@ class HardlinkManager:
"""
roots = self._resolve_roots(categories, category_roots)
plan = self._plan_links(
files, roots, category_transforms, category_filename_templates)
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: 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]]:
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.
@@ -543,7 +558,7 @@ class HardlinkManager:
Returns:
List of tuples (link_path, source_path) for obsolete links
"""
obsolete: List[Tuple[Path, Path]] = []
obsolete: list[tuple[Path, Path]] = []
if not self.output_dir.exists():
return obsolete
@@ -580,7 +595,8 @@ class HardlinkManager:
# 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)
files, roots, category_transforms, reserved_subfolders
)
for top in self.output_dir.iterdir():
if not top.is_dir():
continue
@@ -599,22 +615,20 @@ class HardlinkManager:
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)
)
obsolete.append((link_file, source.file_path if source else link_file))
return obsolete
def remove_obsolete_links(
self,
files: List[File],
categories: Optional[List[str]] = None,
files: list[File],
categories: list[str] | None = 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]]:
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.
@@ -633,8 +647,13 @@ class HardlinkManager:
Tuple of (removed_count, list_of_removed_paths)
"""
obsolete = self.find_obsolete_links(
files, categories, category_roots, category_transforms,
category_filename_templates, reserved_subfolders)
files,
categories,
category_roots,
category_transforms,
category_filename_templates,
reserved_subfolders,
)
removed_paths = []
if dry_run:
@@ -654,14 +673,14 @@ class HardlinkManager:
def sync_structure(
self,
files: List[File],
categories: Optional[List[str]] = None,
files: list[File],
categories: list[str] | None = 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]:
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.
@@ -684,31 +703,45 @@ class HardlinkManager:
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))
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
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
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]]]:
files: list[File], output_dir: Path, categories: list[str] | None = None
) -> tuple[int, int, list[tuple[Path, str]]]:
"""
Convenience function to create hardlink structure.