CadHubManage/app/config.py

import os
from pathlib import Path
from typing import List, Optional

import yaml
from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


_raw_base_dir = os.environ.get("CADHUB_BASE_DIR")
if _raw_base_dir:
    BASE_DIR = Path(_raw_base_dir)
else:
    BASE_DIR = Path(__file__).resolve().parents[1]


class Settings(BaseSettings):
    """Application settings."""

    model_config = SettingsConfigDict(
        env_file=str(BASE_DIR / ".env"),
        env_file_encoding="utf-8",
    )

    app_name: str = "CadHubManage"
    debug: bool = False
    host: str = "0.0.0.0"
    port: int = 8000

    secret_key: str = "your-secret-key-change-in-production"
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    refresh_token_expire_days: int = 7

    software_config_path: str = str(BASE_DIR / "configs" / "software_config.yaml")
    users_config_path: str = str(BASE_DIR / "configs" / "users.json")

    cad_files_path: str = r"C:\Users\Public\Documents"
    batch_callback_base_url: str = "http://localhost:8000"

    cors_origins: List[str] = ["*"]

    log_level: str = "INFO"
    log_file: str = str(BASE_DIR / "logs" / "app.log")

    allowed_ips: List[str] = ["*"]

    @field_validator("debug", mode="before")
    @classmethod
    def parse_debug_flag(cls, value):
        if isinstance(value, bool):
            return value
        if isinstance(value, str):
            lowered = value.strip().lower()
            if lowered in {"1", "true", "yes", "on", "debug"}:
                return True
            if lowered in {"0", "false", "no", "off", "release", "prod", "production"}:
                return False
        return value


class SoftwareConfig:
    """Software config manager."""

    def __init__(self, config_path: str):
        self.config_path = config_path
        self._config = None
        self._config_mtime: Optional[float] = None

    def load_config(self) -> dict:
        try:
            with open(self.config_path, "r", encoding="utf-8") as f:
                self._config = yaml.safe_load(f) or {}
            try:
                self._config_mtime = Path(self.config_path).stat().st_mtime
            except OSError:
                self._config_mtime = None
            return self._config
        except FileNotFoundError as exc:
            raise FileNotFoundError(f"Software config not found: {self.config_path}") from exc
        except yaml.YAMLError as exc:
            raise ValueError(f"Invalid software config YAML: {exc}") from exc

    def save_config(self) -> None:
        try:
            with open(self.config_path, "w", encoding="utf-8") as f:
                yaml.dump(self._config, f, allow_unicode=True, default_flow_style=False)
        except Exception as exc:
            raise IOError(f"Failed to save software config: {exc}") from exc

    def _ensure_loaded(self):
        if self._config is None:
            self.load_config()
            return

        try:
            current_mtime = Path(self.config_path).stat().st_mtime
        except OSError:
            current_mtime = None

        if self._config_mtime is None or current_mtime is None:
            return

        if current_mtime > self._config_mtime:
            self.load_config()

    def get_software_list(self) -> List[str]:
        self._ensure_loaded()
        return list(self._config.get("software", {}).keys())

    def get_software_config(self, software_id: str) -> Optional[dict]:
        self._ensure_loaded()
        return self._config.get("software", {}).get(software_id)

    def validate_software_exists(self, software_id: str) -> bool:
        return software_id in self.get_software_list()

    def get_cad_files_path(self) -> str:
        self._ensure_loaded()
        file_storage = self._config.get("file_storage", {})
        return file_storage.get("cad_files_path", r"C:\Users\Public\Documents")

    def set_cad_files_path(self, path: str) -> None:
        self._ensure_loaded()

        if "file_storage" not in self._config:
            self._config["file_storage"] = {}

        self._config["file_storage"]["cad_files_path"] = path
        self.save_config()

    def get_file_extensions(self) -> dict:
        self._ensure_loaded()
        file_storage = self._config.get("file_storage", {})
        default_extensions = {
            "creo": [".prt", ".asm", ".drw"],
            "pdms": [".rvm", ".dri"],
            "revit": [".rvt", ".rfa", ".rte"],
        }
        return file_storage.get("file_extensions", default_extensions)

    def set_file_extensions(self, extensions: dict) -> None:
        self._ensure_loaded()

        if "file_storage" not in self._config:
            self._config["file_storage"] = {}

        self._config["file_storage"]["file_extensions"] = extensions
        self.save_config()

    def get_extension_routing(self) -> dict:
        self._ensure_loaded()
        routing = self._config.get("routing", {})
        mapping = routing.get("extension_to_software", {})
        return mapping if isinstance(mapping, dict) else {}

    def get_plugin_config(self, software_id: str) -> Optional[dict]:
        self._ensure_loaded()
        plugins = self._config.get("plugins", {})
        if not isinstance(plugins, dict):
            return None
        return plugins.get(software_id)

    def get_plugin_task_config(self, software_id: str, task_type: str) -> Optional[dict]:
        plugin = self.get_plugin_config(software_id)
        if not plugin:
            return None
        tasks = plugin.get("tasks", {})
        if not isinstance(tasks, dict):
            return None
        task_cfg = tasks.get(task_type)
        return task_cfg if isinstance(task_cfg, dict) else None


settings = Settings()
software_config = SoftwareConfig(settings.software_config_path)