forked from Support/sp-hydra-veil-gui
92 lines
4.2 KiB
Python
Executable file
92 lines
4.2 KiB
Python
Executable file
from core.services.subscriptions.subscriptions_by_protocol import get_profile_subscription
|
|
from core.models.Result import Result, ResultError
|
|
from core.errors.logger import logger
|
|
|
|
|
|
from functools import partial
|
|
from typing import Any, Iterable, Mapping
|
|
|
|
from PyQt6.QtCore import Qt
|
|
from gui.v2.actions.menu_results import MenuCallback
|
|
from gui.v2.actions.profile_operations import execute_menu_task
|
|
from gui.v2.infrastructure.screen_size import get_max_screensize
|
|
from gui.v2.workers.worker import Worker
|
|
|
|
|
|
_active_workers: set[Worker] = set()
|
|
|
|
|
|
def _release_worker(worker: Worker) -> None:
|
|
"""Retain workers until their native thread finishes, including long-running browser sessions."""
|
|
_active_workers.discard(worker)
|
|
worker.deleteLater()
|
|
|
|
|
|
def _start_task(request: Mapping[str, Any], on_result: MenuCallback) -> Worker:
|
|
"""Own worker creation and queued GUI delivery so pages only supply task inputs."""
|
|
operation = partial(execute_menu_task, dict(request), max_resolution=get_max_screensize())
|
|
worker = Worker(operation)
|
|
worker.output.connect(on_result, Qt.ConnectionType.QueuedConnection)
|
|
worker.finished.connect(partial(_release_worker, worker))
|
|
_active_workers.add(worker)
|
|
worker.start()
|
|
return worker
|
|
|
|
|
|
def pre_connect_clear(profile_id: int) -> Result:
|
|
subscription = get_profile_subscription(profile_id=profile_id)
|
|
if not subscription or not subscription.billing_code:
|
|
return Result(valid=False, error_type=ResultError.SUBSCRIPTION, message="No Subscription")
|
|
|
|
return Result(valid=True)
|
|
|
|
|
|
def enable_profile(profile_data: Mapping[str, Any], on_result: MenuCallback, allow_uninstalled_tor: bool = False) -> Worker:
|
|
"""Start the complete activation workflow with billing and retry data kept in one request."""
|
|
return _start_task({**profile_data, "task": "enable", "allow_uninstalled_tor": allow_uninstalled_tor}, on_result)
|
|
|
|
|
|
def disable_profile(profile_id: int, connected_profile_ids: Iterable[int], on_result: MenuCallback) -> Worker:
|
|
"""Let the disable task decide whether dependent sessions require confirmation."""
|
|
return _start_task({"task": "disable", "id": profile_id, "connected_ids": list(connected_profile_ids)}, on_result)
|
|
|
|
|
|
def delete_profile(profile_id: int, on_result: MenuCallback) -> Worker:
|
|
"""Capture the deletion target before showing a confirmation."""
|
|
return _start_task({"task": "delete", "id": profile_id}, on_result)
|
|
|
|
|
|
def create_profile(gui_config_file: str | None, is_synced: bool, on_result: MenuCallback) -> Worker:
|
|
"""Let the create task select fast registration and perform any required synchronization."""
|
|
return _start_task({"task": "create", "gui_config_file": gui_config_file, "is_synced": is_synced}, on_result)
|
|
|
|
|
|
def edit_profile(gui_config_file: str | None, is_synced: bool, on_result: MenuCallback) -> Worker:
|
|
"""Let the edit task finish required synchronization before requesting navigation."""
|
|
return _start_task({"task": "edit", "gui_config_file": gui_config_file, "is_synced": is_synced}, on_result)
|
|
|
|
|
|
def download_update(on_result: MenuCallback) -> Worker:
|
|
"""Keep update startup in the action layer rather than the confirmation dialog."""
|
|
return _start_task({"task": "update"}, on_result)
|
|
|
|
|
|
def continue_menu_task(request: Mapping[str, Any], choice: str, on_result: MenuCallback) -> Worker | None:
|
|
"""Resume the captured task without losing accepted exceptions or spending a ticket again."""
|
|
if choice == "abort":
|
|
on_result({"status": "cancelled", "message": "Operation cancelled"})
|
|
return None
|
|
updates = {
|
|
("tor_download", "continue"): {"allow_uninstalled_tor": True},
|
|
("billing_tor", "continue"): {"allow_clearweb": True},
|
|
("force", "continue"): {"ignore_profile_state_conflict": True},
|
|
("endpoint", "continue"): {"ignore_endpoint_verification": True},
|
|
("endpoint", "sync"): {"sync_before_enable": True},
|
|
("disable_all", "continue"): {"confirmed": True},
|
|
("delete", "continue"): {"confirmed": True},
|
|
}
|
|
flags = updates.get((request.get("stage"), choice))
|
|
if flags is None:
|
|
on_result({"status": "error", "message": "Unknown confirmation response"})
|
|
return None
|
|
return _start_task({**request, **flags}, on_result)
|