Files
HS-Rename/engine/pipeline.py
T
2026-03-03 21:58:28 -06:00

107 lines
3.2 KiB
Python

"""
Apply a list of rules to a list of filenames; produce preview and perform renames.
"""
import os
from pathlib import Path
from typing import List, Optional
from .rules import Rule
def _split_name(name: str) -> tuple[str, str]:
if "." in name and not name.startswith("."):
idx = name.rfind(".")
return name[:idx], name[idx:]
return name, ""
def apply_pipeline(
name: str,
rules: List[Rule],
index: int,
total: int,
) -> str:
"""Apply all enabled rules to a single filename (no path). Returns new filename."""
stem, ext = _split_name(name)
for r in rules:
if not r.enabled:
continue
stem, ext = r.apply(stem, ext, index, total)
return stem + ext
def compute_preview(
names: List[str],
rules: List[Rule],
) -> List[tuple[str, str]]:
"""
For each name (filename only), compute the new name after rules.
Returns list of (original_name, new_name). Detects collisions.
"""
result = []
for i, name in enumerate(names):
new_name = apply_pipeline(name, rules, i, len(names))
result.append((name, new_name))
return result
def perform_renames(
base_dir: str,
renames: List[tuple[str, str]],
dry_run: bool = False,
) -> List[tuple[str, str, Optional[str]]]:
"""
Perform renames in base_dir. Each item is (old_name, new_name).
Returns list of (old_path, new_path, error_message or None).
Uses two-pass: first rename to temp names to avoid collisions.
"""
base = Path(base_dir)
results = []
# Two-pass to avoid overwriting: first move to temporary names, then to final
temp_suffix = ".bru_tmp"
step1 = []
for old_name, new_name in renames:
if old_name == new_name:
continue
old_path = base / old_name
if not old_path.exists():
results.append((str(old_path), "", "File not found"))
continue
step1.append((old_path, new_name))
# Assign temp names that don't clash with each other or final names
final_names = {n for _, n in step1}
temp_map = []
for i, (old_path, new_name) in enumerate(step1):
temp_name = f"__temp_{i}_{old_path.name}{temp_suffix}"
while temp_name in final_names or (base / temp_name).exists():
i += 1
temp_name = f"__temp_{i}_{old_path.name}{temp_suffix}"
temp_map.append((old_path, base / temp_name, base / new_name))
if dry_run:
for old_p, temp_p, new_p in temp_map:
results.append((str(old_p), str(new_p), None))
return results
# Execute: first to temp, then to final
for old_p, temp_p, new_p in temp_map:
try:
old_p.rename(temp_p)
except OSError as e:
results.append((str(old_p), str(new_p), str(e)))
for old_p, temp_p, new_p in temp_map:
if not temp_p.exists():
continue
try:
temp_p.rename(new_p)
results.append((str(old_p), str(new_p), None))
except OSError as e:
results.append((str(old_p), str(new_p), str(e)))
try:
temp_p.rename(old_p)
except OSError:
pass
return results