feat: add backup system
This commit is contained in:
286
src/illogical/modules/backup_manager.py
Normal file
286
src/illogical/modules/backup_manager.py
Normal file
@@ -0,0 +1,286 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import hashlib
|
||||||
|
import json
|
||||||
|
import shutil
|
||||||
|
from datetime import UTC, datetime, timedelta
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
from logic_plugin_manager.defaults import tags_path
|
||||||
|
|
||||||
|
from illogical.modules.backup_models import (
|
||||||
|
BackupChanges,
|
||||||
|
BackupInfo,
|
||||||
|
BackupManifest,
|
||||||
|
BackupSettings,
|
||||||
|
BackupTrigger,
|
||||||
|
)
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
MANIFEST_VERSION = 1
|
||||||
|
MANIFEST_FILENAME = "manifest.json"
|
||||||
|
BACKUP_INDEX_FILENAME = ".backup_index.json"
|
||||||
|
|
||||||
|
TAGS_PATH = tags_path
|
||||||
|
BACKUPS_PATH = tags_path.parent / ".nothing_to_see_here_just_illogical_saving_ur_ass"
|
||||||
|
|
||||||
|
|
||||||
|
def _compute_file_checksum(file_path: Path) -> str:
|
||||||
|
sha256 = hashlib.sha256()
|
||||||
|
with file_path.open("rb") as f:
|
||||||
|
for chunk in iter(lambda: f.read(65536), b""):
|
||||||
|
sha256.update(chunk)
|
||||||
|
return f"sha256:{sha256.hexdigest()}"
|
||||||
|
|
||||||
|
|
||||||
|
def _get_backup_files() -> list[Path]:
|
||||||
|
if not TAGS_PATH.exists():
|
||||||
|
return []
|
||||||
|
return [f for f in TAGS_PATH.iterdir() if f.is_file()]
|
||||||
|
|
||||||
|
|
||||||
|
def _generate_backup_name(trigger: BackupTrigger) -> str:
|
||||||
|
timestamp = datetime.now(UTC).strftime("%Y-%m-%d_%H-%M-%S")
|
||||||
|
return f"{timestamp}_{trigger.value}"
|
||||||
|
|
||||||
|
|
||||||
|
def _load_manifest(backup_path: Path) -> BackupManifest | None:
|
||||||
|
manifest_file = backup_path / MANIFEST_FILENAME
|
||||||
|
if not manifest_file.exists():
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
with manifest_file.open() as f:
|
||||||
|
return BackupManifest.from_dict(json.load(f))
|
||||||
|
except (json.JSONDecodeError, KeyError, ValueError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _save_manifest(backup_path: Path, manifest: BackupManifest) -> None:
|
||||||
|
manifest_file = backup_path / MANIFEST_FILENAME
|
||||||
|
with manifest_file.open("w") as f:
|
||||||
|
json.dump(manifest.to_dict(), f, indent=2)
|
||||||
|
|
||||||
|
|
||||||
|
def _get_latest_backup() -> BackupInfo | None:
|
||||||
|
backups = list_backups()
|
||||||
|
return backups[0] if backups else None
|
||||||
|
|
||||||
|
|
||||||
|
def _compute_changes_between(
|
||||||
|
current_files: dict[str, str], previous_files: dict[str, str]
|
||||||
|
) -> BackupChanges:
|
||||||
|
added = [f for f in current_files if f not in previous_files]
|
||||||
|
deleted = [f for f in previous_files if f not in current_files]
|
||||||
|
modified = [
|
||||||
|
f
|
||||||
|
for f in current_files
|
||||||
|
if f in previous_files and current_files[f] != previous_files[f]
|
||||||
|
]
|
||||||
|
return BackupChanges(added=added, modified=modified, deleted=deleted)
|
||||||
|
|
||||||
|
|
||||||
|
def create_backup(
|
||||||
|
trigger: BackupTrigger = BackupTrigger.MANUAL, description: str = ""
|
||||||
|
) -> BackupInfo:
|
||||||
|
BACKUPS_PATH.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
backup_name = _generate_backup_name(trigger)
|
||||||
|
backup_path = BACKUPS_PATH / backup_name
|
||||||
|
backup_path.mkdir()
|
||||||
|
|
||||||
|
source_files = _get_backup_files()
|
||||||
|
checksums: dict[str, str] = {}
|
||||||
|
total_size = 0
|
||||||
|
|
||||||
|
for src_file in source_files:
|
||||||
|
dst_file = backup_path / src_file.name
|
||||||
|
shutil.copy2(src_file, dst_file)
|
||||||
|
checksums[src_file.name] = _compute_file_checksum(src_file)
|
||||||
|
total_size += src_file.stat().st_size
|
||||||
|
|
||||||
|
latest = _get_latest_backup()
|
||||||
|
previous_backup = latest.name if latest else None
|
||||||
|
|
||||||
|
if latest:
|
||||||
|
previous_manifest = _load_manifest(latest.path)
|
||||||
|
if previous_manifest:
|
||||||
|
changes = _compute_changes_between(checksums, previous_manifest.checksums)
|
||||||
|
else:
|
||||||
|
changes = BackupChanges()
|
||||||
|
else:
|
||||||
|
changes = BackupChanges(added=list(checksums.keys()))
|
||||||
|
|
||||||
|
manifest = BackupManifest(
|
||||||
|
version=MANIFEST_VERSION,
|
||||||
|
timestamp=datetime.now(UTC),
|
||||||
|
trigger=trigger,
|
||||||
|
description=description,
|
||||||
|
file_count=len(source_files),
|
||||||
|
total_size_bytes=total_size,
|
||||||
|
previous_backup=previous_backup,
|
||||||
|
changes=changes,
|
||||||
|
checksums=checksums,
|
||||||
|
)
|
||||||
|
_save_manifest(backup_path, manifest)
|
||||||
|
|
||||||
|
return BackupInfo(
|
||||||
|
name=backup_name,
|
||||||
|
path=backup_path,
|
||||||
|
timestamp=manifest.timestamp,
|
||||||
|
trigger=trigger,
|
||||||
|
file_count=manifest.file_count,
|
||||||
|
total_size_bytes=manifest.total_size_bytes,
|
||||||
|
description=description,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def list_backups() -> list[BackupInfo]:
|
||||||
|
if not BACKUPS_PATH.exists():
|
||||||
|
return []
|
||||||
|
|
||||||
|
backups: list[BackupInfo] = []
|
||||||
|
for backup_dir in BACKUPS_PATH.iterdir():
|
||||||
|
if not backup_dir.is_dir() or backup_dir.name.startswith("."):
|
||||||
|
continue
|
||||||
|
|
||||||
|
manifest = _load_manifest(backup_dir)
|
||||||
|
if manifest:
|
||||||
|
backups.append(
|
||||||
|
BackupInfo(
|
||||||
|
name=backup_dir.name,
|
||||||
|
path=backup_dir,
|
||||||
|
timestamp=manifest.timestamp,
|
||||||
|
trigger=manifest.trigger,
|
||||||
|
file_count=manifest.file_count,
|
||||||
|
total_size_bytes=manifest.total_size_bytes,
|
||||||
|
description=manifest.description,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
try:
|
||||||
|
parts = backup_dir.name.rsplit("_", 1)
|
||||||
|
timestamp_str = parts[0]
|
||||||
|
trigger_str = parts[1] if len(parts) > 1 else "manual"
|
||||||
|
timestamp = datetime.strptime(
|
||||||
|
timestamp_str, "%Y-%m-%d_%H-%M-%S"
|
||||||
|
).replace(tzinfo=UTC)
|
||||||
|
trigger = BackupTrigger(trigger_str)
|
||||||
|
except (ValueError, IndexError):
|
||||||
|
continue
|
||||||
|
|
||||||
|
files = list(backup_dir.glob("*"))
|
||||||
|
total_size = sum(f.stat().st_size for f in files if f.is_file())
|
||||||
|
backups.append(
|
||||||
|
BackupInfo(
|
||||||
|
name=backup_dir.name,
|
||||||
|
path=backup_dir,
|
||||||
|
timestamp=timestamp,
|
||||||
|
trigger=trigger,
|
||||||
|
file_count=len(files),
|
||||||
|
total_size_bytes=total_size,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
backups.sort(key=lambda b: b.timestamp, reverse=True)
|
||||||
|
return backups
|
||||||
|
|
||||||
|
|
||||||
|
def compute_changes(backup_name: str) -> BackupChanges:
|
||||||
|
backup_path = BACKUPS_PATH / backup_name
|
||||||
|
if not backup_path.exists():
|
||||||
|
return BackupChanges()
|
||||||
|
|
||||||
|
backup_manifest = _load_manifest(backup_path)
|
||||||
|
if not backup_manifest:
|
||||||
|
return BackupChanges()
|
||||||
|
|
||||||
|
current_files = _get_backup_files()
|
||||||
|
current_checksums = {f.name: _compute_file_checksum(f) for f in current_files}
|
||||||
|
backup_checksums = backup_manifest.checksums
|
||||||
|
|
||||||
|
added = [f for f in current_checksums if f not in backup_checksums]
|
||||||
|
deleted = [f for f in backup_checksums if f not in current_checksums]
|
||||||
|
modified = [
|
||||||
|
f
|
||||||
|
for f in current_checksums
|
||||||
|
if f in backup_checksums and current_checksums[f] != backup_checksums[f]
|
||||||
|
]
|
||||||
|
|
||||||
|
return BackupChanges(added=added, modified=modified, deleted=deleted)
|
||||||
|
|
||||||
|
|
||||||
|
def restore_backup(backup_name: str) -> bool:
|
||||||
|
backup_path = BACKUPS_PATH / backup_name
|
||||||
|
if not backup_path.exists():
|
||||||
|
return False
|
||||||
|
|
||||||
|
create_backup(BackupTrigger.AUTO, f"Auto-backup before restoring {backup_name}")
|
||||||
|
|
||||||
|
for existing_file in TAGS_PATH.iterdir():
|
||||||
|
if existing_file.is_file():
|
||||||
|
existing_file.unlink()
|
||||||
|
|
||||||
|
for backup_file in backup_path.iterdir():
|
||||||
|
if backup_file.is_file() and backup_file.name != MANIFEST_FILENAME:
|
||||||
|
shutil.copy2(backup_file, TAGS_PATH / backup_file.name)
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def purge_old_backups(settings: BackupSettings) -> int:
|
||||||
|
if not BACKUPS_PATH.exists():
|
||||||
|
return 0
|
||||||
|
|
||||||
|
backups = list_backups()
|
||||||
|
if not backups:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
to_delete: list[BackupInfo] = []
|
||||||
|
cutoff_date = datetime.now(UTC) - timedelta(days=settings.retention_days)
|
||||||
|
|
||||||
|
for i, backup in enumerate(backups):
|
||||||
|
if i >= settings.max_backups or (
|
||||||
|
settings.retention_days > 0 and backup.timestamp < cutoff_date
|
||||||
|
):
|
||||||
|
to_delete.append(backup)
|
||||||
|
|
||||||
|
for backup in to_delete:
|
||||||
|
shutil.rmtree(backup.path)
|
||||||
|
|
||||||
|
return len(to_delete)
|
||||||
|
|
||||||
|
|
||||||
|
def get_storage_usage() -> tuple[int, int]:
|
||||||
|
if not BACKUPS_PATH.exists():
|
||||||
|
return (0, 0)
|
||||||
|
|
||||||
|
total_bytes = 0
|
||||||
|
count = 0
|
||||||
|
|
||||||
|
for backup_dir in BACKUPS_PATH.iterdir():
|
||||||
|
if not backup_dir.is_dir() or backup_dir.name.startswith("."):
|
||||||
|
continue
|
||||||
|
count += 1
|
||||||
|
for file in backup_dir.iterdir():
|
||||||
|
if file.is_file():
|
||||||
|
total_bytes += file.stat().st_size
|
||||||
|
|
||||||
|
return (total_bytes, count)
|
||||||
|
|
||||||
|
|
||||||
|
def get_last_auto_backup_time() -> datetime | None:
|
||||||
|
backups = list_backups()
|
||||||
|
for backup in backups:
|
||||||
|
if backup.trigger == BackupTrigger.AUTO:
|
||||||
|
return backup.timestamp
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def should_create_auto_backup(min_interval_seconds: int = 300) -> bool:
|
||||||
|
last_auto = get_last_auto_backup_time()
|
||||||
|
if last_auto is None:
|
||||||
|
return True
|
||||||
|
elapsed = (datetime.now(UTC) - last_auto).total_seconds()
|
||||||
|
return elapsed >= min_interval_seconds
|
||||||
122
src/illogical/modules/backup_models.py
Normal file
122
src/illogical/modules/backup_models.py
Normal file
@@ -0,0 +1,122 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
from enum import Enum
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
class BackupTrigger(Enum):
|
||||||
|
MANUAL = "manual"
|
||||||
|
AUTO = "auto"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class BackupChanges:
|
||||||
|
added: list[str] = field(default_factory=list)
|
||||||
|
modified: list[str] = field(default_factory=list)
|
||||||
|
deleted: list[str] = field(default_factory=list)
|
||||||
|
|
||||||
|
def to_dict(self) -> dict[str, list[str]]:
|
||||||
|
return {"added": self.added, "modified": self.modified, "deleted": self.deleted}
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_dict(cls, data: dict[str, list[str]]) -> BackupChanges:
|
||||||
|
return cls(
|
||||||
|
added=data.get("added", []),
|
||||||
|
modified=data.get("modified", []),
|
||||||
|
deleted=data.get("deleted", []),
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_empty(self) -> bool:
|
||||||
|
return not self.added and not self.modified and not self.deleted
|
||||||
|
|
||||||
|
@property
|
||||||
|
def total_count(self) -> int:
|
||||||
|
return len(self.added) + len(self.modified) + len(self.deleted)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class BackupManifest:
|
||||||
|
version: int
|
||||||
|
timestamp: datetime
|
||||||
|
trigger: BackupTrigger
|
||||||
|
description: str
|
||||||
|
file_count: int
|
||||||
|
total_size_bytes: int
|
||||||
|
previous_backup: str | None
|
||||||
|
changes: BackupChanges
|
||||||
|
checksums: dict[str, str]
|
||||||
|
|
||||||
|
def to_dict(self) -> dict:
|
||||||
|
return {
|
||||||
|
"version": self.version,
|
||||||
|
"timestamp": self.timestamp.isoformat(),
|
||||||
|
"trigger": self.trigger.value,
|
||||||
|
"description": self.description,
|
||||||
|
"file_count": self.file_count,
|
||||||
|
"total_size_bytes": self.total_size_bytes,
|
||||||
|
"previous_backup": self.previous_backup,
|
||||||
|
"changes": self.changes.to_dict(),
|
||||||
|
"checksums": self.checksums,
|
||||||
|
}
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_dict(cls, data: dict) -> BackupManifest:
|
||||||
|
ts = datetime.fromisoformat(data["timestamp"])
|
||||||
|
if ts.tzinfo is None:
|
||||||
|
ts = ts.replace(tzinfo=UTC)
|
||||||
|
return cls(
|
||||||
|
version=data["version"],
|
||||||
|
timestamp=ts,
|
||||||
|
trigger=BackupTrigger(data["trigger"]),
|
||||||
|
description=data.get("description", ""),
|
||||||
|
file_count=data["file_count"],
|
||||||
|
total_size_bytes=data["total_size_bytes"],
|
||||||
|
previous_backup=data.get("previous_backup"),
|
||||||
|
changes=BackupChanges.from_dict(data.get("changes", {})),
|
||||||
|
checksums=data.get("checksums", {}),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class BackupInfo:
|
||||||
|
name: str
|
||||||
|
path: Path
|
||||||
|
timestamp: datetime
|
||||||
|
trigger: BackupTrigger
|
||||||
|
file_count: int
|
||||||
|
total_size_bytes: int
|
||||||
|
description: str = ""
|
||||||
|
|
||||||
|
@property
|
||||||
|
def display_name(self) -> str:
|
||||||
|
trigger_label = "Manual" if self.trigger == BackupTrigger.MANUAL else "Auto"
|
||||||
|
return f"{self.timestamp.strftime('%Y-%m-%d %H:%M:%S')} ({trigger_label})"
|
||||||
|
|
||||||
|
@property
|
||||||
|
def size_display(self) -> str:
|
||||||
|
return _format_size(self.total_size_bytes)
|
||||||
|
|
||||||
|
|
||||||
|
_BYTES_PER_KB = 1024
|
||||||
|
|
||||||
|
|
||||||
|
def _format_size(size_bytes: int) -> str:
|
||||||
|
size = float(size_bytes)
|
||||||
|
for unit in ("B", "KB", "MB", "GB"):
|
||||||
|
if size < _BYTES_PER_KB:
|
||||||
|
return f"{size:.1f} {unit}" if unit != "B" else f"{int(size)} {unit}"
|
||||||
|
size /= _BYTES_PER_KB
|
||||||
|
return f"{size:.1f} TB"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class BackupSettings:
|
||||||
|
retention_days: int = 30
|
||||||
|
max_backups: int = 100
|
||||||
|
auto_purge: bool = True
|
||||||
134
src/illogical/modules/backup_service.py
Normal file
134
src/illogical/modules/backup_service.py
Normal file
@@ -0,0 +1,134 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from PySide6.QtCore import QObject, QThread, Signal
|
||||||
|
|
||||||
|
from illogical.modules import backup_manager
|
||||||
|
from illogical.modules.backup_models import BackupSettings, BackupTrigger
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class BackupWorker(QObject):
|
||||||
|
backup_created = Signal(object)
|
||||||
|
backup_list_ready = Signal(list)
|
||||||
|
restore_completed = Signal(bool, str)
|
||||||
|
changes_computed = Signal(str, object)
|
||||||
|
storage_usage_ready = Signal(int, int)
|
||||||
|
purge_completed = Signal(int)
|
||||||
|
error_occurred = Signal(str)
|
||||||
|
|
||||||
|
def create_backup(
|
||||||
|
self, trigger: BackupTrigger = BackupTrigger.MANUAL, description: str = ""
|
||||||
|
) -> None:
|
||||||
|
try:
|
||||||
|
backup_info = backup_manager.create_backup(trigger, description)
|
||||||
|
self.backup_created.emit(backup_info)
|
||||||
|
except OSError as e:
|
||||||
|
logger.exception("Backup creation failed")
|
||||||
|
self.error_occurred.emit(str(e))
|
||||||
|
|
||||||
|
def list_backups(self) -> None:
|
||||||
|
try:
|
||||||
|
backups = backup_manager.list_backups()
|
||||||
|
self.backup_list_ready.emit(backups)
|
||||||
|
except OSError as e:
|
||||||
|
logger.exception("Listing backups failed")
|
||||||
|
self.error_occurred.emit(str(e))
|
||||||
|
|
||||||
|
def restore_backup(self, backup_name: str) -> None:
|
||||||
|
try:
|
||||||
|
success = backup_manager.restore_backup(backup_name)
|
||||||
|
self.restore_completed.emit(success, backup_name)
|
||||||
|
except OSError as e:
|
||||||
|
logger.exception("Restore failed")
|
||||||
|
self.error_occurred.emit(str(e))
|
||||||
|
|
||||||
|
def compute_changes(self, backup_name: str) -> None:
|
||||||
|
try:
|
||||||
|
changes = backup_manager.compute_changes(backup_name)
|
||||||
|
self.changes_computed.emit(backup_name, changes)
|
||||||
|
except OSError as e:
|
||||||
|
logger.exception("Computing changes failed")
|
||||||
|
self.error_occurred.emit(str(e))
|
||||||
|
|
||||||
|
def get_storage_usage(self) -> None:
|
||||||
|
try:
|
||||||
|
total_bytes, count = backup_manager.get_storage_usage()
|
||||||
|
self.storage_usage_ready.emit(total_bytes, count)
|
||||||
|
except OSError as e:
|
||||||
|
logger.exception("Getting storage usage failed")
|
||||||
|
self.error_occurred.emit(str(e))
|
||||||
|
|
||||||
|
def purge_old_backups(self, settings: BackupSettings) -> None:
|
||||||
|
try:
|
||||||
|
deleted_count = backup_manager.purge_old_backups(settings)
|
||||||
|
self.purge_completed.emit(deleted_count)
|
||||||
|
except OSError as e:
|
||||||
|
logger.exception("Purging backups failed")
|
||||||
|
self.error_occurred.emit(str(e))
|
||||||
|
|
||||||
|
def ensure_backup_before_change(self) -> None:
|
||||||
|
try:
|
||||||
|
if backup_manager.should_create_auto_backup():
|
||||||
|
backup_info = backup_manager.create_backup(
|
||||||
|
BackupTrigger.AUTO, "Auto-backup before plugin modification"
|
||||||
|
)
|
||||||
|
self.backup_created.emit(backup_info)
|
||||||
|
except OSError as e:
|
||||||
|
logger.exception("Auto-backup before change failed")
|
||||||
|
self.error_occurred.emit(str(e))
|
||||||
|
|
||||||
|
|
||||||
|
class BackupService(QObject):
|
||||||
|
backup_created = Signal(object)
|
||||||
|
backup_list_ready = Signal(list)
|
||||||
|
restore_completed = Signal(bool, str)
|
||||||
|
changes_computed = Signal(str, object)
|
||||||
|
storage_usage_ready = Signal(int, int)
|
||||||
|
purge_completed = Signal(int)
|
||||||
|
error_occurred = Signal(str)
|
||||||
|
|
||||||
|
def __init__(self, parent: QObject | None = None) -> None:
|
||||||
|
super().__init__(parent)
|
||||||
|
self._thread = QThread()
|
||||||
|
self._worker = BackupWorker()
|
||||||
|
self._worker.moveToThread(self._thread)
|
||||||
|
|
||||||
|
self._worker.backup_created.connect(self.backup_created)
|
||||||
|
self._worker.backup_list_ready.connect(self.backup_list_ready)
|
||||||
|
self._worker.restore_completed.connect(self.restore_completed)
|
||||||
|
self._worker.changes_computed.connect(self.changes_computed)
|
||||||
|
self._worker.storage_usage_ready.connect(self.storage_usage_ready)
|
||||||
|
self._worker.purge_completed.connect(self.purge_completed)
|
||||||
|
self._worker.error_occurred.connect(self.error_occurred)
|
||||||
|
|
||||||
|
self._thread.start()
|
||||||
|
|
||||||
|
def create_backup(
|
||||||
|
self, trigger: BackupTrigger = BackupTrigger.MANUAL, description: str = ""
|
||||||
|
) -> None:
|
||||||
|
self._worker.create_backup(trigger, description)
|
||||||
|
|
||||||
|
def list_backups(self) -> None:
|
||||||
|
self._worker.list_backups()
|
||||||
|
|
||||||
|
def restore_backup(self, backup_name: str) -> None:
|
||||||
|
self._worker.restore_backup(backup_name)
|
||||||
|
|
||||||
|
def compute_changes(self, backup_name: str) -> None:
|
||||||
|
self._worker.compute_changes(backup_name)
|
||||||
|
|
||||||
|
def get_storage_usage(self) -> None:
|
||||||
|
self._worker.get_storage_usage()
|
||||||
|
|
||||||
|
def purge_old_backups(self, settings: BackupSettings) -> None:
|
||||||
|
self._worker.purge_old_backups(settings)
|
||||||
|
|
||||||
|
def ensure_backup_before_change(self) -> None:
|
||||||
|
self._worker.ensure_backup_before_change()
|
||||||
|
|
||||||
|
def shutdown(self) -> None:
|
||||||
|
self._thread.quit()
|
||||||
|
self._thread.wait()
|
||||||
54
src/illogical/modules/settings.py
Normal file
54
src/illogical/modules/settings.py
Normal file
@@ -0,0 +1,54 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from PySide6.QtCore import QSettings
|
||||||
|
|
||||||
|
from illogical.modules.backup_models import BackupSettings
|
||||||
|
|
||||||
|
_BACKUP_RETENTION_DAYS = "backup/retention_days"
|
||||||
|
_BACKUP_MAX_COUNT = "backup/max_count"
|
||||||
|
_BACKUP_AUTO_PURGE = "backup/auto_purge"
|
||||||
|
|
||||||
|
|
||||||
|
class Settings:
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self._settings = QSettings("com.kotikot.illogical", "illogical")
|
||||||
|
|
||||||
|
@property
|
||||||
|
def backup_retention_days(self) -> int:
|
||||||
|
value = self._settings.value(_BACKUP_RETENTION_DAYS, 30)
|
||||||
|
return int(str(value)) if value is not None else 30
|
||||||
|
|
||||||
|
@backup_retention_days.setter
|
||||||
|
def backup_retention_days(self, value: int) -> None:
|
||||||
|
self._settings.setValue(_BACKUP_RETENTION_DAYS, value)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def backup_max_count(self) -> int:
|
||||||
|
value = self._settings.value(_BACKUP_MAX_COUNT, 100)
|
||||||
|
return int(str(value)) if value is not None else 100
|
||||||
|
|
||||||
|
@backup_max_count.setter
|
||||||
|
def backup_max_count(self, value: int) -> None:
|
||||||
|
self._settings.setValue(_BACKUP_MAX_COUNT, value)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def backup_auto_purge(self) -> bool:
|
||||||
|
default_value = True
|
||||||
|
value = self._settings.value(_BACKUP_AUTO_PURGE, default_value, type=bool) # type: ignore[call-overload]
|
||||||
|
return bool(value)
|
||||||
|
|
||||||
|
@backup_auto_purge.setter
|
||||||
|
def backup_auto_purge(self, value: bool) -> None:
|
||||||
|
self._settings.setValue(_BACKUP_AUTO_PURGE, value)
|
||||||
|
|
||||||
|
def get_backup_settings(self) -> BackupSettings:
|
||||||
|
return BackupSettings(
|
||||||
|
retention_days=self.backup_retention_days,
|
||||||
|
max_backups=self.backup_max_count,
|
||||||
|
auto_purge=self.backup_auto_purge,
|
||||||
|
)
|
||||||
|
|
||||||
|
def save_backup_settings(self, settings: BackupSettings) -> None:
|
||||||
|
self.backup_retention_days = settings.retention_days
|
||||||
|
self.backup_max_count = settings.max_backups
|
||||||
|
self.backup_auto_purge = settings.auto_purge
|
||||||
133
src/illogical/ui/backup_settings_window.py
Normal file
133
src/illogical/ui/backup_settings_window.py
Normal file
@@ -0,0 +1,133 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from PySide6.QtCore import Qt, Signal
|
||||||
|
from PySide6.QtWidgets import (
|
||||||
|
QCheckBox,
|
||||||
|
QComboBox,
|
||||||
|
QDialog,
|
||||||
|
QGroupBox,
|
||||||
|
QHBoxLayout,
|
||||||
|
QLabel,
|
||||||
|
QPushButton,
|
||||||
|
QSpinBox,
|
||||||
|
QVBoxLayout,
|
||||||
|
QWidget,
|
||||||
|
)
|
||||||
|
|
||||||
|
from illogical.modules.backup_models import BackupSettings, _format_size
|
||||||
|
|
||||||
|
RETENTION_OPTIONS = [(7, "7 days"), (30, "30 days"), (90, "90 days"), (0, "Forever")]
|
||||||
|
|
||||||
|
|
||||||
|
class BackupSettingsWindow(QDialog):
|
||||||
|
settings_saved = Signal(object)
|
||||||
|
purge_requested = Signal()
|
||||||
|
|
||||||
|
def __init__(self, settings: BackupSettings, parent: QWidget | None = None) -> None:
|
||||||
|
super().__init__(parent)
|
||||||
|
self._settings = settings
|
||||||
|
self._setup_ui()
|
||||||
|
self._load_settings()
|
||||||
|
|
||||||
|
def _setup_ui(self) -> None:
|
||||||
|
self.setWindowTitle("Backup Settings")
|
||||||
|
self.setFixedSize(400, 300)
|
||||||
|
self.setWindowModality(Qt.WindowModality.WindowModal)
|
||||||
|
|
||||||
|
layout = QVBoxLayout(self)
|
||||||
|
layout.setSpacing(16)
|
||||||
|
layout.setContentsMargins(20, 20, 20, 20)
|
||||||
|
|
||||||
|
layout.addWidget(self._create_retention_group())
|
||||||
|
layout.addWidget(self._create_storage_group())
|
||||||
|
layout.addStretch()
|
||||||
|
layout.addWidget(self._create_buttons())
|
||||||
|
|
||||||
|
def _create_retention_group(self) -> QGroupBox:
|
||||||
|
group = QGroupBox("Retention Policy")
|
||||||
|
layout = QVBoxLayout(group)
|
||||||
|
layout.setSpacing(12)
|
||||||
|
|
||||||
|
retention_row = QHBoxLayout()
|
||||||
|
retention_row.addWidget(QLabel("Keep backups for:"))
|
||||||
|
self._retention_combo = QComboBox()
|
||||||
|
for _, label in RETENTION_OPTIONS:
|
||||||
|
self._retention_combo.addItem(label)
|
||||||
|
retention_row.addWidget(self._retention_combo)
|
||||||
|
retention_row.addStretch()
|
||||||
|
layout.addLayout(retention_row)
|
||||||
|
|
||||||
|
max_row = QHBoxLayout()
|
||||||
|
max_row.addWidget(QLabel("Maximum backups:"))
|
||||||
|
self._max_spin = QSpinBox()
|
||||||
|
self._max_spin.setRange(5, 500)
|
||||||
|
self._max_spin.setValue(100)
|
||||||
|
max_row.addWidget(self._max_spin)
|
||||||
|
max_row.addStretch()
|
||||||
|
layout.addLayout(max_row)
|
||||||
|
|
||||||
|
self._auto_purge_check = QCheckBox("Automatically purge old backups")
|
||||||
|
layout.addWidget(self._auto_purge_check)
|
||||||
|
|
||||||
|
return group
|
||||||
|
|
||||||
|
def _create_storage_group(self) -> QGroupBox:
|
||||||
|
group = QGroupBox("Storage")
|
||||||
|
layout = QVBoxLayout(group)
|
||||||
|
layout.setSpacing(12)
|
||||||
|
|
||||||
|
self._storage_label = QLabel("Loading...")
|
||||||
|
layout.addWidget(self._storage_label)
|
||||||
|
|
||||||
|
self._purge_button = QPushButton("Purge Old Backups Now")
|
||||||
|
self._purge_button.clicked.connect(self._on_purge_clicked)
|
||||||
|
layout.addWidget(self._purge_button)
|
||||||
|
|
||||||
|
return group
|
||||||
|
|
||||||
|
def _create_buttons(self) -> QWidget:
|
||||||
|
container = QWidget()
|
||||||
|
layout = QHBoxLayout(container)
|
||||||
|
layout.setContentsMargins(0, 0, 0, 0)
|
||||||
|
|
||||||
|
layout.addStretch()
|
||||||
|
|
||||||
|
cancel_btn = QPushButton("Cancel")
|
||||||
|
cancel_btn.clicked.connect(self.reject)
|
||||||
|
layout.addWidget(cancel_btn)
|
||||||
|
|
||||||
|
save_btn = QPushButton("Save")
|
||||||
|
save_btn.setDefault(True)
|
||||||
|
save_btn.clicked.connect(self._on_save)
|
||||||
|
layout.addWidget(save_btn)
|
||||||
|
|
||||||
|
return container
|
||||||
|
|
||||||
|
def _load_settings(self) -> None:
|
||||||
|
retention_days = self._settings.retention_days
|
||||||
|
for i, (days, _) in enumerate(RETENTION_OPTIONS):
|
||||||
|
if days == retention_days:
|
||||||
|
self._retention_combo.setCurrentIndex(i)
|
||||||
|
break
|
||||||
|
|
||||||
|
self._max_spin.setValue(self._settings.max_backups)
|
||||||
|
self._auto_purge_check.setChecked(self._settings.auto_purge)
|
||||||
|
|
||||||
|
def _on_save(self) -> None:
|
||||||
|
idx = self._retention_combo.currentIndex()
|
||||||
|
retention_days = RETENTION_OPTIONS[idx][0]
|
||||||
|
|
||||||
|
settings = BackupSettings(
|
||||||
|
retention_days=retention_days,
|
||||||
|
max_backups=self._max_spin.value(),
|
||||||
|
auto_purge=self._auto_purge_check.isChecked(),
|
||||||
|
)
|
||||||
|
self.settings_saved.emit(settings)
|
||||||
|
self.accept()
|
||||||
|
|
||||||
|
def _on_purge_clicked(self) -> None:
|
||||||
|
self.purge_requested.emit()
|
||||||
|
|
||||||
|
def update_storage_info(self, total_bytes: int, count: int) -> None:
|
||||||
|
size_display = _format_size(total_bytes)
|
||||||
|
self._storage_label.setText(f"{count} backups using {size_display}")
|
||||||
@@ -4,17 +4,28 @@ from typing import TYPE_CHECKING
|
|||||||
|
|
||||||
import pyqt_liquidglass as glass
|
import pyqt_liquidglass as glass
|
||||||
from PySide6.QtCore import Qt, QTimer
|
from PySide6.QtCore import Qt, QTimer
|
||||||
from PySide6.QtWidgets import QHBoxLayout, QMainWindow, QSplitter, QWidget
|
from PySide6.QtWidgets import QHBoxLayout, QMainWindow, QMessageBox, QSplitter, QWidget
|
||||||
|
|
||||||
|
from illogical.modules.backup_service import BackupService
|
||||||
from illogical.modules.plugin_service import PluginService
|
from illogical.modules.plugin_service import PluginService
|
||||||
|
from illogical.modules.settings import Settings
|
||||||
|
from illogical.ui.backup_settings_window import BackupSettingsWindow
|
||||||
from illogical.ui.loading_overlay import LoadingOverlay
|
from illogical.ui.loading_overlay import LoadingOverlay
|
||||||
|
from illogical.ui.menu_bar import MenuBar
|
||||||
from illogical.ui.plugin_table import PluginTableView
|
from illogical.ui.plugin_table import PluginTableView
|
||||||
|
from illogical.ui.restore_backup_window import RestoreBackupWindow
|
||||||
from illogical.ui.sidebar import Sidebar
|
from illogical.ui.sidebar import Sidebar
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from logic_plugin_manager import Logic, SearchResult
|
from logic_plugin_manager import Logic, SearchResult
|
||||||
from PySide6.QtGui import QCloseEvent, QKeyEvent, QShowEvent
|
from PySide6.QtGui import QCloseEvent, QKeyEvent, QShowEvent
|
||||||
|
|
||||||
|
from illogical.modules.backup_models import (
|
||||||
|
BackupChanges,
|
||||||
|
BackupInfo,
|
||||||
|
BackupSettings,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class MainWindow(QMainWindow):
|
class MainWindow(QMainWindow):
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
@@ -25,12 +36,22 @@ class MainWindow(QMainWindow):
|
|||||||
|
|
||||||
self._logic: Logic | None = None
|
self._logic: Logic | None = None
|
||||||
self._glass_applied = False
|
self._glass_applied = False
|
||||||
|
self._settings = Settings()
|
||||||
|
|
||||||
self._setup_ui()
|
self._setup_ui()
|
||||||
self._setup_service()
|
self._setup_service()
|
||||||
|
self._setup_backup_service()
|
||||||
|
self._setup_menu_bar()
|
||||||
|
|
||||||
glass.prepare_window_for_glass(self)
|
glass.prepare_window_for_glass(self)
|
||||||
|
|
||||||
|
def _setup_menu_bar(self) -> None:
|
||||||
|
self._menu_bar = MenuBar(self)
|
||||||
|
|
||||||
|
self._menu_bar.backup_now_triggered.connect(self._on_backup_now)
|
||||||
|
self._menu_bar.restore_backup_triggered.connect(self._on_restore_backup)
|
||||||
|
self._menu_bar.backup_settings_triggered.connect(self._on_backup_settings)
|
||||||
|
|
||||||
def _setup_ui(self) -> None:
|
def _setup_ui(self) -> None:
|
||||||
self._central = QWidget()
|
self._central = QWidget()
|
||||||
layout = QHBoxLayout(self._central)
|
layout = QHBoxLayout(self._central)
|
||||||
@@ -71,6 +92,19 @@ class MainWindow(QMainWindow):
|
|||||||
self._service.search_results.connect(self._on_search_results)
|
self._service.search_results.connect(self._on_search_results)
|
||||||
self._service.error_occurred.connect(self._on_error)
|
self._service.error_occurred.connect(self._on_error)
|
||||||
|
|
||||||
|
def _setup_backup_service(self) -> None:
|
||||||
|
self._backup_service = BackupService(self)
|
||||||
|
self._backup_service.backup_created.connect(self._on_backup_created)
|
||||||
|
self._backup_service.backup_list_ready.connect(self._on_backup_list_ready)
|
||||||
|
self._backup_service.restore_completed.connect(self._on_restore_completed)
|
||||||
|
self._backup_service.changes_computed.connect(self._on_changes_computed)
|
||||||
|
self._backup_service.storage_usage_ready.connect(self._on_storage_usage_ready)
|
||||||
|
self._backup_service.purge_completed.connect(self._on_purge_completed)
|
||||||
|
self._backup_service.error_occurred.connect(self._on_backup_error)
|
||||||
|
|
||||||
|
self._restore_window: RestoreBackupWindow | None = None
|
||||||
|
self._settings_window: BackupSettingsWindow | None = None
|
||||||
|
|
||||||
def showEvent(self, event: QShowEvent) -> None: # noqa: N802
|
def showEvent(self, event: QShowEvent) -> None: # noqa: N802
|
||||||
super().showEvent(event)
|
super().showEvent(event)
|
||||||
if not self._glass_applied:
|
if not self._glass_applied:
|
||||||
@@ -125,6 +159,83 @@ class MainWindow(QMainWindow):
|
|||||||
def _on_error(self, message: str) -> None:
|
def _on_error(self, message: str) -> None:
|
||||||
self._loading_overlay.set_message(f"Error: {message}")
|
self._loading_overlay.set_message(f"Error: {message}")
|
||||||
|
|
||||||
|
def _on_backup_now(self) -> None:
|
||||||
|
self._backup_service.create_backup()
|
||||||
|
|
||||||
|
def _on_restore_backup(self) -> None:
|
||||||
|
self._restore_window = RestoreBackupWindow(self)
|
||||||
|
self._restore_window.backup_selected.connect(self._on_restore_backup_selected)
|
||||||
|
self._restore_window.restore_requested.connect(self._on_restore_requested)
|
||||||
|
self._backup_service.list_backups()
|
||||||
|
self._restore_window.show()
|
||||||
|
|
||||||
|
def _on_backup_settings(self) -> None:
|
||||||
|
settings = self._settings.get_backup_settings()
|
||||||
|
self._settings_window = BackupSettingsWindow(settings, self)
|
||||||
|
self._settings_window.settings_saved.connect(self._on_settings_saved)
|
||||||
|
self._settings_window.purge_requested.connect(self._on_purge_requested)
|
||||||
|
self._backup_service.get_storage_usage()
|
||||||
|
self._settings_window.show()
|
||||||
|
|
||||||
|
def _on_backup_created(self, backup_info: BackupInfo) -> None:
|
||||||
|
QMessageBox.information(
|
||||||
|
self,
|
||||||
|
"Backup Created",
|
||||||
|
f"Backup created successfully.\n\n"
|
||||||
|
f"Files: {backup_info.file_count}\n"
|
||||||
|
f"Size: {backup_info.size_display}",
|
||||||
|
)
|
||||||
|
|
||||||
|
def _on_backup_list_ready(self, backups: list[BackupInfo]) -> None:
|
||||||
|
if self._restore_window:
|
||||||
|
self._restore_window.set_backups(backups)
|
||||||
|
|
||||||
|
def _on_restore_backup_selected(self, backup_name: str) -> None:
|
||||||
|
self._backup_service.compute_changes(backup_name)
|
||||||
|
|
||||||
|
def _on_changes_computed(self, backup_name: str, changes: BackupChanges) -> None:
|
||||||
|
if self._restore_window:
|
||||||
|
self._restore_window.set_changes(backup_name, changes)
|
||||||
|
|
||||||
|
def _on_restore_requested(self, backup_name: str) -> None:
|
||||||
|
self._backup_service.restore_backup(backup_name)
|
||||||
|
|
||||||
|
def _on_restore_completed(self, success: bool, backup_name: str) -> None: # noqa: FBT001
|
||||||
|
if success:
|
||||||
|
QMessageBox.information(
|
||||||
|
self,
|
||||||
|
"Restore Complete",
|
||||||
|
f"Backup '{backup_name}' has been restored.\n\n"
|
||||||
|
"Please restart the application to see the changes.",
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
QMessageBox.warning(
|
||||||
|
self, "Restore Failed", f"Failed to restore backup '{backup_name}'."
|
||||||
|
)
|
||||||
|
|
||||||
|
def _on_storage_usage_ready(self, total_bytes: int, count: int) -> None:
|
||||||
|
if self._settings_window:
|
||||||
|
self._settings_window.update_storage_info(total_bytes, count)
|
||||||
|
|
||||||
|
def _on_settings_saved(self, settings: BackupSettings) -> None:
|
||||||
|
self._settings.save_backup_settings(settings)
|
||||||
|
if settings.auto_purge:
|
||||||
|
self._backup_service.purge_old_backups(settings)
|
||||||
|
|
||||||
|
def _on_purge_requested(self) -> None:
|
||||||
|
settings = self._settings.get_backup_settings()
|
||||||
|
self._backup_service.purge_old_backups(settings)
|
||||||
|
|
||||||
|
def _on_purge_completed(self, deleted_count: int) -> None:
|
||||||
|
self._backup_service.get_storage_usage()
|
||||||
|
if deleted_count > 0:
|
||||||
|
QMessageBox.information(
|
||||||
|
self, "Purge Complete", f"Deleted {deleted_count} old backup(s)."
|
||||||
|
)
|
||||||
|
|
||||||
|
def _on_backup_error(self, message: str) -> None:
|
||||||
|
QMessageBox.warning(self, "Backup Error", message)
|
||||||
|
|
||||||
def keyPressEvent(self, event: QKeyEvent) -> None: # noqa: N802
|
def keyPressEvent(self, event: QKeyEvent) -> None: # noqa: N802
|
||||||
mods = event.modifiers()
|
mods = event.modifiers()
|
||||||
key = event.key()
|
key = event.key()
|
||||||
@@ -175,4 +286,5 @@ class MainWindow(QMainWindow):
|
|||||||
|
|
||||||
def closeEvent(self, event: QCloseEvent) -> None: # noqa: N802
|
def closeEvent(self, event: QCloseEvent) -> None: # noqa: N802
|
||||||
self._service.shutdown()
|
self._service.shutdown()
|
||||||
|
self._backup_service.shutdown()
|
||||||
super().closeEvent(event)
|
super().closeEvent(event)
|
||||||
|
|||||||
93
src/illogical/ui/menu_bar.py
Normal file
93
src/illogical/ui/menu_bar.py
Normal file
@@ -0,0 +1,93 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
from PySide6.QtCore import Signal
|
||||||
|
from PySide6.QtGui import QAction, QKeySequence
|
||||||
|
from PySide6.QtWidgets import QMenuBar
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from PySide6.QtWidgets import QMainWindow
|
||||||
|
|
||||||
|
|
||||||
|
class MenuBar(QMenuBar):
|
||||||
|
backup_now_triggered = Signal()
|
||||||
|
restore_backup_triggered = Signal()
|
||||||
|
backup_settings_triggered = Signal()
|
||||||
|
|
||||||
|
def __init__(self, main_window: QMainWindow | None = None) -> None:
|
||||||
|
super().__init__()
|
||||||
|
self._main_window = main_window
|
||||||
|
self.setNativeMenuBar(True)
|
||||||
|
self._setup_menus()
|
||||||
|
|
||||||
|
def _setup_menus(self) -> None:
|
||||||
|
self._setup_file_menu()
|
||||||
|
self._setup_edit_menu()
|
||||||
|
self._setup_backup_menu()
|
||||||
|
self._setup_help_menu()
|
||||||
|
|
||||||
|
def _setup_file_menu(self) -> None:
|
||||||
|
file_menu = self.addMenu("File")
|
||||||
|
|
||||||
|
close_action = QAction("Close Window", self)
|
||||||
|
close_action.setShortcut(QKeySequence.StandardKey.Close)
|
||||||
|
if self._main_window is not None:
|
||||||
|
close_action.triggered.connect(self._main_window.close)
|
||||||
|
file_menu.addAction(close_action)
|
||||||
|
|
||||||
|
def _setup_edit_menu(self) -> None:
|
||||||
|
edit_menu = self.addMenu("Edit")
|
||||||
|
|
||||||
|
undo_action = QAction("Undo", self)
|
||||||
|
undo_action.setShortcut(QKeySequence.StandardKey.Undo)
|
||||||
|
undo_action.setEnabled(False)
|
||||||
|
edit_menu.addAction(undo_action)
|
||||||
|
|
||||||
|
redo_action = QAction("Redo", self)
|
||||||
|
redo_action.setShortcut(QKeySequence.StandardKey.Redo)
|
||||||
|
redo_action.setEnabled(False)
|
||||||
|
edit_menu.addAction(redo_action)
|
||||||
|
|
||||||
|
edit_menu.addSeparator()
|
||||||
|
|
||||||
|
cut_action = QAction("Cut", self)
|
||||||
|
cut_action.setShortcut(QKeySequence.StandardKey.Cut)
|
||||||
|
edit_menu.addAction(cut_action)
|
||||||
|
|
||||||
|
copy_action = QAction("Copy", self)
|
||||||
|
copy_action.setShortcut(QKeySequence.StandardKey.Copy)
|
||||||
|
edit_menu.addAction(copy_action)
|
||||||
|
|
||||||
|
paste_action = QAction("Paste", self)
|
||||||
|
paste_action.setShortcut(QKeySequence.StandardKey.Paste)
|
||||||
|
edit_menu.addAction(paste_action)
|
||||||
|
|
||||||
|
select_all_action = QAction("Select All", self)
|
||||||
|
select_all_action.setShortcut(QKeySequence.StandardKey.SelectAll)
|
||||||
|
edit_menu.addAction(select_all_action)
|
||||||
|
|
||||||
|
def _setup_backup_menu(self) -> None:
|
||||||
|
backup_menu = self.addMenu("Backup")
|
||||||
|
|
||||||
|
backup_now_action = QAction("Backup Now", self)
|
||||||
|
backup_now_action.setShortcut(QKeySequence("Ctrl+B"))
|
||||||
|
backup_now_action.triggered.connect(self.backup_now_triggered)
|
||||||
|
backup_menu.addAction(backup_now_action)
|
||||||
|
|
||||||
|
backup_menu.addSeparator()
|
||||||
|
|
||||||
|
restore_action = QAction("Restore Backup...", self)
|
||||||
|
restore_action.setShortcut(QKeySequence("Ctrl+Shift+R"))
|
||||||
|
restore_action.triggered.connect(self.restore_backup_triggered)
|
||||||
|
backup_menu.addAction(restore_action)
|
||||||
|
|
||||||
|
backup_menu.addSeparator()
|
||||||
|
|
||||||
|
settings_action = QAction("Backup Settings...", self)
|
||||||
|
settings_action.setShortcut(QKeySequence.StandardKey.Preferences)
|
||||||
|
settings_action.triggered.connect(self.backup_settings_triggered)
|
||||||
|
backup_menu.addAction(settings_action)
|
||||||
|
|
||||||
|
def _setup_help_menu(self) -> None:
|
||||||
|
self.addMenu("Help")
|
||||||
204
src/illogical/ui/restore_backup_window.py
Normal file
204
src/illogical/ui/restore_backup_window.py
Normal file
@@ -0,0 +1,204 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from PySide6.QtCore import Qt, Signal
|
||||||
|
from PySide6.QtWidgets import (
|
||||||
|
QDialog,
|
||||||
|
QHBoxLayout,
|
||||||
|
QLabel,
|
||||||
|
QListWidget,
|
||||||
|
QListWidgetItem,
|
||||||
|
QMessageBox,
|
||||||
|
QPushButton,
|
||||||
|
QSplitter,
|
||||||
|
QTreeWidget,
|
||||||
|
QTreeWidgetItem,
|
||||||
|
QVBoxLayout,
|
||||||
|
QWidget,
|
||||||
|
)
|
||||||
|
|
||||||
|
from illogical.modules.backup_models import BackupChanges, BackupInfo, BackupTrigger
|
||||||
|
from illogical.modules.sf_symbols import sf_symbol
|
||||||
|
|
||||||
|
|
||||||
|
class RestoreBackupWindow(QDialog):
|
||||||
|
restore_requested = Signal(str)
|
||||||
|
backup_selected = Signal(str)
|
||||||
|
|
||||||
|
def __init__(self, parent: QWidget | None = None) -> None:
|
||||||
|
super().__init__(parent)
|
||||||
|
self._backups: list[BackupInfo] = []
|
||||||
|
self._changes_cache: dict[str, BackupChanges] = {}
|
||||||
|
self._setup_ui()
|
||||||
|
|
||||||
|
def _setup_ui(self) -> None:
|
||||||
|
self.setWindowTitle("Restore Backup")
|
||||||
|
self.resize(700, 500)
|
||||||
|
self.setWindowModality(Qt.WindowModality.WindowModal)
|
||||||
|
|
||||||
|
layout = QVBoxLayout(self)
|
||||||
|
layout.setSpacing(12)
|
||||||
|
layout.setContentsMargins(16, 16, 16, 16)
|
||||||
|
|
||||||
|
splitter = QSplitter(Qt.Orientation.Horizontal)
|
||||||
|
splitter.addWidget(self._create_backup_list())
|
||||||
|
splitter.addWidget(self._create_details_panel())
|
||||||
|
splitter.setStretchFactor(0, 1)
|
||||||
|
splitter.setStretchFactor(1, 2)
|
||||||
|
layout.addWidget(splitter, 1)
|
||||||
|
|
||||||
|
layout.addWidget(self._create_buttons())
|
||||||
|
|
||||||
|
def _create_backup_list(self) -> QWidget:
|
||||||
|
container = QWidget()
|
||||||
|
layout = QVBoxLayout(container)
|
||||||
|
layout.setContentsMargins(0, 0, 0, 0)
|
||||||
|
layout.setSpacing(8)
|
||||||
|
|
||||||
|
layout.addWidget(QLabel("Available Backups"))
|
||||||
|
|
||||||
|
self._backup_list = QListWidget()
|
||||||
|
self._backup_list.currentItemChanged.connect(self._on_backup_selected)
|
||||||
|
layout.addWidget(self._backup_list)
|
||||||
|
|
||||||
|
return container
|
||||||
|
|
||||||
|
def _create_details_panel(self) -> QWidget:
|
||||||
|
container = QWidget()
|
||||||
|
layout = QVBoxLayout(container)
|
||||||
|
layout.setContentsMargins(0, 0, 0, 0)
|
||||||
|
layout.setSpacing(8)
|
||||||
|
|
||||||
|
layout.addWidget(QLabel("Details"))
|
||||||
|
|
||||||
|
self._details_label = QLabel("Select a backup to view details")
|
||||||
|
self._details_label.setWordWrap(True)
|
||||||
|
layout.addWidget(self._details_label)
|
||||||
|
|
||||||
|
layout.addWidget(QLabel("Changes if restored:"))
|
||||||
|
|
||||||
|
self._changes_tree = QTreeWidget()
|
||||||
|
self._changes_tree.setHeaderHidden(True)
|
||||||
|
self._changes_tree.setIndentation(16)
|
||||||
|
layout.addWidget(self._changes_tree, 1)
|
||||||
|
|
||||||
|
return container
|
||||||
|
|
||||||
|
def _create_buttons(self) -> QWidget:
|
||||||
|
container = QWidget()
|
||||||
|
layout = QHBoxLayout(container)
|
||||||
|
layout.setContentsMargins(0, 0, 0, 0)
|
||||||
|
|
||||||
|
layout.addStretch()
|
||||||
|
|
||||||
|
cancel_btn = QPushButton("Cancel")
|
||||||
|
cancel_btn.clicked.connect(self.reject)
|
||||||
|
layout.addWidget(cancel_btn)
|
||||||
|
|
||||||
|
self._restore_btn = QPushButton("Restore")
|
||||||
|
self._restore_btn.setEnabled(False)
|
||||||
|
self._restore_btn.clicked.connect(self._on_restore_clicked)
|
||||||
|
layout.addWidget(self._restore_btn)
|
||||||
|
|
||||||
|
return container
|
||||||
|
|
||||||
|
def set_backups(self, backups: list[BackupInfo]) -> None:
|
||||||
|
self._backups = backups
|
||||||
|
self._backup_list.clear()
|
||||||
|
|
||||||
|
for backup in backups:
|
||||||
|
item = QListWidgetItem()
|
||||||
|
item.setText(backup.display_name)
|
||||||
|
item.setData(Qt.ItemDataRole.UserRole, backup.name)
|
||||||
|
|
||||||
|
if backup.trigger == BackupTrigger.MANUAL:
|
||||||
|
icon_name = "hand.tap"
|
||||||
|
else:
|
||||||
|
icon_name = "clock.arrow.circlepath"
|
||||||
|
icon = sf_symbol(icon_name, 16)
|
||||||
|
if not icon.isNull():
|
||||||
|
item.setIcon(icon)
|
||||||
|
|
||||||
|
self._backup_list.addItem(item)
|
||||||
|
|
||||||
|
def _on_backup_selected(self, current: QListWidgetItem | None) -> None:
|
||||||
|
if current is None:
|
||||||
|
self._restore_btn.setEnabled(False)
|
||||||
|
self._details_label.setText("Select a backup to view details")
|
||||||
|
self._changes_tree.clear()
|
||||||
|
return
|
||||||
|
|
||||||
|
backup_name = current.data(Qt.ItemDataRole.UserRole)
|
||||||
|
self._restore_btn.setEnabled(True)
|
||||||
|
|
||||||
|
backup = next((b for b in self._backups if b.name == backup_name), None)
|
||||||
|
if backup:
|
||||||
|
details = (
|
||||||
|
f"Created: {backup.timestamp.strftime('%Y-%m-%d %H:%M:%S')}\n"
|
||||||
|
f"Files: {backup.file_count}\n"
|
||||||
|
f"Size: {backup.size_display}"
|
||||||
|
)
|
||||||
|
if backup.description:
|
||||||
|
details += f"\nNote: {backup.description}"
|
||||||
|
self._details_label.setText(details)
|
||||||
|
|
||||||
|
self.backup_selected.emit(backup_name)
|
||||||
|
|
||||||
|
def set_changes(self, backup_name: str, changes: BackupChanges) -> None:
|
||||||
|
self._changes_cache[backup_name] = changes
|
||||||
|
|
||||||
|
current = self._backup_list.currentItem()
|
||||||
|
if current and current.data(Qt.ItemDataRole.UserRole) == backup_name:
|
||||||
|
self._display_changes(changes)
|
||||||
|
|
||||||
|
def _display_changes(self, changes: BackupChanges) -> None:
|
||||||
|
self._changes_tree.clear()
|
||||||
|
|
||||||
|
if changes.is_empty:
|
||||||
|
item = QTreeWidgetItem(["No changes (identical to current)"])
|
||||||
|
self._changes_tree.addTopLevelItem(item)
|
||||||
|
return
|
||||||
|
|
||||||
|
self._add_change_group(changes.added, "Files to remove", "minus.circle")
|
||||||
|
self._add_change_group(
|
||||||
|
changes.modified, "Files to revert", "arrow.uturn.backward.circle"
|
||||||
|
)
|
||||||
|
self._add_change_group(changes.deleted, "Files to restore", "plus.circle")
|
||||||
|
|
||||||
|
def _add_change_group(self, files: list[str], label: str, icon_name: str) -> None:
|
||||||
|
if not files:
|
||||||
|
return
|
||||||
|
group_item = QTreeWidgetItem([f"{label} ({len(files)})"])
|
||||||
|
icon = sf_symbol(icon_name, 14)
|
||||||
|
if not icon.isNull():
|
||||||
|
group_item.setIcon(0, icon)
|
||||||
|
for filename in sorted(files):
|
||||||
|
QTreeWidgetItem(group_item, [filename])
|
||||||
|
self._changes_tree.addTopLevelItem(group_item)
|
||||||
|
group_item.setExpanded(True)
|
||||||
|
|
||||||
|
def _on_restore_clicked(self) -> None:
|
||||||
|
current = self._backup_list.currentItem()
|
||||||
|
if not current:
|
||||||
|
return
|
||||||
|
|
||||||
|
backup_name = current.data(Qt.ItemDataRole.UserRole)
|
||||||
|
backup = next((b for b in self._backups if b.name == backup_name), None)
|
||||||
|
if not backup:
|
||||||
|
return
|
||||||
|
|
||||||
|
ts = backup.timestamp.strftime("%Y-%m-%d %H:%M:%S")
|
||||||
|
msg = (
|
||||||
|
f"Are you sure you want to restore the backup from {ts}?\n\n"
|
||||||
|
"An automatic backup will be created before restoring."
|
||||||
|
)
|
||||||
|
result = QMessageBox.question(
|
||||||
|
self,
|
||||||
|
"Restore Backup",
|
||||||
|
msg,
|
||||||
|
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||||||
|
QMessageBox.StandardButton.No,
|
||||||
|
)
|
||||||
|
|
||||||
|
if result == QMessageBox.StandardButton.Yes:
|
||||||
|
self.restore_requested.emit(backup_name)
|
||||||
|
self.accept()
|
||||||
Reference in New Issue
Block a user