from essentials.observers.ConnectionObserver import ConnectionObserver from core.observers.ClientObserver import ClientObserver from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType from core.controllers.SyncController import new_sync from core.models.Result import Result, ResultError from PyQt6.QtWidgets import (QDialog, QVBoxLayout, QHBoxLayout, QTextEdit, QLabel, QPushButton, QApplication) from PyQt6.QtGui import QFont, QColor, QTextCursor from PyQt6.QtCore import Qt, QTimer, QThread, pyqtSignal, pyqtSlot from PyQt6.QtWidgets import QGraphicsDropShadowEffect import sys import traceback from typing import Callable, Any from PyQt6.QtGui import QTextCharFormat from gui.v2.infrastructure.ThreadSafetyTool import ThreadSafetyTool import threading from threading import Thread # mainthread = ThreadSafetyTool() class NotificationTerminalUI(QDialog): def __init__(self, connection_observer: ConnectionObserver, client_observer: ClientObserver): super().__init__() self.mainthread = ThreadSafetyTool() self.connection_observer = connection_observer self.client_observer = client_observer self.synchronized = False self.worker_thread = None self.task_success = False self.connection_observer.subscribe("connecting", lambda msg: self.add_notification("CONNECT", msg, "#00ff88")) self.connection_observer.subscribe("tor_bootstrapping", lambda msg: self.add_notification("BOOTSTRAP", msg, "#00d4ff")) self.connection_observer.subscribe("tor_bootstrap_progressing", lambda msg: self.add_notification("PROGRESS", msg, "#ff0088")) self.connection_observer.subscribe("tor_bootstrapped", lambda msg: self.add_notification("SUCCESS", msg, "#00ff88")) self.connection_observer.subscribe("custom_message", lambda msg: self.add_notification("INFO", msg, "#ffff00")) # Subscribe to client_observer events self.client_observer.subscribe("synchronizing", lambda event: self.add_notification("SYNC", event.subject if event.subject else "Sync in progress...", "#00d4ff")) self.client_observer.subscribe("synchronized", lambda event: self.on_synchronized()) self.client_observer.subscribe("updating", lambda event: self.add_notification("UPDATE", "Updating client...", "#ff0088")) self.client_observer.subscribe("update_progressing", lambda event: self.add_notification("PROGRESS", f'Current progress: {event.meta.get("progress", 0):.2f}%', "#ffff00")) self.client_observer.subscribe("updated", lambda event: self.add_notification("COMPLETE", "Restart client to apply update.", "#00ff88")) self.client_observer.subscribe("custom_message", lambda event: self.add_notification("ERROR", event.subject if event.subject else "Error, check logs", "#ff0088")) self.init_ui() def init_ui(self): self.setWindowTitle("SYNC PROCESS") self.setGeometry(100, 100, 700, 600) self.setModal(True) layout = QVBoxLayout(self) layout.setSpacing(15) layout.setContentsMargins(30, 30, 30, 30) self.setStyleSheet(""" QDialog { background-color: #0a0e27; color: #00ff88; } """) # Title Label title_label = QLabel("▓░ NOTIFICATION TERMINAL ░▓") title_label.setFont(QFont("Courier New", 14, QFont.Weight.Bold)) title_label.setAlignment(Qt.AlignmentFlag.AlignCenter) title_label.setStyleSheet(""" QLabel { color: #00ff88; background-color: transparent; padding: 15px; border: 2px solid #00ff88; border-radius: 5px; letter-spacing: 2px; } """) title_shadow = QGraphicsDropShadowEffect() title_shadow.setBlurRadius(15) title_shadow.setColor(QColor(0, 255, 136, 200)) title_shadow.setOffset(0, 0) title_label.setGraphicsEffect(title_shadow) layout.addWidget(title_label) # Scrollable text display self.output_text = QTextEdit() self.output_text.setReadOnly(True) self.output_text.setFont(QFont("Courier New", 10)) self.output_text.setMinimumHeight(400) self.output_text.setStyleSheet(""" QTextEdit { background-color: #1a1f3a; color: #00ff88; border: 2px solid #00ff88; border-radius: 5px; padding: 15px; font-size: 16pt; font-family: 'Courier New', monospace; selection-background-color: #ff0088; } QScrollBar:vertical { background-color: #0a0e27; width: 12px; border-radius: 6px; } QScrollBar::handle:vertical { background-color: #00ff88; border-radius: 6px; min-height: 20px; } QScrollBar::handle:vertical:hover { background-color: #00d4ff; } """) layout.addWidget(self.output_text) # Status label self.status_label = QLabel("◄ WAITING FOR TASK START ►") self.status_label.setFont(QFont("Courier New", 9)) self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self.status_label.setStyleSheet(""" QLabel { color: #ff0088; background-color: transparent; padding: 8px; letter-spacing: 1px; } """) layout.addWidget(self.status_label) # Button layout button_layout = QHBoxLayout() clear_btn = QPushButton("CLEAR") clear_btn.setFont(QFont("Courier New", 10, QFont.Weight.Bold)) clear_btn.setMinimumHeight(40) clear_btn.setCursor(Qt.CursorShape.PointingHandCursor) clear_btn.setStyleSheet(""" QPushButton { background-color: #1a1f3a; color: #00d4ff; border: 2px solid #00d4ff; border-radius: 5px; padding: 8px 20px; font-weight: bold; } QPushButton:hover { background-color: #00d4ff; color: #0a0e27; } QPushButton:pressed { background-color: #0a0e27; border: 3px solid #00d4ff; } """) clear_btn.clicked.connect(self.clear_output) shadow = QGraphicsDropShadowEffect() shadow.setBlurRadius(15) shadow.setColor(QColor(0, 212, 255)) shadow.setOffset(0, 0) clear_btn.setGraphicsEffect(shadow) button_layout.addWidget(clear_btn) button_layout.addStretch() layout.addLayout(button_layout) def add_notification(self, prefix: str, message: str, color: str = "#00ff88"): """Queue to main thread safely.""" self.mainthread(lambda: self._add_notification_impl(prefix, message, color))() def _add_notification_impl(self, prefix: str, message: str, color: str): """Actual implementation on main thread.""" cursor = self.output_text.textCursor() cursor.movePosition(cursor.MoveOperation.End) fmt = QTextCharFormat() fmt.setForeground(QColor(color)) cursor.insertText(f"[{prefix}] {message}\n", fmt) self.output_text.setTextCursor(cursor) self.output_text.ensureCursorVisible() def set_status(self, text: str): """This runs on the main thread, safely.""" self.status_label.setText(text) def on_synchronized(self): """Handle synchronized notification and exit.""" self.add_notification("SYNC", "Sync complete ✓", "#00ff88") self.set_status("◄ SYNCHRONIZED - CLOSING ►") self.synchronized = True self.task_success = True QTimer.singleShot(500, self.accept) def run_task(self, task_func: Callable): """ Purpose: Insert any function for a pure python background thread. Which then pushes to the main UI via ThreadSafety """ if self.worker_thread is not None and self.worker_thread.is_alive(): self.add_notification("SYSTEM", "Task already running!", "#ff0088") return self.set_status("◄ TASK RUNNING ►") self.add_notification("SYSTEM", "Background task started", "#00d4ff") # Start in pure Python thread (daemon=True so it doesn't block exit) self.worker_thread = Thread(target=self._run_task_wrapper, args=(task_func,), daemon=True) self.worker_thread.start() def _run_task_wrapper(self, task_func: Callable): """Wrapper that runs task and handles errors.""" try: task_func() self.task_success = True except Exception as e: self.add_notification("ERROR", f"Task failed: {str(e)}", "#ff0088") self.set_status("◄ TASK FAILED ►") self.task_success = False def clear_output(self): """Clear the terminal output.""" self.output_text.clear() self.add_notification("SYSTEM", "Terminal cleared", "#00d4ff") def get_result(self) -> bool: """Return True if synchronized, False otherwise.""" return self.synchronized and self.task_success def closeEvent(self, event): """Ensure thread is properly cleaned up on close.""" if self.worker_thread is not None and self.worker_thread.is_alive(): self.worker_thread.join(timeout=1) event.accept() def show_terminal( do_this_function: Callable[[Any, Any], None], connection_observer: ConnectionObserver, client_observer: ClientObserver ) -> bool: app = QApplication.instance() or QApplication(sys.argv) # Create the UI dialog = NotificationTerminalUI(connection_observer=connection_observer, client_observer=client_observer) # Start the background task dialog.run_task(do_this_function) # Show dialog and wait dialog.exec() # Return success return dialog.get_result()