Files
dl-organizer/src/vlm/config.py
T

366 lines
15 KiB
Python

"""Configuration management for Video Library Manager."""
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import yaml
@dataclass
class Config:
"""Configuration for Video Library Manager."""
library_root: Path
video_extensions: list[str] = field(default_factory=lambda: [
".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".m4v"
])
movie_template: str = "movie/{title} ({year})/"
series_template: str = "series/{title}/Season {season:02d}/"
movie_filename_template: str = "{title} ({year}){ext}"
series_filename_template: str = "S{season:02d}E{episode:02d}{ext}"
log_level: str = "INFO"
quarantine_dir: str = ".quarantine"
categories: dict[str, list[str]] = field(default_factory=lambda: {
"movie": ["movie", "movies"],
"series": ["series", "tv", "shows"],
"anime": ["anime"]
})
# Enrichment settings
enrichment_enabled: bool = True
enrichment_incremental: bool = True
enrichment_refresh_mode: str = "manual"
enrichment_providers: list[str] = field(default_factory=lambda: ["tmdb"])
enrichment_cache_db: Path = field(default_factory=lambda: Path.home() / ".vlm" / "enrichment_cache.db")
enrichment_max_concurrency: int = 6
enrichment_min_match_score: float = 0.75
translation_mode: str = "bidirectional"
translation_fallback_machine: bool = True
tmdb_api_key: Optional[str] = None
tmdb_bearer_token: Optional[str] = None
tmdb_language: str = "zh-CN"
tmdb_region: Optional[str] = None
tmdb_include_adult: bool = False
openai_api_key: Optional[str] = None
reputation_min_votes: int = 50
reputation_low_score_threshold: float = 6.0
reputation_policy: str = "flag_for_review"
naming_title_format: str = "{title_zh} {title_en}"
# Plan settings (e.g. duplicate handling when consuming analysis)
duplicate_keep: str = "by_reputation"
plan_max_season: int = 15
plan_max_episode: int = 100
plan_include_sample_files: bool = False
def load_config(path: Path) -> Config:
"""Load configuration from YAML file."""
if not path.exists():
raise FileNotFoundError(f"Configuration file not found: {path}")
try:
with open(path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
except yaml.YAMLError as e:
raise yaml.YAMLError(f"Invalid YAML syntax in configuration file: {e}")
if data is None:
data = {}
library_root_str = data.get("library_root")
if not library_root_str:
raise ValueError("Configuration must specify 'library_root'")
library_root = Path(library_root_str).expanduser()
video_extensions = data.get("video_extensions", [
".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".m4v"
])
templates = data.get("templates", {})
movie_template = templates.get("movie_dir", "movie/{title} ({year})/")
series_template = templates.get("series_dir", "series/{title}/Season {season:02d}/")
movie_filename_template = templates.get("movie_filename", "{title} ({year}){ext}")
series_filename_template = templates.get("series_filename", "S{season:02d}E{episode:02d}{ext}")
quarantine_dir = data.get("quarantine_dir", ".quarantine")
log_level = data.get("log_level", "INFO")
categories = data.get("categories", {
"movie": ["movie", "movies"],
"series": ["series", "tv", "shows"],
"anime": ["anime"]
})
plan = data.get("plan", {})
duplicate_keep = plan.get("duplicate_keep", "by_reputation")
plan_max_season = int(plan.get("max_season", 15))
plan_max_episode = int(plan.get("max_episode", 100))
plan_include_sample_files = bool(plan.get("include_sample_files", False))
enrichment = data.get("enrichment")
if enrichment is None:
enrichment = data.get("enrich", {})
translation = enrichment.get("translation", {})
api_keys = enrichment.get("api_keys", {})
reputation = enrichment.get("reputation", {})
naming = enrichment.get("naming", {})
tmdb = enrichment.get("tmdb", {})
return Config(
library_root=library_root,
video_extensions=video_extensions,
movie_template=movie_template,
series_template=series_template,
movie_filename_template=movie_filename_template,
series_filename_template=series_filename_template,
log_level=log_level,
quarantine_dir=quarantine_dir,
categories=categories,
enrichment_enabled=enrichment.get("enabled", True),
enrichment_incremental=enrichment.get("incremental", True),
enrichment_refresh_mode=enrichment.get("refresh_mode", "manual"),
enrichment_providers=enrichment.get("providers", ["tmdb"]),
enrichment_cache_db=Path(
enrichment.get("cache_db", str(Path.home() / ".vlm" / "enrichment_cache.db"))
).expanduser(),
enrichment_max_concurrency=enrichment.get("max_concurrency", 6),
enrichment_min_match_score=enrichment.get("min_match_score", 0.75),
translation_mode=translation.get("mode", "bidirectional"),
translation_fallback_machine=translation.get("fallback_machine", True),
tmdb_api_key=api_keys.get("tmdb"),
tmdb_bearer_token=api_keys.get("tmdb_bearer"),
tmdb_language=tmdb.get("language", "zh-CN"),
tmdb_region=tmdb.get("region"),
tmdb_include_adult=tmdb.get("include_adult", False),
openai_api_key=api_keys.get("openai"),
reputation_min_votes=reputation.get("min_votes", 50),
reputation_low_score_threshold=reputation.get("low_score_threshold", 6.0),
reputation_policy=reputation.get("policy", "flag_for_review"),
naming_title_format=naming.get("title_format", "{title_zh} {title_en}"),
duplicate_keep=duplicate_keep,
plan_max_season=plan_max_season,
plan_max_episode=plan_max_episode,
plan_include_sample_files=plan_include_sample_files,
)
def create_default_config(path: Path) -> Config:
"""Create a default configuration file and return the Config object."""
default_config = Config(
library_root=Path.home() / "Videos",
video_extensions=[".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".m4v"],
)
enrichment_content = {
"enabled": default_config.enrichment_enabled,
"incremental": default_config.enrichment_incremental,
"refresh_mode": default_config.enrichment_refresh_mode,
"providers": default_config.enrichment_providers,
"cache_db": str(default_config.enrichment_cache_db),
"max_concurrency": default_config.enrichment_max_concurrency,
"min_match_score": default_config.enrichment_min_match_score,
"translation": {
"mode": default_config.translation_mode,
"fallback_machine": default_config.translation_fallback_machine,
},
"api_keys": {
"tmdb": default_config.tmdb_api_key,
"tmdb_bearer": default_config.tmdb_bearer_token,
"openai": default_config.openai_api_key,
},
"tmdb": {
"language": default_config.tmdb_language,
"region": default_config.tmdb_region,
"include_adult": default_config.tmdb_include_adult,
},
"reputation": {
"min_votes": default_config.reputation_min_votes,
"low_score_threshold": default_config.reputation_low_score_threshold,
"policy": default_config.reputation_policy,
},
"naming": {
"title_format": default_config.naming_title_format,
},
}
plan_content = {
"duplicate_keep": default_config.duplicate_keep,
"max_season": default_config.plan_max_season,
"max_episode": default_config.plan_max_episode,
"include_sample_files": default_config.plan_include_sample_files,
}
yaml_content = {
"library_root": str(default_config.library_root),
"video_extensions": default_config.video_extensions,
"plan": plan_content,
"templates": {
"movie_dir": default_config.movie_template,
"series_dir": default_config.series_template,
"movie_filename": default_config.movie_filename_template,
"series_filename": default_config.series_filename_template,
},
"quarantine_dir": default_config.quarantine_dir,
"log_level": default_config.log_level,
"categories": default_config.categories,
"enrichment": enrichment_content,
# Backward-compatible alias for users who prefer `enrich`.
"enrich": enrichment_content,
}
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
yaml.dump(yaml_content, f, default_flow_style=False, sort_keys=False)
return default_config
def validate_config(config: Config) -> list[str]:
"""Validate configuration and return list of error messages."""
errors = []
if not isinstance(config.library_root, Path):
errors.append("library_root must be a Path object")
elif not str(config.library_root) or str(config.library_root) == ".":
errors.append("library_root cannot be empty")
if not config.video_extensions:
errors.append("video_extensions cannot be empty")
elif not isinstance(config.video_extensions, list):
errors.append("video_extensions must be a list")
else:
for ext in config.video_extensions:
if not isinstance(ext, str):
errors.append(f"video_extensions must contain strings, found: {type(ext)}")
break
if not ext.startswith("."):
errors.append(f"video extension must start with '.': {ext}")
if not config.movie_template:
errors.append("movie_template cannot be empty")
elif not isinstance(config.movie_template, str):
errors.append("movie_template must be a string")
if not config.series_template:
errors.append("series_template cannot be empty")
elif not isinstance(config.series_template, str):
errors.append("series_template must be a string")
if not config.movie_filename_template:
errors.append("movie_filename_template cannot be empty")
elif not isinstance(config.movie_filename_template, str):
errors.append("movie_filename_template must be a string")
if not config.series_filename_template:
errors.append("series_filename_template cannot be empty")
elif not isinstance(config.series_filename_template, str):
errors.append("series_filename_template must be a string")
if not isinstance(config.enrichment_max_concurrency, int):
errors.append("enrichment_max_concurrency must be an integer")
elif config.enrichment_max_concurrency < 1:
errors.append("enrichment_max_concurrency must be >= 1")
valid_log_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
if not config.log_level:
errors.append("log_level cannot be empty")
elif not isinstance(config.log_level, str):
errors.append("log_level must be a string")
elif config.log_level.upper() not in valid_log_levels:
errors.append(f"log_level must be one of {valid_log_levels}, got: {config.log_level}")
if not config.quarantine_dir:
errors.append("quarantine_dir cannot be empty")
elif not isinstance(config.quarantine_dir, str):
errors.append("quarantine_dir must be a string")
elif config.quarantine_dir.startswith("/") or config.quarantine_dir.startswith("\\"):
errors.append("quarantine_dir must be relative to category root, not absolute")
if not config.categories:
errors.append("categories cannot be empty")
elif not isinstance(config.categories, dict):
errors.append("categories must be a dictionary")
else:
required_categories = {"movie", "series", "anime"}
missing = required_categories - set(config.categories.keys())
if missing:
errors.append(f"categories must include keys: {sorted(missing)}")
seen_dirs = {}
for category, dir_list in config.categories.items():
if not isinstance(dir_list, list):
errors.append(f"categories['{category}'] must be a list")
continue
if not dir_list:
errors.append(f"categories['{category}'] cannot be empty")
continue
for dir_name in dir_list:
if not isinstance(dir_name, str):
errors.append(f"categories['{category}'] must contain strings")
break
if not dir_name.strip():
errors.append(f"categories['{category}'] contains empty directory name")
break
dir_lower = dir_name.lower()
if dir_lower in seen_dirs:
errors.append(
f"Duplicate directory name '{dir_name}' in categories "
f"'{category}' and '{seen_dirs[dir_lower]}'"
)
else:
seen_dirs[dir_lower] = category
if not isinstance(config.enrichment_cache_db, Path):
errors.append("enrichment_cache_db must be a Path object")
if not isinstance(config.enrichment_providers, list) or not config.enrichment_providers:
errors.append("enrichment_providers must be a non-empty list")
else:
allowed_providers = {"tmdb"}
invalid = [provider for provider in config.enrichment_providers if provider.lower() not in allowed_providers]
if invalid:
errors.append(
f"enrichment_providers contains unsupported providers: {invalid}; supported providers: ['tmdb']"
)
if config.enrichment_max_concurrency < 1:
errors.append("enrichment_max_concurrency must be >= 1")
if not (0.0 <= config.enrichment_min_match_score <= 1.0):
errors.append("enrichment_min_match_score must be between 0.0 and 1.0")
if config.enrichment_refresh_mode not in {"manual"}:
errors.append("enrichment_refresh_mode must be 'manual'")
if config.reputation_min_votes < 0:
errors.append("reputation_min_votes must be >= 0")
if not (0.0 <= config.reputation_low_score_threshold <= 10.0):
errors.append("reputation_low_score_threshold must be between 0.0 and 10.0")
if not isinstance(config.tmdb_language, str) or not config.tmdb_language.strip():
errors.append("tmdb_language must be a non-empty string")
if config.tmdb_region is not None and not isinstance(config.tmdb_region, str):
errors.append("tmdb_region must be a string when set")
if not isinstance(config.tmdb_include_adult, bool):
errors.append("tmdb_include_adult must be a boolean")
if config.duplicate_keep not in (
"by_reputation",
"by_reputation_quality_time",
"first_seen",
"manual",
"by_quality",
):
errors.append(
"duplicate_keep must be one of "
"'by_reputation', 'by_reputation_quality_time', 'first_seen', 'manual', 'by_quality', "
f"got: {config.duplicate_keep!r}"
)
if not isinstance(config.plan_max_season, int) or config.plan_max_season < 1:
errors.append("plan_max_season must be an integer >= 1")
if not isinstance(config.plan_max_episode, int) or config.plan_max_episode < 1:
errors.append("plan_max_episode must be an integer >= 1")
if not isinstance(config.plan_include_sample_files, bool):
errors.append("plan_include_sample_files must be a boolean")
return errors