Lukas
5f46d2e802
Runs after NFO refresh during scheduled rescans. Renames folders that are missing a year (e.g. 'Naruto' → 'Naruto (1999)') using the year from the database record. Safety: _build_target_folder() always strips any existing year suffix first, preventing double/triple year accumulation like 'Naruto (1999) (1999) (1999)'. Changes: - New FolderNamingService (folder_naming_service.py) with safe target name construction, DB update, and in-memory cache update - New SchedulerConfig field: folder_naming_after_nfo_scan (default True) - Integrated as step 3 in scheduler _perform_rescan() after NFO scan - Runtime UI: existing 'folder-scan-enabled' checkbox in index.html now wired to toggle the feature (app.js + scheduler-config.js) - Setup screen: new checkbox in setup.html Scheduler Settings section - API: scheduler config endpoint returns all scan toggles - Tests: 39 unit tests covering static helpers, rename logic, safety guard, and integration cases (folder_naming_service.py) - Docs: testing guide updated with FolderNamingService examples
206 lines
7.6 KiB
Python
206 lines
7.6 KiB
Python
"""Folder naming service for fixing missing years in anime folder names."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import re
|
|
import shutil
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import structlog
|
|
|
|
from src.config.settings import settings
|
|
from src.server.database.connection import get_db_session as _get_db_session
|
|
from src.server.database.service import AnimeSeriesService
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class FolderRenameResult:
|
|
key: str
|
|
old_folder: str
|
|
new_folder: Optional[str]
|
|
success: bool
|
|
skipped: bool = False
|
|
reason: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class FolderNamingReport:
|
|
total: int
|
|
renamed: int
|
|
skipped: int
|
|
errors: int
|
|
results: List[FolderRenameResult]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"total": self.total,
|
|
"renamed": self.renamed,
|
|
"skipped": self.skipped,
|
|
"errors": self.errors,
|
|
"results": [
|
|
{
|
|
"key": r.key,
|
|
"old_folder": r.old_folder,
|
|
"new_folder": r.new_folder,
|
|
"success": r.success,
|
|
"skipped": r.skipped,
|
|
"reason": r.reason,
|
|
}
|
|
for r in self.results
|
|
],
|
|
}
|
|
|
|
|
|
class FolderNamingService:
|
|
def __init__(self) -> None:
|
|
self._is_running = False
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def run(self) -> FolderNamingReport:
|
|
async with self._lock:
|
|
if self._is_running:
|
|
logger.warning("FolderNamingService.run() called while already running")
|
|
return FolderNamingReport(total=0, renamed=0, skipped=0, errors=0, results=[])
|
|
self._is_running = True
|
|
|
|
try:
|
|
logger.info("FolderNamingService: starting folder naming scan")
|
|
results: List[FolderRenameResult] = []
|
|
|
|
async with _get_db_session() as db:
|
|
all_series = await AnimeSeriesService.get_all(db)
|
|
|
|
for series in all_series:
|
|
result = await self._process_series(series)
|
|
results.append(result)
|
|
|
|
renamed = sum(1 for r in results if r.success and not r.skipped)
|
|
skipped = sum(1 for r in results if r.skipped)
|
|
errors = sum(1 for r in results if not r.skipped and not r.success)
|
|
|
|
report = FolderNamingReport(
|
|
total=len(results),
|
|
renamed=renamed,
|
|
skipped=skipped,
|
|
errors=errors,
|
|
results=results,
|
|
)
|
|
logger.info(
|
|
"FolderNamingService: scan complete — total=%d renamed=%d skipped=%d errors=%d",
|
|
report.total, report.renamed, report.skipped, report.errors,
|
|
)
|
|
return report
|
|
finally:
|
|
self._is_running = False
|
|
|
|
async def _process_series(self, series) -> FolderRenameResult:
|
|
key = series.key
|
|
folder = series.folder or ""
|
|
year = getattr(series, "year", None)
|
|
|
|
if year is None:
|
|
return FolderRenameResult(key=key, old_folder=folder, new_folder=None, success=True, skipped=True, reason="no year in DB record")
|
|
|
|
if self._folder_has_year(folder):
|
|
return FolderRenameResult(key=key, old_folder=folder, new_folder=None, success=True, skipped=True, reason="folder already has year")
|
|
|
|
target_folder = self._build_target_folder(folder, year)
|
|
|
|
# Safety: re-extract year from target to prevent double-year
|
|
if target_folder != folder:
|
|
extracted = self._extract_year_from_folder_name(target_folder)
|
|
if extracted != year:
|
|
logger.error("Safety guard for %s: target '%s' year=%s != DB year=%s — skipping", key, target_folder, extracted, year)
|
|
return FolderRenameResult(key=key, old_folder=folder, new_folder=None, success=False, skipped=True, reason="safety guard: target year mismatch")
|
|
|
|
return await self._execute_rename(series, folder, target_folder)
|
|
|
|
async def _execute_rename(self, series, old_folder: str, target_folder: str) -> FolderRenameResult:
|
|
key = series.key
|
|
|
|
if old_folder == target_folder:
|
|
return FolderRenameResult(key=key, old_folder=old_folder, new_folder=None, success=True, skipped=True, reason="same folder name")
|
|
|
|
anime_dir = settings.anime_directory
|
|
old_path = os.path.join(anime_dir, old_folder)
|
|
target_path = os.path.join(anime_dir, target_folder)
|
|
|
|
if not os.path.isdir(old_path):
|
|
return FolderRenameResult(key=key, old_folder=old_folder, new_folder=None, success=False, skipped=False, reason="source folder does not exist on disk")
|
|
|
|
if os.path.isdir(target_path):
|
|
return FolderRenameResult(key=key, old_folder=old_folder, new_folder=None, success=False, skipped=False, reason="target folder already exists on disk")
|
|
|
|
try:
|
|
shutil.move(old_path, target_path)
|
|
logger.info("Renamed folder %s -> %s for series %s", old_folder, target_folder, key)
|
|
|
|
# Update in-memory cache
|
|
try:
|
|
from src.server.SeriesApp import get_series_app
|
|
series_app = get_series_app()
|
|
if key in series_app.list.keyDict:
|
|
series_app.list.keyDict[key].folder = target_folder
|
|
except Exception as exc:
|
|
logger.warning("Failed to update in-memory cache for %s: %s", key, exc)
|
|
|
|
# Update database
|
|
async with _get_db_session() as db:
|
|
db_series = await AnimeSeriesService.get_by_key(db, key)
|
|
if db_series:
|
|
await AnimeSeriesService.update(db, series_id=db_series.id, folder=target_folder)
|
|
logger.debug("Updated DB folder for %s to %s", key, target_folder)
|
|
|
|
return FolderRenameResult(key=key, old_folder=old_folder, new_folder=target_folder, success=True, skipped=False)
|
|
|
|
except Exception as exc:
|
|
logger.error("Failed to rename folder for %s (%s -> %s): %s", key, old_folder, target_folder, exc)
|
|
return FolderRenameResult(key=key, old_folder=old_folder, new_folder=None, success=False, skipped=False, reason=str(exc))
|
|
|
|
# Static helpers — public for direct testing
|
|
@staticmethod
|
|
def _folder_has_year(folder_name: str) -> bool:
|
|
if not folder_name:
|
|
return False
|
|
return bool(re.search(r"\(\d{4}\)", folder_name))
|
|
|
|
@staticmethod
|
|
def _extract_year_from_folder_name(folder_name: str) -> Optional[int]:
|
|
if not folder_name:
|
|
return None
|
|
match = re.search(r"\((\d{4})\)", folder_name)
|
|
if match:
|
|
try:
|
|
year = int(match.group(1))
|
|
if 1900 <= year <= 2100:
|
|
return year
|
|
except ValueError:
|
|
pass
|
|
return None
|
|
|
|
@staticmethod
|
|
def _extract_title_from_folder_name(folder_name: str) -> str:
|
|
return re.sub(r"\s*\(\d{4}\)\s*$", "", folder_name).strip()
|
|
|
|
@staticmethod
|
|
def _build_target_folder(folder_name: str, year: int) -> str:
|
|
title = FolderNamingService._extract_title_from_folder_name(folder_name)
|
|
return f"{title} ({year})"
|
|
|
|
|
|
_folder_naming_service: Optional[FolderNamingService] = None
|
|
|
|
def get_folder_naming_service() -> FolderNamingService:
|
|
global _folder_naming_service
|
|
if _folder_naming_service is None:
|
|
_folder_naming_service = FolderNamingService()
|
|
return _folder_naming_service
|
|
|
|
def reset_folder_naming_service() -> None:
|
|
global _folder_naming_service
|
|
_folder_naming_service = None
|