240 lines
8.3 KiB
Python
240 lines
8.3 KiB
Python
"""Domänenmodelle und validierte Konfigurationswerte."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import asdict, dataclass, field
|
|
from enum import StrEnum
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
__all__ = [
|
|
"AppConfig",
|
|
"AppSettings",
|
|
"ConflictMode",
|
|
"DeleteMode",
|
|
"SyncDirection",
|
|
"SyncTask",
|
|
"TaskRuntimeState",
|
|
"TriggerMode",
|
|
]
|
|
|
|
|
|
class SyncDirection(StrEnum):
|
|
"""Unterstützte Synchronisationsrichtungen."""
|
|
|
|
ONE_WAY = "one_way"
|
|
BIDIRECTIONAL = "bidirectional"
|
|
|
|
|
|
class TriggerMode(StrEnum):
|
|
"""Unterstützte Auslöser für eine Aufgabe."""
|
|
|
|
REALTIME = "realtime"
|
|
INTERVAL = "interval"
|
|
MANUAL = "manual"
|
|
|
|
|
|
class DeleteMode(StrEnum):
|
|
"""Verhalten bei erkannten Löschungen."""
|
|
|
|
MIRROR = "mirror"
|
|
KEEP = "keep"
|
|
|
|
|
|
class ConflictMode(StrEnum):
|
|
"""Verhalten bei beidseitig geänderten Dateien."""
|
|
|
|
NEWER_WINS = "newer_wins"
|
|
CONFLICT_COPY = "conflict_copy"
|
|
|
|
|
|
class TaskRuntimeState(StrEnum):
|
|
"""Nicht persistierter, tatsächlicher Laufzeitzustand einer Aufgabe."""
|
|
|
|
ACTIVE = "active"
|
|
PAUSED = "paused"
|
|
WAITING_FOR_RELEASE = "waiting_for_release"
|
|
ERROR = "error"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SyncTask:
|
|
"""Persistierbare Konfiguration eines Ordnerpaars."""
|
|
|
|
name: str
|
|
source: str
|
|
target: str
|
|
id: str = field(default_factory=lambda: str(uuid4()))
|
|
direction: SyncDirection = SyncDirection.ONE_WAY
|
|
trigger: TriggerMode = TriggerMode.MANUAL
|
|
delete_mode: DeleteMode = DeleteMode.KEEP
|
|
conflict_mode: ConflictMode = ConflictMode.NEWER_WINS
|
|
interval_minutes: int = 30
|
|
selected_paths: list[str] = field(default_factory=list)
|
|
ignored_paths: list[str] = field(default_factory=list)
|
|
enabled: bool = True
|
|
last_run: str | None = None
|
|
|
|
def validate(self) -> None:
|
|
"""Validiert sicherheitsrelevante Aufgabeneigenschaften frühzeitig."""
|
|
if not self.name.strip():
|
|
raise ValueError("Bitte einen Namen für die Aufgabe eingeben.")
|
|
if not self.source.strip() or not self.target.strip():
|
|
raise ValueError("Quell- und Zielordner müssen angegeben werden.")
|
|
source = _safe_resolve(Path(self.source).expanduser())
|
|
target = _safe_resolve(Path(self.target).expanduser())
|
|
source_norm = os.path.normcase(str(source))
|
|
target_norm = os.path.normcase(str(target))
|
|
if source_norm == target_norm:
|
|
raise ValueError("Quell- und Zielordner dürfen nicht identisch sein.")
|
|
if _is_relative_to(source, target) or _is_relative_to(target, source):
|
|
raise ValueError("Quell- und Zielordner dürfen nicht ineinander liegen.")
|
|
if not 1 <= self.interval_minutes <= 10080:
|
|
raise ValueError("Das Intervall muss zwischen 1 Minute und 7 Tagen liegen.")
|
|
self.name = self.name.strip()
|
|
self.source = str(source)
|
|
self.target = str(target)
|
|
self.selected_paths = sorted({_normalize_relative_path(item) for item in self.selected_paths if item.strip()})
|
|
self.ignored_paths = sorted({_normalize_relative_path(item) for item in self.ignored_paths if item.strip()})
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Serialisiert die Aufgabe in ein JSON-kompatibles Dictionary."""
|
|
data = asdict(self)
|
|
data["direction"] = self.direction.value
|
|
data["trigger"] = self.trigger.value
|
|
data["delete_mode"] = self.delete_mode.value
|
|
data["conflict_mode"] = self.conflict_mode.value
|
|
data["ignored_paths"] = list(self.ignored_paths)
|
|
return data
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> SyncTask:
|
|
"""Erzeugt eine Aufgabe aus persistierten Daten."""
|
|
task = cls(
|
|
id=str(data.get("id") or uuid4()),
|
|
name=str(data.get("name", "Unbenannte Aufgabe")),
|
|
source=str(data.get("source", "")),
|
|
target=str(data.get("target", "")),
|
|
direction=SyncDirection(data.get("direction", SyncDirection.ONE_WAY.value)),
|
|
trigger=TriggerMode(data.get("trigger", TriggerMode.MANUAL.value)),
|
|
delete_mode=DeleteMode(data.get("delete_mode", DeleteMode.KEEP.value)),
|
|
conflict_mode=ConflictMode(data.get("conflict_mode", ConflictMode.NEWER_WINS.value)),
|
|
interval_minutes=int(data.get("interval_minutes", 30)),
|
|
selected_paths=[str(item) for item in data.get("selected_paths", [])],
|
|
ignored_paths=[str(item) for item in data.get("ignored_paths", [])],
|
|
enabled=bool(data.get("enabled", True)),
|
|
last_run=data.get("last_run"),
|
|
)
|
|
try:
|
|
task.validate()
|
|
except ValueError:
|
|
task.enabled = False
|
|
return task
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AppSettings:
|
|
"""Globale, persistierbare Anwendungseinstellungen."""
|
|
|
|
autostart: bool = False
|
|
font_family: str = "Segoe UI Variable"
|
|
font_size: int = 10
|
|
font_weight: str = "regular"
|
|
ui_scale: int = 100
|
|
paused: bool = False
|
|
pending_queue: list[str] = field(default_factory=list)
|
|
check_updates_on_start: bool = True
|
|
last_update_check: str | None = None
|
|
|
|
def validate(self) -> None:
|
|
"""Begrenzt Darstellungswerte auf brauchbare Bereiche."""
|
|
self.font_size = max(8, min(18, int(self.font_size)))
|
|
self.ui_scale = max(80, min(150, int(self.ui_scale)))
|
|
if self.font_weight not in {"regular", "bold"}:
|
|
self.font_weight = "regular"
|
|
if not self.font_family.strip():
|
|
self.font_family = "Segoe UI Variable"
|
|
self.pending_queue = [str(item) for item in self.pending_queue if str(item).strip()]
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Serialisiert die Einstellungen."""
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> AppSettings:
|
|
"""Erzeugt Einstellungen aus einem Dictionary."""
|
|
settings = cls(
|
|
autostart=bool(data.get("autostart", False)),
|
|
font_family=str(data.get("font_family", "Segoe UI Variable")),
|
|
font_size=int(data.get("font_size", 10)),
|
|
font_weight=str(data.get("font_weight", "regular")),
|
|
ui_scale=int(data.get("ui_scale", 100)),
|
|
paused=bool(data.get("paused", False)),
|
|
pending_queue=[str(item) for item in data.get("pending_queue", []) if str(item).strip()],
|
|
check_updates_on_start=bool(data.get("check_updates_on_start", True)),
|
|
last_update_check=data.get("last_update_check"),
|
|
)
|
|
settings.validate()
|
|
return settings
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AppConfig:
|
|
"""Wurzelobjekt der lokalen JSON-Konfiguration."""
|
|
|
|
tasks: list[SyncTask] = field(default_factory=list)
|
|
settings: AppSettings = field(default_factory=AppSettings)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Serialisiert die vollständige Konfiguration."""
|
|
return {
|
|
"schema_version": 1,
|
|
"tasks": [task.to_dict() for task in self.tasks],
|
|
"settings": self.settings.to_dict(),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> AppConfig:
|
|
"""Erzeugt eine vollständige Konfiguration aus JSON-Daten."""
|
|
tasks: list[SyncTask] = []
|
|
for item in data.get("tasks", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
try:
|
|
tasks.append(SyncTask.from_dict(item))
|
|
except Exception:
|
|
continue
|
|
return cls(
|
|
tasks=tasks,
|
|
settings=AppSettings.from_dict(data.get("settings", {})),
|
|
)
|
|
|
|
|
|
def _safe_resolve(path: Path) -> Path:
|
|
"""Löst Pfade sicher auf; fängt Netzwerkfehler und unvollständige Pfade ab."""
|
|
try:
|
|
return path.resolve(strict=False)
|
|
except OSError:
|
|
return path.absolute()
|
|
|
|
|
|
def _normalize_relative_path(value: str) -> str:
|
|
normalized = value.replace("/", os.sep).replace("\\", os.sep).strip(os.sep)
|
|
path = Path(normalized)
|
|
if path.is_absolute() or ".." in path.parts:
|
|
raise ValueError(f"Ungültige Unterauswahl: {value}")
|
|
return str(path)
|
|
|
|
|
|
def _is_relative_to(path: Path, base: Path) -> bool:
|
|
try:
|
|
return path.is_relative_to(base)
|
|
except (ValueError, OSError, AttributeError):
|
|
try:
|
|
path.relative_to(base)
|
|
return True
|
|
except ValueError:
|
|
return False
|