bak-and-rest/src/bakrest/config.py

from __future__ import annotations

import shutil
import tomllib
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import tomli_w

from .default_config import DEFAULT_CONFIG_TOML
from .paths import default_config_path, expand_path


@dataclass(frozen=True)
class ServerCheckConfig:
    type: str
    port: int
    interval_seconds: int
    timeout_seconds: int


@dataclass(frozen=True)
class BackupConfig:
    engine: str
    server_host: str
    remote_destination: str
    remote_destinations: tuple[str, ...]
    rsync_path: str
    robocopy_path: str
    shutdown_on_success: bool
    shutdown_command: str
    server_check: ServerCheckConfig


@dataclass(frozen=True)
class WatchConfig:
    include_dirs: tuple[Path, ...]
    include_extensions: frozenset[str]
    exclude_dirs: tuple[Path | str, ...]
    exclude_patterns: tuple[str, ...]


@dataclass(frozen=True)
class StorageConfig:
    change_registry: Path
    logs_dir: Path
    status_file: Path


@dataclass(frozen=True)
class StabilityConfig:
    min_age_seconds: int
    unchanged_check_interval_seconds: int
    unchanged_checks_required: int


@dataclass(frozen=True)
class AppConfig:
    path: Path
    service_autostart: bool
    tray_autostart: bool
    watch: WatchConfig
    backup: BackupConfig
    storage: StorageConfig
    stability: StabilityConfig


def ensure_default_config(path: Path | None = None) -> Path:
    config_path = path or default_config_path()
    if config_path.exists():
        return config_path

    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(DEFAULT_CONFIG_TOML, encoding="utf-8")
    return config_path


def copy_example_config(destination: Path) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    shutil.copyfile(Path("config/config.example.toml"), destination)


def load_config(path: Path | None = None) -> AppConfig:
    config_path = ensure_default_config(path)
    data = load_config_data(config_path)
    return parse_config(data, config_path)


def load_config_data(path: Path | None = None) -> dict[str, Any]:
    config_path = ensure_default_config(path)
    data = tomllib.loads(config_path.read_text(encoding="utf-8"))
    migrated = migrate_config_data(data)
    if migrated:
        save_config_data(data, config_path)
    return data


def save_config_data(data: dict[str, Any], path: Path | None = None) -> Path:
    config_path = path or default_config_path()
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(tomli_w.dumps(data), encoding="utf-8")
    return config_path


def migrate_config_data(data: dict[str, Any]) -> bool:
    backup = data.get("backup")
    if not isinstance(backup, dict):
        return False

    changed = False
    engine = str(backup.get("engine", "robocopy")).lower()
    if "engine" not in backup:
        backup["engine"] = engine
        changed = True

    if engine == "robocopy":
        if "robocopy_path" not in backup:
            backup["robocopy_path"] = "robocopy"
            changed = True

        server_check = backup.setdefault("server_check", {})
        if isinstance(server_check, dict) and int(server_check.get("port", 22)) == 22:
            server_check["port"] = 445
            changed = True
        if isinstance(server_check, dict) and int(server_check.get("interval_seconds", 300)) == 300:
            server_check["interval_seconds"] = 600
            changed = True

        destinations = backup.get("remote_destinations")
        if not isinstance(destinations, list):
            legacy_destination = str(backup.get("remote_destination", "")).strip()
            backup["remote_destinations"] = (
                [legacy_destination] if _is_windows_destination(legacy_destination) else []
            )
            changed = True

    return changed


def parse_config(data: dict[str, Any], path: Path) -> AppConfig:
    service = data.get("service", {})
    tray = data.get("tray", {})
    watch = data.get("watch", {})
    backup = data.get("backup", {})
    server_check = backup.get("server_check", {})
    storage = data.get("storage", {})
    stability = data.get("stability", {})

    include_dirs = tuple(expand_path(item) for item in watch.get("include_dirs", []))
    include_extensions = frozenset(
        _normalize_extension(item) for item in watch.get("include_extensions", [])
    )
    exclude_dirs = tuple(_normalize_exclude_dir(item) for item in watch.get("exclude_dirs", []))

    storage_config = StorageConfig(
        change_registry=expand_path(
            storage.get("change_registry", "%PROGRAMDATA%\\BakRest\\changes.sqlite")
        ),
        logs_dir=expand_path(storage.get("logs_dir", "%PROGRAMDATA%\\BakRest\\logs")),
        status_file=expand_path(storage.get("status_file", "%PROGRAMDATA%\\BakRest\\status.json")),
    )

    return AppConfig(
        path=path,
        service_autostart=bool(service.get("autostart", True)),
        tray_autostart=bool(tray.get("autostart", True)),
        watch=WatchConfig(
            include_dirs=include_dirs,
            include_extensions=include_extensions,
            exclude_dirs=exclude_dirs,
            exclude_patterns=tuple(watch.get("exclude_patterns", [])),
        ),
        backup=BackupConfig(
            engine=str(backup.get("engine", "robocopy")).lower(),
            server_host=str(backup.get("server_host", "backup-server")),
            remote_destination=str(backup.get("remote_destination", "")),
            remote_destinations=_remote_destinations(backup),
            rsync_path=str(backup.get("rsync_path", "rsync")),
            robocopy_path=str(backup.get("robocopy_path", "robocopy")),
            shutdown_on_success=bool(backup.get("shutdown_on_success", True)),
            shutdown_command=str(backup.get("shutdown_command", "shutdown /s /t 0")),
            server_check=ServerCheckConfig(
                type=str(server_check.get("type", "tcp")).lower(),
                port=int(server_check.get("port", 22)),
                interval_seconds=int(server_check.get("interval_seconds", 300)),
                timeout_seconds=int(server_check.get("timeout_seconds", 3)),
            ),
        ),
        storage=storage_config,
        stability=StabilityConfig(
            min_age_seconds=int(stability.get("min_age_seconds", 60)),
            unchanged_check_interval_seconds=int(
                stability.get("unchanged_check_interval_seconds", 5)
            ),
            unchanged_checks_required=int(stability.get("unchanged_checks_required", 2)),
        ),
    )


def _normalize_extension(value: str) -> str:
    value = value.strip().lower()
    if not value:
        return value
    return value if value.startswith(".") else f".{value}"


def _normalize_exclude_dir(value: str) -> Path | str:
    expanded = expand_path(value)
    if "%" in value or not expanded.is_absolute():
        return value.lower()
    return expanded


def _remote_destinations(backup: dict[str, Any]) -> tuple[str, ...]:
    destinations = backup.get("remote_destinations")
    if isinstance(destinations, list):
        return tuple(str(item).strip() for item in destinations if str(item).strip())

    destination = str(backup.get("remote_destination", "")).strip()
    engine = str(backup.get("engine", "robocopy")).lower()
    if engine == "robocopy" and not _is_windows_destination(destination):
        return ()
    return (destination,) if destination else ()


def _is_windows_destination(destination: str) -> bool:
    if destination.startswith("\\\\"):
        return True
    if len(destination) >= 3 and destination[1:3] == ":\\":
        return True
    return False