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
416 lines
16 KiB
Python
416 lines
16 KiB
Python
import ast
|
||
import json
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator
|
||
|
||
_VALID_DAYS = frozenset(["mon", "tue", "wed", "thu", "fri", "sat", "sun"])
|
||
_ALL_DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]
|
||
|
||
|
||
class SchedulerConfig(BaseModel):
|
||
"""Scheduler related configuration.
|
||
|
||
Cron-based scheduling is configured via ``schedule_time`` and
|
||
``schedule_days``. The legacy ``interval_minutes`` field is kept for
|
||
backward compatibility but is **deprecated** and ignored when
|
||
``schedule_time`` is set.
|
||
"""
|
||
|
||
enabled: bool = Field(
|
||
default=True, description="Whether the scheduler is enabled"
|
||
)
|
||
interval_minutes: int = Field(
|
||
default=60,
|
||
ge=1,
|
||
description="[Deprecated] Scheduler interval in minutes. "
|
||
"Use schedule_time + schedule_days instead.",
|
||
)
|
||
schedule_time: str = Field(
|
||
default="03:00",
|
||
description="Daily run time in 24-hour HH:MM format (e.g. '03:00')",
|
||
)
|
||
schedule_days: List[str] = Field(
|
||
default_factory=lambda: list(_ALL_DAYS),
|
||
description="Days of week to run the scheduler (3-letter lowercase "
|
||
"abbreviations: mon, tue, wed, thu, fri, sat, sun). "
|
||
"Empty list means disabled.",
|
||
)
|
||
auto_download_after_rescan: bool = Field(
|
||
default=False,
|
||
description="Automatically queue and start downloads for all missing "
|
||
"episodes after a scheduled rescan completes.",
|
||
)
|
||
nfo_scan_after_rescan: bool = Field(
|
||
default=True,
|
||
description="Run NFO validation and creation after a scheduled rescan "
|
||
"completes. Checks each series folder for tvshow.nfo and "
|
||
"creates or fills missing properties.",
|
||
)
|
||
image_scan_after_rescan: bool = Field(
|
||
default=True,
|
||
description="Download series images (poster.jpg, fanart.jpg, logo.png) "
|
||
"from TMDB after a scheduled rescan completes.",
|
||
)
|
||
folder_naming_after_nfo_scan: bool = Field(
|
||
default=True,
|
||
description="Fix missing years in folder names after NFO refresh. "
|
||
"Renames folders (e.g. 'Naruto' -> 'Naruto (1999)') using "
|
||
"the year from the database record.",
|
||
)
|
||
# Legacy alias fields — read via Pydantic alias
|
||
auto_download: Optional[bool] = Field(default=None, alias="auto_download")
|
||
|
||
def __init__(self, **data):
|
||
super().__init__(**data)
|
||
# Map legacy keys to primary fields only when primary key absent from data.
|
||
# "key in data" checks for explicit presence (even False/None), not just truthiness.
|
||
if self.auto_download is not None and "auto_download_after_rescan" not in data:
|
||
object.__setattr__(self, "auto_download_after_rescan", self.auto_download)
|
||
|
||
@field_validator("schedule_time")
|
||
@classmethod
|
||
def validate_schedule_time(cls, v: str) -> str:
|
||
"""Validate HH:MM format within 00:00–23:59."""
|
||
import re
|
||
if not re.fullmatch(r"([01]\d|2[0-3]):[0-5]\d", v or ""):
|
||
raise ValueError(
|
||
f"Invalid schedule_time '{v}'. "
|
||
"Expected HH:MM in 24-hour format (00:00–23:59)."
|
||
)
|
||
return v
|
||
|
||
@classmethod
|
||
def _parse_schedule_days(cls, v):
|
||
"""Parse schedule_days that may arrive as a malformed string.
|
||
|
||
Robot Framework's Create Dictionary converts Python-style lists
|
||
like ['monday', 'tuesday'] into strings. Handle that here before
|
||
Pydantic's type validation runs.
|
||
"""
|
||
if not isinstance(v, str):
|
||
return v
|
||
# Try JSON first (double-quoted), then Python literal (single-quoted)
|
||
for parse_fn in (json.loads, ast.literal_eval):
|
||
try:
|
||
parsed = parse_fn(v)
|
||
if isinstance(parsed, list):
|
||
return parsed
|
||
except Exception:
|
||
pass
|
||
# Cannot parse - let Pydantic handle the error
|
||
return v
|
||
|
||
@model_validator(mode="before")
|
||
@classmethod
|
||
def _pre_validate(cls, data):
|
||
"""Handle malformed schedule_days from Robot Framework before type validation."""
|
||
if isinstance(data, dict):
|
||
sd = data.get("schedule_days")
|
||
if isinstance(sd, str):
|
||
parsed = cls._parse_schedule_days(sd)
|
||
if isinstance(parsed, list):
|
||
data = dict(data)
|
||
data["schedule_days"] = parsed
|
||
return data
|
||
|
||
@field_validator("schedule_days")
|
||
@classmethod
|
||
def validate_schedule_days(cls, v: List[str]) -> List[str]:
|
||
"""Validate each entry is a valid 3-letter lowercase day abbreviation."""
|
||
invalid = [d for d in v if d not in _VALID_DAYS]
|
||
if invalid:
|
||
raise ValueError(
|
||
f"Invalid day(s) in schedule_days: {invalid}. "
|
||
f"Allowed values: {sorted(_VALID_DAYS)}"
|
||
)
|
||
return v
|
||
|
||
def model_dump(self, **kwargs) -> Dict[str, object]:
|
||
"""Serialize, excluding legacy alias fields when they are None.
|
||
|
||
The alias fields (auto_download, folder_scan) must not be written to
|
||
config.json as null entries, otherwise a roundtrip load sees the key
|
||
present (哪怕 value is None) and skips the alias-to-primary mapping.
|
||
"""
|
||
data = super().model_dump(**kwargs)
|
||
# Drop None alias fields so they don't pollute config.json.
|
||
# They are still settable via the constructor for backward compatibility.
|
||
if data.get("auto_download") is None:
|
||
data.pop("auto_download", None)
|
||
if data.get("folder_scan") is None:
|
||
data.pop("folder_scan", None)
|
||
return data
|
||
|
||
|
||
class BackupConfig(BaseModel):
|
||
"""Configuration for automatic backups of application data."""
|
||
|
||
enabled: bool = Field(
|
||
default=False, description="Whether backups are enabled"
|
||
)
|
||
path: Optional[str] = Field(
|
||
default="data/backups", description="Path to store backups"
|
||
)
|
||
keep_days: int = Field(
|
||
default=30, ge=0, description="How many days to keep backups"
|
||
)
|
||
|
||
|
||
class LoggingConfig(BaseModel):
|
||
"""Logging configuration with basic validation for level."""
|
||
|
||
level: str = Field(
|
||
default="INFO", description="Logging level"
|
||
)
|
||
file: Optional[str] = Field(
|
||
default=None, description="Optional file path for log output"
|
||
)
|
||
max_bytes: Optional[int] = Field(
|
||
default=None, description="Max bytes per log file for rotation"
|
||
)
|
||
backup_count: Optional[int] = Field(
|
||
default=3, description="Number of rotated log files to keep"
|
||
)
|
||
|
||
@field_validator("level")
|
||
@classmethod
|
||
def validate_level(cls, v: str) -> str:
|
||
allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
|
||
lvl = (v or "").upper()
|
||
if lvl not in allowed:
|
||
raise ValueError(f"invalid logging level: {v}")
|
||
return lvl
|
||
|
||
|
||
class NFOConfig(BaseModel):
|
||
"""NFO metadata configuration."""
|
||
|
||
tmdb_api_key: Optional[str] = Field(
|
||
default=None, description="TMDB API key for metadata scraping"
|
||
)
|
||
auto_create: bool = Field(
|
||
default=False, description="Auto-create NFO files for new series"
|
||
)
|
||
update_on_scan: bool = Field(
|
||
default=False, description="Update existing NFO files on rescan"
|
||
)
|
||
download_poster: bool = Field(
|
||
default=True, description="Download poster.jpg"
|
||
)
|
||
download_logo: bool = Field(
|
||
default=True, description="Download logo.png"
|
||
)
|
||
download_fanart: bool = Field(
|
||
default=True, description="Download fanart.jpg"
|
||
)
|
||
image_size: str = Field(
|
||
default="original", description="Image size (original or w500)"
|
||
)
|
||
|
||
@field_validator("image_size")
|
||
@classmethod
|
||
def validate_image_size(cls, v: str) -> str:
|
||
allowed = {"original", "w500"}
|
||
size = (v or "").lower()
|
||
if size not in allowed:
|
||
raise ValueError(
|
||
f"invalid image size: {v}. Must be 'original' or 'w500'"
|
||
)
|
||
return size
|
||
|
||
|
||
class ValidationResult(BaseModel):
|
||
"""Result of a configuration validation attempt."""
|
||
|
||
valid: bool = Field(..., description="Whether the configuration is valid")
|
||
errors: List[str] = Field(
|
||
default_factory=lambda: [],
|
||
description="List of validation error messages"
|
||
)
|
||
|
||
|
||
class AppConfig(BaseModel):
|
||
"""Top-level application configuration model used by the web layer.
|
||
|
||
This model intentionally keeps things small and serializable to JSON.
|
||
"""
|
||
|
||
name: str = Field(default="Aniworld", description="Application name")
|
||
data_dir: str = Field(default="data", description="Base data directory")
|
||
scheduler: SchedulerConfig = Field(
|
||
default_factory=SchedulerConfig
|
||
)
|
||
logging: LoggingConfig = Field(default_factory=LoggingConfig)
|
||
backup: BackupConfig = Field(default_factory=BackupConfig)
|
||
nfo: NFOConfig = Field(default_factory=NFOConfig)
|
||
scan_key_overrides: Dict[str, str] = Field(
|
||
default_factory=dict,
|
||
description="Map of folder names to provider keys for scan overrides. "
|
||
"Used when auto-generated keys from folder names are incorrect. "
|
||
"Format: {\"Folder Name\": \"actual-provider-key\"}"
|
||
)
|
||
other: Dict[str, object] = Field(
|
||
default_factory=dict, description="Arbitrary other settings"
|
||
)
|
||
|
||
def validate_config(self) -> ValidationResult:
|
||
"""Perform light-weight validation and return a ValidationResult.
|
||
|
||
This method intentionally avoids performing IO (no filesystem checks)
|
||
so it remains fast and side-effect free for unit tests and API use.
|
||
"""
|
||
errors: List[str] = []
|
||
|
||
# Pydantic field validators already run on construction; re-run a
|
||
# quick check for common constraints and collect messages.
|
||
try:
|
||
# Reconstruct to ensure nested validators are executed
|
||
AppConfig(**self.model_dump())
|
||
except ValidationError as exc:
|
||
for e in exc.errors():
|
||
loc = ".".join(str(x) for x in e.get("loc", []))
|
||
msg = f"{loc}: {e.get('msg')}"
|
||
errors.append(msg)
|
||
|
||
# backup.path must be set when backups are enabled
|
||
backup_data = self.model_dump().get("backup", {})
|
||
if backup_data.get("enabled") and not backup_data.get("path"):
|
||
errors.append(
|
||
"backup.path must be set when backups.enabled is true"
|
||
)
|
||
|
||
return ValidationResult(valid=(len(errors) == 0), errors=errors)
|
||
|
||
|
||
class ConfigUpdate(BaseModel):
|
||
name: Optional[str] = None
|
||
data_dir: Optional[str] = None
|
||
scheduler: Optional[Dict[str, Any]] = None
|
||
logging: Optional[Dict[str, Any]] = None
|
||
backup: Optional[Dict[str, Any]] = None
|
||
nfo: Optional[Dict[str, Any]] = None
|
||
scan_key_overrides: Optional[Dict[str, str]] = None
|
||
other: Optional[Dict[str, Any]] = None
|
||
|
||
@classmethod
|
||
def _parse_dict_field(cls, v):
|
||
"""Parse a field that may arrive as a malformed string from Robot Framework.
|
||
|
||
Robot Framework's Create Dictionary converts Python-style nested dicts
|
||
like {'enabled': False} into their string representation. Handle that here
|
||
before Pydantic's type validation runs. Also handles Pydantic models being
|
||
passed directly (from unit tests).
|
||
"""
|
||
# Pydantic model - convert to dict first
|
||
if hasattr(v, 'model_dump'):
|
||
return v.model_dump()
|
||
if hasattr(v, 'dict'):
|
||
return v.dict()
|
||
# Already a dict
|
||
if isinstance(v, dict):
|
||
return v
|
||
# String - try parsing
|
||
if isinstance(v, str):
|
||
for parse_fn in (json.loads, ast.literal_eval):
|
||
try:
|
||
parsed = parse_fn(v)
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
except Exception:
|
||
pass
|
||
return v
|
||
|
||
@model_validator(mode="before")
|
||
@classmethod
|
||
def _pre_validate(cls, data):
|
||
"""Handle malformed dict strings from Robot Framework and Pydantic models passed directly.
|
||
|
||
Robot Framework's Create Dictionary converts Python-style nested dicts
|
||
like {'enabled': False} into their string representation.
|
||
Unit tests may pass Pydantic model instances directly.
|
||
Both cases need conversion before type validation.
|
||
"""
|
||
if isinstance(data, dict):
|
||
data = dict(data) # make mutable
|
||
for field in ("name", "data_dir", "scheduler", "logging", "backup", "nfo", "scan_key_overrides", "other"):
|
||
if field in data:
|
||
v = data[field]
|
||
# Pydantic model - convert to dict
|
||
if hasattr(v, "model_dump"):
|
||
data[field] = v.model_dump()
|
||
# String from Robot Framework - try parsing
|
||
elif isinstance(v, str):
|
||
parsed = cls._parse_dict_field(v)
|
||
if isinstance(parsed, dict):
|
||
data[field] = parsed
|
||
return data
|
||
|
||
def apply_to(self, current: AppConfig) -> AppConfig:
|
||
"""Return a new AppConfig with updates applied to the current config.
|
||
|
||
Performs a shallow merge for `other`.
|
||
"""
|
||
data = current.model_dump()
|
||
if self.name is not None:
|
||
data["name"] = self.name
|
||
if self.data_dir is not None:
|
||
data["data_dir"] = self.data_dir
|
||
if self.scheduler is not None:
|
||
scheduler_data = self.scheduler
|
||
if isinstance(scheduler_data, str):
|
||
try:
|
||
scheduler_data = json.loads(scheduler_data)
|
||
except json.JSONDecodeError:
|
||
scheduler_data = ast.literal_eval(scheduler_data)
|
||
if isinstance(scheduler_data, dict):
|
||
try:
|
||
scheduler_data = SchedulerConfig(**scheduler_data)
|
||
except ValidationError:
|
||
raise
|
||
data["scheduler"] = scheduler_data.model_dump()
|
||
if self.logging is not None:
|
||
logging_data = self.logging
|
||
if isinstance(logging_data, str):
|
||
try:
|
||
logging_data = json.loads(logging_data)
|
||
except json.JSONDecodeError:
|
||
logging_data = ast.literal_eval(logging_data)
|
||
if isinstance(logging_data, dict):
|
||
logging_data = LoggingConfig(**logging_data)
|
||
data["logging"] = logging_data.model_dump()
|
||
if self.backup is not None:
|
||
backup_data = self.backup
|
||
if isinstance(backup_data, str):
|
||
try:
|
||
backup_data = json.loads(backup_data)
|
||
except json.JSONDecodeError:
|
||
backup_data = ast.literal_eval(backup_data)
|
||
if isinstance(backup_data, dict):
|
||
backup_data = BackupConfig(**backup_data)
|
||
data["backup"] = backup_data.model_dump()
|
||
if self.nfo is not None:
|
||
nfo_data = self.nfo
|
||
if isinstance(nfo_data, str):
|
||
try:
|
||
nfo_data = json.loads(nfo_data)
|
||
except json.JSONDecodeError:
|
||
nfo_data = ast.literal_eval(nfo_data)
|
||
if isinstance(nfo_data, dict):
|
||
nfo_data = NFOConfig(**nfo_data)
|
||
data["nfo"] = nfo_data.model_dump()
|
||
if self.scan_key_overrides is not None:
|
||
data["scan_key_overrides"] = self.scan_key_overrides
|
||
if self.other is not None:
|
||
merged = dict(current.other or {})
|
||
other_data = self.other
|
||
if isinstance(other_data, str):
|
||
try:
|
||
other_data = json.loads(other_data)
|
||
except json.JSONDecodeError:
|
||
other_data = ast.literal_eval(other_data)
|
||
if isinstance(other_data, dict):
|
||
merged.update(other_data)
|
||
data["other"] = merged
|
||
return AppConfig(**data)
|