import os import sys import threading from urllib import parse, request import requests from PySide6.QtCore import QTimer, Qt, QObject, Slot from PySide6.QtWidgets import QMainWindow, QWidget, QVBoxLayout, QMessageBox from app.application.services import AppConfig, BackendRuntime, FrontendRuntime from app.presentation.shell import LoadingWindowController, WebViewShellController print(f"[GUI] Python executable: {sys.executable}") try: from PySide6.QtWebEngineWidgets import QWebEngineView from PySide6.QtWebChannel import QWebChannel WEB_ENGINE_AVAILABLE = True except Exception: WEB_ENGINE_AVAILABLE = False QWebEngineView = None QWebChannel = None class WindowBridge(QObject): def __init__(self, main_window): super().__init__() self.main_window = main_window @Slot() def open_devtools(self): self.main_window.open_devtools() @Slot() def minimize(self): self.main_window.showMinimized() @Slot() def maximize_or_restore(self): if self.main_window.isMaximized(): self.main_window.showNormal() else: self.main_window.showMaximized() @Slot() def close_window(self): self.main_window.close() @Slot() def start_move(self): wh = self.main_window.windowHandle() if wh is not None: wh.startSystemMove() class MainWindow(QMainWindow): def post_settings(self, **kwargs): if not self.base_url: return try: payload = {"action": "settings_set"} for k, v in kwargs.items(): payload[k] = "1" if v is True else "0" if v is False else str(v) data = parse.urlencode(payload).encode("utf-8") req = request.Request(self.base_url + "/api/rules", data=data, method="POST") with request.urlopen(req, timeout=2): pass except Exception: pass def __init__(self): super().__init__() self.setWindowTitle("识流 AI 助手") self.resize(1280, 920) self.setWindowFlags(Qt.FramelessWindowHint | Qt.Window) self.hide() self.config = AppConfig() self.backend_runtime = BackendRuntime(self.config) self.frontend_runtime = FrontendRuntime(self.config, self.backend_runtime) self.base_url = os.getenv("APP_BASE_URL", self.backend_runtime.base_url) self.backend_process = None self.backend_started_by_gui = False self.backend_boot_thread = None self.web_channel = None self.window_bridge = None self.loading = LoadingWindowController() self.web_shell = None self.frontend_entry_url = "" central = QWidget() central.setStyleSheet("background:#eef2f7;") self.setCentralWidget(central) root = QVBoxLayout(central) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) self.web = None self.devtools_view = None if WEB_ENGINE_AVAILABLE: self.web = QWebEngineView() self.web.setStyleSheet("background:#eef2f7;") root.addWidget(self.web, 1) self.web_channel = QWebChannel(self.web.page()) self.window_bridge = WindowBridge(self) self.web_channel.registerObject("windowBridge", self.window_bridge) self.web.page().setWebChannel(self.web_channel) self.web.loadFinished.connect(self.on_main_page_loaded) self.web_shell = WebViewShellController(self.web) self.listener_sync_timer = QTimer(self) self.listener_sync_timer.setInterval(30000) self.listener_sync_timer.timeout.connect(self.sync_listener_state) self._syncing_listener_state = False self.backend_ready_timer = QTimer(self) self.backend_ready_timer.setInterval(100) self.backend_ready_timer.timeout.connect(self.consume_backend_boot_result) self._backend_boot_result = None self.loading.show() self.start_backend() def get_settings(self): if not self.base_url: return {} try: url = self.base_url + "/api/rules?action=settings_get" with request.urlopen(url, timeout=2) as resp: data = resp.read().decode("utf-8") import json payload = json.loads(data) return payload if isinstance(payload, dict) else {} except Exception: return {} def sync_listener_state(self): if self._syncing_listener_state: return self._syncing_listener_state = True try: if not self.base_url: return requests.get(self.base_url + "/api/bot/status", timeout=2) except Exception: pass finally: self._syncing_listener_state = False def start_backend(self): self.base_url = self.backend_runtime.base_url self._backend_boot_result = None self.backend_boot_thread = threading.Thread(target=self._boot_backend_worker, daemon=True) self.backend_boot_thread.start() self.backend_ready_timer.start() def _boot_backend_worker(self): try: if self.backend_runtime.is_backend_alive(): self.backend_started_by_gui = False self._backend_boot_result = {"ok": True, "url": self.base_url} return process = self.backend_runtime.launch_backend_process() self.backend_process = process self.backend_started_by_gui = True ok, message = self.frontend_runtime.wait_for_backend(process=process) if ok: self._backend_boot_result = {"ok": True, "url": self.base_url} else: self._backend_boot_result = {"ok": False, "message": message} except Exception as exc: self._backend_boot_result = {"ok": False, "message": str(exc)} def consume_backend_boot_result(self): if self._backend_boot_result is None: return result = self._backend_boot_result self._backend_boot_result = None self.backend_ready_timer.stop() if result.get("ok"): self.on_backend_started(result["url"]) return self.on_backend_failed(result.get("message") or "后台启动失败") def on_backend_started(self, url): self.base_url = url self.frontend_entry_url = self.frontend_runtime.resolve_frontend_entry_url(url) self.reload_admin_page() self.listener_sync_timer.start() def on_backend_failed(self, msg): self.loading.close() QMessageBox.critical(self, "后台错误", msg) def reload_admin_page(self): self.web_shell.load(self.frontend_entry_url) def on_main_page_loaded(self, ok): if not self.web_shell.mark_loaded(ok): return self.loading.finish_with(self._finish_show_main) def _finish_show_main(self): if not self.web_shell.can_finish(): return self.web_shell.finish() self.show() self.raise_() self.activateWindow() QTimer.singleShot(60, self.loading.close) def open_devtools(self): if self.web is None: return if self.devtools_view is None: self.devtools_view = QWebEngineView() self.devtools_view.setWindowTitle("页面调试控制台") self.devtools_view.resize(1100, 760) self.web.page().setDevToolsPage(self.devtools_view.page()) self.devtools_view.show() self.devtools_view.raise_() self.devtools_view.activateWindow() def closeEvent(self, event): try: self.listener_sync_timer.stop() except Exception: pass try: self.backend_ready_timer.stop() except Exception: pass try: if self.backend_started_by_gui: self.backend_runtime.terminate_backend_process(self.backend_process) except Exception: pass super().closeEvent(event)