|
|
|
@@ -30,11 +30,12 @@ class ConfigStore:
|
|
|
|
|
|
|
|
|
|
def __init__(self, path: Path | None = None) -> None:
|
|
|
|
|
self.path = path or default_data_directory() / "config.json"
|
|
|
|
|
self._recovery_marker = self.path.with_name("config.recovery.json")
|
|
|
|
|
|
|
|
|
|
def load(self) -> AppConfig:
|
|
|
|
|
"""Lädt die Konfiguration; beschädigte Dateien werden gesichert."""
|
|
|
|
|
if not self.path.exists():
|
|
|
|
|
return AppConfig()
|
|
|
|
|
return self._recover_from_backup(None) or AppConfig()
|
|
|
|
|
|
|
|
|
|
content: str | None = None
|
|
|
|
|
last_io_exc: OSError | None = None
|
|
|
|
@@ -58,25 +59,132 @@ class ConfigStore:
|
|
|
|
|
except (ValueError, TypeError, json.JSONDecodeError) as exc:
|
|
|
|
|
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
|
|
|
recovery_path = self.path.with_name(f"config.defekt-{timestamp}.json")
|
|
|
|
|
backup_created = False
|
|
|
|
|
try:
|
|
|
|
|
self.path.replace(recovery_path)
|
|
|
|
|
backup_created = True
|
|
|
|
|
except OSError:
|
|
|
|
|
recovery_path = self.path
|
|
|
|
|
if backup_created:
|
|
|
|
|
recovered = self._recover_from_backup(None)
|
|
|
|
|
if recovered is not None:
|
|
|
|
|
return recovered
|
|
|
|
|
raise ConfigError(
|
|
|
|
|
f"Die Konfiguration war syntaktisch beschädigt ({exc}). Sicherung: {recovery_path}"
|
|
|
|
|
) from exc
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
return AppConfig.from_dict(raw)
|
|
|
|
|
config = AppConfig.from_dict(raw)
|
|
|
|
|
except Exception as exc:
|
|
|
|
|
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
|
|
|
backup_path = self.path.with_name(f"config.bak-{timestamp}.json")
|
|
|
|
|
with suppress(OSError):
|
|
|
|
|
try:
|
|
|
|
|
shutil.copy2(self.path, backup_path)
|
|
|
|
|
except OSError:
|
|
|
|
|
pass
|
|
|
|
|
else:
|
|
|
|
|
recovered = self._recover_from_backup(None)
|
|
|
|
|
if recovered is not None:
|
|
|
|
|
return recovered
|
|
|
|
|
raise ConfigError(
|
|
|
|
|
f"Die Konfigurationsdaten konnten nicht verarbeitet werden ({exc}). Sicherungskopie: {backup_path}"
|
|
|
|
|
) from exc
|
|
|
|
|
|
|
|
|
|
if config.tasks:
|
|
|
|
|
self._mark_backups_handled(self._recovery_backups())
|
|
|
|
|
return config
|
|
|
|
|
return self._recover_from_backup(config) or config
|
|
|
|
|
|
|
|
|
|
def _recovery_backups(self) -> list[Path]:
|
|
|
|
|
"""Findet mögliche Konfigurationssicherungen in absteigender Zeitfolge."""
|
|
|
|
|
try:
|
|
|
|
|
backups = [
|
|
|
|
|
*self.path.parent.glob("config.defekt-*.json"),
|
|
|
|
|
*self.path.parent.glob("config.bak-*.json"),
|
|
|
|
|
]
|
|
|
|
|
except OSError:
|
|
|
|
|
return []
|
|
|
|
|
return sorted(
|
|
|
|
|
backups,
|
|
|
|
|
key=lambda path: path.stem.removeprefix("config.defekt-").removeprefix("config.bak-"),
|
|
|
|
|
reverse=True,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
def _latest_recovery_backup(self) -> tuple[Path, AppConfig] | None:
|
|
|
|
|
"""Liest die neueste syntaktisch gültige Sicherung."""
|
|
|
|
|
for backup_path in self._recovery_backups():
|
|
|
|
|
try:
|
|
|
|
|
raw = json.loads(backup_path.read_text(encoding="utf-8"))
|
|
|
|
|
if not isinstance(raw, dict):
|
|
|
|
|
continue
|
|
|
|
|
return backup_path, AppConfig.from_dict(raw)
|
|
|
|
|
except (OSError, UnicodeError, ValueError, TypeError, AttributeError, OverflowError):
|
|
|
|
|
continue
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
def _handled_backup_names(self) -> set[str]:
|
|
|
|
|
"""Lädt die Namen bereits geprüfter Sicherungen."""
|
|
|
|
|
try:
|
|
|
|
|
raw = json.loads(self._recovery_marker.read_text(encoding="utf-8"))
|
|
|
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
|
|
|
return set()
|
|
|
|
|
if not isinstance(raw, dict):
|
|
|
|
|
return set()
|
|
|
|
|
names = raw.get("handled_backups", [])
|
|
|
|
|
if not isinstance(names, list):
|
|
|
|
|
return set()
|
|
|
|
|
return {name for name in names if isinstance(name, str) and Path(name).name == name}
|
|
|
|
|
|
|
|
|
|
def _mark_backups_handled(self, backups: list[Path]) -> None:
|
|
|
|
|
"""Speichert, welche Sicherungen nicht erneut automatisch geladen werden sollen."""
|
|
|
|
|
if not backups:
|
|
|
|
|
return
|
|
|
|
|
handled_names = self._handled_backup_names()
|
|
|
|
|
handled_names.update(path.name for path in backups)
|
|
|
|
|
temporary = self._recovery_marker.with_suffix(".json.tmp")
|
|
|
|
|
try:
|
|
|
|
|
temporary.write_text(
|
|
|
|
|
json.dumps({"handled_backups": sorted(handled_names)}, ensure_ascii=False, indent=2),
|
|
|
|
|
encoding="utf-8",
|
|
|
|
|
)
|
|
|
|
|
os.replace(temporary, self._recovery_marker)
|
|
|
|
|
except OSError:
|
|
|
|
|
with suppress(OSError):
|
|
|
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
|
|
|
|
def _recover_from_backup(self, current_config: AppConfig | None) -> AppConfig | None:
|
|
|
|
|
"""Stellt Aufgaben aus der neuesten Sicherung wieder her, wenn die Liste leer ist."""
|
|
|
|
|
latest = self._latest_recovery_backup()
|
|
|
|
|
if latest is None:
|
|
|
|
|
return None
|
|
|
|
|
backup_path, backup_config = latest
|
|
|
|
|
handled_names = self._handled_backup_names()
|
|
|
|
|
if backup_path.name in handled_names:
|
|
|
|
|
return None
|
|
|
|
|
if not backup_config.tasks:
|
|
|
|
|
self._mark_backups_handled(self._recovery_backups())
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
if current_config is None:
|
|
|
|
|
recovered_config = backup_config
|
|
|
|
|
else:
|
|
|
|
|
snapshot_path = self.path.with_name(
|
|
|
|
|
f"config.vor-wiederherstellung-{datetime.now().strftime('%Y%m%d-%H%M%S-%f')}.json"
|
|
|
|
|
)
|
|
|
|
|
if self.path.exists():
|
|
|
|
|
try:
|
|
|
|
|
shutil.copy2(self.path, snapshot_path)
|
|
|
|
|
except OSError:
|
|
|
|
|
return None
|
|
|
|
|
recovered_config = AppConfig(tasks=backup_config.tasks, settings=current_config.settings)
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
self.save(recovered_config)
|
|
|
|
|
except ConfigError:
|
|
|
|
|
return recovered_config
|
|
|
|
|
self._mark_backups_handled(self._recovery_backups())
|
|
|
|
|
return recovered_config
|
|
|
|
|
|
|
|
|
|
def save(self, config: AppConfig) -> None:
|
|
|
|
|
"""Speichert die Konfiguration atomar und UTF-8-kodiert."""
|
|
|
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
|