loop-uh
1e6268b6a8
All checks were successful
Windows project source guards / test (push) Successful in 1m19s
1733 lines
80 KiB
Python
1733 lines
80 KiB
Python
from __future__ import annotations
|
||
|
||
|
||
from pathlib import Path
|
||
import sys
|
||
import time
|
||
import logging
|
||
|
||
from PyQt6.QtCore import QTimer, QUrl
|
||
from PyQt6.QtGui import QAction, QActionGroup, QCloseEvent, QDesktopServices, QGuiApplication, QIcon
|
||
from PyQt6.QtWidgets import QApplication, QDialog, QFileDialog, QMenu, QSystemTrayIcon
|
||
from qfluentwidgets import (
|
||
FluentIcon as FIF,
|
||
FluentWindow,
|
||
InfoBar,
|
||
InfoBarIcon,
|
||
InfoBarPosition,
|
||
NavigationItemPosition,
|
||
)
|
||
|
||
from ..diagnostics.connection_message import connection_message
|
||
from ..application.controller import AppController
|
||
from ..profiles.storage import PassphraseRequired
|
||
from ..constants import APP_ICON_PATH, APP_NAME, APP_VERSION, BASE_DIR, LOG_DIR
|
||
from ..profiles.models import AppSettings, Node, RoutingSettings, Subscription, SubscriptionUpdateResult
|
||
from ..importer.subscription_http import mask_subscription_url
|
||
from ..updates.app_updater import AppUpdate, UpdateChecker, UpdateDownloader
|
||
from ..engines.xray import XrayCoreUpdateResult
|
||
from .dashboard_page import DashboardPage
|
||
from .deferred_page import DeferredPage
|
||
from .lock_dialog import PasswordDialog
|
||
from .logs_page import LogsPage
|
||
from .nodes_page import NodesPage
|
||
from .configs_page import ConfigsPage
|
||
from .settings_page import SettingsPage
|
||
from .subscriptions_page import SubscriptionDeleteDialog, SubscriptionsPage
|
||
from .about_page import AboutPage
|
||
from .history_page import HistoryPage
|
||
from .theme import apply_theme, sync_system_theme_listener
|
||
from .updates_page import UpdatesPage
|
||
from .zapret_page import ZapretPage
|
||
|
||
|
||
APP_UPDATE_INITIAL_DELAY_MS = 2500
|
||
APP_UPDATE_INTERVAL_MS = 30 * 60 * 1000
|
||
|
||
|
||
def _runtime_identity_log_line() -> str:
|
||
"""Return the non-secret runtime identity written to the regular app log."""
|
||
executable = Path(sys.executable).resolve()
|
||
base_dir = BASE_DIR.resolve()
|
||
return (
|
||
f"[runtime] app_version={APP_VERSION} "
|
||
f"executable={executable} "
|
||
f"base_dir={base_dir}"
|
||
)
|
||
|
||
|
||
class MainWindow(FluentWindow):
|
||
def __init__(self, defer_init: bool = False):
|
||
super().__init__()
|
||
self._quitting = False
|
||
self._tray_notified = False
|
||
self._state_loaded = False
|
||
self._initialized = False
|
||
self._startup_loader = None
|
||
self._metadata_worker = None
|
||
self._startup_started = time.perf_counter()
|
||
self._bulk_task_tip: InfoBar | None = None
|
||
self._bulk_task_type: str | None = None
|
||
self._speed_test_was_cancelled = False
|
||
self._tray_available = QSystemTrayIcon.isSystemTrayAvailable()
|
||
self._deferred_dashboard_metrics: tuple[float, float, int | None] | None = None
|
||
self._deferred_process_stats: list | None = None
|
||
self._has_deferred_process_stats = False
|
||
self._geometry_persistence_ready = False
|
||
self._restoring_geometry = False
|
||
self._geometry_applied = False
|
||
self._app_update_scheduler_ready = False
|
||
self._app_update_timer = QTimer(self)
|
||
self._app_update_timer.setSingleShot(True)
|
||
self._app_update_timer.timeout.connect(self._on_app_update_timer_timeout)
|
||
self.tray: QSystemTrayIcon | None = None
|
||
self.tray_show_action: QAction | None = None
|
||
self.tray_connect_action: QAction | None = None
|
||
self.tray_next_action: QAction | None = None
|
||
self.tray_quit_action: QAction | None = None
|
||
self.tray_mode_menu: QMenu | None = None
|
||
self.tray_mode_group: QActionGroup | None = None
|
||
self.tray_mode_global: QAction | None = None
|
||
self.tray_mode_rule: QAction | None = None
|
||
self.tray_mode_direct: QAction | None = None
|
||
|
||
self._app_icon = QIcon(str(APP_ICON_PATH))
|
||
self.setWindowTitle(APP_NAME)
|
||
self.setWindowIcon(self._app_icon)
|
||
self.resize(1000, 720)
|
||
|
||
if not defer_init:
|
||
self.initialize()
|
||
|
||
def initialize(self, *, load_state: bool = True) -> None:
|
||
if self._initialized:
|
||
return
|
||
|
||
self._initialized = True
|
||
self._nodes_view_prefs_applied = False
|
||
self.controller = AppController(self)
|
||
self.dashboard_page = DashboardPage(self)
|
||
self.nodes_page = DeferredPage(NodesPage, "nodes", self)
|
||
self.subscriptions_page = DeferredPage(SubscriptionsPage, "subscriptions", self)
|
||
self.configs_page = DeferredPage(ConfigsPage, "configs", self)
|
||
self.zapret_page = DeferredPage(ZapretPage, "zapret", self)
|
||
self.logs_page = DeferredPage(LogsPage, "logs", self)
|
||
self.settings_page = DeferredPage(SettingsPage, "settings", self)
|
||
self.updates_page = DeferredPage(UpdatesPage, "updates", self)
|
||
self.history_page = DeferredPage(HistoryPage, "history", self)
|
||
self.about_page = DeferredPage(AboutPage, "about", self)
|
||
|
||
self._create_navigation()
|
||
self._create_tray()
|
||
self._connect_signals()
|
||
# Keep this in the ordinary application log as well as startup.log so
|
||
# a support report can identify the exact portable copy in use.
|
||
self.controller._log(_runtime_identity_log_line())
|
||
self._init_window()
|
||
|
||
self.dashboard_page.set_initializing(True)
|
||
self.navigationInterface.setEnabled(False)
|
||
for action in (self.tray_connect_action, self.tray_next_action):
|
||
if action:
|
||
action.setEnabled(False)
|
||
self.configs_page.created.connect(self._load_config_editor_documents)
|
||
if load_state:
|
||
from ..application.startup_service import StartupLoader
|
||
self._startup_loader = StartupLoader(self)
|
||
self._startup_loader.early.connect(lambda settings, _locked: self._apply_window_geometry(settings))
|
||
self._startup_loader.loaded.connect(self.complete_initialization)
|
||
self._startup_loader.password_needed.connect(self._ask_startup_password)
|
||
self._startup_loader.failed.connect(self._startup_failed)
|
||
QTimer.singleShot(0, self._startup_loader.start)
|
||
|
||
def _ask_startup_password(self):
|
||
if self._quitting:
|
||
return
|
||
dialog = PasswordDialog("Зашифрованные данные", self)
|
||
if dialog.exec() != int(QDialog.DialogCode.Accepted):
|
||
self._quit_app()
|
||
return
|
||
self._startup_loader.start(dialog.password())
|
||
|
||
def _startup_failed(self, message):
|
||
if self._quitting:
|
||
return
|
||
self.dashboard_page.set_initializing(False)
|
||
self.dashboard_page.set_runtime_status("error", message)
|
||
self.dashboard_page.set_transition_busy(True)
|
||
self._show_status("error", message)
|
||
|
||
def complete_initialization(self, state, history):
|
||
if self._quitting or self._geometry_persistence_ready:
|
||
return
|
||
if self._startup_loader is not None:
|
||
self.controller.storage.passphrase = self._startup_loader.storage.passphrase
|
||
self.controller.storage._known_encrypted = self._startup_loader.storage.is_encrypted()
|
||
self.controller.load(state, history)
|
||
self._state_loaded = True
|
||
self._apply_theme(state.settings.theme, state.settings.accent_color)
|
||
self._apply_window_geometry(state.settings)
|
||
if state.security.enabled:
|
||
self.controller.locked = True
|
||
if not self._ensure_unlocked(startup=True):
|
||
return
|
||
self._geometry_persistence_ready = True
|
||
self.dashboard_page.set_initializing(False)
|
||
self.navigationInterface.setEnabled(True)
|
||
for action in (self.tray_connect_action, self.tray_next_action):
|
||
if action:
|
||
action.setEnabled(True)
|
||
self.history_page.set_storage(self.controller.traffic_history)
|
||
self.controller.auto_connect_if_needed()
|
||
self._app_update_scheduler_ready = True
|
||
self._sync_app_update_timer(state.settings)
|
||
self._consume_update_error_log()
|
||
from ..application.startup_service import MetadataWorker
|
||
from ..profiles.path_utils import resolve_configured_path
|
||
from ..constants import SINGBOX_PATH_DEFAULT
|
||
sb_path = resolve_configured_path(state.settings.singbox_path, default_path=SINGBOX_PATH_DEFAULT, use_default_if_empty=True) or SINGBOX_PATH_DEFAULT
|
||
self._metadata_worker = MetadataWorker(state.settings.xray_path, str(sb_path), self)
|
||
self._metadata_worker.ready.connect(self._metadata_ready)
|
||
self._metadata_worker.start()
|
||
if state.settings.xray_auto_update:
|
||
QTimer.singleShot(4500, lambda: self.controller.run_xray_core_update(True, silent=True) if not self._quitting else None)
|
||
logging.getLogger("xray_fluent.bootstrap").info("controls_ready_ms=%.1f", (time.perf_counter() - self._startup_started) * 1000)
|
||
|
||
def _metadata_ready(self, result):
|
||
if self._quitting:
|
||
return
|
||
self.updates_page.set_xray_version(result["xray"])
|
||
self.updates_page.set_singbox_version(result["singbox"])
|
||
self._init_zapret_page(result["presets"])
|
||
|
||
def finish_background_shutdown(self):
|
||
if self._startup_loader:
|
||
self._startup_loader.finish_shutdown()
|
||
if self._metadata_worker and self._metadata_worker.isRunning():
|
||
self._metadata_worker.requestInterruption()
|
||
self._metadata_worker.wait()
|
||
worker = getattr(getattr(self, "controller", None), "_country_resolver", None)
|
||
if worker and worker.isRunning():
|
||
worker.requestInterruption()
|
||
worker.wait()
|
||
|
||
def _create_navigation(self) -> None:
|
||
self.navigationInterface.setMinimumExpandWidth(1100)
|
||
self.navigationInterface.setExpandWidth(200)
|
||
self.addSubInterface(self.dashboard_page, FIF.SPEED_HIGH, "Панель")
|
||
self.addSubInterface(self.nodes_page, FIF.LINK, "Серверы")
|
||
self.addSubInterface(self.subscriptions_page, FIF.CLOUD, "Подписки")
|
||
self.addSubInterface(self.configs_page, FIF.CODE, "Конфиги")
|
||
self.addSubInterface(self.zapret_page, FIF.COMMAND_PROMPT, "Zapret")
|
||
self.addSubInterface(self.logs_page, FIF.DOCUMENT, "Логи")
|
||
self.addSubInterface(self.history_page, FIF.HISTORY, "История")
|
||
self.addSubInterface(self.about_page, FIF.INFO, "О проекте", NavigationItemPosition.BOTTOM)
|
||
self.addSubInterface(self.updates_page, FIF.UPDATE, "Обновления", NavigationItemPosition.BOTTOM)
|
||
self.addSubInterface(self.settings_page, FIF.SETTING, "Настройки", NavigationItemPosition.BOTTOM)
|
||
|
||
def _create_tray(self) -> None:
|
||
if not self._tray_available:
|
||
return
|
||
|
||
self.tray = QSystemTrayIcon(self)
|
||
self.tray.setIcon(self._app_icon)
|
||
self.tray.setToolTip(APP_NAME)
|
||
|
||
menu = QMenu()
|
||
self.tray_show_action = QAction("Показать", self)
|
||
self.tray_connect_action = QAction("Подключить", self)
|
||
self.tray_next_action = QAction("Следующий сервер", self)
|
||
self.tray_quit_action = QAction("Выход", self)
|
||
|
||
self.tray_mode_menu = QMenu("Режим", menu)
|
||
self.tray_mode_group = QActionGroup(self)
|
||
self.tray_mode_group.setExclusive(True)
|
||
self.tray_mode_global = QAction("Глобальный", self)
|
||
self.tray_mode_global.setCheckable(True)
|
||
self.tray_mode_rule = QAction("По правилам", self)
|
||
self.tray_mode_rule.setCheckable(True)
|
||
self.tray_mode_direct = QAction("Прямой", self)
|
||
self.tray_mode_direct.setCheckable(True)
|
||
|
||
self.tray_mode_group.addAction(self.tray_mode_global)
|
||
self.tray_mode_group.addAction(self.tray_mode_rule)
|
||
self.tray_mode_group.addAction(self.tray_mode_direct)
|
||
|
||
self.tray_mode_menu.addAction(self.tray_mode_global)
|
||
self.tray_mode_menu.addAction(self.tray_mode_rule)
|
||
self.tray_mode_menu.addAction(self.tray_mode_direct)
|
||
|
||
menu.addAction(self.tray_show_action)
|
||
menu.addAction(self.tray_connect_action)
|
||
menu.addAction(self.tray_next_action)
|
||
menu.addMenu(self.tray_mode_menu)
|
||
menu.addSeparator()
|
||
menu.addAction(self.tray_quit_action)
|
||
|
||
self.tray.setContextMenu(menu)
|
||
self.tray.show()
|
||
|
||
self.tray.activated.connect(self._on_tray_activated)
|
||
self.tray_show_action.triggered.connect(self._toggle_window_visible)
|
||
self.tray_connect_action.triggered.connect(self.controller.toggle_connection)
|
||
self.tray_next_action.triggered.connect(self.controller.switch_next_node)
|
||
self.tray_quit_action.triggered.connect(self._quit_app)
|
||
self.tray_mode_global.triggered.connect(lambda: self._set_mode_from_tray("global"))
|
||
self.tray_mode_rule.triggered.connect(lambda: self._set_mode_from_tray("rule"))
|
||
self.tray_mode_direct.triggered.connect(lambda: self._set_mode_from_tray("direct"))
|
||
|
||
def _connect_signals(self) -> None:
|
||
self.dashboard_page.logs_requested.connect(lambda: self.switchTo(self.logs_page))
|
||
self.dashboard_page.mode_changed.connect(self._set_mode_only)
|
||
self.dashboard_page.toggle_connection_requested.connect(self.controller.toggle_connection)
|
||
self.dashboard_page.tun_toggled.connect(self._on_dashboard_tun_toggled)
|
||
self.dashboard_page.proxy_toggled.connect(self._on_dashboard_proxy_toggled)
|
||
self.nodes_page.import_clipboard_requested.connect(self._import_nodes_from_clipboard)
|
||
self.nodes_page.delete_requested.connect(self.controller.remove_nodes)
|
||
self.nodes_page.hide_subscription_nodes_requested.connect(self.controller.hide_subscription_nodes)
|
||
self.nodes_page.reorder_requested.connect(self.controller.reorder_nodes)
|
||
self.nodes_page.selected_node_changed.connect(self.controller.set_selected_node)
|
||
self.nodes_page.ping_requested.connect(self._ping_requested)
|
||
self.nodes_page.export_outbound_json_requested.connect(self._export_outbound_json)
|
||
self.nodes_page.export_runtime_json_requested.connect(self._export_runtime_json)
|
||
self.nodes_page.edit_node_requested.connect(self._on_edit_node)
|
||
self.nodes_page.node_edit_saved.connect(self._on_node_edit_saved)
|
||
self.nodes_page.bulk_edit_requested.connect(self._on_bulk_edit_nodes)
|
||
self.nodes_page.bulk_edit_applied.connect(self._on_bulk_edit_applied)
|
||
self.nodes_page.favorite_requested.connect(self._set_favorites)
|
||
self.nodes_page.view_prefs_changed.connect(self._on_nodes_view_prefs_changed)
|
||
|
||
self.subscriptions_page.add_requested.connect(self._add_subscription)
|
||
self.subscriptions_page.edit_requested.connect(self._edit_subscription)
|
||
self.subscriptions_page.editor_save_requested.connect(self._save_subscription_from_editor)
|
||
self.subscriptions_page.update_requested.connect(
|
||
lambda sid, mode: self.controller.update_subscription(sid, mode=mode)
|
||
)
|
||
self.subscriptions_page.force_update_requested.connect(
|
||
lambda sid, mode: self.controller.update_subscription(
|
||
sid,
|
||
mode=mode,
|
||
force_refresh=True,
|
||
)
|
||
)
|
||
self.subscriptions_page.check_requested.connect(
|
||
lambda sid, mode: self.controller.check_subscription(sid, mode=mode)
|
||
)
|
||
self.subscriptions_page.update_all_requested.connect(
|
||
lambda mode: self.controller.update_all_subscriptions(mode=mode)
|
||
)
|
||
self.subscriptions_page.delete_requested.connect(self._delete_subscription)
|
||
self.subscriptions_page.auto_update_changed.connect(self._set_subscription_auto_update)
|
||
self.subscriptions_page.open_url_requested.connect(
|
||
lambda url: QDesktopServices.openUrl(QUrl(url))
|
||
)
|
||
self.subscriptions_page.show_url_requested.connect(self._show_subscription_url)
|
||
self.subscriptions_page.reset_hidden_requested.connect(self._reset_subscription_hidden)
|
||
self.subscriptions_page.accept_pending_url_requested.connect(self._accept_subscription_url)
|
||
|
||
self.configs_page.open_requested.connect(self._open_core_config)
|
||
self.configs_page.config_selected.connect(self._select_core_config)
|
||
self.configs_page.template_selected.connect(self._select_core_template)
|
||
self.configs_page.reset_requested.connect(self._reset_core_config_to_template)
|
||
self.configs_page.save_requested.connect(self._save_core_config)
|
||
self.configs_page.validate_requested.connect(self._validate_core_config)
|
||
self.configs_page.apply_requested.connect(self._apply_core_config)
|
||
|
||
self.zapret_page.start_requested.connect(self._on_zapret_start)
|
||
self.zapret_page.stop_requested.connect(self._on_zapret_stop)
|
||
self.zapret_page.target_settings_changed.connect(self._on_zapret_target_settings_changed)
|
||
self.controller.zapret.started.connect(self._on_zapret_started)
|
||
self.controller.zapret.stopped.connect(self._on_zapret_stopped)
|
||
self.controller.zapret.error.connect(self._on_zapret_error)
|
||
self.controller.zapret.log_line.connect(self.logs_page.append_line)
|
||
self.controller.zapret.target_profile_changed.connect(self._on_zapret_target_changed)
|
||
self.controller.transition_state_changed.connect(self._on_zapret_transition_state)
|
||
|
||
self.logs_page.clear_requested.connect(self._clear_logs_view)
|
||
self.logs_page.export_diag_requested.connect(self._export_diagnostics)
|
||
|
||
self.settings_page.save_requested.connect(self.controller.update_settings)
|
||
self.settings_page.auto_lock_minutes_changed.connect(self._update_auto_lock_minutes)
|
||
self.settings_page.set_password_requested.connect(self._set_password)
|
||
self.settings_page.disable_password_requested.connect(self.controller.disable_master_password)
|
||
self.settings_page.lock_now_requested.connect(self.controller.lock)
|
||
self.updates_page.check_app_requested.connect(self._check_updates)
|
||
self.updates_page.check_xray_requested.connect(self._check_xray_updates)
|
||
self.updates_page.update_xray_requested.connect(self._update_xray_core)
|
||
self.updates_page.check_singbox_requested.connect(self._check_singbox_updates)
|
||
self.updates_page.update_singbox_requested.connect(self._update_singbox_core)
|
||
self.settings_page.export_backup_requested.connect(self._export_backup)
|
||
self.settings_page.import_backup_requested.connect(self._import_backup)
|
||
self.settings_page.set_encryption_requested.connect(self._set_encryption)
|
||
self.settings_page.disable_encryption_requested.connect(self._disable_encryption)
|
||
|
||
# Тест скорости
|
||
self.nodes_page.speed_test_requested.connect(self._speed_test_requested)
|
||
self.nodes_page.cancel_speed_test_requested.connect(self._cancel_speed_test_requested)
|
||
self.controller.speed_updated.connect(self._on_speed_updated)
|
||
self.controller.speed_test_cancelled.connect(self._on_speed_test_cancelled)
|
||
|
||
self.controller.nodes_changed.connect(self._on_nodes_changed)
|
||
self.controller.countries_changed.connect(self._on_countries_changed)
|
||
self.controller.subscriptions_changed.connect(self._on_subscriptions_changed)
|
||
self.controller.subscription_update_started.connect(self._on_subscription_update_started)
|
||
self.controller.subscription_update_finished.connect(self._on_subscription_update_finished)
|
||
self.controller.selection_changed.connect(self._on_selection_changed)
|
||
self.controller.connection_changed.connect(self._on_connection_changed)
|
||
self.controller.connection_status_changed.connect(self.dashboard_page.set_runtime_status)
|
||
self.controller.routing_changed.connect(self._on_routing_changed)
|
||
self.controller.settings_changed.connect(self._on_settings_changed)
|
||
self.controller.log_line.connect(self.logs_page.append_line)
|
||
self.controller.runtime_errors_changed.connect(self.logs_page.set_error_records)
|
||
self.logs_page.set_error_records(self.controller.runtime_errors.snapshot())
|
||
self.controller.status.connect(self._show_status)
|
||
self.controller.bulk_task_progress.connect(self._on_bulk_task_progress)
|
||
self.controller.ping_updated.connect(self._on_ping_updated)
|
||
self.controller.speed_progress_updated.connect(self._on_speed_progress_updated)
|
||
self.controller.connectivity_test_done.connect(self._on_connectivity_test_done)
|
||
self.controller.live_metrics_updated.connect(self._on_live_metrics_updated)
|
||
self.controller.xray_update_result.connect(self._on_xray_update_result)
|
||
self.controller.lock_state_changed.connect(self._on_lock_state_changed)
|
||
self.controller.auto_switch_triggered.connect(self._on_auto_switch)
|
||
self.controller.transition_state_changed.connect(self._on_transition_state_changed)
|
||
self.stackedWidget.currentChanged.connect(self._on_current_interface_changed)
|
||
|
||
def _init_window(self) -> None:
|
||
self.setMinimumSize(860, 560)
|
||
self.setWindowTitle(APP_NAME)
|
||
self.setWindowIcon(self._app_icon)
|
||
self._apply_window_geometry(self.controller.state.settings)
|
||
|
||
def _apply_window_geometry(self, settings: AppSettings) -> None:
|
||
from .window_geometry import fitted_geometry
|
||
screens = QGuiApplication.screens()
|
||
primary = screens.index(QGuiApplication.primaryScreen()) if screens else 0
|
||
rect, minimum = fitted_geometry(settings, [screen.availableGeometry() for screen in screens], primary)
|
||
self._restoring_geometry = True
|
||
try:
|
||
self.setMinimumSize(minimum)
|
||
self.setGeometry(rect)
|
||
self._geometry_applied = True
|
||
finally:
|
||
self._restoring_geometry = False
|
||
|
||
def _fit_current_screen(self, *_):
|
||
if not self._geometry_persistence_ready or self._restoring_geometry:
|
||
return
|
||
from copy import copy
|
||
settings = copy(self.controller.state.settings)
|
||
geo = self.normalGeometry() if self.isMaximized() else self.geometry()
|
||
settings.window_x, settings.window_y = geo.x(), geo.y()
|
||
settings.window_width, settings.window_height = geo.width(), geo.height()
|
||
if not self.isMaximized():
|
||
self._apply_window_geometry(settings)
|
||
else:
|
||
from .window_geometry import MINIMUM_SIZE
|
||
area = self.screen().availableGeometry()
|
||
self.setMinimumSize(min(MINIMUM_SIZE.width(), area.width()), min(MINIMUM_SIZE.height(), area.height()))
|
||
|
||
def showEvent(self, event):
|
||
super().showEvent(event)
|
||
if not getattr(self, "_first_show_logged", False):
|
||
self._first_show_logged = True
|
||
QTimer.singleShot(0, lambda: logging.getLogger("xray_fluent.bootstrap").info("first_window_ms=%.1f", (time.perf_counter() - self._startup_started) * 1000))
|
||
if not getattr(self, "_screen_signals_connected", False) and self.windowHandle():
|
||
self._screen_signals_connected = True
|
||
self.windowHandle().screenChanged.connect(self._fit_current_screen)
|
||
for screen in QGuiApplication.screens():
|
||
screen.availableGeometryChanged.connect(self._fit_current_screen)
|
||
screen.logicalDotsPerInchChanged.connect(self._fit_current_screen)
|
||
|
||
def _on_countries_changed(self, node_ids) -> None:
|
||
self.nodes_page.update_countries(node_ids)
|
||
if self.controller.state.selected_node_id in node_ids:
|
||
self.dashboard_page.set_selected_node(self.controller.selected_node)
|
||
|
||
def _on_nodes_changed(self, nodes: list[Node]) -> None:
|
||
if not self._nodes_view_prefs_applied:
|
||
# Apply persisted list view preferences before the first set_nodes.
|
||
self._nodes_view_prefs_applied = True
|
||
self.nodes_page.apply_view_settings(self.controller.state.settings)
|
||
selected_id = self.controller.state.selected_node_id
|
||
self.nodes_page.set_nodes(nodes, selected_id)
|
||
self.dashboard_page.set_nodes(nodes, selected_id)
|
||
self._refresh_tray_tooltip()
|
||
self.subscriptions_page.set_data(self.controller.state.subscriptions, nodes)
|
||
|
||
def _on_subscriptions_changed(self, subscriptions: list[Subscription]) -> None:
|
||
self.subscriptions_page.set_data(subscriptions, self.controller.state.nodes)
|
||
self.nodes_page.set_subscriptions(subscriptions)
|
||
|
||
def _add_subscription(self) -> None:
|
||
self.subscriptions_page.open_editor(None)
|
||
|
||
def _edit_subscription(self, subscription_id: str) -> None:
|
||
subscription = self.controller.get_subscription(subscription_id)
|
||
if subscription is None:
|
||
return
|
||
provider_names = [
|
||
node.provider_name or node.name
|
||
for node in self.controller.state.nodes
|
||
if node.subscription_id == subscription_id
|
||
]
|
||
self.subscriptions_page.open_editor(subscription, provider_names)
|
||
|
||
def _save_subscription_from_editor(self, subscription_id: str) -> None:
|
||
"""Commit the in-page subscription form (empty id = a new subscription).
|
||
|
||
On failure the sub-page stays open so the entered URL survives the error.
|
||
"""
|
||
editor = self.subscriptions_page.editor
|
||
if not subscription_id:
|
||
try:
|
||
started = self.controller.add_subscription(editor.subscription_value())
|
||
except Exception as exc:
|
||
self._show_status("error", str(exc))
|
||
return
|
||
self.subscriptions_page.close_editor()
|
||
if started:
|
||
self._show_status("info", "Проверяю подписку…")
|
||
return
|
||
|
||
try:
|
||
changed = self.controller.update_subscription_definition(
|
||
subscription_id, editor.update_values()
|
||
)
|
||
except Exception as exc:
|
||
self._show_status("error", str(exc))
|
||
return
|
||
self.subscriptions_page.close_editor()
|
||
if changed:
|
||
self.controller.update_subscription(subscription_id)
|
||
|
||
def _delete_subscription(self, subscription_id: str) -> None:
|
||
subscription = self.controller.get_subscription(subscription_id)
|
||
if subscription is None:
|
||
return
|
||
dialog = SubscriptionDeleteDialog(subscription.name or "Подписка", self)
|
||
if dialog.exec() == int(QDialog.DialogCode.Accepted):
|
||
self.controller.remove_subscription(
|
||
subscription_id,
|
||
keep_nodes=dialog.keep_nodes_check.isChecked(),
|
||
)
|
||
|
||
def _set_subscription_auto_update(self, subscription_id: str, enabled: bool) -> None:
|
||
subscription = self.controller.get_subscription(subscription_id)
|
||
if subscription is None or subscription.auto_update == enabled:
|
||
return
|
||
self.controller.update_subscription_definition(subscription_id, {"auto_update": enabled})
|
||
|
||
def _show_subscription_url(self, subscription_id: str) -> None:
|
||
subscription = self.controller.get_subscription(subscription_id)
|
||
if subscription is not None:
|
||
self._show_status("info-long", mask_subscription_url(subscription.url))
|
||
|
||
def _reset_subscription_hidden(self, subscription_id: str) -> None:
|
||
if self.controller.reset_subscription_hidden_nodes(subscription_id):
|
||
self.controller.update_subscription(subscription_id)
|
||
|
||
def _accept_subscription_url(self, subscription_id: str) -> None:
|
||
if self.controller.accept_subscription_pending_url(subscription_id):
|
||
self.controller.update_subscription(subscription_id)
|
||
|
||
def _on_subscription_update_started(self, subscription_id: str) -> None:
|
||
self.subscriptions_page.set_updating(subscription_id, True)
|
||
|
||
def _on_subscription_update_finished(self, result: SubscriptionUpdateResult) -> None:
|
||
self.subscriptions_page.set_updating(result.subscription_id, False)
|
||
if result.success:
|
||
message = result.message
|
||
if result.check_only:
|
||
# Проверка ничего не сохраняет: счётчики изменений у неё всегда
|
||
# нулевые, а расхождение с сохранённым списком выглядело как
|
||
# потерянные серверы.
|
||
if not result.not_modified and result.stored_count != result.source_count:
|
||
message += (
|
||
f"; сохранено {result.stored_count} — "
|
||
"нажмите «Обновить», чтобы применить"
|
||
)
|
||
if result.skipped:
|
||
message += f"; пропущено {result.skipped}"
|
||
elif not result.not_modified:
|
||
message += (
|
||
f"; добавлено {result.added}, обновлено {result.updated}, "
|
||
f"удалено {result.removed}, пропущено {result.skipped}"
|
||
)
|
||
if result.warnings:
|
||
message += f"; предупреждения: {'; '.join(result.warnings[:2])}"
|
||
self._show_status("warning-long" if result.warnings else "success", message)
|
||
else:
|
||
self._show_status("error", result.message)
|
||
|
||
def _set_favorites(self, node_ids: set[str], favorite: bool) -> None:
|
||
for node in self.controller.state.nodes:
|
||
if node.id in node_ids:
|
||
node.is_favorite = favorite
|
||
self.controller.nodes_changed.emit(self.controller.state.nodes)
|
||
self.controller.schedule_save()
|
||
|
||
def _on_nodes_view_prefs_changed(self, prefs: dict) -> None:
|
||
controller = getattr(self, "controller", None)
|
||
if controller is None:
|
||
return
|
||
settings = controller.state.settings
|
||
for key, value in prefs.items():
|
||
if hasattr(settings, key):
|
||
setattr(settings, key, value)
|
||
controller.schedule_save()
|
||
|
||
def _on_selection_changed(self, node: Node | None) -> None:
|
||
self.nodes_page.set_active_node(node.id if node else None)
|
||
self.dashboard_page.set_selected_node(node)
|
||
if node:
|
||
self.dashboard_page.set_selected_latency(node.ping_ms)
|
||
else:
|
||
self.dashboard_page.set_selected_latency(None)
|
||
self.zapret_page.set_target(node, self.controller.zapret.resolved_target)
|
||
self._refresh_tray_tooltip()
|
||
|
||
def _on_connection_changed(self, connected: bool) -> None:
|
||
self.dashboard_page.set_connection(connected)
|
||
if connected and not self.controller.state.settings.tun_mode:
|
||
socks_port, http_port = self.controller.get_effective_proxy_ports()
|
||
self.dashboard_page.set_proxy_ports(socks_port, http_port)
|
||
else:
|
||
self.dashboard_page.set_proxy_ports(0, 0)
|
||
# Подключение/отключение меняет системный прокси — обновляем снимок.
|
||
self.dashboard_page.set_system_proxy_state(self.controller.query_system_proxy_state())
|
||
if not connected:
|
||
self._deferred_dashboard_metrics = None
|
||
self._deferred_process_stats = None
|
||
self._has_deferred_process_stats = False
|
||
if self.tray_connect_action is not None:
|
||
self.tray_connect_action.setText("Отключить" if connected else "Подключить")
|
||
self._refresh_tray_tooltip()
|
||
|
||
def _on_transition_state_changed(self, busy: bool, _message: str) -> None:
|
||
self.dashboard_page.set_transition_busy(busy)
|
||
if self.tray_connect_action is not None:
|
||
self.tray_connect_action.setEnabled(not busy)
|
||
|
||
def _on_routing_changed(self, routing: RoutingSettings) -> None:
|
||
self.dashboard_page.set_routing_snapshot(routing)
|
||
if self.tray_mode_global is not None:
|
||
self.tray_mode_global.setChecked(routing.mode == "global")
|
||
if self.tray_mode_rule is not None:
|
||
self.tray_mode_rule.setChecked(routing.mode == "rule")
|
||
if self.tray_mode_direct is not None:
|
||
self.tray_mode_direct.setChecked(routing.mode == "direct")
|
||
|
||
def _on_settings_changed(self, settings: AppSettings) -> None:
|
||
self.settings_page.set_values(settings, self.controller.state.security)
|
||
self.zapret_page.set_target_settings(settings.zapret_target)
|
||
self.settings_page.set_encryption_active(self.controller.is_data_encrypted())
|
||
# Сверяем дашборд с реальным состоянием прокси Windows (быстрое
|
||
# чтение реестра; на не-Windows вернётся supported=False).
|
||
self.dashboard_page.set_system_proxy_state(self.controller.query_system_proxy_state())
|
||
self.dashboard_page.set_settings_snapshot(settings)
|
||
self._apply_theme(settings.theme, settings.accent_color)
|
||
routing_controls_enabled = bool(settings.tun_mode and settings.tun_engine == "tun2socks")
|
||
for action in (self.tray_mode_global, self.tray_mode_rule, self.tray_mode_direct):
|
||
if action is not None:
|
||
action.setEnabled(routing_controls_enabled)
|
||
self._refresh_tray_tooltip()
|
||
if self._app_update_scheduler_ready:
|
||
self._sync_app_update_timer(settings)
|
||
|
||
def _on_ping_updated(self, node_id: str, ping_ms: int | None) -> None:
|
||
self.nodes_page.update_ping(node_id, ping_ms)
|
||
self.nodes_page.refresh_detail(node_id)
|
||
node = self.controller.selected_node
|
||
if node and node.id == node_id:
|
||
self.dashboard_page.set_selected_latency(ping_ms)
|
||
|
||
def _on_live_metrics_updated(self, payload: dict[str, object]) -> None:
|
||
down_bps = float(payload.get("down_bps") or 0.0)
|
||
up_bps = float(payload.get("up_bps") or 0.0)
|
||
latency = payload.get("latency_ms")
|
||
latency_ms = int(latency) if isinstance(latency, int) else None
|
||
process_stats = payload.get("process_stats")
|
||
if not self._is_dashboard_active():
|
||
self._deferred_dashboard_metrics = (down_bps, up_bps, latency_ms)
|
||
if process_stats is not None:
|
||
self._deferred_process_stats = list(process_stats)
|
||
self._has_deferred_process_stats = True
|
||
return
|
||
|
||
self.dashboard_page.set_live_metrics(down_bps, up_bps, latency_ms)
|
||
if process_stats is not None:
|
||
self.dashboard_page.set_process_stats(process_stats)
|
||
|
||
def _is_dashboard_active(self) -> bool:
|
||
return self.stackedWidget.currentWidget() is self.dashboard_page
|
||
|
||
def _on_current_interface_changed(self, _index: int) -> None:
|
||
if not self._is_dashboard_active():
|
||
return
|
||
if self._deferred_dashboard_metrics is not None:
|
||
down_bps, up_bps, latency_ms = self._deferred_dashboard_metrics
|
||
self.dashboard_page.set_live_metrics(down_bps, up_bps, latency_ms)
|
||
self._deferred_dashboard_metrics = None
|
||
if self._has_deferred_process_stats:
|
||
self.dashboard_page.set_process_stats(self._deferred_process_stats or [])
|
||
self._deferred_process_stats = None
|
||
self._has_deferred_process_stats = False
|
||
|
||
def _on_xray_update_result(self, result: XrayCoreUpdateResult) -> None:
|
||
self.logs_page.append_line(f"[core-update] {result.status}: {result.message}")
|
||
if result.status == "error":
|
||
self.updates_page.set_xray_error(result.message)
|
||
elif result.status in {"updated", "up_to_date"}:
|
||
self.updates_page.set_xray_success(result.message)
|
||
else:
|
||
self.updates_page.set_xray_status(result.message)
|
||
|
||
if result.updated:
|
||
self.updates_page.set_xray_version(result.latest_version)
|
||
|
||
def _on_connectivity_test_done(self, ok: bool, message: str, elapsed_ms: int | None) -> None:
|
||
if ok and elapsed_ms is not None:
|
||
self.logs_page.append_line(f"[test] ok {elapsed_ms} ms | {message}")
|
||
else:
|
||
self.logs_page.append_line(f"[test] fail | {message}")
|
||
|
||
def _on_lock_state_changed(self, locked: bool) -> None:
|
||
if locked:
|
||
self._show_status("warning", "Приложение заблокировано")
|
||
self._ensure_unlocked(startup=False)
|
||
|
||
def _on_auto_switch(self, node_name: str) -> None:
|
||
InfoBar.warning(
|
||
"Авто-переключение",
|
||
f"Скорость упала. Переключение на {node_name}...",
|
||
position=InfoBarPosition.TOP_RIGHT,
|
||
duration=30000,
|
||
parent=self,
|
||
)
|
||
|
||
def _on_bulk_task_progress(self, task: str, current: int, total: int, completed: bool) -> None:
|
||
task = task.strip().lower()
|
||
if task not in {"ping", "speed"}:
|
||
return
|
||
|
||
total = max(1, total)
|
||
current = max(0, min(current, total))
|
||
title = "Пинг серверов" if task == "ping" else "Тест скорости"
|
||
|
||
if completed:
|
||
if task == "ping":
|
||
self.nodes_page.finish_ping_activity()
|
||
else:
|
||
self.nodes_page.finish_speed_activity()
|
||
if self._bulk_task_tip and self._bulk_task_type == task:
|
||
if task == "speed" and self._speed_test_was_cancelled:
|
||
self._set_bulk_task_tip_content(f"Остановлено {current}/{total}")
|
||
self._speed_test_was_cancelled = False
|
||
else:
|
||
self._set_bulk_task_tip_content(f"Завершено {current}/{total}")
|
||
QTimer.singleShot(1000, self._clear_bulk_task_tip)
|
||
return
|
||
|
||
content = f"{current}/{total}"
|
||
|
||
if self._bulk_task_tip is None or self._bulk_task_type != task:
|
||
self._clear_bulk_task_tip()
|
||
tip = InfoBar(
|
||
icon=InfoBarIcon.INFORMATION,
|
||
title=title,
|
||
content=content,
|
||
isClosable=True,
|
||
duration=-1,
|
||
position=InfoBarPosition.TOP_RIGHT,
|
||
parent=self,
|
||
)
|
||
tip.setMinimumWidth(220)
|
||
tip.closedSignal.connect(self._on_bulk_task_tip_closed)
|
||
tip.show()
|
||
self._bulk_task_tip = tip
|
||
self._bulk_task_type = task
|
||
return
|
||
|
||
update_step = max(1, (total + 99) // 100)
|
||
if current != total and current % update_step:
|
||
return
|
||
self._set_bulk_task_tip_content(content)
|
||
|
||
def _on_bulk_task_tip_closed(self) -> None:
|
||
self._bulk_task_tip = None
|
||
self._bulk_task_type = None
|
||
|
||
def _set_bulk_task_tip_content(self, content: str) -> None:
|
||
if self._bulk_task_tip is None:
|
||
return
|
||
self._bulk_task_tip.content = content
|
||
self._bulk_task_tip.contentLabel.setText(content)
|
||
|
||
def _clear_bulk_task_tip(self) -> None:
|
||
if self._bulk_task_tip is None:
|
||
return
|
||
tip = self._bulk_task_tip
|
||
self._bulk_task_tip = None
|
||
self._bulk_task_type = None
|
||
tip.close()
|
||
|
||
def _show_status(self, level: str, message: str) -> None:
|
||
message = connection_message(message)
|
||
level = level.lower().strip()
|
||
long_duration = level.endswith("-long")
|
||
if long_duration:
|
||
level = level[:-5]
|
||
dashboard = getattr(self, "dashboard_page", None)
|
||
if (level == "error" and dashboard is not None
|
||
and dashboard._connection_phase == "error"
|
||
and dashboard._connection_message == message
|
||
and self._is_dashboard_active()):
|
||
return
|
||
if level == "error":
|
||
InfoBar.error("Ошибка", message, position=InfoBarPosition.TOP_RIGHT, duration=6000, parent=self)
|
||
elif level == "warning":
|
||
InfoBar.warning(
|
||
"Внимание",
|
||
message,
|
||
position=InfoBarPosition.TOP_RIGHT,
|
||
duration=10000 if long_duration else 3000,
|
||
parent=self,
|
||
)
|
||
elif level == "success":
|
||
InfoBar.success("Успешно", message, position=InfoBarPosition.TOP_RIGHT, duration=2200, parent=self)
|
||
else:
|
||
InfoBar.info(
|
||
"Инфо",
|
||
message,
|
||
position=InfoBarPosition.TOP_RIGHT,
|
||
duration=10000 if long_duration else 2200,
|
||
parent=self,
|
||
)
|
||
|
||
def _import_nodes_from_clipboard(self) -> None:
|
||
clipboard = QApplication.clipboard()
|
||
text = clipboard.text().strip() if clipboard is not None else ""
|
||
if not text:
|
||
self._show_status("warning", "Буфер обмена пуст")
|
||
return
|
||
|
||
added, errors = self.controller.import_nodes_from_text(text)
|
||
if added:
|
||
self._show_status("success", f"Импортировано серверов: {added}")
|
||
if errors:
|
||
# Раньше показывались только первые две ошибки, остальные терялись молча —
|
||
# при импорте большой подписки это скрывало реальный масштаб проблемы.
|
||
for line in errors:
|
||
self.controller._log(f"[import] {line}")
|
||
preview = "; ".join(errors[:5])
|
||
if len(errors) > 5:
|
||
preview = f"{preview}; …ещё {len(errors) - 5} (полный список — на странице логов)"
|
||
self._show_status(
|
||
"warning-long",
|
||
f"Не удалось импортировать ссылок: {len(errors)}. {preview}",
|
||
)
|
||
if not added and not errors:
|
||
self._show_status("warning", "Новых серверов не импортировано")
|
||
|
||
def _ping_requested(self, ids: set[str]) -> None:
|
||
self.nodes_page.start_ping_activity(ids or None)
|
||
if ids:
|
||
self.controller.ping_nodes(ids)
|
||
else:
|
||
self.controller.ping_nodes(None)
|
||
|
||
def _speed_test_requested(self, ids: set[str]) -> None:
|
||
started = self.controller.speed_test_nodes(ids or None)
|
||
if started:
|
||
self._speed_test_was_cancelled = False
|
||
self.nodes_page.start_speed_activity()
|
||
|
||
def _cancel_speed_test_requested(self) -> None:
|
||
if self.controller.cancel_speed_test():
|
||
self.nodes_page.mark_speed_test_stopping()
|
||
if self._bulk_task_tip and self._bulk_task_type == "speed":
|
||
self._set_bulk_task_tip_content("Останавливаю...")
|
||
|
||
def _on_speed_test_cancelled(self, _completed: int, _total: int) -> None:
|
||
self._speed_test_was_cancelled = True
|
||
|
||
def _on_speed_progress_updated(self, node_id: str, percent: int) -> None:
|
||
self.nodes_page.update_speed_progress(node_id, percent)
|
||
|
||
def _on_speed_updated(self, node_id: str, speed_mbps: float | None, _is_alive: bool) -> None:
|
||
self.nodes_page.update_speed(node_id, speed_mbps)
|
||
self.nodes_page.refresh_detail(node_id)
|
||
|
||
def _on_edit_node(self, node_id: str) -> None:
|
||
node = self.controller._get_node_by_id(node_id)
|
||
if not node:
|
||
return
|
||
self.nodes_page.open_node_editor(node, self.controller.get_all_groups())
|
||
|
||
def _on_node_edit_saved(self, node_id: str, fields: dict) -> None:
|
||
self.controller.update_node(node_id, fields)
|
||
self.nodes_page.close_editor()
|
||
self._show_status("success", "Сервер обновлён")
|
||
|
||
def _on_bulk_edit_nodes(self, node_ids: set[str]) -> None:
|
||
if not node_ids:
|
||
return
|
||
self.nodes_page.open_bulk_editor(node_ids, self.controller.get_all_groups())
|
||
|
||
def _on_bulk_edit_applied(self, node_ids: set[str], operations: dict) -> None:
|
||
self.nodes_page.close_editor()
|
||
if operations["group"] or operations["add_tags"] or operations["remove_tags"]:
|
||
count = self.controller.bulk_update_nodes(set(node_ids), operations)
|
||
self._show_status("success", f"Обновлено серверов: {count}")
|
||
|
||
def _export_outbound_json(self, node_id: str) -> None:
|
||
payload = self.controller.export_node_outbound_json(node_id)
|
||
if not payload:
|
||
self._show_status("warning", "Выберите сервер для экспорта")
|
||
return
|
||
self._save_json_payload(payload, "outbound.json")
|
||
|
||
def _export_runtime_json(self, node_id: str) -> None:
|
||
payload = self.controller.export_runtime_config_json(node_id)
|
||
if not payload:
|
||
self._show_status("warning", "Выберите сервер для экспорта")
|
||
return
|
||
suggested_name = "singbox_config.json" if self.controller.is_singbox_editor_mode() else "xray_config.json"
|
||
self._save_json_payload(payload, suggested_name)
|
||
|
||
def _save_json_payload(self, payload: str, suggested_name: str) -> None:
|
||
file_path, _ = QFileDialog.getSaveFileName(self, "Экспорт JSON", suggested_name, "JSON files (*.json)")
|
||
if file_path:
|
||
with open(file_path, "w", encoding="utf-8") as file:
|
||
file.write(payload)
|
||
self._show_status("success", f"JSON экспортирован: {file_path}")
|
||
return
|
||
|
||
clipboard = QApplication.clipboard()
|
||
if clipboard is not None:
|
||
clipboard.setText(payload)
|
||
self._show_status("info", "Экспорт отменён, JSON скопирован в буфер обмена")
|
||
|
||
@staticmethod
|
||
def _core_title(core: str) -> str:
|
||
return "sing-box" if core == "singbox" else "xray"
|
||
|
||
def _get_core_profile_dir(self, core: str, kind: str) -> Path:
|
||
if core == "singbox":
|
||
return self.controller.get_singbox_config_dir() if kind == "config" else self.controller.get_singbox_template_dir()
|
||
return self.controller.get_xray_config_dir() if kind == "config" else self.controller.get_xray_template_dir()
|
||
|
||
def _get_active_core_profile_relative(self, core: str, kind: str) -> str:
|
||
base_dir = self._get_core_profile_dir(core, kind).resolve()
|
||
if core == "singbox":
|
||
path = (
|
||
self.controller.get_active_singbox_config_path()
|
||
if kind == "config"
|
||
else self.controller.get_active_singbox_template_path()
|
||
)
|
||
else:
|
||
path = (
|
||
self.controller.get_active_xray_config_path()
|
||
if kind == "config"
|
||
else self.controller.get_active_xray_template_path()
|
||
)
|
||
if path is None:
|
||
return ""
|
||
try:
|
||
return path.resolve().relative_to(base_dir).as_posix()
|
||
except ValueError:
|
||
return ""
|
||
|
||
def _list_core_profile_items(self, core: str, kind: str) -> list[tuple[str, str]]:
|
||
base_dir = self._get_core_profile_dir(core, kind).resolve()
|
||
items: list[tuple[str, str]] = []
|
||
for path in sorted(base_dir.rglob("*.json")):
|
||
relative = path.relative_to(base_dir).as_posix()
|
||
items.append((relative, relative))
|
||
return items
|
||
|
||
def _refresh_core_profile_choices(self, core: str) -> None:
|
||
config_items = self._list_core_profile_items(core, "config")
|
||
template_items = self._list_core_profile_items(core, "template")
|
||
self.configs_page.set_available_configs(
|
||
core,
|
||
config_items,
|
||
self._get_active_core_profile_relative(core, "config"),
|
||
)
|
||
self.configs_page.set_available_templates(
|
||
core,
|
||
template_items,
|
||
self._get_active_core_profile_relative(core, "template"),
|
||
)
|
||
|
||
def _sync_core_template_for_config(self, core: str, config_path: Path) -> Path | None:
|
||
if core == "singbox":
|
||
template_path = self.controller._default_singbox_template_path_for_config(config_path)
|
||
if template_path is not None:
|
||
self.controller._set_active_singbox_template_path(template_path, emit_signal=False)
|
||
return template_path
|
||
template_path = self.controller._default_xray_template_path_for_config(config_path)
|
||
if template_path is not None:
|
||
self.controller._set_active_xray_template_path(template_path, emit_signal=False)
|
||
return template_path
|
||
|
||
def _load_config_editor_documents(self) -> None:
|
||
for core in ("singbox",):
|
||
self._load_core_config_document(core)
|
||
|
||
def _load_core_config_document(self, core: str) -> None:
|
||
try:
|
||
if core == "singbox":
|
||
path, text = self.controller.load_active_singbox_config_text()
|
||
else:
|
||
path, text = self.controller.load_active_xray_config_text()
|
||
except Exception as exc:
|
||
self.configs_page.set_status(core, "error", str(exc))
|
||
return
|
||
self.configs_page.set_document(core, path, text)
|
||
template_path = (
|
||
self.controller.get_active_singbox_template_path()
|
||
if core == "singbox"
|
||
else self.controller.get_active_xray_template_path()
|
||
)
|
||
self._refresh_core_profile_choices(core)
|
||
self.configs_page.set_template_source(core, template_path)
|
||
self.configs_page.set_status(core, "info", f"Открыта активная копия: {path.name}")
|
||
|
||
def _select_core_config(self, core: str, relative_path: str) -> None:
|
||
try:
|
||
if core == "singbox":
|
||
path, text = self.controller.load_singbox_config_text(relative_path)
|
||
else:
|
||
path, text = self.controller.load_xray_config_text(relative_path)
|
||
template_path = self._sync_core_template_for_config(core, path)
|
||
if template_path is None:
|
||
template_path = (
|
||
self.controller.get_active_singbox_template_path()
|
||
if core == "singbox"
|
||
else self.controller.get_active_xray_template_path()
|
||
)
|
||
except Exception as exc:
|
||
self._refresh_core_profile_choices(core)
|
||
self.configs_page.set_status(core, "error", str(exc))
|
||
self._show_status("error", str(exc).splitlines()[0])
|
||
return
|
||
self.configs_page.set_document(core, path, text)
|
||
self._refresh_core_profile_choices(core)
|
||
self.configs_page.set_template_source(core, template_path)
|
||
self.configs_page.set_status(core, "info", f"Открыт конфиг: {path.name}")
|
||
self._show_status("success", f"Открыт конфиг: {path.name}")
|
||
|
||
def _select_core_template(self, core: str, relative_path: str) -> None:
|
||
try:
|
||
if core == "singbox":
|
||
path, text = self.controller.import_singbox_template(relative_path)
|
||
template_path = self.controller.get_active_singbox_template_path()
|
||
else:
|
||
path, text = self.controller.import_xray_template(relative_path)
|
||
template_path = self.controller.get_active_xray_template_path()
|
||
except Exception as exc:
|
||
self._refresh_core_profile_choices(core)
|
||
self.configs_page.set_status(core, "error", str(exc))
|
||
self._show_status("error", str(exc).splitlines()[0])
|
||
return
|
||
self.configs_page.set_document(core, path, text)
|
||
self._refresh_core_profile_choices(core)
|
||
self.configs_page.set_template_source(core, template_path)
|
||
self.configs_page.set_status(core, "info", f"Применён шаблон: {Path(relative_path).name}")
|
||
self._show_status("success", f"Применён шаблон: {Path(relative_path).name}")
|
||
|
||
def _open_core_config(self, core: str) -> None:
|
||
title = self._core_title(core)
|
||
base_dir = str(
|
||
self.controller.get_singbox_template_dir()
|
||
if core == "singbox"
|
||
else self.controller.get_xray_template_dir()
|
||
)
|
||
file_path, _ = QFileDialog.getOpenFileName(self, f"Импортировать {title} template", base_dir, "JSON files (*.json)")
|
||
if not file_path:
|
||
return
|
||
try:
|
||
if core == "singbox":
|
||
path, text = self.controller.import_singbox_template(file_path)
|
||
template_path = self.controller.get_active_singbox_template_path()
|
||
else:
|
||
path, text = self.controller.import_xray_template(file_path)
|
||
template_path = self.controller.get_active_xray_template_path()
|
||
except Exception as exc:
|
||
self.configs_page.set_status(core, "error", str(exc))
|
||
self._show_status("error", str(exc).splitlines()[0])
|
||
return
|
||
self.configs_page.set_document(core, path, text)
|
||
self._refresh_core_profile_choices(core)
|
||
self.configs_page.set_template_source(core, template_path)
|
||
self.configs_page.set_status(core, "info", f"Импортирован template и обновлена активная копия: {path.name}")
|
||
self._show_status("success", f"Импортирован template и обновлён активный конфиг: {Path(file_path).name}")
|
||
|
||
def _reset_core_config_to_template(self, core: str) -> None:
|
||
if core == "singbox":
|
||
ok, path, message = self.controller.reset_active_singbox_config_to_template()
|
||
template_path = self.controller.get_active_singbox_template_path()
|
||
loader = self.controller.load_active_singbox_config_text
|
||
else:
|
||
ok, path, message = self.controller.reset_active_xray_config_to_template()
|
||
template_path = self.controller.get_active_xray_template_path()
|
||
loader = self.controller.load_active_xray_config_text
|
||
if not ok or path is None:
|
||
self.configs_page.set_status(core, "error", message)
|
||
self._show_status("error", message.splitlines()[0])
|
||
return
|
||
loaded_path, text = loader()
|
||
self.configs_page.set_document(core, loaded_path, text)
|
||
self._refresh_core_profile_choices(core)
|
||
self.configs_page.set_template_source(core, template_path)
|
||
self.configs_page.set_status(core, "success", message)
|
||
self._show_status("success", message)
|
||
|
||
def _save_core_config(self, core: str, text: str) -> None:
|
||
try:
|
||
if core == "singbox":
|
||
path = self.controller.save_singbox_config_text(text)
|
||
path, saved_text = self.controller.load_active_singbox_config_text()
|
||
else:
|
||
path = self.controller.save_xray_config_text(text)
|
||
saved_text = text
|
||
except Exception as exc:
|
||
self.configs_page.set_status(core, "error", str(exc))
|
||
self._show_status("error", str(exc).splitlines()[0])
|
||
return
|
||
self.configs_page.mark_saved(core, path, saved_text)
|
||
if saved_text != text:
|
||
self.configs_page.set_document(core, path, saved_text)
|
||
self.configs_page.set_status(core, "success", f"Сохранено: {path.name}")
|
||
self._show_status("success", f"Сохранено: {path.name}")
|
||
|
||
def _validate_core_config(self, core: str, text: str) -> None:
|
||
if core == "singbox":
|
||
repair = self.controller.try_repair_singbox_json_text(text)
|
||
if repair is not None:
|
||
self.configs_page.replace_editor_text(core, repair.repaired_text)
|
||
message = (
|
||
f"Конфиг восстановлен автоматически: {repair.description}. "
|
||
"Проверьте результат и нажмите «Сохранить» или «Применить»."
|
||
)
|
||
self.configs_page.set_status(core, "warning", message)
|
||
self._show_status("warning-long", message)
|
||
return
|
||
ok, message = self.controller.validate_singbox_json_text(text)
|
||
else:
|
||
ok, message = self.controller.validate_xray_json_text(text)
|
||
self.configs_page.set_status(core, "success" if ok else "error", message)
|
||
if ok:
|
||
self._show_status("success", "JSON корректен")
|
||
|
||
def _apply_core_config(self, core: str, text: str) -> None:
|
||
if core == "singbox":
|
||
ok, path, message = self.controller.apply_singbox_config_text(text)
|
||
else:
|
||
ok, path, message = self.controller.apply_xray_config_text(text)
|
||
if not ok:
|
||
self.configs_page.set_status(core, "error", message)
|
||
self._show_status("error", message.splitlines()[0])
|
||
return
|
||
if path is not None:
|
||
if core == "singbox":
|
||
loaded_path, saved_text = self.controller.load_active_singbox_config_text()
|
||
self.configs_page.set_document(core, loaded_path, saved_text)
|
||
else:
|
||
self.configs_page.mark_saved(core, path, text)
|
||
level = "info" if "Применяю" in message else "success"
|
||
self.configs_page.set_status(core, level, message)
|
||
self._show_status(level, message.splitlines()[0])
|
||
|
||
def _on_dashboard_tun_toggled(self, checked: bool) -> None:
|
||
from copy import deepcopy
|
||
settings = deepcopy(self.controller.state.settings)
|
||
settings.tun_mode = checked
|
||
if checked:
|
||
settings.enable_system_proxy = False
|
||
self.controller.update_settings(settings)
|
||
|
||
def _on_dashboard_proxy_toggled(self, checked: bool) -> None:
|
||
from copy import deepcopy
|
||
settings = deepcopy(self.controller.state.settings)
|
||
settings.enable_system_proxy = checked
|
||
self.controller.update_settings(settings)
|
||
|
||
def _set_mode_only(self, mode: str) -> None:
|
||
from copy import deepcopy
|
||
routing = deepcopy(self.controller.state.routing)
|
||
routing.mode = mode
|
||
self.controller.update_routing(routing)
|
||
|
||
def _set_mode_from_tray(self, mode: str) -> None:
|
||
self._set_mode_only(mode)
|
||
|
||
def _set_password(self, password: str) -> None:
|
||
self.controller.set_master_password(password)
|
||
self._show_status("success", "Мастер-пароль включён")
|
||
|
||
def _update_auto_lock_minutes(self, minutes: int) -> None:
|
||
self.controller.state.security.auto_lock_minutes = max(1, minutes)
|
||
self.controller.save()
|
||
|
||
def _clear_logs_view(self) -> None:
|
||
self.logs_page.clear_view()
|
||
|
||
def _export_diagnostics(self) -> None:
|
||
path = self.controller.build_diagnostics()
|
||
self._show_status("success", f"Диагностика экспортирована: {path}")
|
||
from .file_manager import reveal_file
|
||
try:
|
||
reveal_file(path)
|
||
except OSError:
|
||
self._show_status("warning", f"Архив сохранён, но папку открыть не удалось: {path}")
|
||
|
||
# ── Zapret ───────────────────────────────────────────────
|
||
|
||
def _init_zapret_page(self, infos=None) -> None:
|
||
if infos is None:
|
||
from ..engines.zapret.manager import ZapretManager
|
||
infos = ZapretManager.list_preset_infos()
|
||
saved = self.controller.state.settings.zapret_preset
|
||
self.zapret_page.set_presets(infos, saved)
|
||
self.zapret_page.set_target_settings(self.controller.state.settings.zapret_target)
|
||
self.zapret_page.set_target(
|
||
self.controller.selected_node,
|
||
self.controller.zapret.resolved_target,
|
||
)
|
||
if self.controller.state.settings.zapret_autostart and saved:
|
||
if any(p.name == saved for p in infos):
|
||
QTimer.singleShot(1000, lambda: self._on_zapret_start(saved))
|
||
|
||
def _on_zapret_start(self, preset_name: str) -> None:
|
||
self.controller.start_zapret(preset_name)
|
||
|
||
def _on_zapret_stop(self) -> None:
|
||
self.controller.zapret.stop()
|
||
|
||
def _on_zapret_started(self) -> None:
|
||
active = self.controller.state.settings.zapret_preset
|
||
self.zapret_page.set_running(True, active)
|
||
|
||
def _on_zapret_stopped(self) -> None:
|
||
self.zapret_page.set_running(False)
|
||
|
||
def _on_zapret_error(self, message: str) -> None:
|
||
self.zapret_page.set_error(message)
|
||
self._show_status("error", f"Zapret: {message}")
|
||
|
||
def _on_zapret_target_changed(self, resolved) -> None:
|
||
self.zapret_page.set_target(
|
||
self.controller.selected_node,
|
||
resolved,
|
||
"Профиль подготовлен" if resolved else "Профиль отключён",
|
||
)
|
||
|
||
def _on_zapret_transition_state(self, busy: bool, text: str) -> None:
|
||
if busy:
|
||
self.zapret_page.set_target_runtime_state(text)
|
||
return
|
||
ready = self.controller.zapret.target_profile_is_ready(self.controller.selected_node)
|
||
self.zapret_page.set_target_runtime_state("Готов" if ready else "Не готов")
|
||
|
||
def _on_zapret_target_settings_changed(self, settings) -> None:
|
||
self.controller.state.settings.zapret_target = settings
|
||
self.controller.zapret.set_target_settings(settings)
|
||
self.controller.schedule_save()
|
||
self.zapret_page.set_target_settings(settings)
|
||
self.zapret_page.set_target(
|
||
self.controller.selected_node,
|
||
self.controller.zapret.resolved_target,
|
||
)
|
||
if self.controller.connected or self.controller._desired_connected:
|
||
self.controller._desired_connected = True
|
||
self.controller._request_transition("Zapret target strategy changed")
|
||
elif self.controller.zapret.running and self.controller.state.settings.zapret_preset:
|
||
self.controller.start_zapret(self.controller.state.settings.zapret_preset)
|
||
else:
|
||
self.zapret_page.set_target_runtime_state("Настройки сохранены")
|
||
|
||
def _check_updates(self, silent: bool = False) -> None:
|
||
if getattr(self, "_update_in_progress", False):
|
||
return
|
||
self._update_in_progress = True
|
||
self._pending_update: AppUpdate | None = None
|
||
checker = UpdateChecker(proxy_url=self._active_update_proxy_url(), parent=self)
|
||
self._update_checker = checker
|
||
checker.result.connect(
|
||
lambda update: self._on_update_check_result(update, silent, checker)
|
||
)
|
||
checker.error.connect(
|
||
lambda error: self._on_update_check_error(error, silent, checker)
|
||
)
|
||
checker.finished.connect(checker.deleteLater)
|
||
checker.start()
|
||
if not silent:
|
||
self.updates_page.show_checking()
|
||
|
||
def _sync_app_update_timer(self, settings: AppSettings) -> None:
|
||
"""Keep the one-shot/periodic application update timer in sync."""
|
||
if self._quitting or not settings.check_updates:
|
||
self._app_update_timer.stop()
|
||
return
|
||
if self._app_update_timer.isActive():
|
||
return
|
||
# The first check remains the existing startup check. After it fires,
|
||
# _on_app_update_timer_timeout switches this same timer to the 30-minute
|
||
# cadence instead of creating another timer.
|
||
self._app_update_timer.setInterval(APP_UPDATE_INITIAL_DELAY_MS)
|
||
self._app_update_timer.start()
|
||
|
||
def _on_app_update_timer_timeout(self) -> None:
|
||
if self._quitting or not self._app_update_scheduler_ready:
|
||
return
|
||
settings = self.controller.state.settings
|
||
if not settings.check_updates:
|
||
return
|
||
|
||
# _check_updates owns the concurrent-check guard. Even if a user
|
||
# started a manual check at the same moment, this timer callback cannot
|
||
# create a second UpdateChecker thread.
|
||
self._check_updates(silent=True)
|
||
|
||
if self._quitting or not self.controller.state.settings.check_updates:
|
||
return
|
||
self._app_update_timer.setInterval(APP_UPDATE_INTERVAL_MS)
|
||
self._app_update_timer.start()
|
||
|
||
def _stop_app_update_timer(self) -> None:
|
||
self._app_update_scheduler_ready = False
|
||
self._app_update_timer.stop()
|
||
|
||
def _active_update_proxy_url(self) -> str | None:
|
||
if not self.controller.connected:
|
||
return None
|
||
from ..constants import PROXY_HOST
|
||
port = self.controller.get_effective_http_proxy_port()
|
||
return f"http://{PROXY_HOST}:{port}" if port else None
|
||
|
||
def _on_update_check_error(
|
||
self,
|
||
err: str,
|
||
silent: bool,
|
||
checker: UpdateChecker | None = None,
|
||
) -> None:
|
||
if checker is not None and checker is not getattr(self, "_update_checker", None):
|
||
return
|
||
self._update_in_progress = False
|
||
if silent:
|
||
message = " ".join(str(err).split()) or "неизвестная ошибка"
|
||
# Use the file logger directly: the failure is retained in
|
||
# data/logs/app.log without producing a toast or a repeated UI
|
||
# notification for every 30-minute background attempt.
|
||
self.controller._logger.warning(
|
||
"[update] Фоновая проверка обновлений не выполнена: %s",
|
||
message,
|
||
)
|
||
return
|
||
if not silent:
|
||
self.updates_page.show_idle()
|
||
self.updates_page.set_app_error(f"Ошибка проверки: {err}")
|
||
self._show_status("error", f"Ошибка проверки обновлений: {err}")
|
||
|
||
def _on_update_check_result(
|
||
self,
|
||
update: AppUpdate | None,
|
||
silent: bool,
|
||
checker: UpdateChecker | None = None,
|
||
) -> None:
|
||
if checker is not None and checker is not getattr(self, "_update_checker", None):
|
||
return
|
||
self._update_in_progress = False
|
||
if update is None:
|
||
if not silent:
|
||
self.updates_page.show_up_to_date()
|
||
self._show_status("info", "У вас установлена последняя версия")
|
||
return
|
||
|
||
if not self.controller.state.settings.allow_updates:
|
||
self._pending_update = None
|
||
self.updates_page.show_idle()
|
||
self.updates_page.set_app_status(
|
||
f"Доступна новая версия: v{update.version}. Установка отключена в настройках"
|
||
)
|
||
if not silent:
|
||
self._show_status("warning", "Доступно обновление, но установка отключена в настройках")
|
||
return
|
||
|
||
self._pending_update = update
|
||
self.updates_page.show_update_available(update.version)
|
||
try:
|
||
self.updates_page.download_btn.clicked.disconnect()
|
||
except TypeError:
|
||
pass
|
||
self.updates_page.download_btn.clicked.connect(
|
||
lambda: self._start_update_download(self._pending_update)
|
||
)
|
||
|
||
# A silent check only suppresses routine progress, errors and the
|
||
# "up to date" message. A newer version is actionable and must still
|
||
# be shown to the user.
|
||
self._show_update_available_dialog(update, open_updates_page=not silent)
|
||
|
||
def _show_update_available_dialog(
|
||
self,
|
||
update: AppUpdate,
|
||
*,
|
||
open_updates_page: bool,
|
||
) -> None:
|
||
was_hidden = not self.isVisible()
|
||
was_minimized = self.isMinimized()
|
||
|
||
# qfluentwidgets MessageBox stays hidden when its parent is hidden.
|
||
# Bring the window forward for the prompt and restore its previous
|
||
# state when the user postpones the update.
|
||
if was_minimized:
|
||
self.showNormal()
|
||
elif was_hidden:
|
||
self.show()
|
||
self.activateWindow()
|
||
self.raise_()
|
||
|
||
if open_updates_page:
|
||
self.switchTo(self.updates_page)
|
||
|
||
from qfluentwidgets import MessageBox
|
||
box = MessageBox(
|
||
"Доступно обновление",
|
||
f"Доступна новая версия v{update.version}.\n"
|
||
f"Текущая: v{APP_VERSION}\n\n"
|
||
f"Рекомендуется обновить приложение.\n\n"
|
||
f"Приложение скачает обновление, закроется и перезапустится автоматически.",
|
||
self,
|
||
)
|
||
box.yesButton.setText("Скачать и установить")
|
||
box.cancelButton.setText("Позже")
|
||
if box.exec():
|
||
self._start_update_download(update)
|
||
return
|
||
|
||
if was_hidden:
|
||
self.hide()
|
||
elif was_minimized:
|
||
self.showMinimized()
|
||
|
||
def _start_update_download(self, update: AppUpdate) -> None:
|
||
if not self.controller.state.settings.allow_updates:
|
||
self.updates_page.show_idle()
|
||
self.updates_page.set_app_status("Установка обновлений отключена в настройках")
|
||
self._show_status("warning", "Установка обновлений отключена в настройках")
|
||
return
|
||
|
||
self._update_in_progress = True
|
||
self.switchTo(self.updates_page)
|
||
self.updates_page.show_download_progress(0)
|
||
|
||
proxy_url = self._active_update_proxy_url()
|
||
|
||
restart_in_tray = self._tray_available and not self.isVisible()
|
||
|
||
self._update_downloader = UpdateDownloader(
|
||
update,
|
||
proxy_url=proxy_url,
|
||
restart_in_tray=restart_in_tray,
|
||
parent=self,
|
||
)
|
||
self._update_downloader.progress.connect(self.updates_page.show_download_progress)
|
||
self._update_downloader.status.connect(self.updates_page.set_app_status)
|
||
self._update_downloader.finished_ok.connect(self._on_update_ready)
|
||
self._update_downloader.error.connect(self._on_update_error)
|
||
self._update_downloader.start()
|
||
|
||
def _on_update_ready(self) -> None:
|
||
self.updates_page.set_app_status("Обновление загружено. Перезапуск...")
|
||
self._show_status("success", "Обновление загружено. Перезапуск...")
|
||
QTimer.singleShot(1500, self._quit_for_update)
|
||
|
||
def _on_update_error(self, err: str) -> None:
|
||
self._update_in_progress = False
|
||
self.updates_page.show_idle()
|
||
self.updates_page.set_app_error(f"Ошибка: {err}")
|
||
self._show_status("error", err)
|
||
|
||
def _quit_for_update(self) -> None:
|
||
self._quitting = True
|
||
self._save_geometry()
|
||
self._shutdown_controller()
|
||
app = QApplication.instance()
|
||
if app is not None:
|
||
app.quit()
|
||
|
||
def _consume_update_error_log(self) -> None:
|
||
error_log = LOG_DIR / "update_error.log"
|
||
if not error_log.exists():
|
||
return
|
||
archived_log = LOG_DIR / "update_error.last.log"
|
||
try:
|
||
content = error_log.read_text(encoding="utf-8").strip()
|
||
except Exception:
|
||
content = ""
|
||
try:
|
||
if archived_log.exists():
|
||
archived_log.unlink()
|
||
error_log.replace(archived_log)
|
||
except Exception:
|
||
pass
|
||
|
||
message = "Предыдущее обновление не завершилось. См. data/logs/update_error.last.log"
|
||
if content:
|
||
self.logs_page.append_line("[update] previous install failed")
|
||
for line in content.splitlines()[:10]:
|
||
if line.strip():
|
||
self.logs_page.append_line(f"[update] {line.strip()}")
|
||
self.updates_page.set_app_error(message)
|
||
QTimer.singleShot(0, lambda: self._show_status("error", message))
|
||
|
||
def _check_xray_updates(self) -> None:
|
||
self.updates_page.set_xray_status("Проверка обновлений Xray...")
|
||
self.controller.run_xray_core_update(False, silent=False)
|
||
|
||
def _update_xray_core(self) -> None:
|
||
self.updates_page.set_xray_status("Обновление Xray...")
|
||
self.controller.run_xray_core_update(True, silent=False)
|
||
|
||
def _check_singbox_updates(self) -> None:
|
||
self.updates_page.set_singbox_status("Проверка обновлений sing-box...")
|
||
self._start_singbox_core_update(apply_update=False)
|
||
|
||
def _update_singbox_core(self) -> None:
|
||
self.updates_page.set_singbox_status("Обновление sing-box...")
|
||
self._start_singbox_core_update(apply_update=True)
|
||
|
||
def _start_singbox_core_update(self, *, apply_update: bool) -> None:
|
||
from ..engines.singbox.core_updater import SingboxCoreUpdateWorker
|
||
|
||
existing = getattr(self, "_singbox_core_worker", None)
|
||
if existing is not None and existing.isRunning():
|
||
return
|
||
worker = SingboxCoreUpdateWorker(
|
||
self.controller.state.settings.singbox_path,
|
||
apply_update,
|
||
parent=self,
|
||
)
|
||
worker.finished_with_result.connect(self._on_singbox_update_result)
|
||
self._singbox_core_worker = worker
|
||
worker.start()
|
||
|
||
def _on_singbox_update_result(self, result) -> None:
|
||
self.logs_page.append_line(f"[singbox-core-update] {result.status}: {result.message}")
|
||
if result.status == "error":
|
||
self.updates_page.set_singbox_error(result.message)
|
||
elif result.updated:
|
||
self.updates_page.set_singbox_success(result.message)
|
||
else:
|
||
self.updates_page.set_singbox_status(result.message)
|
||
version = result.current_version or result.latest_version
|
||
if version:
|
||
self.updates_page.set_singbox_version(version)
|
||
|
||
def _apply_theme(self, theme_name: str, accent_color: str) -> None:
|
||
# Deduplicated: theme.apply_theme skips setTheme/setThemeColor when
|
||
# the requested values match the last applied ones.
|
||
apply_theme(theme_name, accent_color)
|
||
# "system" mode follows the OS via SystemThemeListener; any other
|
||
# mode stops the listener.
|
||
sync_system_theme_listener(theme_name, self)
|
||
|
||
def _refresh_tray_tooltip(self) -> None:
|
||
if self.tray is None:
|
||
return
|
||
node = self.controller.selected_node
|
||
status = "Подключено" if self.controller.connected else "Отключено"
|
||
if node is not None:
|
||
node_text = node.name
|
||
elif self.controller.is_singbox_editor_mode():
|
||
node_text = self.controller.get_active_singbox_config_name()
|
||
elif self.controller.uses_xray_raw_config():
|
||
node_text = self.controller.get_active_xray_config_name()
|
||
else:
|
||
node_text = "Нет сервера"
|
||
self.tray.setToolTip(f"{APP_NAME}\n{status}\n{node_text}")
|
||
|
||
def _ensure_unlocked(self, startup: bool) -> bool:
|
||
if not self.controller.state.security.enabled:
|
||
return True
|
||
|
||
while True:
|
||
dialog = PasswordDialog("Разблокировка", self)
|
||
result = dialog.exec()
|
||
if result != int(QDialog.DialogCode.Accepted):
|
||
if startup:
|
||
self._quit_app()
|
||
return False
|
||
|
||
if self.controller.unlock(dialog.password()):
|
||
self._show_status("success", "Разблокировано")
|
||
return True
|
||
|
||
self._show_status("error", "Неверный пароль")
|
||
|
||
def _load_with_passphrase(self) -> bool:
|
||
if self.controller.load():
|
||
self._apply_window_geometry(self.controller.state.settings)
|
||
self._geometry_persistence_ready = True
|
||
return True
|
||
|
||
# State file is encrypted — ask for passphrase
|
||
while True:
|
||
dialog = PasswordDialog("Зашифрованные данные", self)
|
||
dialog.password_edit.setPlaceholderText("Введите пароль шифрования")
|
||
result = dialog.exec()
|
||
if result != int(QDialog.DialogCode.Accepted):
|
||
self._quit_app()
|
||
return False
|
||
|
||
passphrase = dialog.password()
|
||
if not passphrase:
|
||
continue
|
||
|
||
self.controller.storage.passphrase = passphrase
|
||
try:
|
||
self.controller.load()
|
||
self._geometry_persistence_ready = True
|
||
return True
|
||
except Exception:
|
||
self._show_status("error", "Неверный пароль шифрования")
|
||
|
||
def _export_backup(self) -> None:
|
||
file_path, _ = QFileDialog.getSaveFileName(
|
||
self, "Экспорт резервной копии", "xray_fluent_backup.json", "Backup files (*.json)"
|
||
)
|
||
if not file_path:
|
||
return
|
||
|
||
passphrase = ""
|
||
dialog = PasswordDialog("Зашифровать резервную копию?", self)
|
||
dialog.password_edit.setPlaceholderText("Пароль (оставьте пустым для открытого формата)")
|
||
if dialog.exec() == int(QDialog.DialogCode.Accepted):
|
||
passphrase = dialog.password()
|
||
|
||
try:
|
||
from pathlib import Path
|
||
self.controller.export_backup(Path(file_path), passphrase)
|
||
self._show_status("success", f"Резервная копия экспортирована: {file_path}")
|
||
except Exception as exc:
|
||
self._show_status("error", f"Ошибка экспорта: {exc}")
|
||
|
||
def _import_backup(self) -> None:
|
||
file_path, _ = QFileDialog.getOpenFileName(
|
||
self, "Импорт резервной копии", "", "Backup files (*.json);;All files (*)"
|
||
)
|
||
if not file_path:
|
||
return
|
||
|
||
from pathlib import Path
|
||
path = Path(file_path)
|
||
raw = path.read_text(encoding="utf-8").strip()
|
||
|
||
passphrase = ""
|
||
from ..platform.windows.security import is_passphrase_encrypted
|
||
if is_passphrase_encrypted(raw):
|
||
dialog = PasswordDialog("Расшифровать резервную копию", self)
|
||
dialog.password_edit.setPlaceholderText("Введите пароль резервной копии")
|
||
if dialog.exec() != int(QDialog.DialogCode.Accepted):
|
||
return
|
||
passphrase = dialog.password()
|
||
|
||
try:
|
||
self.controller.import_backup(path, passphrase)
|
||
self._load_config_editor_documents()
|
||
self._show_status("success", "Резервная копия успешно импортирована")
|
||
except PassphraseRequired:
|
||
self._show_status("error", "Для этой резервной копии требуется пароль")
|
||
except Exception as exc:
|
||
self._show_status("error", f"Ошибка импорта: {exc}")
|
||
|
||
def _set_encryption(self, passphrase: str) -> None:
|
||
self.controller.set_data_passphrase(passphrase)
|
||
self.settings_page.set_encryption_active(True)
|
||
|
||
def _disable_encryption(self) -> None:
|
||
self.controller.clear_data_passphrase()
|
||
self.settings_page.set_encryption_active(False)
|
||
|
||
def _toggle_window_visible(self) -> None:
|
||
if self.isVisible():
|
||
self.hide()
|
||
else:
|
||
self.show()
|
||
self.activateWindow()
|
||
self.raise_()
|
||
|
||
def _on_tray_activated(self, reason: QSystemTrayIcon.ActivationReason) -> None:
|
||
if reason in {QSystemTrayIcon.ActivationReason.Trigger, QSystemTrayIcon.ActivationReason.DoubleClick}:
|
||
self._toggle_window_visible()
|
||
|
||
def _shutdown_controller(self):
|
||
if self._startup_loader:
|
||
self._startup_loader.cancel()
|
||
if self._state_loaded:
|
||
self.controller.shutdown()
|
||
else:
|
||
# Cancelling startup must never overwrite the real state with defaults.
|
||
self.controller.network_monitor.stop()
|
||
|
||
def _quit_app(self) -> None:
|
||
self._quitting = True
|
||
if self._startup_loader:
|
||
self._startup_loader.cancel()
|
||
self._stop_app_update_timer()
|
||
self._save_geometry()
|
||
self._shutdown_controller()
|
||
app = QApplication.instance()
|
||
if app is not None:
|
||
app.quit()
|
||
|
||
def _save_geometry(self) -> None:
|
||
controller = getattr(self, "controller", None)
|
||
if controller is None or self._restoring_geometry or not self._geometry_persistence_ready:
|
||
return
|
||
if self.isMinimized():
|
||
return
|
||
geo = self.normalGeometry() if self.isMaximized() else self.geometry()
|
||
s = controller.state.settings
|
||
s.window_x = geo.x()
|
||
s.window_y = geo.y()
|
||
s.window_width = geo.width()
|
||
s.window_height = geo.height()
|
||
if self._geometry_persistence_ready:
|
||
controller.schedule_save()
|
||
|
||
def moveEvent(self, e) -> None:
|
||
super().moveEvent(e)
|
||
self._save_geometry()
|
||
|
||
def resizeEvent(self, e) -> None:
|
||
super().resizeEvent(e)
|
||
self._save_geometry()
|
||
|
||
def closeEvent(self, e: QCloseEvent) -> None:
|
||
if self._quitting:
|
||
e.accept()
|
||
return
|
||
|
||
if not self._tray_available:
|
||
self._quitting = True
|
||
self._stop_app_update_timer()
|
||
self._save_geometry()
|
||
self._shutdown_controller()
|
||
e.accept()
|
||
app = QApplication.instance()
|
||
if app is not None:
|
||
app.quit()
|
||
return
|
||
|
||
self._save_geometry()
|
||
if self._state_loaded:
|
||
self.controller.save()
|
||
e.ignore()
|
||
self.hide()
|
||
if self.tray is not None and not self._tray_notified:
|
||
self.tray.showMessage(APP_NAME, "Приложение свёрнуто в системный трей", QSystemTrayIcon.MessageIcon.Information, 2000)
|
||
self._tray_notified = True
|