import sys import json import socket import subprocess import threading import time import os from PyQt6.QtCore import Qt, pyqtSignal, QObject from PyQt6.QtWidgets import (QApplication, QVBoxLayout, QHBoxLayout, QWidget, QFrame, QLabel, QLineEdit, QDialog, QFormLayout, QSystemTrayIcon, QMenu) from PyQt6.QtGui import QIcon, QPixmap, QPainter, QColor from qfluentwidgets import (PrimaryPushButton, PushButton, SubtitleLabel, MessageBox, InfoBar, InfoBarPosition, BodyLabel, setTheme, Theme, ToolButton, FluentIcon) import websocket # Файл конфигурации для сохранения IP и имени CONFIG_FILE = "config.json" def load_config(): if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: return json.load(f) except: pass return {"server_ip": "192.168.1.50", "client_name": socket.gethostname()} def save_config(config): with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(config, f, ensure_ascii=False, indent=4) config = load_config() SERVER_IP = config.get("server_ip", "192.168.1.50") CLIENT_NAME = config.get("client_name", socket.gethostname()) SERVER_PORT = "8000" VH_PATH = "vhui64.exe" VH_DEVICE_ADDRESS = "Rutoken.31" class WorkerSignals(QObject): state_changed = pyqtSignal(dict) command_received = pyqtSignal(str) prompt_received = pyqtSignal(str) notify_received = pyqtSignal(str) class WsClient(threading.Thread): def __init__(self, signals: WorkerSignals): super().__init__() self.signals = signals self.ws = None self.running = True def run(self): while self.running: try: self.ws = websocket.WebSocketApp( f"ws://{SERVER_IP}:{SERVER_PORT}/ws/{CLIENT_NAME}", on_message=self.on_message, on_error=lambda w, e: None, on_close=lambda w, c, m: None ) self.ws.run_forever() except: time.sleep(2) def on_message(self, ws, message): data = json.loads(message) msg_type = data.get("type") if msg_type == "state": self.signals.state_changed.emit(data) elif msg_type == "command": self.signals.command_received.emit(data.get("cmd")) elif msg_type == "prompt_request": self.signals.prompt_received.emit(data.get("from")) elif msg_type == "notify": self.signals.notify_received.emit(data.get("text")) def send_action(self, action: str, extra: dict = None): if self.ws and self.ws.sock and self.ws.sock.connected: msg = {"action": action} if extra: msg.update(extra) self.ws.send(json.dumps(msg)) class SettingsDialog(QDialog): def __init__(self, parent=None): super().__init__(parent) self.setWindowTitle("Настройки") self.resize(350, 180) self.setStyleSheet("background-color: #202020; color: white;") layout = QFormLayout(self) layout.setContentsMargins(20, 20, 20, 20) self.ip_input = QLineEdit(SERVER_IP, self) self.name_input = QLineEdit(CLIENT_NAME, self) layout.addRow(BodyLabel("IP сервера LXC:", self), self.ip_input) layout.addRow(BodyLabel("Имя ПК / Отдела:", self), self.name_input) btn_layout = QHBoxLayout() save_btn = PrimaryPushButton("Сохранить", self) cancel_btn = PushButton("Отмена", self) save_btn.clicked.connect(self.save) cancel_btn.clicked.connect(self.reject) btn_layout.addWidget(save_btn) btn_layout.addWidget(cancel_btn) layout.addRow(btn_layout) def save(self): global SERVER_IP, CLIENT_NAME SERVER_IP = self.ip_input.text().strip() CLIENT_NAME = self.name_input.text().strip() save_config({"server_ip": SERVER_IP, "client_name": CLIENT_NAME}) self.accept() class EcpClientWindow(QWidget): def __init__(self): super().__init__() self.current_owner = None self.is_connected_to_us = False self.signals = WorkerSignals() self.ws_thread = WsClient(self.signals) self.setup_ui() self.setup_tray() self.setup_signals() self.ws_thread.start() def setup_ui(self): self.setWindowTitle("Менеджер ЭЦП Рутокен") self.resize(420, 230) self.setStyleSheet("background-color: #202020; color: white;") layout = QVBoxLayout(self) layout.setContentsMargins(20, 20, 20, 20) layout.setSpacing(15) # Верхняя панель с заголовком и кнопкой настроек (шестеренка) top_layout = QHBoxLayout() title_label = SubtitleLabel("Управление ЭЦП", self) self.btn_settings = ToolButton(FluentIcon.SETTING, self) self.btn_settings.clicked.connect(self.open_settings) top_layout.addWidget(title_label) top_layout.addStretch() top_layout.addWidget(self.btn_settings) layout.addLayout(top_layout) # Карточка-индикатор со сменой цвета фона рамки self.card = QFrame(self) self.update_card_style("free") card_layout = QVBoxLayout(self.card) card_layout.setContentsMargins(15, 20, 15, 20) card_layout.setSpacing(8) self.status_label = SubtitleLabel("Подключение к серверу...", self) self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self.owner_label = BodyLabel("", self) self.owner_label.setAlignment(Qt.AlignmentFlag.AlignCenter) card_layout.addWidget(self.status_label) card_layout.addWidget(self.owner_label) layout.addWidget(self.card) # Кнопки btn_layout = QHBoxLayout() self.btn_connect = PrimaryPushButton("Подключить ЭЦП", self) self.btn_disconnect = PushButton("Отключить", self) self.btn_connect.clicked.connect(self.on_connect_clicked) self.btn_disconnect.clicked.connect(self.on_disconnect_clicked) self.btn_connect.setEnabled(False) self.btn_disconnect.setEnabled(False) btn_layout.addWidget(self.btn_connect) btn_layout.addWidget(self.btn_disconnect) layout.addLayout(btn_layout) def update_card_style(self, mode): # Цвета рамок в зависимости от состояния colors = { "free": ("#1e3b22", "#2e7d32"), # Зеленоватый (Свободно) "mine": ("#1a2b3c", "#1565c0"), # Синеватый (У нас) "busy": ("#3c1a1a", "#c62828") # Красноватый (Занята другим) } bg, border = colors.get(mode, colors["free"]) self.card.setStyleSheet(f""" QFrame {{ background-color: {bg}; border: 2px solid {border}; border-radius: 10px; } """) def setup_tray(self): self.tray_icon = QSystemTrayIcon(self) self.update_tray_icon("free") tray_menu = QMenu() show_action = tray_menu.addAction("Открыть") show_action.triggered.connect(self.showNormal) self.tray_connect_action = tray_menu.addAction("Подключить ЭЦП") self.tray_connect_action.triggered.connect(self.on_connect_clicked) self.tray_disconnect_action = tray_menu.addAction("Освободить ЭЦП") self.tray_disconnect_action.triggered.connect(self.on_disconnect_clicked) self.tray_disconnect_action.setEnabled(False) tray_menu.addSeparator() quit_action = tray_menu.addAction("Выход") quit_action.triggered.connect(self.quit_app) self.tray_icon.setContextMenu(tray_menu) self.tray_icon.show() self.tray_icon.activated.connect(self.tray_icon_activated) def update_tray_icon(self, mode): # Рисуем красивый кружок для трея программно pixmap = QPixmap(32, 32) pixmap.fill(Qt.GlobalColor.transparent) painter = QPainter(pixmap) painter.setRenderHint(QPainter.RenderHint.Antialiasing) colors = { "free": QColor(46, 125, 50), # Зеленый "mine": QColor(21, 101, 192), # Синий "busy": QColor(198, 40, 40) # Красный } painter.setBrush(colors.get(mode, colors["free"])) painter.setPen(Qt.PenStyle.NoPen) painter.drawEllipse(4, 4, 24, 24) painter.end() self.tray_icon.setIcon(QIcon(pixmap)) def tray_icon_activated(self, reason): if reason == QSystemTrayIcon.ActivationReason.Trigger: if self.isVisible(): self.hide() else: self.showNormal() self.activateWindow() def setup_signals(self): self.signals.state_changed.connect(self.update_state) self.signals.command_received.connect(self.execute_command) self.signals.prompt_received.connect(self.show_prompt) self.signals.notify_received.connect(self.show_notify) def update_state(self, state: dict): status = state.get("status") self.current_owner = state.get("owner") if status == "free": self.status_label.setText("ЭЦП Свободна") self.owner_label.setText("Нажмите для подключения") self.btn_connect.setEnabled(True) self.btn_disconnect.setEnabled(False) self.tray_connect_action.setEnabled(True) self.tray_disconnect_action.setEnabled(False) self.update_card_style("free") self.update_tray_icon("free") else: if self.current_owner == CLIENT_NAME: self.status_label.setText("ЭЦП подключена к вам") self.owner_label.setText("Не забудьте отключить после работы!") self.btn_connect.setEnabled(False) self.btn_disconnect.setEnabled(True) self.tray_connect_action.setEnabled(False) self.tray_disconnect_action.setEnabled(True) self.update_card_style("mine") self.update_tray_icon("mine") else: self.status_label.setText("ЭЦП Занята") self.owner_label.setText(f"Использует: {self.current_owner}") self.btn_connect.setEnabled(True) self.btn_disconnect.setEnabled(False) self.tray_connect_action.setEnabled(True) self.tray_disconnect_action.setEnabled(False) self.update_card_style("busy") self.update_tray_icon("busy") def open_settings(self): dialog = SettingsDialog(self) if dialog.exec(): # Переподключаемся с новым IP при изменении self.show_notify("Настройки сохранены. Перезапустите приложение для применения IP.", 4000) def on_connect_clicked(self): if not self.current_owner: self.ws_thread.send_action("take_free") else: msg = MessageBox("ЭЦП сейчас используется", f"ЭЦП подключена к {self.current_owner}. Что делаем?", self) msg.yesButton.setText("Принудительно забрать") msg.cancelButton.setText("Отмена") btn_wait = PushButton("Ожидать (2 мин)") msg.buttonGroup.addWidget(btn_wait) msg.buttonLayout.insertWidget(1, btn_wait) msg.yesButton.clicked.disconnect() msg.yesButton.clicked.connect(lambda: [msg.accept(), self.ws_thread.send_action("force_take")]) btn_wait.clicked.connect(lambda: [msg.accept(), self.ws_thread.send_action("request_take"), self.show_notify("Запрос отправлен. Ожидаем ответа...", duration=120000)]) msg.exec() def on_disconnect_clicked(self): self.ws_thread.send_action("release") def show_prompt(self, from_user): self.showNormal() msg = MessageBox("Запрос доступа", f"{from_user} хочет подключить ЭЦП. Отдать?", self) msg.yesButton.setText("Да") msg.cancelButton.setText("Нет") if msg.exec(): self.ws_thread.send_action("request_response", {"accept": True}) else: self.ws_thread.send_action("request_response", {"accept": False}) def show_notify(self, text, duration=5000): InfoBar.info( title="Уведомление", content=text, orient=Qt.Orientation.Horizontal, isClosable=True, position=InfoBarPosition.TOP, duration=duration, parent=self ) def execute_command(self, cmd): if cmd == "attach": try: subprocess.run( [VH_PATH, "-t", f"USE,{VH_DEVICE_ADDRESS}"], creationflags=subprocess.CREATE_NO_WINDOW ) self.show_notify("ЭЦП успешно подключена!", 3000) except Exception as e: self.show_notify(f"Ошибка attach: {e}") elif cmd == "detach": try: subprocess.run( [VH_PATH, "-t", f"STOP USING,{VH_DEVICE_ADDRESS}"], creationflags=subprocess.CREATE_NO_WINDOW ) except Exception as e: pass def closeEvent(self, event): # Сворачиваем в трей вместо закрытия при нажатии на крестик event.ignore() self.hide() self.tray_icon.showMessage( "Менеджер ЭЦП", "Приложение свернуто в трей и продолжает работать.", QSystemTrayIcon.MessageIcon.Information, 2000 ) def quit_app(self): self.ws_thread.running = False if self.current_owner == CLIENT_NAME: self.ws_thread.send_action("release") self.tray_icon.hide() QApplication.quit() if __name__ == '__main__': setTheme(Theme.DARK) app = QApplication(sys.argv) app.setQuitOnLastWindowClosed(False) window = EcpClientWindow() window.show() sys.exit(app.exec())