2116 lines
82 KiB
Python
2116 lines
82 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""PySide6 desktop shell for the WeCom customer-service assistant."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import queue
|
||
import signal
|
||
import sys
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from PySide6.QtCore import (
|
||
QEasingCurve,
|
||
QPoint,
|
||
QPropertyAnimation,
|
||
QRect,
|
||
QTimer,
|
||
Qt,
|
||
QUrl,
|
||
Signal,
|
||
)
|
||
from PySide6.QtGui import QColor, QDesktopServices, QFont, QIcon, QPainter, QPixmap
|
||
from PySide6.QtWidgets import (
|
||
QAbstractItemView,
|
||
QApplication,
|
||
QCheckBox,
|
||
QDialog,
|
||
QDoubleSpinBox,
|
||
QFrame,
|
||
QGraphicsDropShadowEffect,
|
||
QGraphicsOpacityEffect,
|
||
QGridLayout,
|
||
QHBoxLayout,
|
||
QHeaderView,
|
||
QLabel,
|
||
QLineEdit,
|
||
QMainWindow,
|
||
QMessageBox,
|
||
QPlainTextEdit,
|
||
QProgressBar,
|
||
QPushButton,
|
||
QScrollArea,
|
||
QSpinBox,
|
||
QStackedWidget,
|
||
QTableWidget,
|
||
QTableWidgetItem,
|
||
QTextEdit,
|
||
QVBoxLayout,
|
||
QWidget,
|
||
)
|
||
from PySide6.QtWebEngineCore import (
|
||
QWebEnginePage,
|
||
QWebEngineProfile,
|
||
QWebEngineScript,
|
||
QWebEngineSettings,
|
||
)
|
||
from PySide6.QtWebEngineWidgets import QWebEngineView
|
||
|
||
|
||
try:
|
||
import __main__ as _legacy
|
||
|
||
if not hasattr(_legacy, "BotThread"):
|
||
raise ImportError
|
||
except ImportError:
|
||
import wechat_gui as _legacy
|
||
|
||
|
||
BotThread = _legacy.BotThread
|
||
LogQueue = _legacy.LogQueue
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
APP_SETTINGS_FILE = SCRIPT_DIR / "app_settings.json"
|
||
CUSTOMER_SERVICE_URL = "http://kf.zhenyangtang.com.cn/"
|
||
WIN_TITLE = "甄养堂 · 企微客服助手"
|
||
WECOM_WAITING_MESSAGE = (
|
||
"企业微信主界面未显示、未在前台或已最小化，正在尝试切到前台；"
|
||
"本轮失败时下一轮将自动重试。"
|
||
)
|
||
|
||
|
||
COLORS = {
|
||
"canvas": "#F4F7F5",
|
||
"surface": "#FFFFFF",
|
||
"surface_alt": "#F8FAF9",
|
||
"sidebar": "#10271F",
|
||
"sidebar_alt": "#17372C",
|
||
"ink": "#17251F",
|
||
"muted": "#617269",
|
||
"faint": "#8A9991",
|
||
"line": "#DDE7E1",
|
||
"line_soft": "#EAF0ED",
|
||
"accent": "#10A57A",
|
||
"accent_dark": "#087A59",
|
||
"accent_soft": "#DDF5EC",
|
||
"success": "#118764",
|
||
"warning": "#B67820",
|
||
"danger": "#D85260",
|
||
"danger_soft": "#FBEAEC",
|
||
}
|
||
|
||
|
||
APP_QSS = """
|
||
* {
|
||
font-family: "Microsoft YaHei UI", "Segoe UI";
|
||
font-size: 14px;
|
||
color: #17251F;
|
||
}
|
||
QMainWindow, QWidget#AppRoot, QWidget#PageRoot, QScrollArea#PageScroll,
|
||
QScrollArea#PageScroll > QWidget > QWidget {
|
||
background: #F4F7F5;
|
||
}
|
||
QDialog#SessionDetailDialog { background: #F4F7F5; }
|
||
QFrame#Sidebar {
|
||
background: #FCFDFC;
|
||
border: none;
|
||
border-right: 1px solid #DDE7E1;
|
||
}
|
||
QFrame#Sidebar QLabel { color: #42574D; }
|
||
QFrame#Sidebar QLabel#Brand {
|
||
color: #087A59;
|
||
font-family: "Cascadia Mono", "Microsoft YaHei UI";
|
||
font-size: 22px;
|
||
font-weight: 700;
|
||
letter-spacing: 1px;
|
||
}
|
||
QFrame#Sidebar QLabel#BrandSub { color: #6C7D74; font-size: 13px; }
|
||
QFrame#Sidebar QLabel#SectionLabel {
|
||
color: #8A9991;
|
||
font-family: "Cascadia Mono";
|
||
font-size: 11px;
|
||
font-weight: 600;
|
||
letter-spacing: 1px;
|
||
}
|
||
QFrame#SideStatus {
|
||
background: #EEF8F3;
|
||
border: 1px solid #CFE9DE;
|
||
border-radius: 12px;
|
||
}
|
||
QFrame#Sidebar QLabel#SideStatusTitle { color: #17372C; font-size: 15px; font-weight: 700; }
|
||
QFrame#Sidebar QLabel#SideStatusHint { color: #6C7D74; font-size: 12px; }
|
||
QPushButton#NavButton {
|
||
background: transparent;
|
||
color: #5D7067;
|
||
border: none;
|
||
border-left: 3px solid transparent;
|
||
border-radius: 0px;
|
||
text-align: left;
|
||
padding: 13px 16px;
|
||
font-size: 15px;
|
||
}
|
||
QPushButton#NavButton:hover { background: #F0F6F3; color: #17372C; }
|
||
QPushButton#NavButton:checked {
|
||
background: #DDF5EC;
|
||
color: #087A59;
|
||
border-left-color: #10A57A;
|
||
font-weight: 700;
|
||
}
|
||
QPushButton#PrimaryButton, QPushButton#SidePrimary {
|
||
background: #10A57A;
|
||
color: #FFFFFF;
|
||
border: none;
|
||
border-radius: 10px;
|
||
padding: 11px 20px;
|
||
font-weight: 700;
|
||
}
|
||
QPushButton#PrimaryButton:hover, QPushButton#SidePrimary:hover { background: #087A59; }
|
||
QPushButton#PrimaryButton:pressed, QPushButton#SidePrimary:pressed { background: #06684B; }
|
||
QPushButton#PrimaryButton:disabled, QPushButton#SidePrimary:disabled {
|
||
background: #C8D7D0; color: #F7F9F8;
|
||
}
|
||
QPushButton#SecondaryButton {
|
||
background: #FFFFFF;
|
||
color: #42574D;
|
||
border: 1px solid #D7E2DC;
|
||
border-radius: 10px;
|
||
padding: 10px 18px;
|
||
}
|
||
QPushButton#SecondaryButton:hover { background: #F0F6F3; border-color: #BFD3C9; }
|
||
QPushButton#DangerButton {
|
||
background: #FBEAEC;
|
||
color: #C74452;
|
||
border: 1px solid #F2CDD2;
|
||
border-radius: 10px;
|
||
padding: 10px 18px;
|
||
font-weight: 600;
|
||
}
|
||
QPushButton#DangerButton:hover { background: #D85260; color: #FFFFFF; }
|
||
QFrame#Card, QFrame#HeroCard, QFrame#MetricCard {
|
||
background: #FFFFFF;
|
||
border: 1px solid #DDE7E1;
|
||
border-radius: 16px;
|
||
}
|
||
QFrame#CustomerBubble {
|
||
background: #FFFFFF;
|
||
border: 1px solid #DDE7E1;
|
||
border-radius: 13px;
|
||
}
|
||
QFrame#AssistantBubble {
|
||
background: #DDF5EC;
|
||
border: 1px solid #BFE8D8;
|
||
border-radius: 13px;
|
||
}
|
||
QFrame#SystemBubble {
|
||
background: #EEF2F0;
|
||
border: 1px solid #DDE7E1;
|
||
border-radius: 11px;
|
||
}
|
||
QLabel#MessageRole { color: #087A59; font-size: 12px; font-weight: 700; }
|
||
QLabel#MessageTime { color: #91A097; font-size: 11px; }
|
||
QLabel#MessageContent { color: #17251F; font-size: 14px; }
|
||
QFrame#HeroCard { background: #FBFDFC; }
|
||
QLabel#Eyebrow {
|
||
color: #0B8B67;
|
||
font-family: "Cascadia Mono";
|
||
font-size: 12px;
|
||
font-weight: 700;
|
||
letter-spacing: 1px;
|
||
}
|
||
QLabel#PageTitle { color: #17251F; font-size: 28px; font-weight: 700; }
|
||
QLabel#PageSubtitle { color: #687A71; font-size: 14px; }
|
||
QLabel#CardTitle { color: #17251F; font-size: 17px; font-weight: 700; }
|
||
QLabel#CardSubtitle { color: #6E7F76; font-size: 13px; }
|
||
QLabel#MetricValue {
|
||
color: #17251F;
|
||
font-family: "Cascadia Mono";
|
||
font-size: 30px;
|
||
font-weight: 700;
|
||
}
|
||
QLabel#MetricLabel { color: #6C7D74; font-size: 13px; }
|
||
QLabel#MetricMeta { color: #96A39C; font-size: 12px; }
|
||
QLabel#HeroStatus { color: #17251F; font-size: 30px; font-weight: 700; }
|
||
QLabel#HeroHint { color: #687A71; font-size: 14px; }
|
||
QLabel#SuccessText { color: #118764; font-weight: 600; }
|
||
QLabel#WarningText { color: #B67820; font-weight: 600; }
|
||
QLabel#DangerText { color: #D85260; font-weight: 600; }
|
||
QLineEdit, QPlainTextEdit, QTextEdit, QSpinBox, QDoubleSpinBox {
|
||
background: #FBFCFB;
|
||
color: #17251F;
|
||
border: 1px solid #D7E2DC;
|
||
border-radius: 9px;
|
||
padding: 9px 11px;
|
||
selection-background-color: #BDEBDC;
|
||
}
|
||
QLineEdit:focus, QPlainTextEdit:focus, QTextEdit:focus,
|
||
QSpinBox:focus, QDoubleSpinBox:focus {
|
||
background: #FFFFFF;
|
||
border: 2px solid #10A57A;
|
||
padding: 8px 10px;
|
||
}
|
||
QLineEdit[readOnly="true"] { background: #F2F5F3; color: #798880; }
|
||
QCheckBox { spacing: 10px; font-size: 14px; }
|
||
QCheckBox::indicator {
|
||
width: 38px; height: 21px;
|
||
border-radius: 10px;
|
||
background: #C9D6D0;
|
||
}
|
||
QCheckBox::indicator:checked { background: #10A57A; }
|
||
QTableWidget {
|
||
background: #FFFFFF;
|
||
alternate-background-color: #F8FAF9;
|
||
border: 1px solid #DDE7E1;
|
||
border-radius: 12px;
|
||
gridline-color: #EAF0ED;
|
||
selection-background-color: #DDF5EC;
|
||
selection-color: #17251F;
|
||
}
|
||
QHeaderView::section {
|
||
background: #F2F6F4;
|
||
color: #607168;
|
||
border: none;
|
||
border-bottom: 1px solid #DDE7E1;
|
||
padding: 10px;
|
||
font-weight: 600;
|
||
}
|
||
QScrollBar:vertical { width: 10px; background: transparent; margin: 3px; }
|
||
QScrollBar::handle:vertical { background: #C7D5CE; border-radius: 4px; min-height: 30px; }
|
||
QScrollBar::handle:vertical:hover { background: #9EB5AA; }
|
||
QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical { height: 0px; }
|
||
QProgressBar { background: transparent; border: none; max-height: 3px; }
|
||
QProgressBar::chunk { background: #10A57A; }
|
||
QToolTip {
|
||
background: #17372C; color: #FFFFFF; border: none; padding: 7px;
|
||
}
|
||
"""
|
||
|
||
|
||
def _button(text: str, kind: str = "secondary") -> QPushButton:
|
||
button = QPushButton(text)
|
||
names = {
|
||
"primary": "PrimaryButton",
|
||
"danger": "DangerButton",
|
||
"side": "SidePrimary",
|
||
}
|
||
button.setObjectName(names.get(kind, "SecondaryButton"))
|
||
button.setCursor(Qt.PointingHandCursor)
|
||
return button
|
||
|
||
|
||
def _shadow(widget: QWidget, blur: int = 28, y: int = 8, alpha: int = 24) -> None:
|
||
effect = QGraphicsDropShadowEffect(widget)
|
||
effect.setBlurRadius(blur)
|
||
effect.setOffset(0, y)
|
||
effect.setColor(QColor(22, 58, 43, alpha))
|
||
widget.setGraphicsEffect(effect)
|
||
|
||
|
||
def _label(text: str, name: str = "") -> QLabel:
|
||
item = QLabel(text)
|
||
if name:
|
||
item.setObjectName(name)
|
||
item.setWordWrap(True)
|
||
return item
|
||
|
||
|
||
def _page_header(index: str, title: str, subtitle: str) -> QVBoxLayout:
|
||
layout = QVBoxLayout()
|
||
layout.setSpacing(5)
|
||
layout.addWidget(_label(f"{index} CONTROL CENTER", "Eyebrow"))
|
||
layout.addWidget(_label(title, "PageTitle"))
|
||
layout.addWidget(_label(subtitle, "PageSubtitle"))
|
||
return layout
|
||
|
||
|
||
def _card(title: str = "", subtitle: str = "") -> tuple[QFrame, QVBoxLayout]:
|
||
frame = QFrame()
|
||
frame.setObjectName("Card")
|
||
layout = QVBoxLayout(frame)
|
||
layout.setContentsMargins(22, 20, 22, 20)
|
||
layout.setSpacing(11)
|
||
if title:
|
||
layout.addWidget(_label(title, "CardTitle"))
|
||
if subtitle:
|
||
layout.addWidget(_label(subtitle, "CardSubtitle"))
|
||
return frame, layout
|
||
|
||
|
||
class FadingStack(QStackedWidget):
|
||
def setCurrentIndexAnimated(self, index: int) -> None:
|
||
if index == self.currentIndex():
|
||
return
|
||
self.setCurrentIndex(index)
|
||
page = self.currentWidget()
|
||
effect = QGraphicsOpacityEffect(page)
|
||
page.setGraphicsEffect(effect)
|
||
animation = QPropertyAnimation(effect, b"opacity", page)
|
||
animation.setDuration(220)
|
||
animation.setStartValue(0.25)
|
||
animation.setEndValue(1.0)
|
||
animation.setEasingCurve(QEasingCurve.OutCubic)
|
||
animation.finished.connect(lambda: page.setGraphicsEffect(None))
|
||
page._fade_animation = animation
|
||
animation.start()
|
||
|
||
|
||
class PortalWebView(QWebEngineView):
|
||
def createWindow(self, _window_type):
|
||
return self
|
||
|
||
|
||
class PortalPage(QWidget):
|
||
def __init__(self, parent: QWidget | None = None):
|
||
super().__init__(parent)
|
||
self.setObjectName("PageRoot")
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(0, 0, 0, 0)
|
||
layout.setSpacing(0)
|
||
|
||
self.progress = QProgressBar()
|
||
self.progress.setRange(0, 100)
|
||
self.progress.hide()
|
||
layout.addWidget(self.progress)
|
||
|
||
self.view = PortalWebView()
|
||
self.view.setFocusPolicy(Qt.StrongFocus)
|
||
self.view.settings().setAttribute(
|
||
QWebEngineSettings.WebAttribute.LocalStorageEnabled, True
|
||
)
|
||
self.view.settings().setAttribute(
|
||
QWebEngineSettings.WebAttribute.JavascriptCanOpenWindows, True
|
||
)
|
||
|
||
profile_root = Path(os.environ.get("LOCALAPPDATA", SCRIPT_DIR)) / "ZhenYangTangRPA" / "QtPortal"
|
||
profile_root.mkdir(parents=True, exist_ok=True)
|
||
self.profile = QWebEngineProfile("zhen-ai-portal", self)
|
||
self.profile.setPersistentStoragePath(str(profile_root / "storage"))
|
||
self.profile.setCachePath(str(profile_root / "cache"))
|
||
self.profile.setPersistentCookiesPolicy(
|
||
QWebEngineProfile.PersistentCookiesPolicy.ForcePersistentCookies
|
||
)
|
||
self.page = QWebEnginePage(self.profile, self.view)
|
||
self.view.setPage(self.page)
|
||
self._install_light_theme()
|
||
|
||
self.view.loadStarted.connect(self._load_started)
|
||
self.view.loadProgress.connect(self.progress.setValue)
|
||
self.view.loadFinished.connect(self._load_finished)
|
||
layout.addWidget(self.view, 1)
|
||
|
||
if os.environ.get("WECOM_RPA_DISABLE_PORTAL") != "1":
|
||
self.view.setUrl(QUrl(CUSTOMER_SERVICE_URL))
|
||
else:
|
||
self.view.setHtml(
|
||
"<html><body style='margin:0;background:#f4f7f5;font-family:Microsoft YaHei UI'>"
|
||
"<div style='height:100vh;display:grid;place-items:center;color:#617269'>"
|
||
"AI 客服网页在测试模式下未加载</div></body></html>"
|
||
)
|
||
|
||
def _install_light_theme(self) -> None:
|
||
css_path = SCRIPT_DIR / "edge_light_theme" / "light-theme.css"
|
||
try:
|
||
css = css_path.read_text(encoding="utf-8")
|
||
except OSError:
|
||
css = "html,body,#app{background:#f8faf9!important;color:#17251f!important}"
|
||
source = f"""
|
||
(() => {{
|
||
const css = {json.dumps(css)};
|
||
const applyTheme = () => {{
|
||
let style = document.getElementById('zhen-qt-light-theme');
|
||
if (!style) {{
|
||
style = document.createElement('style');
|
||
style.id = 'zhen-qt-light-theme';
|
||
(document.head || document.documentElement).appendChild(style);
|
||
}}
|
||
style.textContent = css;
|
||
}};
|
||
applyTheme();
|
||
new MutationObserver(applyTheme).observe(document.documentElement, {{childList:true}});
|
||
}})();
|
||
"""
|
||
script = QWebEngineScript()
|
||
script.setName("zhen-light-theme")
|
||
script.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentReady)
|
||
script.setWorldId(QWebEngineScript.ScriptWorldId.MainWorld)
|
||
script.setRunsOnSubFrames(True)
|
||
script.setSourceCode(source)
|
||
self.profile.scripts().insert(script)
|
||
|
||
def _load_started(self) -> None:
|
||
self.progress.setValue(2)
|
||
self.progress.show()
|
||
|
||
def _load_finished(self, _ok: bool) -> None:
|
||
self.progress.hide()
|
||
self.view.setFocus(Qt.OtherFocusReason)
|
||
|
||
def reload(self) -> None:
|
||
self.view.reload()
|
||
|
||
def open_external(self) -> None:
|
||
QDesktopServices.openUrl(self.view.url() or QUrl(CUSTOMER_SERVICE_URL))
|
||
|
||
|
||
class MetricCard(QFrame):
|
||
def __init__(self, label: str, value: str, meta: str):
|
||
super().__init__()
|
||
self.setObjectName("MetricCard")
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(19, 17, 19, 17)
|
||
layout.setSpacing(6)
|
||
layout.addWidget(_label(label, "MetricLabel"))
|
||
self.value = _label(value, "MetricValue")
|
||
layout.addWidget(self.value)
|
||
layout.addWidget(_label(meta, "MetricMeta"))
|
||
|
||
|
||
class DashboardPage(QScrollArea):
|
||
startRequested = Signal()
|
||
stopRequested = Signal()
|
||
pageRequested = Signal(int)
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setObjectName("PageScroll")
|
||
self.setWidgetResizable(True)
|
||
self.setFrameShape(QFrame.NoFrame)
|
||
root = QWidget()
|
||
root.setObjectName("PageRoot")
|
||
self.setWidget(root)
|
||
layout = QVBoxLayout(root)
|
||
layout.setContentsMargins(28, 25, 28, 28)
|
||
layout.setSpacing(18)
|
||
layout.addLayout(
|
||
_page_header(
|
||
"02",
|
||
"自动回复控制台",
|
||
"监控企业微信窗口、AI 决策和客户沉淀状态。",
|
||
)
|
||
)
|
||
|
||
hero = QFrame()
|
||
hero.setObjectName("HeroCard")
|
||
hero_layout = QHBoxLayout(hero)
|
||
hero_layout.setContentsMargins(25, 22, 25, 22)
|
||
hero_layout.setSpacing(20)
|
||
status_box = QVBoxLayout()
|
||
status_box.setSpacing(5)
|
||
status_box.addWidget(_label("WECOM LIVE MONITOR", "Eyebrow"))
|
||
self.status = _label("待命", "HeroStatus")
|
||
status_box.addWidget(self.status)
|
||
self.status_hint = _label("确认企业微信已登录后即可启动。", "HeroHint")
|
||
status_box.addWidget(self.status_hint)
|
||
self.timer = _label("运行时长 --:--:--", "CardSubtitle")
|
||
status_box.addWidget(self.timer)
|
||
hero_layout.addLayout(status_box, 1)
|
||
self.hero_button = _button("开始监听", "primary")
|
||
self.hero_button.setMinimumWidth(128)
|
||
self.hero_button.clicked.connect(self.startRequested)
|
||
hero_layout.addWidget(self.hero_button, 0, Qt.AlignVCenter)
|
||
layout.addWidget(hero)
|
||
|
||
metrics = QGridLayout()
|
||
metrics.setSpacing(12)
|
||
self.replied = MetricCard("本次已回复", "0", "自动回复数量")
|
||
self.false_pos = MetricCard("识别误判", "0", "已跳过的会话")
|
||
self.registration = MetricCard("待处理登记", "0", "挂号与回访线索")
|
||
self.sessions = MetricCard("会话档案", "0", "已保存上下文")
|
||
for column, card in enumerate(
|
||
(self.replied, self.false_pos, self.registration, self.sessions)
|
||
):
|
||
metrics.addWidget(card, 0, column)
|
||
metrics.setColumnStretch(column, 1)
|
||
layout.addLayout(metrics)
|
||
|
||
workflow, workflow_layout = _card(
|
||
"实际工作流",
|
||
"每一步都映射到当前软件已经具备的执行能力。",
|
||
)
|
||
steps = (
|
||
("A1", "消息监听", "未读识别 → 会话提取 → 自动回复", 2),
|
||
("A2", "AI 决策", "上下文记忆 → 医疗客服人格 → MCP 工具", 4),
|
||
("A3", "业务沉淀", "挂号登记 → 会话档案 → 运行记录", 3),
|
||
)
|
||
for code, title, text, page_index in steps:
|
||
row = QFrame()
|
||
row.setStyleSheet(
|
||
"QFrame{background:#F8FAF9;border:1px solid #E4ECE8;border-radius:11px;}"
|
||
)
|
||
row_layout = QHBoxLayout(row)
|
||
row_layout.setContentsMargins(15, 12, 15, 12)
|
||
badge = _label(code, "Eyebrow")
|
||
badge.setFixedWidth(38)
|
||
row_layout.addWidget(badge)
|
||
copy = QVBoxLayout()
|
||
copy.setSpacing(2)
|
||
copy.addWidget(_label(title, "CardTitle"))
|
||
copy.addWidget(_label(text, "CardSubtitle"))
|
||
row_layout.addLayout(copy, 1)
|
||
action = _button("打开")
|
||
action.clicked.connect(lambda _checked=False, i=page_index: self.pageRequested.emit(i))
|
||
row_layout.addWidget(action)
|
||
workflow_layout.addWidget(row)
|
||
layout.addWidget(workflow)
|
||
layout.addStretch(1)
|
||
|
||
def set_status(self, state: str, text: str, hint: str) -> None:
|
||
colors = {
|
||
"running": COLORS["success"],
|
||
"waiting": COLORS["warning"],
|
||
"connecting": COLORS["warning"],
|
||
"error": COLORS["danger"],
|
||
"stopping": COLORS["warning"],
|
||
"stopped": COLORS["ink"],
|
||
}
|
||
self.status.setText(text)
|
||
self.status.setStyleSheet(
|
||
f"color:{colors.get(state, COLORS['ink'])};font-size:30px;font-weight:700;"
|
||
)
|
||
self.status_hint.setText(hint)
|
||
active = state in {"running", "waiting", "connecting", "stopping"}
|
||
self.hero_button.setText("监听运行中" if active else "开始监听")
|
||
self.hero_button.setEnabled(not active)
|
||
|
||
|
||
class SettingsPage(QScrollArea):
|
||
saved = Signal(dict)
|
||
|
||
def __init__(self, settings: dict):
|
||
super().__init__()
|
||
self.setObjectName("PageScroll")
|
||
self.setWidgetResizable(True)
|
||
self.setFrameShape(QFrame.NoFrame)
|
||
root = QWidget()
|
||
root.setObjectName("PageRoot")
|
||
self.setWidget(root)
|
||
layout = QVBoxLayout(root)
|
||
layout.setContentsMargins(28, 25, 28, 28)
|
||
layout.setSpacing(18)
|
||
layout.addLayout(_page_header("03", "通用设置", "修改后自动保存，并在下次监听时生效。"))
|
||
|
||
runtime, runtime_layout = _card(
|
||
"运行参数", "参数直接参与企业微信监听，不是静态展示。"
|
||
)
|
||
runtime_grid = QGridLayout()
|
||
runtime_grid.setHorizontalSpacing(18)
|
||
runtime_grid.setVerticalSpacing(8)
|
||
runtime_grid.addWidget(_label("固定回复（关闭 AI 时使用）", "CardSubtitle"), 0, 0, 1, 2)
|
||
self.reply = QLineEdit(str(settings["auto_reply_text"]))
|
||
self.reply.setPlaceholderText("请输入固定回复")
|
||
runtime_grid.addWidget(self.reply, 1, 0, 1, 2)
|
||
runtime_grid.addWidget(_label("轮询间隔（秒）", "CardSubtitle"), 2, 0)
|
||
runtime_grid.addWidget(_label("鼠标静止等待（秒）", "CardSubtitle"), 2, 1)
|
||
self.poll = QDoubleSpinBox()
|
||
self.poll.setRange(0.2, 300.0)
|
||
self.poll.setDecimals(1)
|
||
self.poll.setValue(float(settings["poll_interval"]))
|
||
self.idle_seconds = QDoubleSpinBox()
|
||
self.idle_seconds.setRange(0.0, 3600.0)
|
||
self.idle_seconds.setDecimals(1)
|
||
self.idle_seconds.setValue(float(settings["mouse_idle_seconds"]))
|
||
runtime_grid.addWidget(self.poll, 3, 0)
|
||
runtime_grid.addWidget(self.idle_seconds, 3, 1)
|
||
runtime_grid.setColumnStretch(0, 1)
|
||
runtime_grid.setColumnStretch(1, 1)
|
||
runtime_layout.addLayout(runtime_grid)
|
||
self.save_status = _label("已加载保存配置", "SuccessText")
|
||
self.save_status.setAlignment(Qt.AlignRight)
|
||
runtime_layout.addWidget(self.save_status)
|
||
layout.addWidget(runtime)
|
||
|
||
safety, safety_layout = _card(
|
||
"人机共存",
|
||
"检测到用户正在操作时自动等待，避免与人工客服争抢鼠标。",
|
||
)
|
||
self.mouse_idle = QCheckBox("启用鼠标空闲检测")
|
||
self.mouse_idle.setChecked(bool(settings["mouse_idle_enabled"]))
|
||
safety_layout.addWidget(self.mouse_idle)
|
||
safety_layout.addWidget(
|
||
_label(
|
||
"企业微信未显示或最小化时，机器人会尝试还原到前台；本轮未就绪会暂停并自动重试。",
|
||
"CardSubtitle",
|
||
)
|
||
)
|
||
layout.addWidget(safety)
|
||
|
||
checks, checks_layout = _card("启动前检查")
|
||
for index, text in enumerate(
|
||
(
|
||
"企业微信 PC 端已登录，且账号状态正常。",
|
||
"如启用 AI，API 地址、密钥与模型配置已经保存。",
|
||
"双屏环境可以使用不同缩放比例，Qt 会跟随当前屏幕自动缩放。",
|
||
),
|
||
start=1,
|
||
):
|
||
row = QHBoxLayout()
|
||
badge = _label(f"{index:02d}", "Eyebrow")
|
||
badge.setFixedWidth(32)
|
||
row.addWidget(badge)
|
||
row.addWidget(_label(text, "CardSubtitle"), 1)
|
||
checks_layout.addLayout(row)
|
||
layout.addWidget(checks)
|
||
layout.addStretch(1)
|
||
|
||
self.save_timer = QTimer(self)
|
||
self.save_timer.setSingleShot(True)
|
||
self.save_timer.setInterval(500)
|
||
self.save_timer.timeout.connect(self._emit_save)
|
||
self.reply.textChanged.connect(self._schedule_save)
|
||
self.poll.valueChanged.connect(self._schedule_save)
|
||
self.idle_seconds.valueChanged.connect(self._schedule_save)
|
||
self.mouse_idle.toggled.connect(self._schedule_save)
|
||
|
||
def values(self) -> dict:
|
||
return {
|
||
"auto_reply_text": self.reply.text().strip() or "你好",
|
||
"poll_interval": self.poll.value(),
|
||
"mouse_idle_enabled": self.mouse_idle.isChecked(),
|
||
"mouse_idle_seconds": self.idle_seconds.value(),
|
||
}
|
||
|
||
def _schedule_save(self, *_args) -> None:
|
||
self.save_status.setText("正在保存…")
|
||
self.save_status.setObjectName("WarningText")
|
||
self.save_status.style().unpolish(self.save_status)
|
||
self.save_status.style().polish(self.save_status)
|
||
self.save_timer.start()
|
||
|
||
def _emit_save(self) -> None:
|
||
self.saved.emit(self.values())
|
||
|
||
def mark_saved(self, ok: bool, message: str) -> None:
|
||
self.save_status.setText(message)
|
||
self.save_status.setObjectName("SuccessText" if ok else "DangerText")
|
||
self.save_status.style().unpolish(self.save_status)
|
||
self.save_status.style().polish(self.save_status)
|
||
|
||
|
||
class SessionDetailDialog(QDialog):
|
||
def __init__(
|
||
self,
|
||
session_id: str,
|
||
history: list[dict],
|
||
last_lines: list[str],
|
||
parent: QWidget | None = None,
|
||
):
|
||
super().__init__(parent)
|
||
self.setObjectName("SessionDetailDialog")
|
||
self.setWindowTitle("客户会话记录")
|
||
self.setModal(True)
|
||
self.resize(920, 720)
|
||
self.setMinimumSize(700, 520)
|
||
self.session_id = session_id
|
||
self.history = history
|
||
self.last_lines = last_lines
|
||
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(24, 22, 24, 20)
|
||
layout.setSpacing(16)
|
||
|
||
header = QHBoxLayout()
|
||
title_box = QVBoxLayout()
|
||
title_box.setSpacing(4)
|
||
title_box.addWidget(_label("客户会话记录", "PageTitle"))
|
||
title_box.addWidget(
|
||
_label(
|
||
f"会话 ID {session_id} · 共 {len(history)} 条消息",
|
||
"PageSubtitle",
|
||
)
|
||
)
|
||
header.addLayout(title_box, 1)
|
||
copy_button = _button("复制全部记录")
|
||
copy_button.clicked.connect(self._copy_all)
|
||
header.addWidget(copy_button, 0, Qt.AlignBottom)
|
||
layout.addLayout(header)
|
||
|
||
scroll = QScrollArea()
|
||
scroll.setWidgetResizable(True)
|
||
scroll.setFrameShape(QFrame.NoFrame)
|
||
scroll.setStyleSheet(
|
||
"QScrollArea{background:#EEF3F0;border:1px solid #DDE7E1;border-radius:15px;}"
|
||
"QScrollArea>QWidget>QWidget{background:#EEF3F0;}"
|
||
)
|
||
message_root = QWidget()
|
||
message_layout = QVBoxLayout(message_root)
|
||
message_layout.setContentsMargins(20, 20, 20, 20)
|
||
message_layout.setSpacing(13)
|
||
scroll.setWidget(message_root)
|
||
|
||
if history:
|
||
for message in history:
|
||
self._add_message(message_layout, message)
|
||
elif last_lines:
|
||
self._add_message(
|
||
message_layout,
|
||
{
|
||
"role": "system",
|
||
"content": "仅保留了最近画面快照：\n\n" + "\n".join(last_lines),
|
||
"ts": None,
|
||
},
|
||
)
|
||
else:
|
||
empty = _label("这份会话档案暂时没有可显示的消息。", "PageSubtitle")
|
||
empty.setAlignment(Qt.AlignCenter)
|
||
message_layout.addWidget(empty)
|
||
message_layout.addStretch(1)
|
||
layout.addWidget(scroll, 1)
|
||
|
||
footer = QHBoxLayout()
|
||
footer.addWidget(_label("按时间顺序展示已保存的客户与客服消息。", "CardSubtitle"), 1)
|
||
close_button = _button("关闭")
|
||
close_button.clicked.connect(self.accept)
|
||
footer.addWidget(close_button)
|
||
layout.addLayout(footer)
|
||
QTimer.singleShot(0, lambda: scroll.verticalScrollBar().setValue(0))
|
||
|
||
@staticmethod
|
||
def _message_time(value) -> str:
|
||
try:
|
||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(value)))
|
||
except (TypeError, ValueError, OSError):
|
||
return ""
|
||
|
||
def _add_message(self, parent: QVBoxLayout, message: dict) -> None:
|
||
role = str(message.get("role") or "system").lower()
|
||
role_name = {"user": "客户", "assistant": "客服", "system": "系统记录"}.get(
|
||
role, role
|
||
)
|
||
bubble_name = {
|
||
"user": "CustomerBubble",
|
||
"assistant": "AssistantBubble",
|
||
}.get(role, "SystemBubble")
|
||
bubble = QFrame()
|
||
bubble.setObjectName(bubble_name)
|
||
bubble.setMaximumWidth(690)
|
||
bubble_layout = QVBoxLayout(bubble)
|
||
bubble_layout.setContentsMargins(15, 12, 15, 13)
|
||
bubble_layout.setSpacing(6)
|
||
meta = QHBoxLayout()
|
||
meta.setSpacing(10)
|
||
meta.addWidget(_label(role_name, "MessageRole"))
|
||
stamp = self._message_time(message.get("ts"))
|
||
if stamp:
|
||
meta.addWidget(_label(stamp, "MessageTime"))
|
||
meta.addStretch(1)
|
||
bubble_layout.addLayout(meta)
|
||
content = _label(str(message.get("content") or ""), "MessageContent")
|
||
content.setTextInteractionFlags(Qt.TextSelectableByMouse)
|
||
content.setMinimumWidth(260)
|
||
bubble_layout.addWidget(content)
|
||
|
||
row = QHBoxLayout()
|
||
row.setContentsMargins(0, 0, 0, 0)
|
||
if role == "assistant":
|
||
row.addStretch(1)
|
||
row.addWidget(bubble)
|
||
elif role == "system":
|
||
row.addStretch(1)
|
||
row.addWidget(bubble)
|
||
row.addStretch(1)
|
||
else:
|
||
row.addWidget(bubble)
|
||
row.addStretch(1)
|
||
parent.addLayout(row)
|
||
|
||
def _copy_all(self) -> None:
|
||
lines = [f"客户会话记录\n会话 ID：{self.session_id}\n"]
|
||
role_names = {"user": "客户", "assistant": "客服", "system": "系统记录"}
|
||
if self.history:
|
||
for message in self.history:
|
||
role = str(message.get("role") or "system").lower()
|
||
name = role_names.get(role, role)
|
||
stamp = self._message_time(message.get("ts"))
|
||
heading = f"[{stamp}] {name}" if stamp else name
|
||
lines.append(f"{heading}\n{message.get('content') or ''}")
|
||
elif self.last_lines:
|
||
lines.append("系统记录\n" + "\n".join(self.last_lines))
|
||
QApplication.clipboard().setText("\n\n".join(lines))
|
||
|
||
|
||
class BusinessPage(QScrollArea):
|
||
logMessage = Signal(str, str)
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setObjectName("PageScroll")
|
||
self.setWidgetResizable(True)
|
||
self.setFrameShape(QFrame.NoFrame)
|
||
root = QWidget()
|
||
root.setObjectName("PageRoot")
|
||
self.setWidget(root)
|
||
layout = QVBoxLayout(root)
|
||
layout.setContentsMargins(28, 25, 28, 28)
|
||
layout.setSpacing(18)
|
||
|
||
header = QHBoxLayout()
|
||
header.addLayout(_page_header("04", "业务数据", "管理挂号线索与客户会话档案。"), 1)
|
||
refresh = _button("刷新数据")
|
||
refresh.clicked.connect(self.refresh_data)
|
||
header.addWidget(refresh, 0, Qt.AlignBottom)
|
||
layout.addLayout(header)
|
||
|
||
summary = QGridLayout()
|
||
summary.setSpacing(12)
|
||
self.registration_metric = MetricCard("待处理登记", "0", "需要人工跟进")
|
||
self.session_metric = MetricCard("会话档案", "0", "长期上下文记录")
|
||
summary.addWidget(self.registration_metric, 0, 0)
|
||
summary.addWidget(self.session_metric, 0, 1)
|
||
summary.setColumnStretch(0, 1)
|
||
summary.setColumnStretch(1, 1)
|
||
layout.addLayout(summary)
|
||
|
||
registrations, registrations_layout = _card(
|
||
"挂号与回访登记", "AI 识别出的明确挂号意向会沉淀在这里。"
|
||
)
|
||
self.registration_table = self._table(
|
||
["最近更新", "微信客户", "症状 / 诉求", "状态"]
|
||
)
|
||
registrations_layout.addWidget(self.registration_table)
|
||
registration_actions = QHBoxLayout()
|
||
registration_actions.addStretch(1)
|
||
done = _button("标记已联系", "primary")
|
||
done.clicked.connect(self._mark_registration_done)
|
||
delete = _button("删除所选", "danger")
|
||
delete.clicked.connect(self._delete_registrations)
|
||
registration_actions.addWidget(done)
|
||
registration_actions.addWidget(delete)
|
||
registrations_layout.addLayout(registration_actions)
|
||
layout.addWidget(registrations)
|
||
|
||
sessions, sessions_layout = _card(
|
||
"客户会话档案",
|
||
"机器人重启后仍可继续使用已保存的上下文；双击任意一行查看完整记录。",
|
||
)
|
||
self.session_table = self._table(["最近更新", "会话 ID", "消息数", "内容预览"])
|
||
self.session_table.setToolTip("双击任意会话查看完整记录")
|
||
self.session_table.cellDoubleClicked.connect(self._open_session_detail)
|
||
sessions_layout.addWidget(self.session_table)
|
||
session_actions = QHBoxLayout()
|
||
session_actions.addStretch(1)
|
||
view_session = _button("查看记录", "primary")
|
||
view_session.clicked.connect(self._open_selected_session)
|
||
delete_session = _button("删除所选", "danger")
|
||
delete_session.clicked.connect(self._delete_sessions)
|
||
session_actions.addWidget(view_session)
|
||
session_actions.addWidget(delete_session)
|
||
sessions_layout.addLayout(session_actions)
|
||
layout.addWidget(sessions)
|
||
layout.addStretch(1)
|
||
self.refresh_data()
|
||
|
||
@staticmethod
|
||
def _table(headers: list[str]) -> QTableWidget:
|
||
table = QTableWidget(0, len(headers))
|
||
table.setHorizontalHeaderLabels(headers)
|
||
table.setSelectionBehavior(QAbstractItemView.SelectRows)
|
||
table.setSelectionMode(QAbstractItemView.ExtendedSelection)
|
||
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
|
||
table.setAlternatingRowColors(True)
|
||
table.verticalHeader().hide()
|
||
table.verticalHeader().setDefaultSectionSize(42)
|
||
table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeToContents)
|
||
table.horizontalHeader().setStretchLastSection(True)
|
||
table.setMinimumHeight(230)
|
||
return table
|
||
|
||
@staticmethod
|
||
def _time_text(value) -> str:
|
||
try:
|
||
return time.strftime("%Y-%m-%d %H:%M", time.localtime(float(value)))
|
||
except (TypeError, ValueError, OSError):
|
||
return "--"
|
||
|
||
@staticmethod
|
||
def _selected_ids(table: QTableWidget) -> list[str]:
|
||
rows = sorted({index.row() for index in table.selectionModel().selectedRows()})
|
||
ids = []
|
||
for row in rows:
|
||
item = table.item(row, 0)
|
||
if item:
|
||
ids.append(str(item.data(Qt.UserRole) or ""))
|
||
return [item for item in ids if item]
|
||
|
||
def refresh_data(self) -> None:
|
||
try:
|
||
from registration_store import RegistrationStore
|
||
|
||
self.registration_store = RegistrationStore()
|
||
leads = self.registration_store.list_leads(include_done=True)
|
||
except Exception as exc:
|
||
leads = []
|
||
self.logMessage.emit(f"读取挂号登记失败：{exc}", "err")
|
||
status_map = {"pending_symptom": "待补症状", "booked": "待联系", "done": "已联系"}
|
||
self.registration_table.setRowCount(len(leads))
|
||
for row, lead in enumerate(leads):
|
||
values = (
|
||
self._time_text(lead.get("updated")),
|
||
lead.get("contact") or "未知客户",
|
||
(lead.get("symptom") or "待补充").replace("\n", " "),
|
||
status_map.get(lead.get("status"), lead.get("status") or "--"),
|
||
)
|
||
for column, value in enumerate(values):
|
||
item = QTableWidgetItem(str(value))
|
||
if column == 0:
|
||
item.setData(Qt.UserRole, lead.get("id"))
|
||
self.registration_table.setItem(row, column, item)
|
||
pending = sum(1 for item in leads if item.get("status") in {"pending_symptom", "booked"})
|
||
self.registration_metric.value.setText(str(pending))
|
||
|
||
try:
|
||
from conversation_store import ConversationStore
|
||
|
||
self.conversation_store = ConversationStore(str(SCRIPT_DIR / "conversations.json"))
|
||
sessions = self.conversation_store.list_sessions(limit=500)
|
||
except Exception as exc:
|
||
sessions = []
|
||
self.logMessage.emit(f"读取会话档案失败：{exc}", "err")
|
||
self.session_table.setRowCount(len(sessions))
|
||
for row, session in enumerate(sessions):
|
||
values = (
|
||
self._time_text(session.get("updated")),
|
||
session.get("session_id") or "--",
|
||
session.get("message_count") or 0,
|
||
session.get("preview") or "--",
|
||
)
|
||
for column, value in enumerate(values):
|
||
item = QTableWidgetItem(str(value))
|
||
if column == 0:
|
||
item.setData(Qt.UserRole, session.get("session_id"))
|
||
self.session_table.setItem(row, column, item)
|
||
self.session_metric.value.setText(str(len(sessions)))
|
||
|
||
def _mark_registration_done(self) -> None:
|
||
ids = self._selected_ids(self.registration_table)
|
||
if not ids:
|
||
QMessageBox.information(self, "提示", "请先选择登记记录。")
|
||
return
|
||
count = self.registration_store.set_status_many(ids, "done")
|
||
self.logMessage.emit(f"已将 {count} 条登记标记为已联系", "ok")
|
||
self.refresh_data()
|
||
|
||
def _delete_registrations(self) -> None:
|
||
ids = self._selected_ids(self.registration_table)
|
||
if not ids:
|
||
QMessageBox.information(self, "提示", "请先选择要删除的登记。")
|
||
return
|
||
if QMessageBox.question(self, "确认删除", f"确定删除选中的 {len(ids)} 条登记吗？") != QMessageBox.Yes:
|
||
return
|
||
count = self.registration_store.delete_many(ids)
|
||
self.logMessage.emit(f"已删除 {count} 条挂号登记", "warn")
|
||
self.refresh_data()
|
||
|
||
def _delete_sessions(self) -> None:
|
||
ids = self._selected_ids(self.session_table)
|
||
if not ids:
|
||
QMessageBox.information(self, "提示", "请先选择要删除的会话档案。")
|
||
return
|
||
if QMessageBox.question(self, "确认删除", f"确定删除选中的 {len(ids)} 份档案吗？") != QMessageBox.Yes:
|
||
return
|
||
count = sum(1 for session_id in ids if self.conversation_store.delete(session_id))
|
||
self.logMessage.emit(f"已删除 {count} 份会话档案", "warn")
|
||
self.refresh_data()
|
||
|
||
def _open_selected_session(self) -> None:
|
||
rows = self.session_table.selectionModel().selectedRows()
|
||
if not rows:
|
||
QMessageBox.information(self, "提示", "请先选择要查看的会话档案。")
|
||
return
|
||
self._open_session_detail(rows[0].row(), 0)
|
||
|
||
def _open_session_detail(self, row: int, _column: int) -> None:
|
||
id_item = self.session_table.item(row, 0)
|
||
session_id = str(id_item.data(Qt.UserRole) or "") if id_item else ""
|
||
if not session_id:
|
||
QMessageBox.warning(self, "无法打开", "未找到这条会话的 ID。")
|
||
return
|
||
try:
|
||
history = list(self.conversation_store.history(session_id))
|
||
last_lines = list(self.conversation_store.last_lines(session_id))
|
||
except Exception as exc:
|
||
QMessageBox.critical(self, "无法读取会话", str(exc))
|
||
return
|
||
SessionDetailDialog(session_id, history, last_lines, self).exec()
|
||
|
||
|
||
class BackendLoginDialog(QDialog):
|
||
completed = Signal(bool, str)
|
||
|
||
def __init__(self, parent: QWidget | None = None):
|
||
super().__init__(parent)
|
||
self.setWindowTitle("登录配置后台")
|
||
self.setObjectName("SessionDetailDialog")
|
||
self.setModal(True)
|
||
self.setMinimumWidth(480)
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(26, 24, 26, 24)
|
||
layout.setSpacing(14)
|
||
layout.addWidget(_label("登录配置后台", "CardTitle"))
|
||
layout.addWidget(
|
||
_label("登录成功后，模型与 MCP 参数会自动同步到当前电脑。", "CardSubtitle")
|
||
)
|
||
|
||
import backend_client
|
||
|
||
settings = backend_client.load_settings()
|
||
layout.addWidget(_label("后台地址", "CardSubtitle"))
|
||
self.server_url = QLineEdit(settings.get("server_url", backend_client.DEFAULT_SERVER_URL))
|
||
self.server_url.setPlaceholderText("http://127.0.0.1:8765")
|
||
layout.addWidget(self.server_url)
|
||
layout.addWidget(_label("用户名", "CardSubtitle"))
|
||
self.username = QLineEdit(settings.get("username", ""))
|
||
self.username.setPlaceholderText("请输入后台用户名")
|
||
layout.addWidget(self.username)
|
||
layout.addWidget(_label("密码", "CardSubtitle"))
|
||
self.password = QLineEdit()
|
||
self.password.setEchoMode(QLineEdit.Password)
|
||
self.password.setPlaceholderText("密码不会保存在本地")
|
||
self.password.returnPressed.connect(self.submit)
|
||
layout.addWidget(self.password)
|
||
self.auto_sync = QCheckBox("启动及运行期间自动同步")
|
||
self.auto_sync.setChecked(bool(settings.get("auto_sync", True)))
|
||
layout.addWidget(self.auto_sync)
|
||
self.status = _label("", "DangerText")
|
||
self.status.setWordWrap(True)
|
||
layout.addWidget(self.status)
|
||
actions = QHBoxLayout()
|
||
actions.addStretch(1)
|
||
cancel = _button("取消")
|
||
cancel.clicked.connect(self.reject)
|
||
actions.addWidget(cancel)
|
||
self.login_button = _button("登录并同步", "primary")
|
||
self.login_button.clicked.connect(self.submit)
|
||
actions.addWidget(self.login_button)
|
||
layout.addLayout(actions)
|
||
self.completed.connect(self._finished)
|
||
|
||
def submit(self) -> None:
|
||
if not self.server_url.text().strip() or not self.username.text().strip() or not self.password.text():
|
||
self.status.setText("请填写后台地址、用户名和密码")
|
||
return
|
||
self.login_button.setEnabled(False)
|
||
self.status.setObjectName("WarningText")
|
||
self.status.setText("正在登录并获取模型配置…")
|
||
self.status.style().unpolish(self.status)
|
||
self.status.style().polish(self.status)
|
||
server_url = self.server_url.text().strip()
|
||
username = self.username.text().strip()
|
||
password = self.password.text()
|
||
auto_sync = self.auto_sync.isChecked()
|
||
|
||
def worker() -> None:
|
||
try:
|
||
import backend_client
|
||
|
||
backend_client.login(server_url, username, password, auto_sync=auto_sync)
|
||
result = backend_client.sync_config(force=True)
|
||
message = str(result.get("message") or "登录并同步成功")
|
||
except Exception as exc:
|
||
self.completed.emit(False, str(exc))
|
||
else:
|
||
self.completed.emit(True, message)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _finished(self, ok: bool, message: str) -> None:
|
||
self.login_button.setEnabled(True)
|
||
self.status.setObjectName("SuccessText" if ok else "DangerText")
|
||
self.status.setText(message)
|
||
self.status.style().unpolish(self.status)
|
||
self.status.style().polish(self.status)
|
||
if ok:
|
||
QTimer.singleShot(250, self.accept)
|
||
|
||
|
||
class PersonaPage(QScrollArea):
|
||
saved = Signal(bool, str)
|
||
backendSyncFinished = Signal(bool, str)
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setObjectName("PageScroll")
|
||
self.setWidgetResizable(True)
|
||
self.setFrameShape(QFrame.NoFrame)
|
||
root = QWidget()
|
||
root.setObjectName("PageRoot")
|
||
self.setWidget(root)
|
||
layout = QVBoxLayout(root)
|
||
layout.setContentsMargins(28, 25, 28, 28)
|
||
layout.setSpacing(18)
|
||
layout.addLayout(_page_header("05", "AI 人格与能力", "统一管理模型、客服身份、上下文和 MCP 工具。"))
|
||
|
||
import ai_config
|
||
|
||
remote, remote_layout = _card(
|
||
"后台同步",
|
||
"在管理后台配置模型后，本机启动及运行期间会自动获取并应用。",
|
||
)
|
||
remote_row = QHBoxLayout()
|
||
remote_copy = QVBoxLayout()
|
||
self.remote_status = _label("正在读取后台连接…", "WarningText")
|
||
self.remote_meta = _label("", "CardSubtitle")
|
||
remote_copy.addWidget(self.remote_status)
|
||
remote_copy.addWidget(self.remote_meta)
|
||
remote_row.addLayout(remote_copy, 1)
|
||
self.logout_backend_button = _button("退出后台")
|
||
self.logout_backend_button.clicked.connect(self.logout_backend)
|
||
remote_row.addWidget(self.logout_backend_button)
|
||
self.manage_backend_button = _button("打开管理后台")
|
||
self.manage_backend_button.clicked.connect(self.open_backend_page)
|
||
remote_row.addWidget(self.manage_backend_button)
|
||
self.sync_backend_button = _button("立即同步")
|
||
self.sync_backend_button.clicked.connect(lambda: self.sync_from_backend(silent=False))
|
||
remote_row.addWidget(self.sync_backend_button)
|
||
self.login_backend_button = _button("登录后台", "primary")
|
||
self.login_backend_button.clicked.connect(self.open_backend_login)
|
||
remote_row.addWidget(self.login_backend_button)
|
||
remote_layout.addLayout(remote_row)
|
||
layout.addWidget(remote)
|
||
|
||
overview, overview_layout = _card("能力开关", "关闭 AI 后将使用通用设置中的固定回复。")
|
||
switch_grid = QGridLayout()
|
||
self.ai_enabled = QCheckBox("启用 AI 回复")
|
||
self.ai_enabled.setChecked(bool(ai_config.AI_ENABLED))
|
||
self.context_enabled = QCheckBox("启用会话上下文")
|
||
self.context_enabled.setChecked(bool(ai_config.AI_CONTEXT_ENABLED))
|
||
self.counter_enabled = QCheckBox("启用反辱骂策略")
|
||
self.counter_enabled.setChecked(bool(ai_config.AI_COUNTER_INSULT_ENABLED))
|
||
self.mcp_enabled = QCheckBox("启用 MCP 工具")
|
||
self.mcp_enabled.setChecked(bool(ai_config.AI_MCP_ENABLED))
|
||
for index, switch in enumerate(
|
||
(self.ai_enabled, self.context_enabled, self.counter_enabled, self.mcp_enabled)
|
||
):
|
||
switch_grid.addWidget(switch, index // 2, index % 2)
|
||
overview_layout.addLayout(switch_grid)
|
||
layout.addWidget(overview)
|
||
|
||
identity, identity_layout = _card("模型与身份", "保存后立即写入 ai_settings.json。")
|
||
form = QGridLayout()
|
||
form.setHorizontalSpacing(18)
|
||
form.setVerticalSpacing(8)
|
||
self.api_base = self._field(form, 0, 0, "API 地址", ai_config.AI_API_BASE)
|
||
self.model = self._field(form, 0, 1, "模型名称", ai_config.AI_MODEL)
|
||
self.api_key = self._field(form, 2, 0, "API Key", ai_config.AI_API_KEY)
|
||
self.api_key.setEchoMode(QLineEdit.Password)
|
||
self.agent_name = self._field(form, 2, 1, "客服名称", ai_config.AI_AGENT_NAME)
|
||
self.hospital = self._field(form, 4, 0, "机构名称", ai_config.AI_HOSPITAL_NAME, 1, 2)
|
||
self.rounds = QSpinBox()
|
||
self.rounds.setRange(1, 50)
|
||
self.rounds.setValue(int(ai_config.AI_CONTEXT_MAX_ROUNDS))
|
||
self.max_tokens = QSpinBox()
|
||
self.max_tokens.setRange(50, 32000)
|
||
self.max_tokens.setValue(int(ai_config.AI_MAX_TOKENS))
|
||
self.temperature = QDoubleSpinBox()
|
||
self.temperature.setRange(0.0, 2.0)
|
||
self.temperature.setSingleStep(0.05)
|
||
self.temperature.setValue(float(ai_config.AI_TEMPERATURE))
|
||
form.addWidget(_label("上下文轮数", "CardSubtitle"), 6, 0)
|
||
form.addWidget(_label("最大回复 tokens", "CardSubtitle"), 6, 1)
|
||
form.addWidget(self.rounds, 7, 0)
|
||
form.addWidget(self.max_tokens, 7, 1)
|
||
form.addWidget(_label("温度", "CardSubtitle"), 8, 0)
|
||
form.addWidget(self.temperature, 9, 0)
|
||
form.setColumnStretch(0, 1)
|
||
form.setColumnStretch(1, 1)
|
||
identity_layout.addLayout(form)
|
||
layout.addWidget(identity)
|
||
|
||
mcp, mcp_layout = _card("MCP 服务器", "填写服务器 JSON 数组，配置将与 AI 能力一起保存。")
|
||
self.mcp_json = QPlainTextEdit()
|
||
self.mcp_json.setMinimumHeight(160)
|
||
self.mcp_json.setPlainText(
|
||
json.dumps(getattr(ai_config, "AI_MCP_SERVERS", []) or [], ensure_ascii=False, indent=2)
|
||
)
|
||
mcp_layout.addWidget(self.mcp_json)
|
||
mcp_row = QHBoxLayout()
|
||
mcp_row.addWidget(_label("单次回复最多工具轮数", "CardSubtitle"))
|
||
self.mcp_rounds = QSpinBox()
|
||
self.mcp_rounds.setRange(1, 20)
|
||
self.mcp_rounds.setValue(int(getattr(ai_config, "AI_MCP_MAX_ROUNDS", 5)))
|
||
mcp_row.addWidget(self.mcp_rounds)
|
||
mcp_row.addStretch(1)
|
||
mcp_layout.addLayout(mcp_row)
|
||
layout.addWidget(mcp)
|
||
|
||
actions = QHBoxLayout()
|
||
self.save_status = _label("", "SuccessText")
|
||
actions.addWidget(self.save_status, 1)
|
||
self.save_button = _button("保存 AI 配置", "primary")
|
||
self.save_button.clicked.connect(self.save_config)
|
||
actions.addWidget(self.save_button)
|
||
layout.addLayout(actions)
|
||
layout.addStretch(1)
|
||
self._backend_sync_running = False
|
||
self._backend_sync_silent = False
|
||
self.backendSyncFinished.connect(self._backend_sync_finished)
|
||
self._managed_widgets = (
|
||
self.ai_enabled,
|
||
self.context_enabled,
|
||
self.counter_enabled,
|
||
self.mcp_enabled,
|
||
self.api_base,
|
||
self.model,
|
||
self.api_key,
|
||
self.agent_name,
|
||
self.hospital,
|
||
self.rounds,
|
||
self.max_tokens,
|
||
self.temperature,
|
||
self.mcp_json,
|
||
self.mcp_rounds,
|
||
self.save_button,
|
||
)
|
||
self.refresh_backend_status()
|
||
|
||
@staticmethod
|
||
def _field(
|
||
layout: QGridLayout,
|
||
row: int,
|
||
column: int,
|
||
title: str,
|
||
value,
|
||
row_span: int = 1,
|
||
column_span: int = 1,
|
||
) -> QLineEdit:
|
||
layout.addWidget(_label(title, "CardSubtitle"), row, column, 1, column_span)
|
||
field = QLineEdit(str(value or ""))
|
||
layout.addWidget(field, row + 1, column, row_span, column_span)
|
||
return field
|
||
|
||
def save_config(self) -> None:
|
||
try:
|
||
import backend_client
|
||
|
||
remote_managed = backend_client.is_configured() and bool(
|
||
backend_client.load_settings().get("auto_sync", True)
|
||
)
|
||
except Exception:
|
||
remote_managed = False
|
||
if remote_managed:
|
||
self.save_status.setText("当前由后台统一管理，请在后台修改后点击“立即同步”")
|
||
self.save_status.setObjectName("WarningText")
|
||
self.save_status.style().unpolish(self.save_status)
|
||
self.save_status.style().polish(self.save_status)
|
||
return
|
||
try:
|
||
servers = json.loads(self.mcp_json.toPlainText().strip() or "[]")
|
||
if not isinstance(servers, list):
|
||
raise ValueError("MCP JSON 根节点必须是数组")
|
||
import ai_config
|
||
|
||
ai_config.AI_ENABLED = self.ai_enabled.isChecked()
|
||
ai_config.AI_CONTEXT_ENABLED = self.context_enabled.isChecked()
|
||
ai_config.AI_COUNTER_INSULT_ENABLED = self.counter_enabled.isChecked()
|
||
ai_config.AI_MCP_ENABLED = self.mcp_enabled.isChecked()
|
||
ai_config.AI_API_BASE = self.api_base.text().strip()
|
||
ai_config.AI_API_KEY = self.api_key.text().strip()
|
||
ai_config.AI_MODEL = self.model.text().strip()
|
||
ai_config.AI_AGENT_NAME = self.agent_name.text().strip()
|
||
ai_config.AI_HOSPITAL_NAME = self.hospital.text().strip()
|
||
ai_config.AI_CONTEXT_MAX_ROUNDS = self.rounds.value()
|
||
ai_config.AI_MAX_TOKENS = self.max_tokens.value()
|
||
ai_config.AI_TEMPERATURE = self.temperature.value()
|
||
ai_config.AI_MCP_SERVERS = servers
|
||
ai_config.AI_MCP_MAX_ROUNDS = self.mcp_rounds.value()
|
||
ai_config.AI_SYSTEM_PROMPT = ai_config.build_system_prompt()
|
||
ai_config.save_settings()
|
||
except Exception as exc:
|
||
self.save_status.setText(f"保存失败：{exc}")
|
||
self.save_status.setObjectName("DangerText")
|
||
self.saved.emit(False, str(exc))
|
||
else:
|
||
self.save_status.setText("配置已保存并立即生效")
|
||
self.save_status.setObjectName("SuccessText")
|
||
self.saved.emit(True, "AI 配置已保存并生效")
|
||
self.save_status.style().unpolish(self.save_status)
|
||
self.save_status.style().polish(self.save_status)
|
||
|
||
def open_backend_login(self) -> None:
|
||
dialog = BackendLoginDialog(self)
|
||
if dialog.exec() == QDialog.Accepted:
|
||
self.reload_from_ai_config()
|
||
self.refresh_backend_status()
|
||
self.saved.emit(True, "后台已登录，模型配置已自动同步")
|
||
|
||
def open_backend_page(self) -> None:
|
||
try:
|
||
import backend_client
|
||
|
||
url = backend_client.load_settings().get(
|
||
"server_url", backend_client.DEFAULT_SERVER_URL
|
||
)
|
||
QDesktopServices.openUrl(QUrl(str(url)))
|
||
except Exception as exc:
|
||
self.saved.emit(False, f"无法打开管理后台：{exc}")
|
||
|
||
def logout_backend(self) -> None:
|
||
answer = QMessageBox.question(
|
||
self,
|
||
"退出配置后台",
|
||
"退出后将停止自动同步，但已下发到本机的配置会继续保留。确定退出吗？",
|
||
)
|
||
if answer != QMessageBox.Yes:
|
||
return
|
||
try:
|
||
import backend_client
|
||
|
||
backend_client.logout()
|
||
except Exception as exc:
|
||
self.saved.emit(False, f"退出后台失败：{exc}")
|
||
return
|
||
self.refresh_backend_status()
|
||
self.saved.emit(True, "已退出配置后台，当前配置保留在本机")
|
||
|
||
def refresh_backend_status(self) -> None:
|
||
try:
|
||
import backend_client
|
||
|
||
summary = backend_client.connection_summary()
|
||
except Exception as exc:
|
||
self.remote_status.setObjectName("DangerText")
|
||
self.remote_status.setText(f"后台连接配置读取失败：{exc}")
|
||
return
|
||
configured = bool(summary["configured"])
|
||
authenticated = bool(summary.get("authenticated"))
|
||
managed = configured and bool(summary["auto_sync"])
|
||
if configured:
|
||
last_error = str(summary.get("last_error") or "")
|
||
self.remote_status.setObjectName("DangerText" if last_error else "SuccessText")
|
||
self.remote_status.setText(
|
||
f"最近同步失败 · {last_error}"
|
||
if last_error
|
||
else (
|
||
f"已连接 · {summary['username']} · 配置 v{summary['last_version']}"
|
||
if authenticated
|
||
else f"已发现本机后台 · 配置 v{summary['last_version']}"
|
||
)
|
||
)
|
||
sync_at = summary.get("last_sync_at") or "尚未同步"
|
||
self.remote_meta.setText(f"{summary['server_url']} · 最后同步 {sync_at}")
|
||
else:
|
||
self.remote_status.setObjectName("WarningText")
|
||
self.remote_status.setText("尚未登录配置后台")
|
||
self.remote_meta.setText(f"默认地址 {summary['server_url']} · 当前使用本机配置")
|
||
self.remote_status.style().unpolish(self.remote_status)
|
||
self.remote_status.style().polish(self.remote_status)
|
||
self.login_backend_button.setVisible(not authenticated)
|
||
self.login_backend_button.setText("账号登录" if configured else "登录后台")
|
||
self.logout_backend_button.setVisible(authenticated)
|
||
self.sync_backend_button.setVisible(configured)
|
||
for widget in self._managed_widgets:
|
||
widget.setEnabled(not managed)
|
||
self.save_status.setText("后台统一管理中，本地字段为只读" if managed else "")
|
||
self.save_status.setObjectName("SuccessText" if managed else "CardSubtitle")
|
||
self.save_status.style().unpolish(self.save_status)
|
||
self.save_status.style().polish(self.save_status)
|
||
|
||
def sync_from_backend(self, *, silent: bool = True) -> None:
|
||
if self._backend_sync_running:
|
||
return
|
||
try:
|
||
import backend_client
|
||
|
||
if not backend_client.is_configured():
|
||
return
|
||
except Exception as exc:
|
||
if not silent:
|
||
self.saved.emit(False, f"后台连接配置读取失败：{exc}")
|
||
return
|
||
self._backend_sync_running = True
|
||
self._backend_sync_silent = silent
|
||
self.sync_backend_button.setEnabled(False)
|
||
self.remote_status.setObjectName("WarningText")
|
||
self.remote_status.setText("正在从后台同步模型配置…")
|
||
self.remote_status.style().unpolish(self.remote_status)
|
||
self.remote_status.style().polish(self.remote_status)
|
||
|
||
def worker() -> None:
|
||
try:
|
||
result = backend_client.sync_config(force=not silent)
|
||
message = str(result.get("message") or "后台配置同步完成")
|
||
except Exception as exc:
|
||
self.backendSyncFinished.emit(False, str(exc))
|
||
else:
|
||
self.backendSyncFinished.emit(True, message)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _backend_sync_finished(self, ok: bool, message: str) -> None:
|
||
silent = self._backend_sync_silent
|
||
self._backend_sync_running = False
|
||
self.sync_backend_button.setEnabled(True)
|
||
if ok:
|
||
self.reload_from_ai_config()
|
||
self.refresh_backend_status()
|
||
if not silent or not ok:
|
||
self.saved.emit(ok, message if ok else f"后台同步失败：{message}")
|
||
|
||
def reload_from_ai_config(self) -> None:
|
||
import ai_config
|
||
|
||
self.ai_enabled.setChecked(bool(ai_config.AI_ENABLED))
|
||
self.context_enabled.setChecked(bool(ai_config.AI_CONTEXT_ENABLED))
|
||
self.counter_enabled.setChecked(bool(ai_config.AI_COUNTER_INSULT_ENABLED))
|
||
self.mcp_enabled.setChecked(bool(ai_config.AI_MCP_ENABLED))
|
||
self.api_base.setText(str(ai_config.AI_API_BASE or ""))
|
||
self.model.setText(str(ai_config.AI_MODEL or ""))
|
||
self.api_key.setText(str(ai_config.AI_API_KEY or ""))
|
||
self.agent_name.setText(str(ai_config.AI_AGENT_NAME or ""))
|
||
self.hospital.setText(str(ai_config.AI_HOSPITAL_NAME or ""))
|
||
self.rounds.setValue(int(ai_config.AI_CONTEXT_MAX_ROUNDS))
|
||
self.max_tokens.setValue(int(ai_config.AI_MAX_TOKENS))
|
||
self.temperature.setValue(float(ai_config.AI_TEMPERATURE))
|
||
self.mcp_json.setPlainText(
|
||
json.dumps(getattr(ai_config, "AI_MCP_SERVERS", []) or [], ensure_ascii=False, indent=2)
|
||
)
|
||
self.mcp_rounds.setValue(int(getattr(ai_config, "AI_MCP_MAX_ROUNDS", 5)))
|
||
|
||
|
||
class LogPage(QWidget):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setObjectName("PageRoot")
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(28, 25, 28, 28)
|
||
layout.setSpacing(18)
|
||
header = QHBoxLayout()
|
||
header.addLayout(_page_header("06", "运行日志", "实时查看连接、识别、AI 与业务沉淀事件。"), 1)
|
||
clear = _button("清空日志")
|
||
clear.clicked.connect(self.clear)
|
||
header.addWidget(clear, 0, Qt.AlignBottom)
|
||
layout.addLayout(header)
|
||
card, card_layout = _card()
|
||
self.editor = QTextEdit()
|
||
self.editor.setReadOnly(True)
|
||
self.editor.setAcceptRichText(True)
|
||
self.editor.setStyleSheet(
|
||
"QTextEdit{font-family:'Cascadia Mono','Microsoft YaHei UI';font-size:13px;line-height:1.5;}"
|
||
)
|
||
card_layout.addWidget(self.editor)
|
||
layout.addWidget(card, 1)
|
||
|
||
def append(self, message: str, tag: str = "") -> None:
|
||
colors = {
|
||
"ok": COLORS["success"],
|
||
"warn": COLORS["warning"],
|
||
"err": COLORS["danger"],
|
||
"notify": COLORS["accent_dark"],
|
||
}
|
||
color = colors.get(tag, COLORS["ink"])
|
||
stamp = time.strftime("%H:%M:%S")
|
||
safe = (
|
||
str(message)
|
||
.replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">")
|
||
.replace("\n", "<br>")
|
||
)
|
||
self.editor.append(
|
||
f'<span style="color:#94A29B">[{stamp}]</span> '
|
||
f'<span style="color:{color}">{safe}</span>'
|
||
)
|
||
|
||
def clear(self) -> None:
|
||
self.editor.clear()
|
||
|
||
|
||
class CapsuleWindow(QWidget):
|
||
expandRequested = Signal()
|
||
stopRequested = Signal()
|
||
|
||
def __init__(self):
|
||
super().__init__(None)
|
||
self.setWindowFlags(
|
||
Qt.Tool
|
||
| Qt.FramelessWindowHint
|
||
| Qt.WindowStaysOnTopHint
|
||
| Qt.WindowDoesNotAcceptFocus
|
||
)
|
||
self.setAttribute(Qt.WA_TranslucentBackground)
|
||
self.setAttribute(Qt.WA_ShowWithoutActivating)
|
||
self.setFixedSize(430, 82)
|
||
self._drag_origin: QPoint | None = None
|
||
self._window_origin = QPoint()
|
||
|
||
outer = QVBoxLayout(self)
|
||
outer.setContentsMargins(12, 8, 12, 14)
|
||
panel = QFrame()
|
||
panel.setObjectName("CapsulePanel")
|
||
panel.setStyleSheet(
|
||
"QFrame#CapsulePanel{background:#FFFFFF;border:1px solid #CFE0D8;border-radius:16px;}"
|
||
"QFrame#CapsulePanel QLabel{background:transparent;border:none;}"
|
||
)
|
||
_shadow(panel, 28, 7, 38)
|
||
row = QHBoxLayout(panel)
|
||
row.setContentsMargins(17, 10, 10, 10)
|
||
row.setSpacing(11)
|
||
grip = QLabel("•••")
|
||
grip.setAlignment(Qt.AlignCenter)
|
||
grip.setFixedSize(38, 38)
|
||
grip.setStyleSheet(
|
||
"QLabel{background:#F0F6F3;border:1px solid #D8E6DF;border-radius:19px;"
|
||
"color:#789087;font-family:'Cascadia Mono';font-weight:700;}"
|
||
)
|
||
row.addWidget(grip)
|
||
self.dot = QLabel("●")
|
||
self.dot.setStyleSheet("QLabel{color:#10A57A;font-size:17px;}")
|
||
row.addWidget(self.dot)
|
||
copy = QVBoxLayout()
|
||
copy.setSpacing(1)
|
||
self.status = QLabel("监听中")
|
||
self.status.setStyleSheet(
|
||
"QLabel{color:#087A59;font-size:14px;font-weight:700;}"
|
||
)
|
||
self.hint = QLabel("安全运行中")
|
||
self.hint.setStyleSheet("QLabel{color:#7B8D84;font-size:11px;}")
|
||
copy.addWidget(self.status)
|
||
copy.addWidget(self.hint)
|
||
row.addLayout(copy, 1)
|
||
self.timer = QLabel("00:00:00")
|
||
self.timer.setAlignment(Qt.AlignCenter)
|
||
self.timer.setMinimumWidth(82)
|
||
self.timer.setStyleSheet(
|
||
"QLabel{background:#F2F7F4;border:1px solid #DCE8E2;border-radius:9px;"
|
||
"padding:7px 9px;font-family:'Cascadia Mono';font-size:12px;color:#30473D;}"
|
||
)
|
||
row.addWidget(self.timer)
|
||
expand = QPushButton("↗")
|
||
expand.setToolTip("展开控制台")
|
||
expand.setCursor(Qt.PointingHandCursor)
|
||
expand.setFixedSize(34, 34)
|
||
expand.setStyleSheet(
|
||
"QPushButton{background:#E7F4EE;color:#087A59;border:1px solid #CFE8DD;"
|
||
"border-radius:8px;font-size:17px;font-weight:700;}"
|
||
"QPushButton:hover{background:#D4EFE4;border-color:#AEDCC9;}"
|
||
"QPushButton:pressed{background:#C5E8D9;}"
|
||
)
|
||
expand.clicked.connect(self.expandRequested)
|
||
row.addWidget(expand)
|
||
stop = QPushButton("×")
|
||
stop.setToolTip("停止监听")
|
||
stop.setCursor(Qt.PointingHandCursor)
|
||
stop.setFixedSize(34, 34)
|
||
stop.setStyleSheet(
|
||
"QPushButton{background:#FBEAEC;color:#C74452;border:1px solid #F2CDD2;"
|
||
"border-radius:8px;font-size:18px;}"
|
||
"QPushButton:hover{background:#D85260;color:#FFFFFF;border-color:#D85260;}"
|
||
)
|
||
stop.clicked.connect(self.stopRequested)
|
||
row.addWidget(stop)
|
||
outer.addWidget(panel)
|
||
|
||
def show_near(self, window: QWidget) -> None:
|
||
screen = window.screen() or QApplication.primaryScreen()
|
||
area = screen.availableGeometry()
|
||
x = area.x() + (area.width() - self.width()) // 2
|
||
y = area.y() + 28
|
||
self.move(x, y)
|
||
self.show()
|
||
|
||
def mousePressEvent(self, event) -> None:
|
||
if event.button() == Qt.LeftButton:
|
||
self._drag_origin = event.globalPosition().toPoint()
|
||
self._window_origin = self.pos()
|
||
event.accept()
|
||
|
||
def mouseMoveEvent(self, event) -> None:
|
||
if self._drag_origin is not None and event.buttons() & Qt.LeftButton:
|
||
self.move(self._window_origin + event.globalPosition().toPoint() - self._drag_origin)
|
||
event.accept()
|
||
|
||
def mouseReleaseEvent(self, event) -> None:
|
||
self._drag_origin = None
|
||
super().mouseReleaseEvent(event)
|
||
|
||
def set_status(self, state: str, text: str) -> None:
|
||
colors = {
|
||
"running": "#10A57A",
|
||
"waiting": "#B67820",
|
||
"connecting": "#B67820",
|
||
"stopping": "#B67820",
|
||
"error": "#D85260",
|
||
}
|
||
hints = {
|
||
"running": "安全运行中",
|
||
"waiting": "等待企业微信",
|
||
"connecting": "正在连接窗口",
|
||
"stopping": "正在结束任务",
|
||
"error": "请查看运行日志",
|
||
"stopped": "监听已停止",
|
||
}
|
||
color = colors.get(state, "#7B8D84")
|
||
self.dot.setStyleSheet(f"QLabel{{color:{color};font-size:17px;}}")
|
||
self.status.setText(text)
|
||
self.status.setStyleSheet(
|
||
f"QLabel{{color:{color};font-size:14px;font-weight:700;}}"
|
||
)
|
||
self.hint.setText(hints.get(state, text))
|
||
|
||
|
||
class Sidebar(QFrame):
|
||
pageRequested = Signal(int)
|
||
startRequested = Signal()
|
||
stopRequested = Signal()
|
||
|
||
PAGE_NAMES = ("AI 客服", "自动回复", "通用设置", "业务数据", "AI 人格", "运行日志")
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setObjectName("Sidebar")
|
||
self.setFixedWidth(258)
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(18, 24, 18, 18)
|
||
layout.setSpacing(8)
|
||
layout.addWidget(_label("ZHEN AI DESK", "Brand"))
|
||
layout.addWidget(_label("AI 客服 · 企业微信自动回复", "BrandSub"))
|
||
layout.addSpacing(13)
|
||
|
||
status_card = QFrame()
|
||
status_card.setObjectName("SideStatus")
|
||
status_layout = QVBoxLayout(status_card)
|
||
status_layout.setContentsMargins(15, 13, 15, 13)
|
||
status_layout.setSpacing(3)
|
||
self.status_title = _label("● 已停止", "SideStatusTitle")
|
||
self.status_hint = _label("等待连接企业微信", "SideStatusHint")
|
||
status_layout.addWidget(self.status_title)
|
||
status_layout.addWidget(self.status_hint)
|
||
layout.addWidget(status_card)
|
||
layout.addSpacing(9)
|
||
layout.addWidget(_label("CONTROL CENTER", "SectionLabel"))
|
||
|
||
self.nav_buttons: list[QPushButton] = []
|
||
for index, name in enumerate(self.PAGE_NAMES, start=1):
|
||
button = QPushButton(f"{index:02d} {name}")
|
||
button.setObjectName("NavButton")
|
||
button.setCheckable(True)
|
||
button.setAutoExclusive(True)
|
||
button.setCursor(Qt.PointingHandCursor)
|
||
button.clicked.connect(lambda _checked=False, i=index - 1: self.pageRequested.emit(i))
|
||
layout.addWidget(button)
|
||
self.nav_buttons.append(button)
|
||
self.nav_buttons[0].setChecked(True)
|
||
layout.addStretch(1)
|
||
self.monitor_button = _button("开始监听", "side")
|
||
self.monitor_button.clicked.connect(self.startRequested)
|
||
layout.addWidget(self.monitor_button)
|
||
layout.addWidget(_label("RPA DESKTOP / LOCAL", "SectionLabel"))
|
||
|
||
def select(self, index: int) -> None:
|
||
if 0 <= index < len(self.nav_buttons):
|
||
self.nav_buttons[index].setChecked(True)
|
||
|
||
def set_status(self, state: str, text: str, hint: str) -> None:
|
||
colors = {
|
||
"running": "#43E1AE",
|
||
"waiting": "#F2BC64",
|
||
"connecting": "#F2BC64",
|
||
"stopping": "#F2BC64",
|
||
"error": "#FF818E",
|
||
"stopped": "#A5B9B0",
|
||
}
|
||
self.status_title.setText(f"● {text}")
|
||
self.status_title.setStyleSheet(
|
||
f"color:{colors.get(state, '#A5B9B0')};font-size:15px;font-weight:700;"
|
||
)
|
||
self.status_hint.setText(hint)
|
||
active = state in {"running", "waiting", "connecting", "stopping"}
|
||
self.monitor_button.setText("停止监听" if active else "开始监听")
|
||
try:
|
||
self.monitor_button.clicked.disconnect()
|
||
except RuntimeError:
|
||
pass
|
||
self.monitor_button.clicked.connect(self.stopRequested if active else self.startRequested)
|
||
self.monitor_button.setObjectName("DangerButton" if active else "SidePrimary")
|
||
self.monitor_button.style().unpolish(self.monitor_button)
|
||
self.monitor_button.style().polish(self.monitor_button)
|
||
|
||
|
||
class MainWindow(QMainWindow):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setWindowTitle(WIN_TITLE)
|
||
self.setMinimumSize(1120, 720)
|
||
self._set_initial_geometry()
|
||
self._queue: queue.Queue = queue.Queue()
|
||
self._thread = None
|
||
self._running = False
|
||
self._start_time: float | None = None
|
||
self._status_key = "stopped"
|
||
self._original_stdout = sys.stdout
|
||
self._stdout_proxy = None
|
||
self._saved_geometry = None
|
||
self._was_maximized = False
|
||
self.runtime_settings = self._load_runtime_settings()
|
||
|
||
root = QWidget()
|
||
root.setObjectName("AppRoot")
|
||
self.setCentralWidget(root)
|
||
root_layout = QHBoxLayout(root)
|
||
root_layout.setContentsMargins(0, 0, 0, 0)
|
||
root_layout.setSpacing(0)
|
||
|
||
self.sidebar = Sidebar()
|
||
root_layout.addWidget(self.sidebar)
|
||
self.stack = FadingStack()
|
||
root_layout.addWidget(self.stack, 1)
|
||
|
||
self.portal_page = PortalPage()
|
||
self.dashboard_page = DashboardPage()
|
||
self.settings_page = SettingsPage(self.runtime_settings)
|
||
self.business_page = BusinessPage()
|
||
self.persona_page = PersonaPage()
|
||
self.log_page = LogPage()
|
||
for page in (
|
||
self.portal_page,
|
||
self.dashboard_page,
|
||
self.settings_page,
|
||
self.business_page,
|
||
self.persona_page,
|
||
self.log_page,
|
||
):
|
||
self.stack.addWidget(page)
|
||
|
||
self.capsule = CapsuleWindow()
|
||
self._connect_signals()
|
||
self.business_refresh_timer = QTimer(self)
|
||
self.business_refresh_timer.setSingleShot(True)
|
||
self.business_refresh_timer.setInterval(220)
|
||
self.business_refresh_timer.timeout.connect(self._refresh_counts)
|
||
self.set_status("stopped", "已停止")
|
||
self._refresh_counts()
|
||
self.append_log("新控制台已就绪。请确认企业微信已登录，再开始监听。", "notify")
|
||
|
||
self.queue_timer = QTimer(self)
|
||
self.queue_timer.setInterval(140)
|
||
self.queue_timer.timeout.connect(self._process_queue)
|
||
self.queue_timer.start()
|
||
|
||
try:
|
||
import backend_client
|
||
|
||
sync_seconds = int(
|
||
backend_client.load_settings().get("sync_interval_seconds", 300)
|
||
)
|
||
except Exception:
|
||
sync_seconds = 300
|
||
self.backend_sync_timer = QTimer(self)
|
||
self.backend_sync_timer.setInterval(max(60, sync_seconds) * 1000)
|
||
self.backend_sync_timer.timeout.connect(
|
||
lambda: self.persona_page.sync_from_backend(silent=True)
|
||
)
|
||
self.backend_sync_timer.start()
|
||
QTimer.singleShot(350, lambda: self.persona_page.sync_from_backend(silent=True))
|
||
|
||
def _set_initial_geometry(self) -> None:
|
||
screen = QApplication.primaryScreen()
|
||
area = screen.availableGeometry() if screen else QRect(0, 0, 1440, 900)
|
||
width = min(1480, max(1120, area.width() - 72))
|
||
height = min(940, max(720, area.height() - 72))
|
||
self.resize(width, height)
|
||
self.move(area.x() + (area.width() - width) // 2, area.y() + (area.height() - height) // 2)
|
||
|
||
def _connect_signals(self) -> None:
|
||
self.sidebar.pageRequested.connect(self.show_page)
|
||
self.sidebar.startRequested.connect(self.start_monitoring)
|
||
self.sidebar.stopRequested.connect(self.stop_monitoring)
|
||
self.dashboard_page.startRequested.connect(self.start_monitoring)
|
||
self.dashboard_page.stopRequested.connect(self.stop_monitoring)
|
||
self.dashboard_page.pageRequested.connect(self.show_page)
|
||
self.settings_page.saved.connect(self.save_runtime_settings)
|
||
self.business_page.logMessage.connect(self.append_log)
|
||
self.persona_page.saved.connect(self._persona_saved)
|
||
self.capsule.expandRequested.connect(self.expand_console)
|
||
self.capsule.stopRequested.connect(self.stop_monitoring)
|
||
|
||
@staticmethod
|
||
def _runtime_defaults() -> dict:
|
||
return {
|
||
"auto_reply_text": "你好",
|
||
"poll_interval": 2.0,
|
||
"mouse_idle_enabled": True,
|
||
"mouse_idle_seconds": 20.0,
|
||
}
|
||
|
||
def _load_runtime_settings(self) -> dict:
|
||
settings = self._runtime_defaults()
|
||
try:
|
||
saved = json.loads(APP_SETTINGS_FILE.read_text(encoding="utf-8"))
|
||
if isinstance(saved, dict):
|
||
settings.update({key: saved[key] for key in settings if key in saved})
|
||
except (OSError, ValueError, TypeError):
|
||
pass
|
||
try:
|
||
settings["poll_interval"] = max(0.2, float(settings["poll_interval"]))
|
||
settings["mouse_idle_seconds"] = max(0.0, float(settings["mouse_idle_seconds"]))
|
||
settings["mouse_idle_enabled"] = bool(settings["mouse_idle_enabled"])
|
||
settings["auto_reply_text"] = str(settings["auto_reply_text"]).strip() or "你好"
|
||
except (TypeError, ValueError):
|
||
return self._runtime_defaults()
|
||
return settings
|
||
|
||
def save_runtime_settings(self, settings: dict | None = None) -> None:
|
||
self.runtime_settings = dict(settings or self.settings_page.values())
|
||
try:
|
||
temporary = APP_SETTINGS_FILE.with_suffix(".json.tmp")
|
||
temporary.write_text(
|
||
json.dumps(self.runtime_settings, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
os.replace(temporary, APP_SETTINGS_FILE)
|
||
except OSError as exc:
|
||
self.settings_page.mark_saved(False, f"保存失败：{exc}")
|
||
self.append_log(f"运行参数保存失败：{exc}", "err")
|
||
else:
|
||
self.settings_page.mark_saved(True, "已自动保存")
|
||
|
||
def show_page(self, index: int) -> None:
|
||
self.sidebar.select(index)
|
||
self.stack.setCurrentIndexAnimated(index)
|
||
if index == 3:
|
||
self.business_page.refresh_data()
|
||
if index == 0:
|
||
QTimer.singleShot(80, lambda: self.portal_page.view.setFocus(Qt.OtherFocusReason))
|
||
|
||
def start_monitoring(self) -> None:
|
||
if self._thread is not None and self._thread.is_alive():
|
||
self.append_log("监听线程已经在运行", "warn")
|
||
return
|
||
self.save_runtime_settings()
|
||
self.persona_page.save_config()
|
||
self._stdout_proxy = LogQueue(self._queue)
|
||
sys.stdout = self._stdout_proxy
|
||
self._thread = BotThread(
|
||
self._queue,
|
||
self.runtime_settings["auto_reply_text"],
|
||
float(self.runtime_settings["poll_interval"]),
|
||
mouse_idle_enabled=bool(self.runtime_settings["mouse_idle_enabled"]),
|
||
mouse_idle_seconds=float(self.runtime_settings["mouse_idle_seconds"]),
|
||
)
|
||
self._running = True
|
||
self._start_time = time.time()
|
||
self.set_status("connecting", "连接中")
|
||
self.append_log("正在连接企业微信窗口…", "notify")
|
||
self._thread.start()
|
||
|
||
def stop_monitoring(self) -> None:
|
||
if self._thread is None or not self._thread.is_alive():
|
||
self._finish_thread("stopped")
|
||
return
|
||
self._running = False
|
||
self._thread.stop()
|
||
self.set_status("stopping", "正在停止")
|
||
self.append_log("正在停止监听…", "warn")
|
||
|
||
def set_status(self, state: str, text: str) -> None:
|
||
self._status_key = state
|
||
hints = {
|
||
"connecting": "正在查找企业微信窗口",
|
||
"waiting": WECOM_WAITING_MESSAGE,
|
||
"running": "安全运行中，检测到未读消息后自动处理",
|
||
"stopping": "等待后台任务退出",
|
||
"error": "连接失败，请查看运行日志",
|
||
"stopped": "等待连接企业微信",
|
||
}
|
||
hint = hints.get(state, text)
|
||
self.sidebar.set_status(state, text, hint)
|
||
self.dashboard_page.set_status(state, text, hint)
|
||
self.capsule.set_status(state, text)
|
||
|
||
def enter_capsule_mode(self) -> None:
|
||
if self.capsule.isVisible():
|
||
return
|
||
self._saved_geometry = self.saveGeometry()
|
||
self._was_maximized = self.isMaximized()
|
||
self.capsule.show_near(self)
|
||
self.hide()
|
||
|
||
def expand_console(self) -> None:
|
||
self.capsule.hide()
|
||
if self._saved_geometry is not None:
|
||
self.restoreGeometry(self._saved_geometry)
|
||
if self._was_maximized:
|
||
self.showMaximized()
|
||
else:
|
||
self.showNormal()
|
||
self.raise_()
|
||
self.activateWindow()
|
||
|
||
def _finish_thread(self, final_state: str) -> None:
|
||
self._running = False
|
||
if self.capsule.isVisible():
|
||
self.expand_console()
|
||
if sys.stdout is self._stdout_proxy:
|
||
sys.stdout = self._original_stdout
|
||
self._stdout_proxy = None
|
||
if final_state == "error":
|
||
self.set_status("error", "连接失败")
|
||
self.show_page(5)
|
||
else:
|
||
self.set_status("stopped", "已停止")
|
||
self.dashboard_page.timer.setText("运行时长 --:--:--")
|
||
|
||
def append_log(self, message: str, tag: str = "") -> None:
|
||
self.log_page.append(message, tag)
|
||
|
||
def _persona_saved(self, ok: bool, message: str) -> None:
|
||
self.append_log(message, "ok" if ok else "err")
|
||
|
||
def _refresh_counts(self) -> None:
|
||
try:
|
||
self.business_page.refresh_data()
|
||
self.dashboard_page.registration.value.setText(
|
||
self.business_page.registration_metric.value.text()
|
||
)
|
||
self.dashboard_page.sessions.value.setText(
|
||
self.business_page.session_metric.value.text()
|
||
)
|
||
except Exception as exc:
|
||
self.append_log(f"刷新业务数据失败：{exc}", "err")
|
||
|
||
def _schedule_business_refresh(self) -> None:
|
||
"""合并短时间内的多条挂号日志，避免在队列循环中反复重建表格。"""
|
||
self.business_refresh_timer.start()
|
||
|
||
def _process_queue(self) -> None:
|
||
try:
|
||
while True:
|
||
kind, data = self._queue.get_nowait()
|
||
if kind == "log":
|
||
lowered = str(data).lower()
|
||
if any(token in str(data) for token in ("[+]", "回复完成", "已保存")):
|
||
tag = "ok"
|
||
elif any(token in str(data) for token in ("[!]", "误判", "停止")):
|
||
tag = "warn"
|
||
elif any(token in lowered for token in ("错误", "[-]", "失败", "exception", "traceback")):
|
||
tag = "err"
|
||
elif any(token in str(data) for token in ("[MCP]", "[挂号]", "启动", "连接")):
|
||
tag = "notify"
|
||
else:
|
||
tag = ""
|
||
self.append_log(str(data), tag)
|
||
if "[挂号]" in str(data):
|
||
self._schedule_business_refresh()
|
||
elif kind == "status":
|
||
if data == "running":
|
||
self.set_status("running", "监听中")
|
||
self.enter_capsule_mode()
|
||
elif data == "waiting":
|
||
self.set_status("waiting", "等待企业微信")
|
||
elif data == "error":
|
||
self._finish_thread("error")
|
||
elif data == "stopped":
|
||
self._finish_thread("stopped")
|
||
elif kind == "stats":
|
||
self.dashboard_page.replied.value.setText(str(data.get("replied", 0)))
|
||
self.dashboard_page.false_pos.value.setText(str(data.get("false_pos", 0)))
|
||
except KeyboardInterrupt:
|
||
# Qt 默认会把 Ctrl+C 打断留在当前槽函数中并打印堆栈；转为正常关闭。
|
||
QTimer.singleShot(0, self.close)
|
||
return
|
||
except queue.Empty:
|
||
pass
|
||
|
||
if self._start_time and self._running:
|
||
elapsed = int(time.time() - self._start_time)
|
||
runtime = f"{elapsed // 3600:02d}:{(elapsed % 3600) // 60:02d}:{elapsed % 60:02d}"
|
||
self.dashboard_page.timer.setText(f"运行时长 {runtime}")
|
||
self.capsule.timer.setText(runtime)
|
||
|
||
def closeEvent(self, event) -> None:
|
||
self.save_runtime_settings()
|
||
self._running = False
|
||
if self._thread is not None and self._thread.is_alive():
|
||
self._thread.stop()
|
||
if sys.stdout is self._stdout_proxy:
|
||
sys.stdout = self._original_stdout
|
||
self.capsule.close()
|
||
event.accept()
|
||
|
||
|
||
def _app_icon() -> QIcon:
|
||
pixmap = QPixmap(64, 64)
|
||
pixmap.fill(Qt.transparent)
|
||
painter = QPainter(pixmap)
|
||
painter.setRenderHint(QPainter.Antialiasing)
|
||
painter.setBrush(QColor(COLORS["accent"]))
|
||
painter.setPen(Qt.NoPen)
|
||
painter.drawRoundedRect(5, 5, 54, 54, 17, 17)
|
||
painter.setBrush(QColor("#FFFFFF"))
|
||
painter.drawEllipse(17, 18, 30, 24)
|
||
painter.drawPolygon([QPoint(25, 39), QPoint(22, 49), QPoint(34, 41)])
|
||
painter.end()
|
||
return QIcon(pixmap)
|
||
|
||
|
||
def install_console_shutdown_handler(app: QApplication, window: MainWindow):
|
||
"""将控制台 Ctrl+C / Ctrl+Break 转换为 Qt 的正常关闭流程。"""
|
||
shutdown_requested = False
|
||
|
||
def request_shutdown(_signum=None, _frame=None) -> None:
|
||
nonlocal shutdown_requested
|
||
if shutdown_requested:
|
||
return
|
||
shutdown_requested = True
|
||
QTimer.singleShot(0, window.close)
|
||
QTimer.singleShot(0, app.quit)
|
||
|
||
handled = [signal.SIGINT]
|
||
if hasattr(signal, "SIGBREAK"):
|
||
handled.append(signal.SIGBREAK)
|
||
for signal_number in handled:
|
||
try:
|
||
signal.signal(signal_number, request_shutdown)
|
||
except (OSError, ValueError):
|
||
pass
|
||
return request_shutdown
|
||
|
||
|
||
def main() -> None:
|
||
QApplication.setHighDpiScaleFactorRoundingPolicy(
|
||
Qt.HighDpiScaleFactorRoundingPolicy.PassThrough
|
||
)
|
||
app = QApplication.instance() or QApplication(sys.argv)
|
||
app.setApplicationName(WIN_TITLE)
|
||
app.setApplicationDisplayName(WIN_TITLE)
|
||
app.setWindowIcon(_app_icon())
|
||
app.setStyle("Fusion")
|
||
app.setStyleSheet(APP_QSS)
|
||
font = QFont("Microsoft YaHei UI", 10)
|
||
font.setHintingPreference(QFont.PreferFullHinting)
|
||
app.setFont(font)
|
||
|
||
try:
|
||
import backend_client
|
||
|
||
backend_client.startup_sync_config(timeout=3.0)
|
||
except Exception:
|
||
# 启动检测失败不阻止软件打开，具体原因会显示在后台同步卡片中。
|
||
pass
|
||
|
||
window = MainWindow()
|
||
console_shutdown = install_console_shutdown_handler(app, window)
|
||
window.show()
|
||
|
||
if "--qt-smoke-test" in sys.argv:
|
||
for index in range(window.stack.count()):
|
||
def show_test_page(i=index) -> None:
|
||
window.show_page(i)
|
||
|
||
def capture(i=index) -> None:
|
||
window.grab().save(str(SCRIPT_DIR / f"qt-ui-smoke-{i + 1}.png"))
|
||
|
||
QTimer.singleShot(650 + index * 650, show_test_page)
|
||
QTimer.singleShot(950 + index * 650, capture)
|
||
QTimer.singleShot(1100 + window.stack.count() * 650, app.quit)
|
||
try:
|
||
exit_code = app.exec()
|
||
except KeyboardInterrupt:
|
||
console_shutdown()
|
||
exit_code = 0
|
||
raise SystemExit(exit_code)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|