398 lines
15 KiB
Python
398 lines
15 KiB
Python
"""OMS 小红书 Agent 桌面主窗口。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import sys
|
||
import webbrowser
|
||
from datetime import datetime
|
||
|
||
from PyQt6.QtCore import QTimer, pyqtSignal
|
||
from PyQt6.QtGui import QCloseEvent, QFont
|
||
from PyQt6.QtWidgets import (
|
||
QApplication,
|
||
QFormLayout,
|
||
QGroupBox,
|
||
QHBoxLayout,
|
||
QLabel,
|
||
QLineEdit,
|
||
QMainWindow,
|
||
QMessageBox,
|
||
QPlainTextEdit,
|
||
QPushButton,
|
||
QSpinBox,
|
||
QVBoxLayout,
|
||
QWidget,
|
||
)
|
||
|
||
from app.agent_worker import AgentWorker
|
||
from app.config_store import AgentSettings
|
||
from app.engine_path import ensure_runtime_importable, extension_dir
|
||
from app.pair_server import DEFAULT_PAIR_PORT, PairPayload, PairServer
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PAIR_TIMEOUT_SEC = 120
|
||
|
||
|
||
class MainWindow(QMainWindow):
|
||
pair_received = pyqtSignal(object)
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.setWindowTitle("OMS 小红书 Agent")
|
||
self.setMinimumSize(640, 520)
|
||
self.resize(720, 580)
|
||
|
||
self._settings = AgentSettings.load()
|
||
self._worker: AgentWorker | None = None
|
||
self._bridge = None
|
||
self._pair_server: PairServer | None = None
|
||
self._pair_timer: QTimer | None = None
|
||
|
||
root = QWidget()
|
||
self.setCentralWidget(root)
|
||
layout = QVBoxLayout(root)
|
||
layout.setContentsMargins(16, 16, 16, 16)
|
||
layout.setSpacing(12)
|
||
|
||
layout.addWidget(self._build_intro())
|
||
layout.addWidget(self._build_connection_group())
|
||
layout.addWidget(self._build_local_group())
|
||
layout.addWidget(self._build_status_group())
|
||
layout.addWidget(self._build_log_group(), 1)
|
||
|
||
self.pair_received.connect(self._on_pair_received)
|
||
self._apply_settings_to_form()
|
||
self._try_init_bridge()
|
||
|
||
def _build_intro(self) -> QLabel:
|
||
lbl = QLabel(
|
||
"打开本程序并点击「连接 OMS」即可。"
|
||
"请先在浏览器登录 OMS；若已打开「启动服务」页面，将自动完成授权，无需填写 API 或 Token。"
|
||
)
|
||
lbl.setWordWrap(True)
|
||
lbl.setStyleSheet("color:#64748b;font-size:12px;")
|
||
return lbl
|
||
|
||
def _build_connection_group(self) -> QGroupBox:
|
||
box = QGroupBox("OMS 连接")
|
||
form = QFormLayout(box)
|
||
form.setSpacing(8)
|
||
|
||
self.input_device_name = QLineEdit()
|
||
self.input_device_name.setPlaceholderText("如：办公室电脑")
|
||
form.addRow("设备名称", self.input_device_name)
|
||
|
||
self.input_bridge_port = QSpinBox()
|
||
self.input_bridge_port.setRange(1024, 65535)
|
||
self.input_bridge_port.setValue(9333)
|
||
form.addRow("Bridge 端口", self.input_bridge_port)
|
||
|
||
self.lbl_extension_path = QLabel(extension_dir().as_posix())
|
||
self.lbl_extension_path.setWordWrap(True)
|
||
self.lbl_extension_path.setStyleSheet("color:#475569;font-size:11px;")
|
||
form.addRow("Chrome 扩展", self.lbl_extension_path)
|
||
|
||
row = QHBoxLayout()
|
||
self.btn_save = QPushButton("保存配置")
|
||
self.btn_save.clicked.connect(self._save_settings)
|
||
self.btn_connect = QPushButton("连接 OMS")
|
||
self.btn_connect.setStyleSheet(
|
||
"background:#4F46E5;color:#fff;padding:6px 16px;border-radius:4px;"
|
||
)
|
||
self.btn_connect.clicked.connect(self._toggle_agent)
|
||
self.btn_reauth = QPushButton("重新授权")
|
||
self.btn_reauth.clicked.connect(self._reauthorize)
|
||
row.addWidget(self.btn_save)
|
||
row.addWidget(self.btn_reauth)
|
||
row.addStretch(1)
|
||
row.addWidget(self.btn_connect)
|
||
form.addRow("", row)
|
||
return box
|
||
|
||
def _build_local_group(self) -> QGroupBox:
|
||
box = QGroupBox("本机引擎（也可在网页远程控制）")
|
||
row = QHBoxLayout(box)
|
||
self.btn_engine_start = QPushButton("启动引擎")
|
||
self.btn_engine_stop = QPushButton("停止引擎")
|
||
self.btn_login = QPushButton("浏览器登录")
|
||
self.btn_check = QPushButton("刷新状态")
|
||
for btn in (self.btn_engine_start, self.btn_engine_stop, self.btn_login, self.btn_check):
|
||
row.addWidget(btn)
|
||
self.btn_engine_start.clicked.connect(self._local_start_engine)
|
||
self.btn_engine_stop.clicked.connect(self._local_stop_engine)
|
||
self.btn_login.clicked.connect(self._local_browser_login)
|
||
self.btn_check.clicked.connect(self._refresh_local_status)
|
||
return box
|
||
|
||
def _build_status_group(self) -> QGroupBox:
|
||
box = QGroupBox("状态")
|
||
grid = QFormLayout(box)
|
||
mono = QFont("Consolas", 10)
|
||
self.lbl_device_id = QLabel("-")
|
||
self.lbl_device_id.setFont(mono)
|
||
self.lbl_agent_state = QLabel("未连接")
|
||
self.lbl_engine = QLabel("-")
|
||
self.lbl_extension = QLabel("-")
|
||
self.lbl_login = QLabel("-")
|
||
grid.addRow("设备 ID", self.lbl_device_id)
|
||
grid.addRow("Agent", self.lbl_agent_state)
|
||
grid.addRow("引擎", self.lbl_engine)
|
||
grid.addRow("扩展", self.lbl_extension)
|
||
grid.addRow("小红书", self.lbl_login)
|
||
return box
|
||
|
||
def _build_log_group(self) -> QGroupBox:
|
||
box = QGroupBox("运行日志")
|
||
lay = QVBoxLayout(box)
|
||
self.log_view = QPlainTextEdit()
|
||
self.log_view.setReadOnly(True)
|
||
self.log_view.setMaximumBlockCount(500)
|
||
lay.addWidget(self.log_view)
|
||
return box
|
||
|
||
def _apply_settings_to_form(self) -> None:
|
||
self.input_device_name.setText(self._settings.device_name)
|
||
self.input_bridge_port.setValue(self._settings.bridge_port)
|
||
|
||
def _collect_settings(self) -> AgentSettings:
|
||
return AgentSettings(
|
||
oms_url=self._settings.oms_url,
|
||
token=self._settings.token,
|
||
device_name=self.input_device_name.text().strip(),
|
||
bridge_port=self.input_bridge_port.value(),
|
||
web_origin=self._settings.web_origin,
|
||
)
|
||
|
||
def _save_settings(self) -> None:
|
||
self._settings = self._collect_settings()
|
||
self._settings.save()
|
||
self._append_log("配置已保存")
|
||
|
||
def _try_init_bridge(self) -> None:
|
||
try:
|
||
ensure_runtime_importable()
|
||
from xhs_bridge.service import XhsBridgeService
|
||
|
||
self._bridge = XhsBridgeService()
|
||
self.lbl_device_id.setText(self._bridge.license.device_id)
|
||
self._refresh_local_status()
|
||
self._append_log("内置运行时已就绪")
|
||
except Exception as exc:
|
||
self._append_log(f"初始化失败: {exc}")
|
||
QMessageBox.warning(self, "运行时错误", str(exc))
|
||
|
||
def _toggle_agent(self) -> None:
|
||
if self._worker and self._worker.isRunning():
|
||
self._append_log("正在断开…")
|
||
self._worker.request_stop()
|
||
self.btn_connect.setEnabled(False)
|
||
return
|
||
|
||
self._settings = self._collect_settings()
|
||
if not self._settings.token:
|
||
self._start_pairing()
|
||
return
|
||
|
||
self._connect_with_settings()
|
||
|
||
def _connect_with_settings(self) -> None:
|
||
self._settings = self._collect_settings()
|
||
self._settings.save()
|
||
self._stop_pairing()
|
||
|
||
self._worker = AgentWorker(self._settings)
|
||
self._worker.log_line.connect(self._append_log)
|
||
self._worker.status_changed.connect(self._on_agent_status)
|
||
self._worker.started_ok.connect(self._on_agent_started)
|
||
self._worker.failed.connect(self._on_agent_failed)
|
||
self._worker.stopped.connect(self._on_agent_stopped)
|
||
self._worker.start()
|
||
self.btn_connect.setText("断开连接")
|
||
self.lbl_agent_state.setText("连接中…")
|
||
|
||
def _reauthorize(self) -> None:
|
||
if self._worker and self._worker.isRunning():
|
||
QMessageBox.warning(self, "提示", "请先断开连接再重新授权")
|
||
return
|
||
self._settings.token = ""
|
||
self._settings.save()
|
||
self._append_log("已清除授权，等待重新配对…")
|
||
self._start_pairing()
|
||
|
||
def _start_pairing(self) -> None:
|
||
if self._pair_server and self._pair_server.is_waiting:
|
||
self._append_log("已在等待浏览器授权…")
|
||
return
|
||
|
||
self._stop_pairing()
|
||
self._pair_server = PairServer(DEFAULT_PAIR_PORT)
|
||
|
||
def on_pair(payload: PairPayload) -> None:
|
||
self.pair_received.emit(payload)
|
||
|
||
self._pair_server.start(on_pair)
|
||
self.lbl_agent_state.setText("等待浏览器授权…")
|
||
self._append_log("等待 OMS 网页授权（请在浏览器登录 OMS）…")
|
||
self._open_pair_pages()
|
||
|
||
self._pair_timer = QTimer(self)
|
||
self._pair_timer.setSingleShot(True)
|
||
self._pair_timer.timeout.connect(self._on_pair_timeout)
|
||
self._pair_timer.start(PAIR_TIMEOUT_SEC * 1000)
|
||
|
||
def _open_pair_pages(self) -> None:
|
||
port = DEFAULT_PAIR_PORT
|
||
origins: list[str] = []
|
||
if self._settings.web_origin:
|
||
origins.append(self._settings.web_origin.rstrip("/"))
|
||
origins.extend(
|
||
[
|
||
"http://localhost:5173",
|
||
"http://127.0.0.1:5173",
|
||
]
|
||
)
|
||
seen: set[str] = set()
|
||
opened = False
|
||
for origin in origins:
|
||
if origin in seen:
|
||
continue
|
||
seen.add(origin)
|
||
url = f"{origin}/xhs/agentPair?port={port}"
|
||
try:
|
||
webbrowser.open(url)
|
||
opened = True
|
||
self._append_log(f"已打开授权页：{url}")
|
||
break
|
||
except Exception as exc:
|
||
logger.warning("打开浏览器失败 %s: %s", url, exc)
|
||
if not opened:
|
||
self._append_log(
|
||
"未能自动打开浏览器；请手动访问 OMS「小红书 → Agent 授权」或保持「启动服务」页打开"
|
||
)
|
||
|
||
def _on_pair_received(self, payload: object) -> None:
|
||
if not isinstance(payload, PairPayload):
|
||
return
|
||
self._settings.token = payload.token
|
||
self._settings.oms_url = payload.oms_url.rstrip("/")
|
||
if payload.web_origin:
|
||
self._settings.web_origin = payload.web_origin.rstrip("/")
|
||
self._settings.save()
|
||
self._append_log("已从浏览器收到授权，正在连接 OMS…")
|
||
self._connect_with_settings()
|
||
|
||
def _on_pair_timeout(self) -> None:
|
||
if self._pair_server and self._pair_server.is_waiting:
|
||
self._stop_pairing()
|
||
self.lbl_agent_state.setText("授权超时")
|
||
self._append_log("等待授权超时，请重试或保持「启动服务」页打开后再点连接")
|
||
QMessageBox.information(
|
||
self,
|
||
"授权超时",
|
||
"未收到浏览器授权。\n\n请确认：\n1. 浏览器已登录 OMS\n"
|
||
"2. 打开「小红书 → 启动服务」页面，或允许自动打开的授权页\n"
|
||
"3. 再次点击「连接 OMS」",
|
||
)
|
||
|
||
def _stop_pairing(self) -> None:
|
||
if self._pair_timer:
|
||
self._pair_timer.stop()
|
||
self._pair_timer = None
|
||
if self._pair_server:
|
||
self._pair_server.stop()
|
||
self._pair_server = None
|
||
|
||
def _on_agent_started(self, info: dict) -> None:
|
||
self.lbl_device_id.setText(str(info.get("device_id", "-")))
|
||
self.lbl_agent_state.setText("已连接 · 等待网页下发任务")
|
||
self._append_log("Agent 主循环已启动，请在网页「启动服务」查看在线状态")
|
||
|
||
def _on_agent_status(self, st: dict) -> None:
|
||
self.lbl_engine.setText("运行中" if st.get("engine_running") else "未启动")
|
||
self.lbl_extension.setText("已连接" if st.get("extension_connected") else "未连接")
|
||
self.lbl_login.setText("已登录" if st.get("xhs_logged_in") else "未登录")
|
||
|
||
def _on_agent_failed(self, msg: str) -> None:
|
||
self._append_log(f"错误: {msg}")
|
||
QMessageBox.critical(self, "Agent 异常", msg)
|
||
self._on_agent_stopped()
|
||
|
||
def _on_agent_stopped(self) -> None:
|
||
self.btn_connect.setEnabled(True)
|
||
self.btn_connect.setText("连接 OMS")
|
||
self.lbl_agent_state.setText("未连接")
|
||
self._worker = None
|
||
|
||
def _local_start_engine(self) -> None:
|
||
if not self._bridge:
|
||
return
|
||
try:
|
||
pid = self._bridge.start_engine()
|
||
self._append_log(f"引擎已启动 pid={pid}")
|
||
self._refresh_local_status()
|
||
except Exception as exc:
|
||
QMessageBox.warning(self, "启动失败", str(exc))
|
||
|
||
def _local_stop_engine(self) -> None:
|
||
if not self._bridge:
|
||
return
|
||
try:
|
||
self._bridge.stop_engine()
|
||
self._append_log("引擎已停止")
|
||
self._refresh_local_status()
|
||
except Exception as exc:
|
||
QMessageBox.warning(self, "停止失败", str(exc))
|
||
|
||
def _local_browser_login(self) -> None:
|
||
if not self._bridge:
|
||
return
|
||
try:
|
||
self._bridge.ensure_engine_running()
|
||
result = self._bridge.browser_login()
|
||
self._append_log(str(result.get("message") or result))
|
||
self._refresh_local_status()
|
||
except Exception as exc:
|
||
QMessageBox.warning(self, "登录", str(exc))
|
||
|
||
def _refresh_local_status(self) -> None:
|
||
if not self._bridge:
|
||
return
|
||
try:
|
||
st = self._bridge.status()
|
||
self.lbl_engine.setText("运行中" if st.get("bridge_running") else "未启动")
|
||
self.lbl_extension.setText("已连接" if st.get("extension_connected") else "未连接")
|
||
self.lbl_login.setText("已登录" if st.get("login_ok") else "未登录")
|
||
except Exception as exc:
|
||
self._append_log(f"状态刷新失败: {exc}")
|
||
|
||
def _append_log(self, text: str) -> None:
|
||
ts = datetime.now().strftime("%H:%M:%S")
|
||
self.log_view.appendPlainText(f"[{ts}] {text}")
|
||
|
||
def closeEvent(self, event: QCloseEvent) -> None:
|
||
self._stop_pairing()
|
||
if self._worker and self._worker.isRunning():
|
||
box = QMessageBox.question(
|
||
self,
|
||
"退出",
|
||
"Agent 正在运行，确定退出？",
|
||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||
)
|
||
if box != QMessageBox.StandardButton.Yes:
|
||
event.ignore()
|
||
return
|
||
self._worker.request_stop()
|
||
self._worker.wait(3000)
|
||
event.accept()
|
||
|
||
|
||
def run_app() -> int:
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
app = QApplication(sys.argv)
|
||
app.setApplicationName("OMS XHS Agent")
|
||
win = MainWindow()
|
||
win.show()
|
||
return app.exec()
|