247 lines
8.0 KiB
Python
247 lines
8.0 KiB
Python
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
|
|
notifier_state_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:
|
|
changed = _migrate_storage_config(data)
|
|
backup = data.get("backup")
|
|
if not isinstance(backup, dict):
|
|
return changed
|
|
|
|
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)) in {300, 600}:
|
|
server_check["interval_seconds"] = 1800
|
|
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 _migrate_storage_config(data: dict[str, Any]) -> bool:
|
|
storage = data.get("storage")
|
|
if not isinstance(storage, dict):
|
|
return False
|
|
if "notifier_state_file" in storage:
|
|
return False
|
|
storage["notifier_state_file"] = "%PROGRAMDATA%\\BakRest\\notifier-state.json"
|
|
return True
|
|
|
|
|
|
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")),
|
|
notifier_state_file=expand_path(
|
|
storage.get("notifier_state_file", "%PROGRAMDATA%\\BakRest\\notifier-state.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
|