from __future__ import annotations import re from dataclasses import dataclass from pathlib import Path INVALID_PATH_CHARS_RE = re.compile(r'[<>:"/\\|?*\x00-\x1f]') DEFAULT_WEB_PORT = 18080 def sanitize_path_component(value: str, fallback: str) -> str: cleaned = INVALID_PATH_CHARS_RE.sub("_", (value or "").strip()).rstrip(". ") return cleaned or fallback def pick_first_artist_name(singers: str | None) -> str: for candidate in re.split(r"\s*(?:/|,|&|\|)\s*", singers or ""): if candidate.strip(): return sanitize_path_component(candidate, "Unknown Artist") return "Unknown Artist" def build_download_relative_dir(platform: str, singers: str | None) -> Path: return Path(sanitize_path_component(platform, "unknown")) / pick_first_artist_name( singers ) def parse_web_port(value: str | int | None, fallback: int = DEFAULT_WEB_PORT) -> int: try: parsed = int(value) # type: ignore[arg-type] except (TypeError, ValueError): return fallback if 1 <= parsed <= 65535: return parsed return fallback @dataclass class CatalogSyncRuntimeConfig: root_dir: Path app_home: Path library_dir: Path db_path: Path env_file: Path input_dir: Path log_dir: Path python_bin: str venv_dir: Path web_host: str web_port: int download_layout: str @classmethod def from_mapping(cls, mapping: dict[str, str]) -> "CatalogSyncRuntimeConfig": root_dir = Path(mapping["ROOT_DIR"]) app_home = Path(mapping.get("APP_HOME", root_dir / "catalogsync")) library_dir = Path(mapping.get("LIBRARY_DIR", root_dir / "library")) web_port = parse_web_port(mapping.get("WEB_PORT"), fallback=DEFAULT_WEB_PORT) return cls( root_dir=root_dir, app_home=app_home, library_dir=library_dir, db_path=Path(mapping.get("DB_PATH", app_home / "data" / "catalogsync.db")), env_file=Path(mapping.get("ENV_FILE", app_home / "config" / "catalogsync.env")), input_dir=Path(mapping.get("INPUT_DIR", app_home / "inputs")), log_dir=Path(mapping.get("LOG_DIR", app_home / "logs")), python_bin=mapping.get("PYTHON_BIN", "python3"), venv_dir=Path(mapping.get("VENV_DIR", app_home / "app" / ".venv")), web_host=mapping.get("WEB_HOST", "127.0.0.1"), web_port=web_port, download_layout=mapping.get("DOWNLOAD_LAYOUT", "platform_first_artist"), ) def ensure_directories(self) -> None: for path in ( self.root_dir, self.library_dir, self.app_home / "app", self.app_home / "bin", self.app_home / "config", self.db_path.parent, self.env_file.parent, self.input_dir, self.log_dir, ): path.mkdir(parents=True, exist_ok=True)