import sys import json import socket import subprocess import threading import time from PyQt6.QtCore import Qt, pyqtSignal, QObject from PyQt6.QtWidgets import QApplication, QVBoxLayout, QHBoxLayout, QWidget from qfluentwidgets import (PrimaryPushButton, PushButton, SubtitleLabel, MessageBox, InfoBar, InfoBarPosition, BodyLabel, CardWidget, setTheme, Theme) import websocket # --- НАСТРОЙКИ --- SERVER_IP = "192.168.0.8" # <--- ВПИШИ СЮДА IP ТВОЕГО КОНТЕЙНЕРА! SERVER_PORT = "8000" VH_PATH = "vhui64.exe" # Программа должна лежать в той же папке VH_DEVICE_ADDRESS = "Rutoken.31" # Аппаратный ID твоего Рутокена CLIENT_NAME = socket.gethostname() # Имя текущего компьютера # ----------------- 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=self.on_error, on_close=self.on_close ) self.ws.run_forever() except Exception as e: 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 on_error(self, ws, error): pass def on_close(self, ws, close_status_code, close_msg): pass 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 EcpClientWindow(QWidget): def __init__(self): super().__init__() self.current_owner = None self.signals = WorkerSignals() self.ws_thread = WsClient(self.signals) self.setup_ui() self.setup_signals() # Запускаем фоновый поток для связи с сервером self.ws_thread.start() def setup_ui(self): self.setWindowTitle("Менеджер ЭЦП Рутокен") self.resize(400, 200) self.setStyleSheet("background-color: #202020; color: white;") layout = QVBoxLayout(self) layout.setContentsMargins(20, 20, 20, 20) self.card = CardWidget(self) card_layout = QVBoxLayout(self.card) 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 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) 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) else: self.status_label.setText("🔴 ЭЦП Занята") self.owner_label.setText(f"Сейчас использует: {self.current_owner}") self.btn_connect.setEnabled(True) self.btn_disconnect.setEnabled(False) 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): 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 USE,{VH_DEVICE_ADDRESS}"], creationflags=subprocess.CREATE_NO_WINDOW ) except Exception as e: pass def closeEvent(self, event): self.ws_thread.running = False if self.current_owner == CLIENT_NAME: self.ws_thread.send_action("release") event.accept() if __name__ == '__main__': setTheme(Theme.DARK) app = QApplication(sys.argv) window = EcpClientWindow() window.show() sys.exit(app.exec())