""" Application configuration using Pydantic Settings """ from pydantic_settings import BaseSettings from pydantic import BaseModel, Field from typing import Optional import yaml from pathlib import Path class ThumbnailSettings(BaseModel): """Thumbnail generation settings""" small: int = 240 medium: int = 640 large: int = 1280 quality: int = 85 format: str = "webp" class ScannerSettings(BaseModel): """File scanner settings""" watch: bool = True initial_scan_on_start: bool = True batch_size: int = 100 concurrent_workers: int = 4 class PerformanceSettings(BaseModel): """Performance tuning settings""" max_concurrent_thumbnails: int = 10 cache_ttl: int = 3600 db_pool_size: int = 20 db_pool_recycle: int = 3600 class EmbedderSettings(BaseModel): """CLIP / SigLIP embedding model settings""" name: str = "openclip_vitb32" batch_size: int = 8 class OCRSettings(BaseModel): """PaddleOCR / rapidocr settings""" enabled: bool = True languages: list[str] = ["en"] min_confidence: float = 0.5 class DetectorSettings(BaseModel): """YOLOv8n object detection settings""" enabled: bool = True min_confidence: float = 0.35 max_detections: int = 50 class FacesSettings(BaseModel): """YuNet + SFace face detection/recognition settings""" enabled: bool = True min_face_size: int = 40 recognition_threshold: float = 0.85 cluster_eps: float = 0.25 class VisionSettings(BaseModel): """AI vision pipeline settings. Disabled when running on SQLite (pgvector is required for embedding storage).""" enabled: bool = True backend: str = "onnx" # "onnx" | "rocm" (future) models_dir: str = "/data/models" embedder: EmbedderSettings = EmbedderSettings() ocr: OCRSettings = OCRSettings() detector: DetectorSettings = DetectorSettings() faces: FacesSettings = FacesSettings() worker_concurrency: int = 2 class MulitaConfig(BaseModel): """Main configuration from YAML file. Source roots and the discard workflow are owned by the database now — only operational settings live here.""" thumbnails: ThumbnailSettings = ThumbnailSettings() scanner: ScannerSettings = ScannerSettings() performance: PerformanceSettings = PerformanceSettings() vision: VisionSettings = VisionSettings() class Settings(BaseSettings): """Application settings""" # Database — Postgres + pgvector by default. The SQLite escape hatch # remains supported via the docker-compose.sqlite.yml override and by # setting DATABASE_URL=sqlite+aiosqlite:///... in .env for local dev. database_url: str = Field( default="postgresql+asyncpg://mulita:mulita@db:5432/mulita", env="DATABASE_URL" ) # Redis redis_url: str = Field( default="redis://localhost:6379", env="REDIS_URL" ) # Celery celery_broker_url: str = Field( default="redis://localhost:6379", env="CELERY_BROKER_URL" ) celery_result_backend: str = Field( default="redis://localhost:6379", env="CELERY_RESULT_BACKEND" ) # Photo directories photo_dirs: str = Field( default="/photos", env="PHOTO_DIRS" ) # API settings api_host: str = Field(default="0.0.0.0", env="API_HOST") api_port: int = Field(default=8000, env="API_PORT") # CORS — comma-separated list of allowed origins, or "*" for any. # Same-origin requests (the normal case behind nginx / vite proxy) # never trip CORS, so this is only for direct browser access from # other origins (LAN IP, reverse proxy, dev tools). allowed_origins: str = Field(default="*", env="ALLOWED_ORIGINS") # Logging — accepts standard python levels (DEBUG, INFO, WARNING, # ERROR, CRITICAL). Bumped from INFO when chasing a problem. log_level: str = Field(default="INFO", env="LOG_LEVEL") @property def cors_origins(self) -> list[str]: """Parse the ALLOWED_ORIGINS env var into a list. Accepts: - "*" → wildcard (single-element list ["*"]) - "http://a.com,http://b.com" → split + strip Empty entries are dropped. """ raw = (self.allowed_origins or "").strip() if not raw or raw == "*": return ["*"] return [o.strip() for o in raw.split(",") if o.strip()] # App configuration from YAML _config: Optional[MulitaConfig] = None @property def config(self) -> MulitaConfig: """Load configuration from YAML file""" if self._config is None: config_path = Path("/app/config/mulita.yml") if not config_path.exists(): config_path = Path("mulita.yml") if config_path.exists(): with open(config_path, "r") as f: config_data = yaml.safe_load(f) self._config = MulitaConfig(**config_data) else: self._config = MulitaConfig() return self._config @property def thumbnails(self) -> ThumbnailSettings: return self.config.thumbnails @property def scanner(self) -> ScannerSettings: return self.config.scanner @property def performance(self) -> PerformanceSettings: return self.config.performance @property def vision(self) -> VisionSettings: return self.config.vision class Config: env_file = ".env" case_sensitive = False # Global settings instance settings = Settings()