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

Add "Tipy dne" folder and fix stale genre folder cleanup

Browse Source
This commit is contained in:
Honza
2026-06-29 15:33:44 +02:00
parent a71b209539
commit ee99fc957d
7 changed files with 165 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/hardlink_manager.py
+83 -12
View File
@@ -18,6 +18,7 @@ Example:
└── film.mkv (hardlink)
"""
import os
import random
from pathlib import Path
from typing import List, Tuple, Optional, Dict, Set
from .file import File
@@ -115,25 +116,40 @@ class HardlinkManager:
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).
For a category with a non-empty root the root folder is owned; for a
category placed at the output root (empty root, e.g. genres) each of its
(transformed) tag values is its own top-level folder. This lets cleanup
skip unrelated root entries such as the copy-as-is mirror (Seriály).
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
tops: Set[str] = set()
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 folder:
tops.add(folder)
else:
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(
@@ -258,6 +274,53 @@ class HardlinkManager:
return success_count, fail_count
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)
"""
base = self.output_dir / subfolder
if not dry_run and base.exists():
for child in base.iterdir():
if child.is_file():
try:
child.unlink()
except OSError:
pass
chosen = random.sample(files, min(count, len(files)))
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 _is_same_file(self, path1: Path, path2: Path) -> bool:
"""Check if two paths point to the same file (same inode)."""
try:
@@ -359,6 +422,7 @@ class HardlinkManager:
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.
@@ -411,7 +475,8 @@ class HardlinkManager:
continue
# Scan only the tag-tree's own top-level folders (skip copy-as-is mirrors)
top_dirs = self._managed_top_dirs(files, roots, category_transforms)
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
@@ -441,6 +506,7 @@ class HardlinkManager:
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.
@@ -453,13 +519,15 @@ class HardlinkManager:
``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)
category_filename_templates, reserved_subfolders)
removed_paths = []
if dry_run:
@@ -485,6 +553,7 @@ class HardlinkManager:
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.
@@ -501,6 +570,8 @@ class HardlinkManager:
``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)
@@ -508,12 +579,12 @@ class HardlinkManager:
# First find how many obsolete links there are
obsolete_count = len(self.find_obsolete_links(
files, categories, category_roots, category_transforms,
category_filename_templates))
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
category_filename_templates, reserved_subfolders
)
remove_failed = obsolete_count - removed if not dry_run else 0