forked from Support/sp-hydra-veil-gui
199 lines
11 KiB
Python
Executable file
199 lines
11 KiB
Python
Executable file
from functools import partial
|
|
from typing import Any, Callable, Mapping, Sequence
|
|
|
|
from PyQt6.QtCore import Qt
|
|
|
|
from core.controllers.ProfileController import ProfileController
|
|
from core.controllers.profile_state.update_profile import update_profile
|
|
from gui.v2.actions.database_health import GuiStorageDatabaseError
|
|
from gui.v2.actions.editor_data import (
|
|
EDITABLE_FIELDS, EditorCallback, EditorResult, EditorState,
|
|
copy_editor_state, location_record, prepare_editor_data, render_editor_state, set_editor_value, snapshot_profile,
|
|
)
|
|
from gui.v2.actions.profile_operations import current_connection, synchronize
|
|
from gui.v2.workers.page_data_worker import PageDataWorker
|
|
|
|
|
|
_active_workers: set[PageDataWorker] = set()
|
|
|
|
|
|
def _release_worker(worker: PageDataWorker) -> None:
|
|
"""Keep worker references until native completion so page navigation cannot destroy running tasks."""
|
|
_active_workers.discard(worker)
|
|
worker.deleteLater()
|
|
|
|
|
|
def _report_task_failure(on_result: EditorCallback, request_name: str, message: str) -> None:
|
|
"""Restore GUI controls even if the worker cannot format an unexpected task failure."""
|
|
on_result(request_name, {"status": "error", "message": message})
|
|
|
|
|
|
def _execute(operation: Callable[[], EditorResult], state: EditorState | None) -> EditorResult:
|
|
"""Turn background exceptions into explicit outcomes that leave pending changes available."""
|
|
try:
|
|
return operation()
|
|
except GuiStorageDatabaseError:
|
|
message = "Local storage database could not be read. Restart and recover storage.db."
|
|
except Exception as error:
|
|
message = str(error) or "The editor operation failed."
|
|
result: EditorResult = {"status": "error", "message": message}
|
|
if state is not None:
|
|
result.update(render_editor_state(state, "error"))
|
|
return result
|
|
|
|
|
|
def _start_task(request_name: str, operation: Callable[[], EditorResult], on_result: EditorCallback, state: EditorState | None = None) -> PageDataWorker:
|
|
"""Own worker construction and GUI signal delivery so editor handlers only pass task inputs."""
|
|
worker = PageDataWorker(request_name, _execute, operation, state)
|
|
worker.data_ready.connect(on_result, Qt.ConnectionType.QueuedConnection)
|
|
worker.failed.connect(partial(_report_task_failure, on_result), Qt.ConnectionType.QueuedConnection)
|
|
worker.finished.connect(partial(_release_worker, worker))
|
|
_active_workers.add(worker)
|
|
worker.start()
|
|
return worker
|
|
|
|
|
|
def _load_editor(profile_id: int | None, location_choices: list[str], browser_choices: list[str], resolutions: list[str], is_synced: bool,
|
|
prepared: Mapping[str, dict[str, str]] | None, previous: EditorState | None) -> EditorResult:
|
|
"""Load one editing baseline and location metadata while preserving an existing unsaved draft."""
|
|
if profile_id is None:
|
|
return {"status": "error", "message": "Select a profile before opening the editor."}
|
|
profile_id = int(profile_id)
|
|
baseline = previous["baseline"] if previous is not None and previous["changes"] else None
|
|
profile = ProfileController.get(profile_id) if baseline is None else None
|
|
if baseline is None and profile is None:
|
|
return {"status": "error", "message": f"Profile {profile_id} no longer exists."}
|
|
if baseline is None:
|
|
baseline = snapshot_profile(profile)
|
|
location_map = dict(prepared) if prepared is not None else prepare_editor_data()
|
|
if not set(location_choices).issubset(location_map):
|
|
location_map = prepare_editor_data()
|
|
location_metadata = baseline["location_metadata"]
|
|
if location_metadata:
|
|
location_map[location_metadata["key"]] = location_metadata
|
|
state: EditorState = {
|
|
"profile_id": profile_id, "baseline": dict(baseline),
|
|
"changes": dict(previous["changes"]) if previous is not None else {},
|
|
"locations": location_map, "location_choices": list(dict.fromkeys(location_choices)),
|
|
"browser_choices": list(dict.fromkeys(browser_choices)), "resolutions": list(dict.fromkeys(resolutions)),
|
|
"is_synced": is_synced,
|
|
}
|
|
return render_editor_state(state)
|
|
|
|
|
|
def load_editor(profile_id: int | None, location_choices: Sequence[str], browser_choices: Sequence[str], resolutions: Sequence[str], is_synced: bool,
|
|
prepared: Mapping[str, dict[str, str]] | None, previous: EditorState | None, request_name: str, on_result: EditorCallback) -> PageDataWorker:
|
|
"""Prepare editor content asynchronously without letting the page coordinate controller lookups."""
|
|
captured = copy_editor_state(previous) if previous is not None else None
|
|
operation = partial(_load_editor, profile_id, list(location_choices), list(browser_choices), list(resolutions), is_synced, prepared, captured)
|
|
return _start_task(request_name, operation, on_result, captured)
|
|
|
|
|
|
def _sync_editor(state: EditorState) -> EditorResult:
|
|
"""Replace available choices only after synchronization succeeds while retaining the current draft."""
|
|
synced = synchronize(current_connection())
|
|
if synced["status"] == "error":
|
|
return {**render_editor_state(state, "error"), "message": synced["message"], "sync_data": synced["sync_data"]}
|
|
locations, browsers, _, _, location_models, _ = synced["sync_data"]
|
|
records = [location_record(location) for location in location_models]
|
|
updated: EditorState = {**state, "location_choices": locations, "browser_choices": browsers,
|
|
"locations": {record["key"]: record for record in records if record}, "is_synced": True}
|
|
return {**render_editor_state(updated), "sync_data": synced["sync_data"], "message": "Sync complete."}
|
|
|
|
|
|
def cycle_editor_value(state: EditorState, key: str, direction: int, request_name: str, on_result: EditorCallback) -> PageDataWorker | None:
|
|
"""Share previous and next selection logic and own the synchronization needed to populate choices."""
|
|
captured = copy_editor_state(state)
|
|
field = next((field for field in render_editor_state(captured)["view"]["fields"] if field["key"] == key), None)
|
|
if field is None or not field["enabled"] or not field["visible"] or direction not in (-1, 1):
|
|
on_result(request_name, {"status": "error", "message": "This editor field cannot be changed."})
|
|
return None
|
|
if field["needs_sync"]:
|
|
return _start_task(request_name, partial(_sync_editor, captured), on_result, captured)
|
|
choices = field["choices"]
|
|
if not choices:
|
|
on_result(request_name, {"status": "error", "message": f"No {key} data available. Sync the database and try again."})
|
|
return None
|
|
current = field["value"]
|
|
index = choices.index(current) if current in choices else (-1 if direction == 1 else 0)
|
|
on_result(request_name, set_editor_value(captured, key, choices[(index + direction) % len(choices)]))
|
|
return None
|
|
|
|
|
|
def _confirmation(state: EditorState, task: str, message: str) -> EditorResult:
|
|
"""Capture the profile and draft together so confirmations cannot apply subsequently edited values."""
|
|
return {"status": "confirmation", "message": message,
|
|
"request": {"task": task, "state": copy_editor_state(state)}}
|
|
|
|
|
|
def _apply_warning(state: EditorState) -> str:
|
|
"""Include every destructive change in one confirmation rather than hiding the browser warning behind location changes."""
|
|
warnings = {
|
|
"location": "A new location requires a new subscription code and erases the old one.",
|
|
"browser": "Changing the browser would delete all data associated with it.",
|
|
}
|
|
messages = [message for field, message in warnings.items() if field in state["changes"]]
|
|
return "\n\n".join([*messages, "Do you want to proceed?"]) if messages else ""
|
|
|
|
|
|
def _save_editor(state: EditorState) -> EditorResult:
|
|
"""Track each core save so partial failures retain only the changes still needing attention."""
|
|
pending = state["changes"]
|
|
unsupported = set(pending).difference(EDITABLE_FIELDS)
|
|
if unsupported:
|
|
return {**render_editor_state(state, "error"), "message": "These editor changes are not supported; protocol switching is disabled."}
|
|
updated = copy_editor_state(state)
|
|
saved: list[str] = []
|
|
for key, value in pending.items():
|
|
try:
|
|
outcome = update_profile(state["profile_id"], key, value)
|
|
message = None if outcome.valid else str(outcome.message or outcome.data or f"Could not save {key}.")
|
|
except Exception as error:
|
|
message = str(error) or f"Could not save {key}."
|
|
if message is not None:
|
|
progress = f"Saved {', '.join(saved)}. " if saved else ""
|
|
return {**render_editor_state(updated, "error"), "message": f"{progress}Could not save {key}: {message}",
|
|
"saved_fields": saved, "refresh_menu": bool(saved)}
|
|
updated["baseline"][key] = value
|
|
if key == "browser":
|
|
updated["baseline"]["browser_supported"] = value in updated["browser_choices"]
|
|
updated["changes"].pop(key, None)
|
|
saved.append(key)
|
|
return {"status": "navigate", "destination": "menu", "state": updated, "refresh_menu": True,
|
|
"saved_fields": saved, "message": f"Updated profile {state['profile_id']}"}
|
|
|
|
|
|
def apply_editor_changes(state: EditorState, request_name: str, on_result: EditorCallback, confirmed: bool = False) -> PageDataWorker | None:
|
|
"""Own confirmation and saving as one user task so the GUI never chains individual core updates."""
|
|
captured = copy_editor_state(state)
|
|
if not captured["changes"]:
|
|
on_result(request_name, {"status": "navigate", "destination": "menu", "state": captured})
|
|
return None
|
|
warning = _apply_warning(captured)
|
|
if warning and not confirmed:
|
|
on_result(request_name, _confirmation(captured, "apply", warning))
|
|
return None
|
|
return _start_task(request_name, partial(_save_editor, captured), on_result, captured)
|
|
|
|
|
|
def leave_editor(state: EditorState, request_name: str, on_result: EditorCallback, confirmed: bool = False) -> None:
|
|
"""Discard changes only after the user accepts leaving an editor with an unsaved draft."""
|
|
captured = copy_editor_state(state)
|
|
if captured["changes"] and not confirmed:
|
|
on_result(request_name, _confirmation(captured, "leave", "You have unsaved changes. Do you want to continue?"))
|
|
return
|
|
captured["changes"].clear()
|
|
on_result(request_name, {"status": "navigate", "destination": "menu", "state": captured})
|
|
|
|
|
|
def continue_editor_task(request: Mapping[str, Any], accepted: bool, request_name: str, on_result: EditorCallback) -> PageDataWorker | None:
|
|
"""Return popup decisions to the original task without rebuilding its profile or draft from the page."""
|
|
state = request["state"]
|
|
if not accepted:
|
|
on_result(request_name, render_editor_state(state))
|
|
return None
|
|
handlers = {
|
|
"apply": partial(apply_editor_changes, state, request_name, on_result, confirmed=True),
|
|
"leave": partial(leave_editor, state, request_name, on_result, confirmed=True),
|
|
}
|
|
return handlers[request["task"]]()
|