4838 lines
194 KiB
Python
4838 lines
194 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""PySide6 desktop shell for the WeCom customer-service assistant."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import codecs
|
||
import html
|
||
import json
|
||
import os
|
||
import queue
|
||
import re
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
from PySide6.QtCore import (
|
||
QEasingCurve,
|
||
QPoint,
|
||
QProcess,
|
||
QProcessEnvironment,
|
||
QPropertyAnimation,
|
||
QRect,
|
||
QTimer,
|
||
Qt,
|
||
QUrl,
|
||
Signal,
|
||
)
|
||
from PySide6.QtGui import (
|
||
QColor,
|
||
QDesktopServices,
|
||
QFont,
|
||
QIcon,
|
||
QPainter,
|
||
QPixmap,
|
||
QTextCursor,
|
||
)
|
||
from PySide6.QtWidgets import (
|
||
QAbstractItemView,
|
||
QApplication,
|
||
QCheckBox,
|
||
QComboBox,
|
||
QDialog,
|
||
QDoubleSpinBox,
|
||
QFrame,
|
||
QFileDialog,
|
||
QGraphicsDropShadowEffect,
|
||
QGraphicsOpacityEffect,
|
||
QGridLayout,
|
||
QHBoxLayout,
|
||
QHeaderView,
|
||
QLabel,
|
||
QLineEdit,
|
||
QMainWindow,
|
||
QMessageBox,
|
||
QPlainTextEdit,
|
||
QProgressBar,
|
||
QPushButton,
|
||
QScrollArea,
|
||
QSizePolicy,
|
||
QSpinBox,
|
||
QStackedWidget,
|
||
QTableWidget,
|
||
QTableWidgetItem,
|
||
QTextEdit,
|
||
QVBoxLayout,
|
||
QWidget,
|
||
)
|
||
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"
|
||
WIN_TITLE = "甄养堂 · 企微客服助手"
|
||
WECOM_WAITING_MESSAGE = (
|
||
"企业微信主界面未显示、未在前台或已最小化,正在尝试切到前台;"
|
||
"本轮失败时下一轮将自动重试。"
|
||
)
|
||
HEADLESS_COMPLETED_STOP_REASONS = {"EndTurn", "Refusal"}
|
||
|
||
|
||
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; }
|
||
QWidget#ChatWorkspace { background: #FFFFFF; }
|
||
QFrame#ChatTopBar {
|
||
background: #FFFFFF;
|
||
border-bottom: 1px solid #EEF1EF;
|
||
}
|
||
QLabel#ChatBrand { color: #17251F; font-size: 17px; font-weight: 700; }
|
||
QLabel#ChatStatus { color: #7A8781; font-size: 12px; }
|
||
QLabel#ChatStatusDot { color: #10A57A; font-size: 16px; }
|
||
QFrame#ModeSwitch {
|
||
background: #F2F3F2;
|
||
border: 1px solid #E8EBE9;
|
||
border-radius: 21px;
|
||
}
|
||
QPushButton#ModeTab, QPushButton#ModeTabActive {
|
||
border: none;
|
||
border-radius: 18px;
|
||
padding: 8px 28px;
|
||
color: #69746F;
|
||
background: transparent;
|
||
}
|
||
QPushButton#ModeTab:hover { color: #17251F; background: #E9ECEA; }
|
||
QPushButton#ModeTabActive {
|
||
color: #17251F;
|
||
background: #FFFFFF;
|
||
border: 1px solid #E0E5E2;
|
||
font-weight: 600;
|
||
}
|
||
QWidget#ChatEmptyPage, QWidget#ChatReadingPage { background: #FFFFFF; }
|
||
QLabel#ChatWelcome { color: #17251F; font-size: 27px; font-weight: 500; }
|
||
QLabel#ChatWelcomeHint { color: #8A9490; font-size: 13px; }
|
||
QPushButton#PromptSuggestion {
|
||
background: transparent;
|
||
color: #79837E;
|
||
border: none;
|
||
text-align: left;
|
||
padding: 7px 9px;
|
||
border-radius: 8px;
|
||
font-size: 14px;
|
||
}
|
||
QPushButton#PromptSuggestion:hover { color: #17251F; background: #F4F6F5; }
|
||
QFrame#ChatComposer {
|
||
background: #FFFFFF;
|
||
border: 1px solid #DFE5E1;
|
||
border-radius: 24px;
|
||
}
|
||
QPlainTextEdit#ChatPrompt {
|
||
background: transparent;
|
||
border: none;
|
||
padding: 10px 5px;
|
||
color: #17251F;
|
||
selection-background-color: #DDF5EC;
|
||
}
|
||
QPlainTextEdit#ChatPrompt:focus {
|
||
background: transparent;
|
||
border: none;
|
||
padding: 10px 5px;
|
||
}
|
||
QPushButton#ComposerAction, QPushButton#ChatMetaButton {
|
||
background: transparent;
|
||
color: #71807A;
|
||
border: none;
|
||
border-radius: 18px;
|
||
padding: 7px 9px;
|
||
}
|
||
QPushButton#ComposerAction { font-size: 22px; color: #27332D; }
|
||
QPushButton#ComposerAction:hover, QPushButton#ChatMetaButton:hover {
|
||
background: #F1F4F2;
|
||
color: #17251F;
|
||
}
|
||
QPushButton#ChatSendButton, QPushButton#ChatStopButton {
|
||
border: none;
|
||
border-radius: 20px;
|
||
min-width: 40px;
|
||
max-width: 40px;
|
||
min-height: 40px;
|
||
max-height: 40px;
|
||
font-size: 20px;
|
||
font-weight: 700;
|
||
}
|
||
QPushButton#ChatSendButton { background: #111412; color: #FFFFFF; }
|
||
QPushButton#ChatSendButton:hover { background: #303733; }
|
||
QPushButton#ChatSendButton:disabled { background: #CBD3CE; color: #FFFFFF; }
|
||
QPushButton#ChatStopButton { background: #FBEAEC; color: #C74452; }
|
||
QPushButton#ChatStopButton:hover { background: #D85260; color: #FFFFFF; }
|
||
QLabel#ComposerStatus, QLabel#ChatDisclaimer { color: #9AA49F; font-size: 11px; }
|
||
QScrollArea#ChatMessageScroll, QScrollArea#ChatMessageScroll > QWidget > QWidget {
|
||
background: #FFFFFF;
|
||
border: none;
|
||
}
|
||
QFrame#ChatUserMessage {
|
||
background: #F1F2F1;
|
||
border: none;
|
||
border-radius: 18px;
|
||
}
|
||
QFrame#ChatAssistantMessage, QFrame#ChatThoughtMessage {
|
||
background: transparent;
|
||
border: none;
|
||
}
|
||
QFrame#ChatSystemMessage {
|
||
background: #F7F8F7;
|
||
border: 1px solid #ECEFED;
|
||
border-radius: 10px;
|
||
}
|
||
QLabel#ChatMessageRole { color: #79847E; font-size: 12px; font-weight: 600; }
|
||
QLabel#ChatMessageContent { color: #17251F; font-size: 15px; }
|
||
QPushButton#MessageAction {
|
||
background: transparent;
|
||
border: none;
|
||
color: #9AA49F;
|
||
padding: 4px 7px;
|
||
border-radius: 6px;
|
||
font-size: 12px;
|
||
}
|
||
QPushButton#MessageAction:hover { color: #17251F; background: #F2F4F3; }
|
||
QFrame#WorkPanel {
|
||
background: #FFFFFF;
|
||
border: 1px solid #E3E9E5;
|
||
border-radius: 18px;
|
||
}
|
||
QWidget#ChatWorkspace, QWidget#ChatContent, QWidget#ChatEmptyPage,
|
||
QWidget#ChatConversationPage, QWidget#ChatMessageRoot, QWidget#ChatMessageColumn,
|
||
QWidget#ChatComposerShell, QWidget#ChatWorkPage {
|
||
background: #FCFDFC;
|
||
}
|
||
QFrame#ChatTopBar {
|
||
background: #FCFDFC;
|
||
border: none;
|
||
border-bottom: 1px solid #E8ECEA;
|
||
}
|
||
QLabel#ChatBrand {
|
||
color: #17201C;
|
||
font-size: 16px;
|
||
font-weight: 700;
|
||
}
|
||
QLabel#ChatCompactStatus {
|
||
color: #718078;
|
||
font-size: 12px;
|
||
}
|
||
QFrame#ChatModeSwitch {
|
||
background: #F1F2F1;
|
||
border: none;
|
||
border-radius: 18px;
|
||
}
|
||
QPushButton#ChatModeActive, QPushButton#ChatModeIdle {
|
||
min-width: 78px;
|
||
min-height: 34px;
|
||
border: none;
|
||
border-radius: 17px;
|
||
padding: 0 18px;
|
||
font-size: 13px;
|
||
}
|
||
QPushButton#ChatModeActive {
|
||
background: #FFFFFF;
|
||
color: #111714;
|
||
font-weight: 600;
|
||
border: 1px solid #E1E5E3;
|
||
}
|
||
QPushButton#ChatModeIdle {
|
||
background: transparent;
|
||
color: #6E7873;
|
||
}
|
||
QPushButton#ChatModeIdle:hover { color: #17201C; background: #E8EBE9; }
|
||
QLabel#ChatWelcome {
|
||
color: #151B18;
|
||
font-size: 25px;
|
||
font-weight: 600;
|
||
}
|
||
QLabel#ChatWelcomeHint {
|
||
color: #7A8580;
|
||
font-size: 13px;
|
||
}
|
||
QPushButton#PromptSuggestion {
|
||
background: transparent;
|
||
color: #6C7771;
|
||
border: none;
|
||
border-radius: 8px;
|
||
text-align: left;
|
||
padding: 10px 12px;
|
||
font-size: 14px;
|
||
}
|
||
QPushButton#PromptSuggestion:hover { background: #F2F5F3; color: #17201C; }
|
||
QPushButton#PromptSuggestion:pressed { background: #E8EDEB; }
|
||
QFrame#ChatComposerBar {
|
||
background: #FFFFFF;
|
||
border: 1px solid #E0E5E2;
|
||
border-radius: 24px;
|
||
}
|
||
QPlainTextEdit#ChatPrompt {
|
||
background: transparent;
|
||
color: #151B18;
|
||
border: none;
|
||
border-radius: 0;
|
||
padding: 10px 5px;
|
||
font-size: 15px;
|
||
selection-background-color: #CDEADF;
|
||
}
|
||
QPlainTextEdit#ChatPrompt:focus {
|
||
background: transparent;
|
||
border: none;
|
||
padding: 10px 5px;
|
||
}
|
||
QPushButton#ComposerAction {
|
||
background: transparent;
|
||
color: #5F6A65;
|
||
border: none;
|
||
border-radius: 18px;
|
||
font-size: 22px;
|
||
}
|
||
QPushButton#ComposerAction:hover { background: #F1F4F2; color: #17201C; }
|
||
QPushButton#ComposerAction:pressed { background: #E7ECE9; }
|
||
QPushButton#ChatMetaButton {
|
||
background: transparent;
|
||
color: #707B75;
|
||
border: none;
|
||
border-radius: 8px;
|
||
padding: 7px 8px;
|
||
font-size: 12px;
|
||
}
|
||
QPushButton#ChatMetaButton:hover { background: #F1F4F2; color: #17201C; }
|
||
QPushButton#ChatSendButton, QPushButton#ChatStopButton {
|
||
border: none;
|
||
border-radius: 20px;
|
||
min-width: 40px;
|
||
max-width: 40px;
|
||
min-height: 40px;
|
||
max-height: 40px;
|
||
color: #FFFFFF;
|
||
font-size: 19px;
|
||
font-weight: 700;
|
||
}
|
||
QPushButton#ChatSendButton { background: #111714; }
|
||
QPushButton#ChatSendButton:hover { background: #29312D; }
|
||
QPushButton#ChatSendButton:pressed { background: #050706; }
|
||
QPushButton#ChatSendButton:disabled { background: #D7DDDA; color: #F8F9F8; }
|
||
QPushButton#ChatStopButton { background: #C94D59; }
|
||
QPushButton#ChatStopButton:hover { background: #B33E4A; }
|
||
QLabel#ChatComposerStatus, QLabel#ChatDisclaimer {
|
||
color: #949D98;
|
||
font-size: 11px;
|
||
}
|
||
QScrollArea#ChatMessageScroll,
|
||
QScrollArea#ChatMessageScroll > QWidget > QWidget {
|
||
background: #FCFDFC;
|
||
border: none;
|
||
}
|
||
QFrame#ChatUserMessage {
|
||
background: #F1F3F2;
|
||
border: none;
|
||
border-radius: 16px;
|
||
}
|
||
QFrame#ChatAssistantMessage {
|
||
background: transparent;
|
||
border: none;
|
||
}
|
||
QFrame#ChatThoughtMessage {
|
||
background: #F5F7F6;
|
||
border: none;
|
||
border-left: 2px solid #C9D5CF;
|
||
border-radius: 6px;
|
||
}
|
||
QFrame#ChatSystemMessage {
|
||
background: #F7F8F7;
|
||
border: 1px solid #E6EAE8;
|
||
border-radius: 10px;
|
||
}
|
||
QLabel#ChatMessageRole {
|
||
color: #526059;
|
||
font-size: 12px;
|
||
font-weight: 600;
|
||
}
|
||
QLabel#ChatMessageTime { color: #9AA39E; font-size: 11px; }
|
||
QLabel#ChatMessageContent {
|
||
color: #151B18;
|
||
font-size: 15px;
|
||
}
|
||
QPushButton#MessageAction {
|
||
background: transparent;
|
||
color: #7D8782;
|
||
border: none;
|
||
border-radius: 7px;
|
||
padding: 5px 8px;
|
||
font-size: 12px;
|
||
}
|
||
QPushButton#MessageAction:hover { background: #F0F3F1; color: #17201C; }
|
||
QFrame#ChatWorkPanel {
|
||
background: #FFFFFF;
|
||
border: 1px solid #E2E8E5;
|
||
border-radius: 18px;
|
||
}
|
||
QLabel#ChatWorkTitle {
|
||
color: #17201C;
|
||
font-size: 24px;
|
||
font-weight: 700;
|
||
}
|
||
QLabel#ChatWorkHint {
|
||
color: #718078;
|
||
font-size: 13px;
|
||
}
|
||
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, QComboBox {
|
||
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, QComboBox: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 _safe_markdown(text: str) -> str:
|
||
"""Keep useful Markdown while stripping raw HTML and unsafe image/link targets."""
|
||
value = re.sub(r"<[^>]+>", "", str(text or ""))
|
||
value = re.sub(r"!\[[^\]]*\]\([^)]*\)", "[图片]", value)
|
||
value = re.sub(
|
||
r"\]\((?:javascript|data|vbscript):[^)]*\)",
|
||
"](#)",
|
||
value,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
return value
|
||
|
||
|
||
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 ChatPromptEdit(QPlainTextEdit):
|
||
"""Compact chat composer: Enter sends, Shift+Enter inserts a newline."""
|
||
|
||
submitted = Signal()
|
||
|
||
def keyPressEvent(self, event) -> None:
|
||
if event.key() in (Qt.Key_Return, Qt.Key_Enter):
|
||
modifiers = event.modifiers()
|
||
if not (
|
||
modifiers
|
||
& (Qt.ShiftModifier | Qt.ControlModifier | Qt.AltModifier)
|
||
):
|
||
event.accept()
|
||
self.submitted.emit()
|
||
return
|
||
super().keyPressEvent(event)
|
||
|
||
|
||
class CustomerServicePage(QWidget):
|
||
"""Native, resumable chat UI backed by Grok Build and a custom model."""
|
||
|
||
statusReady = Signal(int, object)
|
||
installFinished = Signal(object)
|
||
chatPreparationFinished = Signal(object)
|
||
directChatChunk = Signal(int, str, bool)
|
||
directChatFinished = Signal(object)
|
||
|
||
def __init__(self, parent: QWidget | None = None):
|
||
super().__init__(parent)
|
||
from grok_build_bridge import GrokBuildManager, MODEL_PROFILE
|
||
|
||
self.manager = GrokBuildManager()
|
||
self.model_profile = MODEL_PROFILE
|
||
self._status_running = False
|
||
self._status_pending_deep = False
|
||
self._status_generation = 0
|
||
self._install_running = False
|
||
self._chat_preparing = False
|
||
self._cancel_requested = False
|
||
self._shutting_down = False
|
||
self._agent_enabled = False
|
||
self._model_compatible = False
|
||
self._direct_model_compatible = False
|
||
self._last_endpoint_probe = None
|
||
self._last_probe_signature: tuple[str, ...] | None = None
|
||
self._active_session_id = ""
|
||
self._session_started = False
|
||
self._pending_prompt = ""
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._pending_route = ""
|
||
self._last_route = "direct"
|
||
self._direct_conversation_id = ""
|
||
self._direct_generation = 0
|
||
self._direct_cancellation = None
|
||
self._chat_history: list[dict[str, str]] = []
|
||
self._agent_history_count = 0
|
||
self._process_buffer = ""
|
||
self._decoder = codecs.getincrementaldecoder("utf-8")("replace")
|
||
self._stderr_buffer = ""
|
||
self._stderr_decoder = codecs.getincrementaldecoder("utf-8")("replace")
|
||
self._assistant_label: QLabel | None = None
|
||
self._assistant_chunks: list[str] = []
|
||
self._thought_label: QLabel | None = None
|
||
self._thought_chunks: list[str] = []
|
||
self._turn_received_end = False
|
||
self._turn_had_error = False
|
||
self._turn_stop_reason = ""
|
||
self.agent_process: QProcess | None = None
|
||
|
||
self._conversation_active = False
|
||
self._stream_render_timer = QTimer(self)
|
||
self._stream_render_timer.setSingleShot(True)
|
||
self._stream_render_timer.setInterval(45)
|
||
self._stream_render_timer.timeout.connect(self._flush_stream_render)
|
||
|
||
self.setObjectName("ChatWorkspace")
|
||
root_layout = QVBoxLayout(self)
|
||
root_layout.setContentsMargins(0, 0, 0, 0)
|
||
root_layout.setSpacing(0)
|
||
|
||
top_bar = QFrame()
|
||
top_bar.setObjectName("ChatTopBar")
|
||
top_layout = QHBoxLayout(top_bar)
|
||
top_layout.setContentsMargins(28, 12, 28, 12)
|
||
top_layout.setSpacing(10)
|
||
brand_box = QHBoxLayout()
|
||
brand_box.setSpacing(7)
|
||
brand_box.addWidget(_label("●", "ChatStatusDot"))
|
||
brand_box.addWidget(_label("AI 客服", "ChatBrand"))
|
||
self.compact_status = _label("正在连接…", "ChatCompactStatus")
|
||
self.compact_status.setMaximumWidth(280)
|
||
self.compact_status.setWordWrap(False)
|
||
brand_box.addWidget(self.compact_status)
|
||
top_layout.addLayout(brand_box, 1)
|
||
|
||
mode_switch = QFrame()
|
||
mode_switch.setObjectName("ChatModeSwitch")
|
||
mode_layout = QHBoxLayout(mode_switch)
|
||
mode_layout.setContentsMargins(2, 2, 2, 2)
|
||
mode_layout.setSpacing(0)
|
||
self.chat_mode_button = QPushButton("聊天")
|
||
self.chat_mode_button.setObjectName("ChatModeActive")
|
||
self.chat_mode_button.setCursor(Qt.PointingHandCursor)
|
||
self.chat_mode_button.clicked.connect(lambda: self._set_workspace_mode(False))
|
||
self.work_mode_button = QPushButton("工作")
|
||
self.work_mode_button.setObjectName("ChatModeIdle")
|
||
self.work_mode_button.setCursor(Qt.PointingHandCursor)
|
||
self.work_mode_button.clicked.connect(lambda: self._set_workspace_mode(True))
|
||
mode_layout.addWidget(self.chat_mode_button)
|
||
mode_layout.addWidget(self.work_mode_button)
|
||
top_layout.addWidget(mode_switch)
|
||
|
||
self.new_chat_button = _button("新对话")
|
||
self.new_chat_button.clicked.connect(self.new_conversation)
|
||
top_layout.addWidget(self.new_chat_button, 1, Qt.AlignRight)
|
||
root_layout.addWidget(top_bar)
|
||
|
||
self.workspace_stack = QStackedWidget()
|
||
self.workspace_stack.setObjectName("ChatContent")
|
||
root_layout.addWidget(self.workspace_stack, 1)
|
||
|
||
chat_page = QWidget()
|
||
chat_page.setObjectName("ChatContent")
|
||
chat_page_layout = QVBoxLayout(chat_page)
|
||
chat_page_layout.setContentsMargins(0, 0, 0, 0)
|
||
chat_page_layout.setSpacing(0)
|
||
self.chat_stack = QStackedWidget()
|
||
self.chat_stack.setObjectName("ChatContent")
|
||
chat_page_layout.addWidget(self.chat_stack, 1)
|
||
|
||
empty_page = QWidget()
|
||
empty_page.setObjectName("ChatEmptyPage")
|
||
empty_outer = QHBoxLayout(empty_page)
|
||
empty_outer.setContentsMargins(28, 10, 28, 20)
|
||
empty_center = QWidget()
|
||
empty_center.setMaximumWidth(900)
|
||
empty_center.setMinimumWidth(700)
|
||
empty_center.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
|
||
empty_layout = QVBoxLayout(empty_center)
|
||
empty_layout.setContentsMargins(0, 0, 0, 0)
|
||
empty_layout.setSpacing(8)
|
||
empty_layout.addStretch(2)
|
||
welcome = _label("准备好了,随时开始", "ChatWelcome")
|
||
welcome.setAlignment(Qt.AlignCenter)
|
||
empty_layout.addWidget(welcome)
|
||
welcome_hint = _label(
|
||
"让后台自有模型通过 Grok Build Agent 分析问题并调用客服工具",
|
||
"ChatWelcomeHint",
|
||
)
|
||
welcome_hint.setAlignment(Qt.AlignCenter)
|
||
empty_layout.addWidget(welcome_hint)
|
||
self.empty_composer_host = QWidget()
|
||
self.empty_composer_host.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
|
||
self.empty_composer_layout = QVBoxLayout(self.empty_composer_host)
|
||
self.empty_composer_layout.setContentsMargins(0, 22, 0, 0)
|
||
empty_layout.addWidget(self.empty_composer_host)
|
||
suggestions = QVBoxLayout()
|
||
suggestions.setContentsMargins(18, 10, 0, 0)
|
||
suggestions.setSpacing(1)
|
||
for icon, title, starter in (
|
||
("▧", "整理客户问题", "请帮我整理下面的客户问题,并给出处理建议:"),
|
||
("✎", "撰写或修改", "请帮我撰写一段专业、简洁的客服回复:"),
|
||
("⌕", "查询业务资料", "请查询相关业务资料,并用要点说明:"),
|
||
):
|
||
shortcut = QPushButton(f"{icon} {title}")
|
||
shortcut.setObjectName("PromptSuggestion")
|
||
shortcut.setCursor(Qt.PointingHandCursor)
|
||
shortcut.clicked.connect(
|
||
lambda _checked=False, value=starter: self._use_suggestion(value)
|
||
)
|
||
suggestions.addWidget(shortcut)
|
||
empty_layout.addLayout(suggestions)
|
||
empty_layout.addStretch(3)
|
||
empty_outer.addWidget(empty_center, 1, Qt.AlignHCenter)
|
||
self.chat_stack.addWidget(empty_page)
|
||
|
||
conversation_page = QWidget()
|
||
conversation_page.setObjectName("ChatConversationPage")
|
||
conversation_page_layout = QVBoxLayout(conversation_page)
|
||
conversation_page_layout.setContentsMargins(0, 0, 0, 0)
|
||
conversation_page_layout.setSpacing(0)
|
||
self.message_scroll = QScrollArea()
|
||
self.message_scroll.setObjectName("ChatMessageScroll")
|
||
self.message_scroll.setWidgetResizable(True)
|
||
self.message_scroll.setFrameShape(QFrame.NoFrame)
|
||
self.message_scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
|
||
self.message_root = QWidget()
|
||
self.message_root.setObjectName("ChatMessageRoot")
|
||
message_root_layout = QHBoxLayout(self.message_root)
|
||
message_root_layout.setContentsMargins(22, 0, 22, 0)
|
||
self.message_column = QWidget()
|
||
self.message_column.setObjectName("ChatMessageColumn")
|
||
self.message_column.setMaximumWidth(900)
|
||
self.message_column.setMinimumWidth(700)
|
||
self.message_column.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred)
|
||
self.message_layout = QVBoxLayout(self.message_column)
|
||
self.message_layout.setContentsMargins(0, 26, 0, 18)
|
||
self.message_layout.setSpacing(22)
|
||
self.message_layout.addStretch(1)
|
||
message_root_layout.addWidget(self.message_column, 1, Qt.AlignHCenter)
|
||
self.message_scroll.setWidget(self.message_root)
|
||
conversation_page_layout.addWidget(self.message_scroll, 1)
|
||
|
||
conversation_footer = QWidget()
|
||
conversation_footer.setObjectName("ChatComposerShell")
|
||
conversation_footer_layout = QVBoxLayout(conversation_footer)
|
||
conversation_footer_layout.setContentsMargins(24, 4, 24, 16)
|
||
conversation_footer_layout.setSpacing(6)
|
||
conversation_composer_host = QWidget()
|
||
conversation_composer_host.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
|
||
conversation_composer_host.setMinimumWidth(700)
|
||
conversation_composer_host.setMaximumWidth(900)
|
||
self.conversation_composer_layout = QVBoxLayout(conversation_composer_host)
|
||
self.conversation_composer_layout.setContentsMargins(0, 0, 0, 0)
|
||
conversation_footer_layout.addWidget(conversation_composer_host, 0, Qt.AlignHCenter)
|
||
self.disclaimer = _label("AI 回复可能有误,请核对重要信息。", "ChatDisclaimer")
|
||
self.disclaimer.setAlignment(Qt.AlignCenter)
|
||
conversation_footer_layout.addWidget(self.disclaimer)
|
||
conversation_page_layout.addWidget(conversation_footer, 0)
|
||
self.chat_stack.addWidget(conversation_page)
|
||
|
||
work_page = QWidget()
|
||
work_page.setObjectName("ChatWorkPage")
|
||
work_outer = QHBoxLayout(work_page)
|
||
work_outer.setContentsMargins(28, 32, 28, 32)
|
||
work_center = QWidget()
|
||
work_center.setMaximumWidth(900)
|
||
work_center.setMinimumWidth(700)
|
||
work_center.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred)
|
||
work_layout = QVBoxLayout(work_center)
|
||
work_layout.setSpacing(10)
|
||
work_layout.addWidget(_label("Agent 工作设置", "ChatWorkTitle"))
|
||
work_layout.addWidget(
|
||
_label(
|
||
"这里保留运行时、模型和工具权限配置;聊天页只保留对话。",
|
||
"ChatWorkHint",
|
||
)
|
||
)
|
||
work_panel = QFrame()
|
||
work_panel.setObjectName("ChatWorkPanel")
|
||
work_panel_layout = QVBoxLayout(work_panel)
|
||
work_panel_layout.setContentsMargins(22, 20, 22, 22)
|
||
work_panel_layout.setSpacing(12)
|
||
self.status_label = _label("正在检测 Grok Build 与后台自有模型…", "WarningText")
|
||
self.status_meta = _label("", "CardSubtitle")
|
||
self.status_meta.setWordWrap(True)
|
||
status_copy = QVBoxLayout()
|
||
status_copy.setSpacing(3)
|
||
status_copy.addWidget(self.status_label)
|
||
status_copy.addWidget(self.status_meta)
|
||
status_actions = QHBoxLayout()
|
||
status_actions.addLayout(status_copy, 1)
|
||
self.session_label = _label("新对话", "CardSubtitle")
|
||
self.session_label.setTextInteractionFlags(Qt.TextSelectableByMouse)
|
||
status_actions.addWidget(self.session_label)
|
||
self.refresh_button = _button("刷新状态")
|
||
self.refresh_button.clicked.connect(lambda: self.refresh_status(deep=True))
|
||
status_actions.addWidget(self.refresh_button)
|
||
self.install_button = _button("安装 / 更新", "primary")
|
||
self.install_button.clicked.connect(self.install_runtime)
|
||
status_actions.addWidget(self.install_button)
|
||
work_panel_layout.addLayout(status_actions)
|
||
model_row = QHBoxLayout()
|
||
self.model_value = _label("模型:检测中", "CardSubtitle")
|
||
self.model_value.setTextInteractionFlags(Qt.TextSelectableByMouse)
|
||
model_row.addWidget(self.model_value, 1)
|
||
self.enabled = QCheckBox("启用本地 Grok Build Agent 对话")
|
||
self.enabled.toggled.connect(self.set_agent_enabled)
|
||
model_row.addWidget(self.enabled)
|
||
self.auto_approve = QCheckBox("自动批准工具(含命令与文件修改)")
|
||
self.auto_approve.setToolTip(
|
||
"开启后,Agent 可在本机工作区内直接执行命令和修改文件;"
|
||
"只应在可信任务中开启。关闭时仍可使用只读工具和受控 MCP 工具。"
|
||
)
|
||
self.auto_approve.setChecked(
|
||
bool(self.manager.load_integration_settings().get("chat_auto_approve", False))
|
||
)
|
||
self.auto_approve.toggled.connect(self.set_chat_auto_approve)
|
||
model_row.addWidget(self.auto_approve)
|
||
work_panel_layout.addLayout(model_row)
|
||
work_layout.addWidget(work_panel)
|
||
work_layout.addStretch(1)
|
||
work_outer.addWidget(work_center, 1, Qt.AlignHCenter)
|
||
|
||
self.workspace_stack.addWidget(chat_page)
|
||
self.workspace_stack.addWidget(work_page)
|
||
|
||
self.composer = QWidget()
|
||
self.composer.setObjectName("ChatComposerShell")
|
||
self.composer.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
|
||
composer_root = QVBoxLayout(self.composer)
|
||
composer_root.setContentsMargins(0, 0, 0, 0)
|
||
composer_root.setSpacing(5)
|
||
composer_bar = QFrame()
|
||
composer_bar.setObjectName("ChatComposerBar")
|
||
_shadow(composer_bar, blur=32, y=7, alpha=18)
|
||
composer_bar_layout = QHBoxLayout(composer_bar)
|
||
composer_bar_layout.setContentsMargins(8, 6, 8, 6)
|
||
composer_bar_layout.setSpacing(4)
|
||
self.composer_new_button = QPushButton("+")
|
||
self.composer_new_button.setObjectName("ComposerAction")
|
||
self.composer_new_button.setFixedSize(36, 36)
|
||
self.composer_new_button.setToolTip("新对话")
|
||
self.composer_new_button.clicked.connect(self.new_conversation)
|
||
composer_bar_layout.addWidget(self.composer_new_button)
|
||
self.prompt = ChatPromptEdit()
|
||
self.prompt.setObjectName("ChatPrompt")
|
||
self.prompt.setFixedHeight(48)
|
||
self.prompt.setPlaceholderText("输入消息,让 Agent 帮你处理…")
|
||
self.prompt.submitted.connect(self.send_message)
|
||
composer_bar_layout.addWidget(self.prompt, 1)
|
||
self.tool_mode_button = QPushButton()
|
||
self.tool_mode_button.setObjectName("ChatMetaButton")
|
||
self.tool_mode_button.setCursor(Qt.PointingHandCursor)
|
||
self.tool_mode_button.setToolTip("切换 Agent 工具权限")
|
||
self.tool_mode_button.clicked.connect(lambda: self.auto_approve.toggle())
|
||
composer_bar_layout.addWidget(self.tool_mode_button)
|
||
self.stop_button = QPushButton("■")
|
||
self.stop_button.setObjectName("ChatStopButton")
|
||
self.stop_button.setToolTip("停止生成")
|
||
self.stop_button.clicked.connect(self.stop_chat)
|
||
self.stop_button.setVisible(False)
|
||
composer_bar_layout.addWidget(self.stop_button)
|
||
self.send_button = QPushButton("↑")
|
||
self.send_button.setObjectName("ChatSendButton")
|
||
self.send_button.setToolTip("发送(Enter)")
|
||
self.send_button.clicked.connect(self.send_message)
|
||
self.send_button.setEnabled(False)
|
||
composer_bar_layout.addWidget(self.send_button)
|
||
composer_root.addWidget(composer_bar)
|
||
composer_meta = QHBoxLayout()
|
||
self.chat_status = _label("等待发送", "ChatComposerStatus")
|
||
composer_meta.addWidget(self.chat_status)
|
||
composer_meta.addStretch(1)
|
||
composer_meta.addWidget(_label("Enter 发送 · Shift+Enter 换行", "ChatComposerStatus"))
|
||
composer_root.addLayout(composer_meta)
|
||
self._sync_tool_mode_button()
|
||
self._place_composer(False)
|
||
|
||
self.statusReady.connect(self._apply_status)
|
||
self.installFinished.connect(self._install_finished)
|
||
self.chatPreparationFinished.connect(self._chat_preparation_finished)
|
||
self.directChatChunk.connect(self._direct_chat_chunk)
|
||
self.directChatFinished.connect(self._direct_chat_finished)
|
||
QTimer.singleShot(0, lambda: self.refresh_status(deep=False))
|
||
QTimer.singleShot(500, lambda: self.refresh_status(deep=True))
|
||
|
||
def _place_composer(self, conversation: bool) -> None:
|
||
"""Move the single composer between the centered empty state and footer."""
|
||
target = (
|
||
self.conversation_composer_layout
|
||
if conversation
|
||
else self.empty_composer_layout
|
||
)
|
||
target.addWidget(self.composer)
|
||
self._conversation_active = bool(conversation)
|
||
self.chat_stack.setCurrentIndex(1 if conversation else 0)
|
||
|
||
def _set_workspace_mode(self, work: bool) -> None:
|
||
self.workspace_stack.setCurrentIndex(1 if work else 0)
|
||
self.chat_mode_button.setObjectName("ChatModeIdle" if work else "ChatModeActive")
|
||
self.work_mode_button.setObjectName("ChatModeActive" if work else "ChatModeIdle")
|
||
for button in (self.chat_mode_button, self.work_mode_button):
|
||
button.style().unpolish(button)
|
||
button.style().polish(button)
|
||
if not work:
|
||
self.prompt.setFocus()
|
||
|
||
def _use_suggestion(self, starter: str) -> None:
|
||
current = self.prompt.toPlainText().strip()
|
||
self.prompt.setPlainText(starter if not current else f"{starter}\n{current}")
|
||
self.prompt.setFocus()
|
||
self.prompt.moveCursor(QTextCursor.End)
|
||
|
||
def _sync_tool_mode_button(self) -> None:
|
||
if not hasattr(self, "tool_mode_button"):
|
||
return
|
||
self.tool_mode_button.setText(
|
||
"工具:自动" if self.auto_approve.isChecked() else "工具:只读"
|
||
)
|
||
|
||
def _flush_stream_render(self) -> None:
|
||
if self._assistant_label is not None and self._assistant_chunks:
|
||
self._assistant_label.setText(_safe_markdown("".join(self._assistant_chunks)))
|
||
if self._thought_label is not None and self._thought_chunks:
|
||
self._thought_label.setText("".join(self._thought_chunks))
|
||
self._scroll_to_latest()
|
||
|
||
@staticmethod
|
||
def _style_status(label: QLabel, state: str) -> None:
|
||
label.setObjectName(state)
|
||
label.style().unpolish(label)
|
||
label.style().polish(label)
|
||
|
||
@staticmethod
|
||
def _clear_layout(layout) -> None:
|
||
while layout.count():
|
||
item = layout.takeAt(0)
|
||
child_layout = item.layout()
|
||
child_widget = item.widget()
|
||
if child_layout is not None:
|
||
CustomerServicePage._clear_layout(child_layout)
|
||
child_layout.deleteLater()
|
||
if child_widget is not None:
|
||
child_widget.deleteLater()
|
||
|
||
def _scroll_to_latest(self, force: bool = False) -> None:
|
||
scrollbar = self.message_scroll.verticalScrollBar()
|
||
if not force and scrollbar.maximum() - scrollbar.value() > 80:
|
||
return
|
||
QTimer.singleShot(
|
||
0,
|
||
lambda: scrollbar.setValue(scrollbar.maximum()),
|
||
)
|
||
|
||
def _append_chat_message(self, role: str, content: str) -> QLabel:
|
||
normalized_role = role if role in {"user", "assistant", "thought"} else "system"
|
||
bubble_name = {
|
||
"user": "ChatUserMessage",
|
||
"assistant": "ChatAssistantMessage",
|
||
"thought": "ChatThoughtMessage",
|
||
"system": "ChatSystemMessage",
|
||
}[normalized_role]
|
||
role_name = {
|
||
"user": "你",
|
||
"assistant": (
|
||
"AI" if self._pending_route == "direct" else "Agent"
|
||
),
|
||
"thought": "思考",
|
||
"system": "系统",
|
||
}[normalized_role]
|
||
bubble = QFrame()
|
||
bubble.setObjectName(bubble_name)
|
||
bubble.setMaximumWidth(820 if normalized_role == "user" else 900)
|
||
bubble_layout = QVBoxLayout(bubble)
|
||
bubble_layout.setContentsMargins(
|
||
16 if normalized_role == "user" else 0,
|
||
11 if normalized_role == "user" else 0,
|
||
16 if normalized_role == "user" else 0,
|
||
12 if normalized_role == "user" else 0,
|
||
)
|
||
bubble_layout.setSpacing(5)
|
||
if normalized_role != "user":
|
||
meta = QHBoxLayout()
|
||
meta.addWidget(_label(role_name, "ChatMessageRole"))
|
||
meta.addWidget(_label(time.strftime("%H:%M:%S"), "ChatMessageTime"))
|
||
meta.addStretch(1)
|
||
bubble_layout.addLayout(meta)
|
||
rendered_content = (
|
||
_safe_markdown(content) if normalized_role == "assistant" else content
|
||
)
|
||
content_label = _label(rendered_content, "ChatMessageContent")
|
||
content_label.setTextFormat(
|
||
Qt.MarkdownText if normalized_role == "assistant" else Qt.PlainText
|
||
)
|
||
if normalized_role == "assistant":
|
||
content_label.setProperty("markdown", True)
|
||
content_label.setOpenExternalLinks(False)
|
||
content_label.setTextInteractionFlags(Qt.TextSelectableByMouse)
|
||
content_label.setWordWrap(True)
|
||
content_label.setMinimumWidth(120 if normalized_role != "user" else 80)
|
||
bubble_layout.addWidget(content_label)
|
||
|
||
if normalized_role == "assistant":
|
||
actions = QHBoxLayout()
|
||
actions.setContentsMargins(0, 2, 0, 0)
|
||
copy_button = QPushButton("复制")
|
||
copy_button.setObjectName("MessageAction")
|
||
copy_button.setCursor(Qt.PointingHandCursor)
|
||
copy_button.clicked.connect(
|
||
lambda _checked=False, label=content_label: QApplication.clipboard().setText(
|
||
label.text()
|
||
)
|
||
)
|
||
actions.addWidget(copy_button)
|
||
actions.addStretch(1)
|
||
bubble_layout.addLayout(actions)
|
||
|
||
row = QHBoxLayout()
|
||
row.setContentsMargins(0, 0, 0, 0)
|
||
if normalized_role == "user":
|
||
row.addStretch(1)
|
||
row.addWidget(bubble)
|
||
elif normalized_role in {"thought", "system"}:
|
||
row.addStretch(1)
|
||
row.addWidget(bubble, 1)
|
||
row.addStretch(1)
|
||
else:
|
||
row.addWidget(bubble, 1)
|
||
if not self._conversation_active:
|
||
self._place_composer(True)
|
||
self.message_layout.insertLayout(max(0, self.message_layout.count() - 1), row)
|
||
self._scroll_to_latest(force=normalized_role == "user")
|
||
return content_label
|
||
|
||
def _update_chat_controls(self) -> None:
|
||
busy = self._chat_preparing or (
|
||
self.agent_process is not None
|
||
and self.agent_process.state() != QProcess.NotRunning
|
||
)
|
||
available = self._direct_model_compatible
|
||
self.send_button.setEnabled(available and not busy and not self._shutting_down)
|
||
self.stop_button.setEnabled(busy and not self._shutting_down)
|
||
self.send_button.setVisible(not busy)
|
||
self.stop_button.setVisible(busy)
|
||
self.new_chat_button.setEnabled(not busy and not self._shutting_down)
|
||
self.composer_new_button.setEnabled(not busy and not self._shutting_down)
|
||
self.prompt.setEnabled(not busy and not self._shutting_down)
|
||
|
||
def refresh_status(self, *, deep: bool = True) -> None:
|
||
if self._status_running:
|
||
self._status_pending_deep = self._status_pending_deep or deep
|
||
return
|
||
self._status_running = True
|
||
self._status_pending_deep = False
|
||
self._status_generation += 1
|
||
generation = self._status_generation
|
||
self.refresh_button.setEnabled(False)
|
||
self.status_label.setText("正在检测 Grok Build 与后台自有模型…")
|
||
self._style_status(self.status_label, "WarningText")
|
||
|
||
def worker() -> None:
|
||
try:
|
||
import ai_config
|
||
|
||
profile = self.manager.agent_model_profile()
|
||
runtime = self.manager.status()
|
||
probe = None
|
||
if deep and bool(getattr(profile, "compatible", False)):
|
||
self.manager.sync_model_configuration()
|
||
profile = self.manager.agent_model_profile()
|
||
runtime = self.manager.status()
|
||
probe = self.manager.probe_agent_model(
|
||
force=True,
|
||
timeout=12.0,
|
||
)
|
||
result = {
|
||
"runtime": runtime,
|
||
"profile": profile,
|
||
"probe": probe,
|
||
"verified": bool(deep),
|
||
"enabled": bool(
|
||
getattr(ai_config, "GROK_CUSTOMER_SERVICE_ENABLED", True)
|
||
),
|
||
}
|
||
except Exception as exc:
|
||
result = exc
|
||
self.statusReady.emit(generation, result)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _apply_status(self, generation: int, result: object) -> None:
|
||
if generation != self._status_generation:
|
||
return
|
||
self._status_running = False
|
||
self.refresh_button.setEnabled(True)
|
||
if isinstance(result, Exception):
|
||
self._model_compatible = False
|
||
self._direct_model_compatible = False
|
||
self.status_label.setText(f"本地 Agent 检测失败:{result}")
|
||
self.status_meta.setText("")
|
||
self.model_value.setText("模型:不可用")
|
||
self._style_status(self.status_label, "DangerText")
|
||
else:
|
||
details = result if isinstance(result, dict) else {}
|
||
runtime = details.get("runtime")
|
||
profile = details.get("profile")
|
||
enabled = bool(details.get("enabled", True))
|
||
installed = bool(getattr(runtime, "installed", False))
|
||
compatible = bool(getattr(profile, "compatible", False))
|
||
model_name = str(getattr(profile, "model", "") or "未配置")
|
||
effective_backend = str(
|
||
getattr(profile, "api_backend", "") or "未知协议"
|
||
)
|
||
effective_base_url = str(getattr(profile, "base_url", "") or "")
|
||
source_backend = str(getattr(profile, "source_backend", "") or "")
|
||
source_base_url = str(getattr(profile, "source_base_url", "") or "")
|
||
insecure_dify = bool(
|
||
source_backend == "dify"
|
||
and source_base_url.lower().startswith("http://")
|
||
and QUrl(source_base_url).host().lower()
|
||
not in {"127.0.0.1", "localhost", "::1"}
|
||
)
|
||
backend = (
|
||
"Dify Chat Messages(本地工具调用适配)"
|
||
if source_backend == "dify"
|
||
else effective_backend
|
||
)
|
||
base_url = source_base_url or effective_base_url
|
||
auth_scheme = str(getattr(profile, "auth_scheme", "") or "")
|
||
signature = (
|
||
effective_base_url,
|
||
effective_backend,
|
||
source_base_url,
|
||
source_backend,
|
||
model_name,
|
||
auth_scheme,
|
||
)
|
||
verified = bool(details.get("verified", False))
|
||
probe = details.get("probe")
|
||
if verified:
|
||
self._last_endpoint_probe = probe
|
||
self._last_probe_signature = signature
|
||
elif self._last_probe_signature == signature:
|
||
probe = self._last_endpoint_probe
|
||
verified = probe is not None
|
||
else:
|
||
self._last_endpoint_probe = None
|
||
self._last_probe_signature = None
|
||
probe_ok = bool(getattr(probe, "ok", False))
|
||
probe_message = str(getattr(probe, "message", "") or "")
|
||
self._agent_enabled = enabled
|
||
self._model_compatible = (
|
||
installed and compatible and verified and probe_ok
|
||
)
|
||
self._direct_model_compatible = compatible and verified and probe_ok
|
||
self.enabled.blockSignals(True)
|
||
self.enabled.setChecked(enabled)
|
||
self.enabled.blockSignals(False)
|
||
self.model_value.setText(f"模型:{model_name} · {backend}")
|
||
if not compatible:
|
||
headline = "后台自有模型不可用"
|
||
meta = str(
|
||
getattr(profile, "reason", "")
|
||
or (
|
||
"请在管理后台启用并配置自有模型地址和 API Key;可选择 "
|
||
"Chat Completions、Responses、Anthropic Messages,"
|
||
"也可选择 Dify Chat Messages 本地工具调用适配。"
|
||
)
|
||
)
|
||
state = "DangerText"
|
||
elif not verified:
|
||
headline = f"自有模型已配置 · 等待端点预检"
|
||
meta = (
|
||
f"{backend} · {base_url}。正在核验真实接口、认证和模型,"
|
||
"核验完成前不会启动 Agent。"
|
||
)
|
||
state = "WarningText"
|
||
elif not probe_ok:
|
||
headline = "后台自有模型端点不可用"
|
||
meta = probe_message or (
|
||
"模型端点预检失败;请核对 API 基址、协议、Key 和模型名称。"
|
||
)
|
||
state = "DangerText"
|
||
elif not enabled:
|
||
headline = f"普通对话已就绪 · 自有模型 {model_name}"
|
||
meta = "Agent 调度已关闭;普通语言对话仍会直接调用自有模型。"
|
||
state = "WarningText"
|
||
elif not installed:
|
||
headline = f"普通对话已就绪 · 自有模型 {model_name}"
|
||
meta = "尚未安装 Grok Build;普通对话可用,需要工具时请先安装。"
|
||
state = "WarningText"
|
||
else:
|
||
headline = (
|
||
f"普通对话直连 · Agent 已就绪 · {model_name}"
|
||
+ ("(Dify HTTP 风险)" if insecure_dify else "")
|
||
)
|
||
meta = probe_message or (
|
||
f"{backend} · {base_url}" if base_url else backend
|
||
)
|
||
state = "WarningText" if insecure_dify else "SuccessText"
|
||
self.status_label.setText(headline)
|
||
self.status_meta.setText(meta)
|
||
self._style_status(self.status_label, state)
|
||
if hasattr(self, "compact_status"):
|
||
if self._model_compatible:
|
||
compact = "对话直连 · Agent 就绪"
|
||
elif self._direct_model_compatible:
|
||
compact = "普通对话直连 · Agent 未就绪"
|
||
else:
|
||
compact = "自有模型 · 检查中"
|
||
self.compact_status.setText(compact)
|
||
self._update_chat_controls()
|
||
if self._status_pending_deep:
|
||
self._status_pending_deep = False
|
||
QTimer.singleShot(0, lambda: self.refresh_status(deep=True))
|
||
|
||
def set_agent_enabled(self, enabled: bool) -> None:
|
||
try:
|
||
import backend_client
|
||
|
||
managed = backend_client.is_configured() and bool(
|
||
backend_client.load_settings().get("auto_sync", True)
|
||
)
|
||
except Exception:
|
||
managed = False
|
||
if managed:
|
||
self.enabled.blockSignals(True)
|
||
self.enabled.setChecked(not enabled)
|
||
self.enabled.blockSignals(False)
|
||
self.status_label.setText("当前由管理后台统一配置,请在后台修改客服开关")
|
||
self._style_status(self.status_label, "WarningText")
|
||
return
|
||
try:
|
||
import ai_config
|
||
|
||
ai_config.apply_settings(
|
||
{"GROK_CUSTOMER_SERVICE_ENABLED": bool(enabled)},
|
||
persist=True,
|
||
)
|
||
except Exception as exc:
|
||
self.enabled.blockSignals(True)
|
||
self.enabled.setChecked(not enabled)
|
||
self.enabled.blockSignals(False)
|
||
self.status_label.setText(f"客服开关保存失败:{exc}")
|
||
self._style_status(self.status_label, "DangerText")
|
||
return
|
||
self._agent_enabled = bool(enabled)
|
||
self._update_chat_controls()
|
||
self.refresh_status(deep=False)
|
||
|
||
def set_chat_auto_approve(self, enabled: bool) -> None:
|
||
try:
|
||
self.manager.save_integration_settings(
|
||
{"chat_auto_approve": bool(enabled)}
|
||
)
|
||
except Exception as exc:
|
||
self.auto_approve.blockSignals(True)
|
||
self.auto_approve.setChecked(not enabled)
|
||
self.auto_approve.blockSignals(False)
|
||
QMessageBox.warning(
|
||
self,
|
||
"工具权限保存失败",
|
||
f"无法保存 Agent 工具权限设置:{exc}",
|
||
)
|
||
return
|
||
mode = "已开启" if enabled else "已关闭"
|
||
self._sync_tool_mode_button()
|
||
self.chat_status.setText(f"自动批准工具{mode}")
|
||
|
||
def install_runtime(self) -> None:
|
||
if self._install_running:
|
||
return
|
||
self._install_running = True
|
||
self.install_button.setEnabled(False)
|
||
self.status_label.setText("正在安装或更新 Grok Build 官方运行时…")
|
||
self._style_status(self.status_label, "WarningText")
|
||
|
||
def worker() -> None:
|
||
try:
|
||
result = self.manager.install_official_release()
|
||
except Exception as exc:
|
||
result = exc
|
||
self.installFinished.emit(result)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _install_finished(self, result: object) -> None:
|
||
self._install_running = False
|
||
self.install_button.setEnabled(True)
|
||
if isinstance(result, Exception):
|
||
self.status_label.setText(f"Grok Build 安装失败:{result}")
|
||
self._style_status(self.status_label, "DangerText")
|
||
return
|
||
self.status_label.setText("Grok Build 安装完成,正在重新核验…")
|
||
self._style_status(self.status_label, "SuccessText")
|
||
self.refresh_status(deep=True)
|
||
|
||
def new_conversation(self) -> None:
|
||
if self._chat_preparing or (
|
||
self.agent_process is not None
|
||
and self.agent_process.state() != QProcess.NotRunning
|
||
):
|
||
return
|
||
self._active_session_id = ""
|
||
self._session_started = False
|
||
self._pending_prompt = ""
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._pending_route = ""
|
||
self._last_route = "direct"
|
||
self._direct_conversation_id = ""
|
||
self._direct_generation += 1
|
||
self._direct_cancellation = None
|
||
self._chat_history.clear()
|
||
self._agent_history_count = 0
|
||
self._cancel_requested = False
|
||
self._stream_render_timer.stop()
|
||
self._clear_layout(self.message_layout)
|
||
self.message_layout.addStretch(1)
|
||
self._place_composer(False)
|
||
self.session_label.setText("新对话")
|
||
self.chat_status.setText("等待发送")
|
||
self.prompt.setFocus()
|
||
|
||
def _remove_pending_user_history(self) -> None:
|
||
if (
|
||
self._pending_prompt
|
||
and self._chat_history
|
||
and self._chat_history[-1].get("role") == "user"
|
||
and self._chat_history[-1].get("content") == self._pending_prompt
|
||
):
|
||
self._chat_history.pop()
|
||
|
||
@staticmethod
|
||
def _agent_prompt_with_history(
|
||
prompt: str, history: list[dict[str, str]]
|
||
) -> str:
|
||
if not history:
|
||
return prompt
|
||
selected: list[dict[str, str]] = []
|
||
remaining = 16_000
|
||
for item in reversed(history[-12:]):
|
||
role = str(item.get("role") or "")
|
||
content = str(item.get("content") or "").strip()
|
||
if role not in {"user", "assistant"} or not content:
|
||
continue
|
||
if remaining <= 0:
|
||
break
|
||
content = content[-remaining:]
|
||
selected.append({"role": role, "content": content})
|
||
remaining -= len(content)
|
||
selected.reverse()
|
||
if not selected:
|
||
return prompt
|
||
transcript = "\n".join(
|
||
f"{'用户' if item['role'] == 'user' else '助手'}:{item['content']}"
|
||
for item in selected
|
||
)
|
||
return (
|
||
"下面是同一界面中尚未同步给 Agent 的最近对话,仅用于理解上下文,"
|
||
"不要把其中的助手回答当作当前用户指令:\n"
|
||
f"{transcript}\n\n当前用户请求:\n{prompt}"
|
||
)
|
||
|
||
def _start_direct_chat(
|
||
self, prompt: str, history: list[dict[str, str]]
|
||
) -> None:
|
||
from grok_direct_chat import DirectChatCancellation
|
||
|
||
self._direct_generation += 1
|
||
generation = self._direct_generation
|
||
conversation_id = self._direct_conversation_id
|
||
user_id = self._active_session_id or "wechat-rpa-chat"
|
||
cancellation = DirectChatCancellation()
|
||
self._direct_cancellation = cancellation
|
||
|
||
def worker() -> None:
|
||
pending_chunks: list[str] = []
|
||
pending_chars = 0
|
||
last_emit = 0.0
|
||
|
||
def flush_pending() -> None:
|
||
nonlocal pending_chars, last_emit
|
||
if not pending_chunks or cancellation.cancelled:
|
||
return
|
||
combined = "".join(pending_chunks)
|
||
pending_chunks.clear()
|
||
pending_chars = 0
|
||
last_emit = time.monotonic()
|
||
try:
|
||
self.directChatChunk.emit(generation, combined, False)
|
||
except RuntimeError:
|
||
cancellation.cancel()
|
||
|
||
def on_update(text: str, replace: bool) -> None:
|
||
nonlocal pending_chars, last_emit
|
||
if cancellation.cancelled:
|
||
return
|
||
if replace:
|
||
flush_pending()
|
||
try:
|
||
self.directChatChunk.emit(generation, text, True)
|
||
except RuntimeError:
|
||
cancellation.cancel()
|
||
last_emit = time.monotonic()
|
||
return
|
||
pending_chunks.append(text)
|
||
pending_chars += len(text)
|
||
now = time.monotonic()
|
||
if (
|
||
last_emit == 0.0
|
||
or pending_chars >= 96
|
||
or "\n" in text
|
||
or now - last_emit >= 0.04
|
||
):
|
||
flush_pending()
|
||
|
||
try:
|
||
from grok_direct_chat import stream_direct_chat
|
||
|
||
reply = stream_direct_chat(
|
||
prompt,
|
||
on_update=on_update,
|
||
history=history,
|
||
user_id=user_id,
|
||
conversation_id=conversation_id,
|
||
manager=self.manager,
|
||
cancellation=cancellation,
|
||
)
|
||
flush_pending()
|
||
result = {"generation": generation, "reply": reply}
|
||
except Exception as exc:
|
||
result = {"generation": generation, "error": exc}
|
||
try:
|
||
self.directChatFinished.emit(result)
|
||
except RuntimeError:
|
||
cancellation.cancel()
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _direct_chat_chunk(
|
||
self, generation: int, text: str, replace: bool
|
||
) -> None:
|
||
if (
|
||
generation != self._direct_generation
|
||
or not self._chat_preparing
|
||
or self._pending_route != "direct"
|
||
or self._cancel_requested
|
||
or self._shutting_down
|
||
or not text
|
||
):
|
||
return
|
||
if replace:
|
||
self._assistant_chunks = [text]
|
||
else:
|
||
self._assistant_chunks.append(text)
|
||
if not self._stream_render_timer.isActive():
|
||
self._stream_render_timer.start()
|
||
self.chat_status.setText("普通对话 · 流式生成中…")
|
||
|
||
def _direct_chat_finished(self, result: object) -> None:
|
||
details = (
|
||
result
|
||
if isinstance(result, dict)
|
||
else {"error": RuntimeError(str(result))}
|
||
)
|
||
if int(details.get("generation") or -1) != self._direct_generation:
|
||
return
|
||
self._direct_generation += 1
|
||
self._direct_cancellation = None
|
||
self._stream_render_timer.stop()
|
||
self._chat_preparing = False
|
||
if self._shutting_down:
|
||
return
|
||
if self._cancel_requested:
|
||
self._assistant_chunks = []
|
||
if self._assistant_label is not None:
|
||
self._assistant_label.setText("本轮已停止。")
|
||
self._remove_pending_user_history()
|
||
if self._pending_created_session:
|
||
self._active_session_id = ""
|
||
self.session_label.setText("新对话")
|
||
self.chat_status.setText("已停止")
|
||
else:
|
||
error = details.get("error")
|
||
reply = details.get("reply")
|
||
text = str(getattr(reply, "text", "") or "").strip()
|
||
if isinstance(error, Exception) or not text:
|
||
self._assistant_chunks = []
|
||
failure = error or RuntimeError("后台自有模型没有返回有效文本")
|
||
if self._assistant_label is not None:
|
||
self._assistant_label.setText(f"普通对话失败:{failure}")
|
||
self._remove_pending_user_history()
|
||
if self._pending_created_session:
|
||
self._active_session_id = ""
|
||
self.session_label.setText("新对话")
|
||
if self._pending_prompt and not self.prompt.toPlainText().strip():
|
||
self.prompt.setPlainText(self._pending_prompt)
|
||
self.chat_status.setText("普通对话失败")
|
||
else:
|
||
self._assistant_chunks = [text]
|
||
if self._assistant_label is not None:
|
||
self._assistant_label.setText(_safe_markdown(text))
|
||
self._chat_history.append({"role": "assistant", "content": text})
|
||
self._direct_conversation_id = str(
|
||
getattr(reply, "conversation_id", "") or ""
|
||
)
|
||
self._last_route = "direct"
|
||
model = str(getattr(reply, "model", "") or "后台自有模型")
|
||
self.chat_status.setText(f"普通对话完成 · {model}")
|
||
self.session_label.setText(
|
||
f"会话 {self._active_session_id[:8]} · 普通对话"
|
||
)
|
||
self._scroll_to_latest()
|
||
self._assistant_label = None
|
||
self._pending_prompt = ""
|
||
self._pending_route = ""
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._cancel_requested = False
|
||
self._update_chat_controls()
|
||
self.prompt.setFocus()
|
||
|
||
def send_message(self) -> None:
|
||
if self._chat_preparing or (
|
||
self.agent_process is not None
|
||
and self.agent_process.state() != QProcess.NotRunning
|
||
):
|
||
return
|
||
prompt = self.prompt.toPlainText().strip()
|
||
if not prompt:
|
||
self.prompt.setFocus()
|
||
return
|
||
|
||
from grok_direct_chat import classify_chat_route
|
||
|
||
route = classify_chat_route(prompt, last_route=self._last_route)
|
||
if route == "direct":
|
||
if not self._direct_model_compatible:
|
||
detail = self.status_meta.text().strip()
|
||
QMessageBox.warning(
|
||
self,
|
||
"后台自有模型不可用",
|
||
detail
|
||
or "后台自有模型尚未通过接口检测,请检查地址、协议和 API Key。",
|
||
)
|
||
return
|
||
else:
|
||
if not self._agent_enabled:
|
||
QMessageBox.warning(
|
||
self,
|
||
"此请求需要 Agent",
|
||
"检测到实时查询或执行型任务,但 Agent 调度已关闭。"
|
||
"普通语言对话仍可直接使用后台自有模型。",
|
||
)
|
||
return
|
||
if not self._model_compatible:
|
||
detail = self.status_meta.text().strip()
|
||
QMessageBox.warning(
|
||
self,
|
||
"Agent 暂不可用",
|
||
detail
|
||
or "此请求需要工具能力,请先安装并启用 Grok Build Agent。",
|
||
)
|
||
return
|
||
|
||
created_session = not self._active_session_id
|
||
if created_session:
|
||
self._active_session_id = str(uuid.uuid4())
|
||
is_new_session = route == "agent" and not self._session_started
|
||
history_snapshot = [dict(item) for item in self._chat_history]
|
||
agent_history = history_snapshot[self._agent_history_count :]
|
||
agent_prompt = self._agent_prompt_with_history(prompt, agent_history)
|
||
self._pending_prompt = prompt
|
||
self._pending_is_new_session = is_new_session
|
||
self._pending_created_session = created_session
|
||
self._pending_route = route
|
||
self._cancel_requested = False
|
||
self._turn_received_end = False
|
||
self._turn_had_error = False
|
||
self._turn_stop_reason = ""
|
||
self._assistant_chunks = []
|
||
self._thought_chunks = []
|
||
self._stream_render_timer.stop()
|
||
self._thought_label = None
|
||
self._chat_history.append({"role": "user", "content": prompt})
|
||
self._append_chat_message("user", prompt)
|
||
waiting_text = (
|
||
"正在调用后台自有模型…"
|
||
if route == "direct"
|
||
else "正在连接 Agent…"
|
||
)
|
||
self._assistant_label = self._append_chat_message(
|
||
"assistant", waiting_text
|
||
)
|
||
self.prompt.clear()
|
||
self._chat_preparing = True
|
||
self.chat_status.setText(
|
||
"普通对话 · 正在回复…"
|
||
if route == "direct"
|
||
else "正在准备 Agent 工具调度…"
|
||
)
|
||
self._update_chat_controls()
|
||
|
||
if route == "direct":
|
||
self._pending_is_new_session = False
|
||
self._start_direct_chat(prompt, history_snapshot)
|
||
return
|
||
|
||
session_id = self._active_session_id
|
||
auto_approve = self.auto_approve.isChecked()
|
||
|
||
def worker() -> None:
|
||
try:
|
||
import ai_config
|
||
|
||
cached_probe = (
|
||
self._last_endpoint_probe
|
||
if self._model_compatible
|
||
and bool(getattr(self._last_endpoint_probe, "ok", False))
|
||
else None
|
||
)
|
||
if cached_probe is not None:
|
||
current_profile = self.manager.agent_model_profile()
|
||
sync_result = SimpleNamespace(
|
||
compatible=True,
|
||
profile=self.model_profile,
|
||
model=str(
|
||
getattr(current_profile, "model", "")
|
||
or self.model_profile
|
||
),
|
||
)
|
||
endpoint_probe = cached_probe
|
||
else:
|
||
sync_result = self.manager.sync_model_configuration()
|
||
if not bool(getattr(sync_result, "compatible", False)):
|
||
reason = str(
|
||
getattr(sync_result, "message", "")
|
||
or "后台自有模型配置不兼容"
|
||
)
|
||
raise RuntimeError(
|
||
f"{reason}。本页面只使用后台自有模型,不会回退到其他模型。"
|
||
)
|
||
synchronized_profile = str(
|
||
getattr(sync_result, "profile", "") or ""
|
||
)
|
||
if synchronized_profile != self.model_profile:
|
||
raise RuntimeError(
|
||
"后台 Agent 模型未同步到受管配置 wecom-backend,已阻止启动"
|
||
)
|
||
endpoint_probe = self.manager.probe_agent_model(
|
||
force=False,
|
||
timeout=12.0,
|
||
cache_ttl=30.0,
|
||
)
|
||
if not bool(getattr(endpoint_probe, "ok", False)):
|
||
reason = str(
|
||
getattr(endpoint_probe, "message", "")
|
||
or "后台自有模型端点预检失败"
|
||
)
|
||
raise RuntimeError(reason)
|
||
selected_model = self.model_profile
|
||
configured_model = str(
|
||
getattr(sync_result, "model", "") or "后台配置的自有模型"
|
||
)
|
||
identity_rules = (
|
||
f"本轮推理使用后台配置的自有模型 {configured_model}。"
|
||
"Grok Build 只负责 Agent 调度,不是底层模型。"
|
||
"不得声称自己是 Grok 或由 xAI 发布;用户询问模型身份时,"
|
||
"必须如实回答上述后台自有模型名称。"
|
||
)
|
||
safety_rules = (
|
||
""
|
||
if auto_approve
|
||
else (
|
||
"当前是非交互安全模式。只调用默认可自动批准的只读工具"
|
||
"或受控 MCP 工具;不要调用终端命令或文件修改工具,"
|
||
"因为需要审批的调用会被取消。若任务确实需要这些操作,"
|
||
"请明确提示用户开启“自动批准工具”开关。"
|
||
)
|
||
)
|
||
workspace = str(SCRIPT_DIR.resolve())
|
||
binary = self.manager.require_binary()
|
||
args = self.manager.build_headless_args(
|
||
agent_prompt,
|
||
workspace=workspace,
|
||
model=selected_model,
|
||
effort=str(
|
||
getattr(ai_config, "GROK_CUSTOMER_SERVICE_EFFORT", "low")
|
||
),
|
||
max_turns=int(
|
||
getattr(ai_config, "GROK_CUSTOMER_SERVICE_MAX_TURNS", 8)
|
||
),
|
||
auto_approve=auto_approve,
|
||
continue_session=False,
|
||
resume_session="" if is_new_session else session_id,
|
||
new_session_id=session_id if is_new_session else "",
|
||
rules=identity_rules + safety_rules,
|
||
)
|
||
if "--continue" in args:
|
||
raise RuntimeError("检测到不安全的最近会话恢复参数,已阻止启动")
|
||
try:
|
||
model_index = args.index("--model")
|
||
actual_model = args[model_index + 1]
|
||
except (ValueError, IndexError) as exc:
|
||
raise RuntimeError("Agent 启动参数缺少受管模型") from exc
|
||
if actual_model != self.model_profile:
|
||
raise RuntimeError("Agent 启动参数试图使用非受管模型,已阻止启动")
|
||
environment = self.manager.runtime_environment(
|
||
include_model_key=True,
|
||
include_mcp_secrets=True,
|
||
custom_model_only=True,
|
||
workspace=workspace,
|
||
)
|
||
result = {
|
||
"binary": binary,
|
||
"args": args,
|
||
"workspace": workspace,
|
||
"environment": environment,
|
||
"is_new_session": is_new_session,
|
||
"session_id": session_id,
|
||
"model": configured_model,
|
||
"endpoint_probe": endpoint_probe,
|
||
}
|
||
except Exception as exc:
|
||
result = {
|
||
"error": exc,
|
||
"is_new_session": is_new_session,
|
||
"session_id": session_id,
|
||
}
|
||
if "endpoint_probe" in locals():
|
||
result["endpoint_probe"] = endpoint_probe
|
||
self.chatPreparationFinished.emit(result)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _chat_preparation_finished(self, result: object) -> None:
|
||
self._chat_preparing = False
|
||
details = (
|
||
result
|
||
if isinstance(result, dict)
|
||
else {"error": RuntimeError(str(result))}
|
||
)
|
||
if self._shutting_down:
|
||
return
|
||
if self._cancel_requested:
|
||
if self._assistant_label is not None and not self._assistant_chunks:
|
||
self._assistant_label.setText("本轮已停止。")
|
||
self._remove_pending_user_history()
|
||
if bool(details.get("is_new_session")):
|
||
self._session_started = False
|
||
if self._pending_created_session:
|
||
self._active_session_id = ""
|
||
self.session_label.setText("新对话")
|
||
self.chat_status.setText("已停止")
|
||
self._assistant_label = None
|
||
self._pending_prompt = ""
|
||
self._pending_route = ""
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._cancel_requested = False
|
||
self._update_chat_controls()
|
||
return
|
||
error = details.get("error")
|
||
if isinstance(error, Exception):
|
||
endpoint_probe = details.get("endpoint_probe")
|
||
if endpoint_probe is not None and not bool(
|
||
getattr(endpoint_probe, "ok", False)
|
||
):
|
||
self._last_endpoint_probe = endpoint_probe
|
||
self._model_compatible = False
|
||
self._direct_model_compatible = False
|
||
if self._assistant_label is not None:
|
||
self._assistant_label.setText(f"无法启动 Agent:{error}")
|
||
if bool(details.get("is_new_session")):
|
||
self._session_started = False
|
||
if self._pending_created_session:
|
||
self._active_session_id = ""
|
||
self.session_label.setText("新对话")
|
||
if self._pending_prompt and not self.prompt.toPlainText().strip():
|
||
self.prompt.setPlainText(self._pending_prompt)
|
||
self._remove_pending_user_history()
|
||
self.chat_status.setText("Agent 启动失败")
|
||
self.status_label.setText(
|
||
"后台自有模型端点不可用"
|
||
if endpoint_probe is not None
|
||
else "后台自有模型 Agent 启动失败"
|
||
)
|
||
self.status_meta.setText(str(error))
|
||
self._style_status(self.status_label, "DangerText")
|
||
self._assistant_label = None
|
||
self._pending_prompt = ""
|
||
self._pending_route = ""
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._update_chat_controls()
|
||
return
|
||
|
||
process = QProcess(self)
|
||
process.setProcessChannelMode(QProcess.SeparateChannels)
|
||
process.setWorkingDirectory(str(details["workspace"]))
|
||
environment = QProcessEnvironment()
|
||
for key, value in dict(details["environment"]).items():
|
||
environment.insert(str(key), str(value))
|
||
process.setProcessEnvironment(environment)
|
||
process.readyReadStandardOutput.connect(self._read_chat_output)
|
||
process.readyReadStandardError.connect(self._read_chat_stderr)
|
||
process.started.connect(self._chat_started)
|
||
process.errorOccurred.connect(self._chat_error)
|
||
process.finished.connect(self._chat_finished)
|
||
self.agent_process = process
|
||
self._process_buffer = ""
|
||
self._decoder = codecs.getincrementaldecoder("utf-8")("replace")
|
||
self._stderr_buffer = ""
|
||
self._stderr_decoder = codecs.getincrementaldecoder("utf-8")("replace")
|
||
self._session_started = True
|
||
self.chat_status.setText(
|
||
f"正在启动 Agent · {details.get('model') or '自有模型'}"
|
||
)
|
||
self._update_chat_controls()
|
||
process.start(str(details["binary"]), list(details["args"]))
|
||
|
||
def _chat_started(self) -> None:
|
||
self.chat_status.setText("Agent 正在生成…")
|
||
if self._assistant_label is not None and not self._assistant_chunks:
|
||
self._assistant_label.setText("Agent 正在处理你的请求…")
|
||
|
||
def _read_chat_output(self) -> None:
|
||
if self.agent_process is None:
|
||
return
|
||
raw = bytes(self.agent_process.readAllStandardOutput())
|
||
self._process_buffer += self._decoder.decode(raw)
|
||
while "\n" in self._process_buffer:
|
||
line, self._process_buffer = self._process_buffer.split("\n", 1)
|
||
self._consume_chat_event(line)
|
||
|
||
def _read_chat_stderr(self) -> None:
|
||
if self.agent_process is None:
|
||
return
|
||
raw = bytes(self.agent_process.readAllStandardError())
|
||
if not raw:
|
||
return
|
||
self._stderr_buffer += self._stderr_decoder.decode(raw)
|
||
self._stderr_buffer = self._stderr_buffer[-4000:]
|
||
|
||
def _consume_chat_event(self, line: str) -> None:
|
||
stripped = line.strip().lstrip("\ufeff")
|
||
if not stripped:
|
||
return
|
||
try:
|
||
event = json.loads(stripped)
|
||
except json.JSONDecodeError:
|
||
self._turn_had_error = True
|
||
self._append_chat_message(
|
||
"system", f"Agent 返回了无法解析的事件:{stripped[:500]}"
|
||
)
|
||
return
|
||
if not isinstance(event, dict):
|
||
return
|
||
event_type = str(event.get("type") or "")
|
||
if event_type == "text":
|
||
chunk = str(event.get("data") or "")
|
||
if chunk:
|
||
self._assistant_chunks.append(chunk)
|
||
if not self._stream_render_timer.isActive():
|
||
self._stream_render_timer.start()
|
||
elif event_type == "thought":
|
||
chunk = str(event.get("data") or "")
|
||
if chunk:
|
||
self._thought_chunks.append(chunk)
|
||
if self._thought_label is None:
|
||
self._thought_label = self._append_chat_message("thought", "")
|
||
if not self._stream_render_timer.isActive():
|
||
self._stream_render_timer.start()
|
||
self.chat_status.setText("Agent 正在思考…")
|
||
elif event_type == "error":
|
||
self._turn_had_error = True
|
||
message = str(
|
||
event.get("message") or event.get("data") or "Agent 执行失败"
|
||
)
|
||
self._append_chat_message("system", f"Agent 错误:{message}")
|
||
self.chat_status.setText("Agent 返回错误")
|
||
elif event_type == "end":
|
||
self._stream_render_timer.stop()
|
||
self._flush_stream_render()
|
||
if self._turn_stop_reason:
|
||
self._turn_had_error = True
|
||
self._append_chat_message("system", "Agent 返回了重复的结束事件。")
|
||
return
|
||
returned_session = str(event.get("sessionId") or "").strip()
|
||
stop_reason = str(event.get("stopReason") or "")
|
||
self._turn_stop_reason = stop_reason
|
||
if not returned_session or returned_session != self._active_session_id:
|
||
self._turn_had_error = True
|
||
self._append_chat_message(
|
||
"system",
|
||
"Agent 返回的会话 ID 与当前对话不一致,本轮结果已拒绝。",
|
||
)
|
||
else:
|
||
self._turn_received_end = True
|
||
self._session_started = True
|
||
self.session_label.setText(f"会话 {returned_session}")
|
||
if stop_reason not in HEADLESS_COMPLETED_STOP_REASONS:
|
||
self._turn_had_error = True
|
||
self._append_chat_message(
|
||
"system",
|
||
f"Agent 已结束本轮({stop_reason or '未知原因'}),"
|
||
"但没有完成可作为最终回答的结果。",
|
||
)
|
||
if self._assistant_label is not None and not self._assistant_chunks:
|
||
self._assistant_label.setText("Agent 本轮没有返回文本。")
|
||
self.chat_status.setText(
|
||
f"本轮完成 · {stop_reason}"
|
||
if (
|
||
self._turn_received_end
|
||
and stop_reason in HEADLESS_COMPLETED_STOP_REASONS
|
||
)
|
||
else f"本轮结束 · {stop_reason or '未知原因'}"
|
||
if self._turn_received_end
|
||
else "Agent 结束状态无效"
|
||
)
|
||
|
||
def _chat_error(self, error) -> None:
|
||
process = self.agent_process
|
||
if process is None:
|
||
return
|
||
self._turn_had_error = True
|
||
if error == QProcess.ProcessError.FailedToStart:
|
||
if self._assistant_label is not None and not self._assistant_chunks:
|
||
self._assistant_label.setText(
|
||
f"Agent 进程启动失败:{process.errorString()}"
|
||
)
|
||
if self._pending_is_new_session:
|
||
self._session_started = False
|
||
if self._pending_created_session:
|
||
self._active_session_id = ""
|
||
self.session_label.setText("新对话")
|
||
self.chat_status.setText("Agent 进程启动失败")
|
||
if self._pending_prompt and not self.prompt.toPlainText().strip():
|
||
self.prompt.setPlainText(self._pending_prompt)
|
||
self._remove_pending_user_history()
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._pending_route = ""
|
||
self._pending_prompt = ""
|
||
self._assistant_label = None
|
||
self.agent_process = None
|
||
process.deleteLater()
|
||
self._update_chat_controls()
|
||
|
||
def _chat_finished(self, exit_code: int, exit_status) -> None:
|
||
process = self.agent_process
|
||
if process is not None:
|
||
raw = bytes(process.readAllStandardOutput())
|
||
self._process_buffer += self._decoder.decode(raw, final=True)
|
||
stderr_raw = bytes(process.readAllStandardError())
|
||
self._stderr_buffer += self._stderr_decoder.decode(
|
||
stderr_raw, final=True
|
||
)
|
||
while "\n" in self._process_buffer:
|
||
line, self._process_buffer = self._process_buffer.split("\n", 1)
|
||
self._consume_chat_event(line)
|
||
if self._process_buffer.strip():
|
||
self._consume_chat_event(self._process_buffer)
|
||
self._process_buffer = ""
|
||
self._stream_render_timer.stop()
|
||
self._flush_stream_render()
|
||
|
||
normal_exit = exit_status == QProcess.NormalExit
|
||
successful = (
|
||
not self._cancel_requested
|
||
and normal_exit
|
||
and exit_code == 0
|
||
and self._turn_received_end
|
||
and self._turn_stop_reason in HEADLESS_COMPLETED_STOP_REASONS
|
||
and not self._turn_had_error
|
||
)
|
||
if successful:
|
||
answer = "".join(self._assistant_chunks).strip()
|
||
if answer:
|
||
self._chat_history.append(
|
||
{"role": "assistant", "content": answer}
|
||
)
|
||
self._last_route = "agent"
|
||
self._direct_conversation_id = ""
|
||
self._agent_history_count = len(self._chat_history)
|
||
self.chat_status.setText("等待继续对话")
|
||
else:
|
||
self._turn_had_error = True
|
||
if self._assistant_label is not None:
|
||
if (
|
||
self._turn_stop_reason == "Cancelled"
|
||
and not self._cancel_requested
|
||
):
|
||
self._assistant_label.setText(
|
||
"本轮工具调用需要批准,已在无交互模式下取消。"
|
||
"请开启上方“自动批准工具”后重试。"
|
||
)
|
||
else:
|
||
self._assistant_label.setText(
|
||
"本轮未正常完成,流式产生的未完成内容已丢弃。"
|
||
)
|
||
if self._cancel_requested:
|
||
detail = "生成已停止;未完成内容不会作为回答使用。"
|
||
self.chat_status.setText("已停止生成")
|
||
elif not normal_exit:
|
||
detail = "Agent 进程异常退出;未完成内容不会作为回答使用。"
|
||
self.chat_status.setText("Agent 进程异常退出")
|
||
elif exit_code != 0:
|
||
detail = (
|
||
f"Agent 退出码为 {exit_code};未完成内容不会作为回答使用。"
|
||
)
|
||
self.chat_status.setText(f"Agent 已退出 · 代码 {exit_code}")
|
||
elif not self._turn_received_end:
|
||
detail = "本轮未收到合法结束事件;未完成内容不会作为回答使用。"
|
||
self.chat_status.setText("Agent 输出不完整")
|
||
elif self._turn_stop_reason not in HEADLESS_COMPLETED_STOP_REASONS:
|
||
detail = (
|
||
"工具调用未获批准;请开启“自动批准工具”后重试。"
|
||
if self._turn_stop_reason == "Cancelled"
|
||
else (
|
||
f"Agent 已返回合法结束事件,但本轮状态为 "
|
||
f"{self._turn_stop_reason or '未知原因'};"
|
||
"未完成内容不会作为回答使用。"
|
||
)
|
||
)
|
||
self.chat_status.setText(
|
||
f"Agent 本轮未完成 · {self._turn_stop_reason or '未知原因'}"
|
||
)
|
||
else:
|
||
detail = "Agent 本轮返回了错误事件;未完成内容不会作为回答使用。"
|
||
self.chat_status.setText("Agent 本轮失败")
|
||
self._append_chat_message("system", detail)
|
||
self._remove_pending_user_history()
|
||
if self._pending_is_new_session:
|
||
self._session_started = False
|
||
if self._pending_created_session:
|
||
self._active_session_id = ""
|
||
self.session_label.setText("新对话")
|
||
|
||
if process is not None:
|
||
process.deleteLater()
|
||
self.agent_process = None
|
||
self._assistant_label = None
|
||
self._thought_label = None
|
||
self._pending_prompt = ""
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._pending_route = ""
|
||
self._cancel_requested = False
|
||
self._stderr_buffer = ""
|
||
self._update_chat_controls()
|
||
self.prompt.setFocus()
|
||
|
||
@staticmethod
|
||
def _kill_windows_process_tree(process_id: int) -> None:
|
||
if os.name != "nt" or process_id <= 0:
|
||
return
|
||
taskkill = (
|
||
Path(os.environ.get("SystemRoot", r"C:\Windows"))
|
||
/ "System32"
|
||
/ "taskkill.exe"
|
||
)
|
||
if not taskkill.is_file():
|
||
return
|
||
try:
|
||
subprocess.run(
|
||
[
|
||
str(taskkill),
|
||
"/PID",
|
||
str(process_id),
|
||
"/T",
|
||
"/F",
|
||
],
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
timeout=5,
|
||
check=False,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
)
|
||
except (OSError, subprocess.SubprocessError):
|
||
pass
|
||
|
||
def stop_chat(self) -> None:
|
||
self._cancel_requested = True
|
||
if self._chat_preparing:
|
||
if self._pending_route == "direct":
|
||
cancellation = self._direct_cancellation
|
||
if cancellation is not None:
|
||
cancellation.cancel()
|
||
self._direct_cancellation = None
|
||
self._direct_generation += 1
|
||
self._chat_preparing = False
|
||
self._stream_render_timer.stop()
|
||
self._assistant_chunks = []
|
||
if self._assistant_label is not None:
|
||
self._assistant_label.setText("本轮已停止。")
|
||
self._remove_pending_user_history()
|
||
if self._pending_created_session:
|
||
self._active_session_id = ""
|
||
self.session_label.setText("新对话")
|
||
self._assistant_label = None
|
||
self._pending_prompt = ""
|
||
self._pending_route = ""
|
||
self._pending_is_new_session = False
|
||
self._pending_created_session = False
|
||
self._cancel_requested = False
|
||
self.chat_status.setText("已停止普通对话")
|
||
self._update_chat_controls()
|
||
self.prompt.setFocus()
|
||
return
|
||
self.chat_status.setText("将在启动前停止本轮…")
|
||
self.stop_button.setEnabled(False)
|
||
return
|
||
process = self.agent_process
|
||
if process is None or process.state() == QProcess.NotRunning:
|
||
self._update_chat_controls()
|
||
return
|
||
self.chat_status.setText("正在停止 Agent…")
|
||
self.stop_button.setEnabled(False)
|
||
process_id = int(process.processId())
|
||
if os.name == "nt" and process_id > 0:
|
||
threading.Thread(
|
||
target=self._kill_windows_process_tree,
|
||
args=(process_id,),
|
||
daemon=True,
|
||
).start()
|
||
else:
|
||
process.terminate()
|
||
|
||
def force_kill(target=process) -> None:
|
||
if self.agent_process is target and target.state() != QProcess.NotRunning:
|
||
target.kill()
|
||
|
||
QTimer.singleShot(1800, force_kill)
|
||
|
||
def shutdown(self) -> None:
|
||
self._shutting_down = True
|
||
self._cancel_requested = True
|
||
cancellation = self._direct_cancellation
|
||
if cancellation is not None:
|
||
cancellation.cancel()
|
||
self._direct_cancellation = None
|
||
self._direct_generation += 1
|
||
process = self.agent_process
|
||
if process is not None and process.state() != QProcess.NotRunning:
|
||
if os.name == "nt":
|
||
self._kill_windows_process_tree(int(process.processId()))
|
||
process.kill()
|
||
process.waitForFinished(1200)
|
||
self.agent_process = None
|
||
|
||
|
||
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": "待补症状",
|
||
"pending_human_confirmation": "待人工确认",
|
||
"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", "pending_human_confirmation", "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)
|
||
grokCustomerStatusReady = Signal(object)
|
||
|
||
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))
|
||
self.grok_customer_enabled = QCheckBox("启用 Grok Build 本地客服 Agent")
|
||
self.grok_customer_enabled.setChecked(
|
||
bool(ai_config.GROK_CUSTOMER_SERVICE_ENABLED)
|
||
)
|
||
for index, switch in enumerate(
|
||
(
|
||
self.ai_enabled,
|
||
self.context_enabled,
|
||
self.counter_enabled,
|
||
self.mcp_enabled,
|
||
self.grok_customer_enabled,
|
||
)
|
||
):
|
||
switch_grid.addWidget(switch, index // 2, index % 2)
|
||
overview_layout.addLayout(switch_grid)
|
||
layout.addWidget(overview)
|
||
|
||
customer_agent, customer_agent_layout = _card(
|
||
"Grok Build 客服 Agent",
|
||
"Grok Build 只负责 Agent 调度,回复由后台自有模型生成;"
|
||
"只能调度项目内置的受控客服 MCP,也不具备消息发送能力。",
|
||
)
|
||
customer_status_row = QHBoxLayout()
|
||
self.grok_customer_status = _label("正在检测 Grok Build…", "WarningText")
|
||
self.grok_customer_status.setWordWrap(True)
|
||
customer_status_row.addWidget(self.grok_customer_status, 1)
|
||
self.grok_customer_refresh_button = _button("刷新状态")
|
||
self.grok_customer_refresh_button.clicked.connect(
|
||
self.refresh_grok_customer_status
|
||
)
|
||
customer_status_row.addWidget(self.grok_customer_refresh_button)
|
||
customer_agent_layout.addLayout(customer_status_row)
|
||
customer_form = QGridLayout()
|
||
customer_form.setHorizontalSpacing(18)
|
||
customer_form.setVerticalSpacing(8)
|
||
self.grok_customer_timeout = QSpinBox()
|
||
self.grok_customer_timeout.setRange(30, 600)
|
||
self.grok_customer_timeout.setValue(
|
||
self._bounded_int(
|
||
ai_config.GROK_CUSTOMER_SERVICE_TIMEOUT, 180, 30, 600
|
||
)
|
||
)
|
||
self.grok_customer_max_turns = QSpinBox()
|
||
self.grok_customer_max_turns.setRange(2, 30)
|
||
self.grok_customer_max_turns.setValue(
|
||
self._bounded_int(
|
||
ai_config.GROK_CUSTOMER_SERVICE_MAX_TURNS, 8, 2, 30
|
||
)
|
||
)
|
||
self.grok_customer_effort = QComboBox()
|
||
self.grok_customer_effort.addItems(("low", "medium", "high"))
|
||
effort = str(
|
||
ai_config.GROK_CUSTOMER_SERVICE_EFFORT or "low"
|
||
).strip().lower()
|
||
self.grok_customer_effort.setCurrentText(
|
||
effort if effort in {"low", "medium", "high"} else "low"
|
||
)
|
||
customer_form.addWidget(_label("单次回复超时(秒)", "CardSubtitle"), 0, 0)
|
||
customer_form.addWidget(_label("最多 Agent 轮数", "CardSubtitle"), 0, 1)
|
||
customer_form.addWidget(self.grok_customer_timeout, 1, 0)
|
||
customer_form.addWidget(self.grok_customer_max_turns, 1, 1)
|
||
customer_form.addWidget(_label("推理强度", "CardSubtitle"), 2, 0)
|
||
customer_form.addWidget(self.grok_customer_effort, 3, 0)
|
||
customer_form.setColumnStretch(0, 1)
|
||
customer_form.setColumnStretch(1, 1)
|
||
customer_agent_layout.addLayout(customer_form)
|
||
layout.addWidget(customer_agent)
|
||
|
||
identity, identity_layout = _card(
|
||
"视觉模型与客服身份",
|
||
"下列普通 API 仅供可选视觉能力使用,不会作为 Grok Agent 的模型回退。",
|
||
)
|
||
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._grok_customer_status_running = False
|
||
self._grok_customer_status_pending = False
|
||
self.backendSyncFinished.connect(self._backend_sync_finished)
|
||
self.grokCustomerStatusReady.connect(self._grok_customer_status_ready)
|
||
self._managed_widgets = (
|
||
self.ai_enabled,
|
||
self.context_enabled,
|
||
self.counter_enabled,
|
||
self.mcp_enabled,
|
||
self.grok_customer_enabled,
|
||
self.grok_customer_timeout,
|
||
self.grok_customer_max_turns,
|
||
self.grok_customer_effort,
|
||
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()
|
||
QTimer.singleShot(0, self.refresh_grok_customer_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
|
||
|
||
@staticmethod
|
||
def _bounded_int(value, default: int, minimum: int, maximum: int) -> int:
|
||
try:
|
||
parsed = int(value)
|
||
except (TypeError, ValueError, OverflowError):
|
||
parsed = default
|
||
return max(minimum, min(maximum, parsed))
|
||
|
||
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
|
||
|
||
settings = {
|
||
"AI_ENABLED": self.ai_enabled.isChecked(),
|
||
"AI_CONTEXT_ENABLED": self.context_enabled.isChecked(),
|
||
"AI_COUNTER_INSULT_ENABLED": self.counter_enabled.isChecked(),
|
||
"AI_MCP_ENABLED": self.mcp_enabled.isChecked(),
|
||
"GROK_CUSTOMER_SERVICE_ENABLED": (
|
||
self.grok_customer_enabled.isChecked()
|
||
),
|
||
"GROK_CUSTOMER_SERVICE_TIMEOUT": (
|
||
self.grok_customer_timeout.value()
|
||
),
|
||
"GROK_CUSTOMER_SERVICE_MAX_TURNS": (
|
||
self.grok_customer_max_turns.value()
|
||
),
|
||
"GROK_CUSTOMER_SERVICE_EFFORT": (
|
||
self.grok_customer_effort.currentText()
|
||
),
|
||
"AI_API_BASE": self.api_base.text().strip(),
|
||
"AI_API_KEY": self.api_key.text().strip(),
|
||
"AI_MODEL": self.model.text().strip(),
|
||
"AI_AGENT_NAME": self.agent_name.text().strip(),
|
||
"AI_HOSPITAL_NAME": self.hospital.text().strip(),
|
||
"AI_CONTEXT_MAX_ROUNDS": self.rounds.value(),
|
||
"AI_MAX_TOKENS": self.max_tokens.value(),
|
||
"AI_TEMPERATURE": self.temperature.value(),
|
||
"AI_MCP_SERVERS": servers,
|
||
"AI_MCP_MAX_ROUNDS": self.mcp_rounds.value(),
|
||
}
|
||
ai_config.apply_settings(settings, persist=True)
|
||
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.refresh_grok_customer_status()
|
||
self.save_status.style().unpolish(self.save_status)
|
||
self.save_status.style().polish(self.save_status)
|
||
|
||
def refresh_grok_customer_status(self) -> None:
|
||
"""Inspect the local Grok runtime without probing any HTTP service."""
|
||
if self._grok_customer_status_running:
|
||
self._grok_customer_status_pending = True
|
||
return
|
||
self._grok_customer_status_running = True
|
||
self._grok_customer_status_pending = False
|
||
self.grok_customer_refresh_button.setEnabled(False)
|
||
self.grok_customer_status.setObjectName("WarningText")
|
||
self.grok_customer_status.setText("正在检测本机 Grok Build 运行时…")
|
||
self.grok_customer_status.style().unpolish(self.grok_customer_status)
|
||
self.grok_customer_status.style().polish(self.grok_customer_status)
|
||
|
||
def worker() -> None:
|
||
try:
|
||
from grok_build_bridge import GrokBuildManager
|
||
|
||
result = GrokBuildManager().status()
|
||
except Exception as exc:
|
||
result = exc
|
||
self.grokCustomerStatusReady.emit(result)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _grok_customer_status_ready(self, result: object) -> None:
|
||
self._grok_customer_status_running = False
|
||
self.grok_customer_refresh_button.setEnabled(True)
|
||
if isinstance(result, Exception):
|
||
self.grok_customer_status.setObjectName("DangerText")
|
||
self.grok_customer_status.setText(f"Grok Build 状态检测失败:{result}")
|
||
else:
|
||
installed = bool(getattr(result, "installed", False))
|
||
model_configured = bool(getattr(result, "model_configured", False))
|
||
model_compatible = bool(getattr(result, "model_compatible", False))
|
||
model_name = str(getattr(result, "model_name", "") or "")
|
||
model_message = str(getattr(result, "model_message", "") or "")
|
||
model_backend = str(
|
||
getattr(result, "model_api_backend", "") or ""
|
||
)
|
||
adapter_live = bool(getattr(result, "adapter_live", False))
|
||
runtime_text = "已安装" if installed else "未安装"
|
||
if (
|
||
model_configured
|
||
and model_compatible
|
||
and model_backend == "dify"
|
||
and not adapter_live
|
||
):
|
||
model_text = (
|
||
"Dify 模型已配置(适配器将在启动 Agent 时自动创建)"
|
||
)
|
||
elif model_configured and model_compatible:
|
||
model_text = f"后台模型兼容{f'({model_name})' if model_name else ''}"
|
||
elif model_configured:
|
||
model_text = f"后台模型不兼容{f'({model_message})' if model_message else ''}"
|
||
else:
|
||
model_text = "后台模型未配置"
|
||
enabled_text = (
|
||
"客服 Agent 已启用"
|
||
if self.grok_customer_enabled.isChecked()
|
||
else "客服 Agent 已关闭"
|
||
)
|
||
ready = (
|
||
installed
|
||
and model_compatible
|
||
and (model_backend != "dify" or adapter_live)
|
||
)
|
||
self.grok_customer_status.setObjectName(
|
||
"SuccessText" if ready else "WarningText"
|
||
)
|
||
self.grok_customer_status.setText(
|
||
f"{enabled_text} · 运行时{runtime_text} · {model_text} · 不使用 xAI 登录"
|
||
)
|
||
self.grok_customer_status.style().unpolish(self.grok_customer_status)
|
||
self.grok_customer_status.style().polish(self.grok_customer_status)
|
||
if self._grok_customer_status_pending:
|
||
QTimer.singleShot(0, self.refresh_grok_customer_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.refresh_grok_customer_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.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:
|
||
self._backend_sync_running = False
|
||
self.sync_backend_button.setEnabled(True)
|
||
if ok:
|
||
self.reload_from_ai_config()
|
||
self.refresh_grok_customer_status()
|
||
self.refresh_backend_status()
|
||
if not ok:
|
||
self.saved.emit(False, 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.grok_customer_enabled.setChecked(
|
||
bool(ai_config.GROK_CUSTOMER_SERVICE_ENABLED)
|
||
)
|
||
self.grok_customer_timeout.setValue(
|
||
self._bounded_int(
|
||
ai_config.GROK_CUSTOMER_SERVICE_TIMEOUT, 180, 30, 600
|
||
)
|
||
)
|
||
self.grok_customer_max_turns.setValue(
|
||
self._bounded_int(
|
||
ai_config.GROK_CUSTOMER_SERVICE_MAX_TURNS, 8, 2, 30
|
||
)
|
||
)
|
||
effort = str(
|
||
ai_config.GROK_CUSTOMER_SERVICE_EFFORT or "low"
|
||
).strip().lower()
|
||
self.grok_customer_effort.setCurrentText(
|
||
effort if effort in {"low", "medium", "high"} else "low"
|
||
)
|
||
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 GrokBuildPage(QScrollArea):
|
||
"""Qt control surface for the project-local Grok Build runtime."""
|
||
|
||
logMessage = Signal(str, str)
|
||
runtimeStatusReady = Signal(object)
|
||
modelSyncReady = Signal(object)
|
||
installProgress = Signal(int, str)
|
||
installFinished = Signal(bool, str)
|
||
diagnosticFinished = Signal(str, bool, str)
|
||
taskPreparationFinished = Signal(object)
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
from grok_build_bridge import GrokBuildManager
|
||
|
||
self.manager = GrokBuildManager()
|
||
self.agent_process: QProcess | None = None
|
||
self._host_bound_tui_processes: list[subprocess.Popen] = []
|
||
self._process_buffer = ""
|
||
self._decoder = codecs.getincrementaldecoder("utf-8")("replace")
|
||
self._refresh_running = False
|
||
self._sync_running = False
|
||
self._sync_pending = False
|
||
self._install_running = False
|
||
self._task_preparing = False
|
||
|
||
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(
|
||
"06",
|
||
"Grok Build Agent",
|
||
"Grok Build 只提供 Agent 调度;所有推理固定使用后台自有模型 wecom-backend。",
|
||
)
|
||
)
|
||
|
||
runtime, runtime_layout = _card(
|
||
"运行时与后台自有模型",
|
||
"官方程序缓存在项目 .grok-build;会话和配置保存在 LocalAppData。"
|
||
"自有模型 API Key 仅在安全核验后注入,不写入 TOML,也不读取 xAI 登录。",
|
||
)
|
||
runtime_row = QHBoxLayout()
|
||
runtime_copy = QVBoxLayout()
|
||
runtime_copy.setSpacing(5)
|
||
self.runtime_status = _label("正在检测 Grok Build…", "WarningText")
|
||
self.runtime_meta = _label("", "CardSubtitle")
|
||
self.model_status = _label("正在同步后台模型…", "WarningText")
|
||
self.model_meta = _label("", "CardSubtitle")
|
||
runtime_copy.addWidget(self.runtime_status)
|
||
runtime_copy.addWidget(self.runtime_meta)
|
||
runtime_copy.addWidget(self.model_status)
|
||
runtime_copy.addWidget(self.model_meta)
|
||
runtime_row.addLayout(runtime_copy, 1)
|
||
self.refresh_button = _button("刷新状态")
|
||
self.refresh_button.clicked.connect(self.refresh_status)
|
||
runtime_row.addWidget(self.refresh_button, 0, Qt.AlignTop)
|
||
self.sync_model_button = _button("同步后台模型")
|
||
self.sync_model_button.clicked.connect(self.sync_model_configuration)
|
||
runtime_row.addWidget(self.sync_model_button, 0, Qt.AlignTop)
|
||
self.install_button = _button("安装运行时", "primary")
|
||
self.install_button.clicked.connect(self.install_runtime)
|
||
runtime_row.addWidget(self.install_button, 0, Qt.AlignTop)
|
||
runtime_layout.addLayout(runtime_row)
|
||
self.install_bar = QProgressBar()
|
||
self.install_bar.setRange(0, 100)
|
||
self.install_bar.setValue(0)
|
||
self.install_bar.setTextVisible(True)
|
||
self.install_bar.hide()
|
||
runtime_layout.addWidget(self.install_bar)
|
||
layout.addWidget(runtime)
|
||
|
||
launcher, launcher_layout = _card(
|
||
"完整功能入口",
|
||
"“打开完整 TUI”保留上游交互能力,但模型仍强制为后台 wecom-backend。",
|
||
)
|
||
launcher_row = QHBoxLayout()
|
||
self.tui_button = _button("打开完整 TUI", "primary")
|
||
self.tui_button.clicked.connect(lambda: self.open_tui(""))
|
||
launcher_row.addWidget(self.tui_button)
|
||
extensions_button = _button("插件与技能")
|
||
extensions_button.clicked.connect(lambda: self.open_tui("/plugins"))
|
||
launcher_row.addWidget(extensions_button)
|
||
mcp_button = _button("MCP 管理")
|
||
mcp_button.clicked.connect(lambda: self.open_tui("/mcps"))
|
||
launcher_row.addWidget(mcp_button)
|
||
inspect_button = _button("运行时检查")
|
||
inspect_button.clicked.connect(
|
||
lambda: self.run_diagnostic(["inspect", "--json"], "运行时检查")
|
||
)
|
||
launcher_row.addWidget(inspect_button)
|
||
docs_button = _button("官方文档")
|
||
docs_button.clicked.connect(
|
||
lambda: QDesktopServices.openUrl(QUrl("https://docs.x.ai/build/overview"))
|
||
)
|
||
launcher_row.addWidget(docs_button)
|
||
launcher_row.addStretch(1)
|
||
launcher_layout.addLayout(launcher_row)
|
||
layout.addWidget(launcher)
|
||
|
||
task, task_layout = _card(
|
||
"图形化无头任务",
|
||
"适合自动化和 CI。需要逐项审批、回退、计划评审或完整终端界面时,请使用上方完整 TUI。",
|
||
)
|
||
workspace_row = QHBoxLayout()
|
||
workspace_copy = QVBoxLayout()
|
||
workspace_copy.setSpacing(6)
|
||
workspace_copy.addWidget(_label("工作目录", "CardSubtitle"))
|
||
integration = self.manager.load_integration_settings()
|
||
self.workspace = QLineEdit(
|
||
str(integration.get("default_workspace") or SCRIPT_DIR)
|
||
)
|
||
workspace_copy.addWidget(self.workspace)
|
||
workspace_row.addLayout(workspace_copy, 1)
|
||
browse_button = _button("选择目录")
|
||
browse_button.clicked.connect(self.choose_workspace)
|
||
workspace_row.addWidget(browse_button, 0, Qt.AlignBottom)
|
||
task_layout.addLayout(workspace_row)
|
||
|
||
options = QGridLayout()
|
||
options.setHorizontalSpacing(16)
|
||
options.setVerticalSpacing(7)
|
||
options.addWidget(_label("模型", "CardSubtitle"), 0, 0)
|
||
options.addWidget(_label("推理强度", "CardSubtitle"), 0, 1)
|
||
options.addWidget(_label("最大轮数", "CardSubtitle"), 0, 2)
|
||
options.addWidget(_label("会话策略", "CardSubtitle"), 0, 3)
|
||
self.model = QLineEdit("wecom-backend")
|
||
self.model.setReadOnly(True)
|
||
options.addWidget(self.model, 1, 0)
|
||
self.effort = QComboBox()
|
||
self.effort.addItems(("minimal", "low", "medium", "high", "xhigh", "max"))
|
||
self.effort.setCurrentText("high")
|
||
options.addWidget(self.effort, 1, 1)
|
||
self.max_turns = QSpinBox()
|
||
self.max_turns.setRange(1, 1000)
|
||
self.max_turns.setValue(50)
|
||
options.addWidget(self.max_turns, 1, 2)
|
||
self.resume_session = QLineEdit()
|
||
self.resume_session.setReadOnly(True)
|
||
self.resume_session.setPlaceholderText("每次创建安全新会话")
|
||
self.resume_session.setToolTip("需要多轮续聊请使用左侧 AI 客服原生对话页。")
|
||
options.addWidget(self.resume_session, 1, 3)
|
||
self.continue_session = QCheckBox("禁止继续未知历史会话")
|
||
self.continue_session.setEnabled(False)
|
||
self.read_only = QCheckBox(
|
||
"只读审查(仅在无 MCP、可执行插件 Hook 与 LSP 时启动)"
|
||
)
|
||
self.auto_approve = QCheckBox("无人值守:自动批准所有工具操作")
|
||
self.auto_approve.setToolTip("Windows 缺少上游系统沙箱;仅在可信工作区和可信提示词中开启。")
|
||
options.addWidget(self.continue_session, 2, 0)
|
||
options.addWidget(self.read_only, 2, 1, 1, 2)
|
||
options.addWidget(self.auto_approve, 2, 3)
|
||
options.setColumnStretch(0, 2)
|
||
options.setColumnStretch(1, 1)
|
||
options.setColumnStretch(2, 1)
|
||
options.setColumnStretch(3, 2)
|
||
task_layout.addLayout(options)
|
||
|
||
task_layout.addWidget(_label("任务内容", "CardSubtitle"))
|
||
self.prompt = QPlainTextEdit()
|
||
self.prompt.setMinimumHeight(120)
|
||
self.prompt.setPlaceholderText(
|
||
"例如:审查当前项目,修复启动时端口冲突和界面卡顿,并运行相关测试。"
|
||
)
|
||
task_layout.addWidget(self.prompt)
|
||
task_actions = QHBoxLayout()
|
||
self.task_status = _label("等待任务", "CardSubtitle")
|
||
task_actions.addWidget(self.task_status, 1)
|
||
clear_button = _button("清空输出")
|
||
clear_button.clicked.connect(self.clear_output)
|
||
task_actions.addWidget(clear_button)
|
||
self.cancel_button = _button("取消任务", "danger")
|
||
self.cancel_button.setEnabled(False)
|
||
self.cancel_button.clicked.connect(self.cancel_task)
|
||
task_actions.addWidget(self.cancel_button)
|
||
self.run_button = _button("开始任务", "primary")
|
||
self.run_button.clicked.connect(self.start_task)
|
||
task_actions.addWidget(self.run_button)
|
||
task_layout.addLayout(task_actions)
|
||
layout.addWidget(task)
|
||
|
||
output_card, output_layout = _card(
|
||
"代理输出",
|
||
"文本、思考事件、错误、会话 ID 和完成状态会实时显示在这里。",
|
||
)
|
||
self.output = QTextEdit()
|
||
self.output.setReadOnly(True)
|
||
self.output.setAcceptRichText(False)
|
||
self.output.setMinimumHeight(260)
|
||
self.output.setStyleSheet(
|
||
"QTextEdit{font-family:'Cascadia Mono','Microsoft YaHei UI';"
|
||
"font-size:13px;line-height:1.5;background:#111A16;color:#DDF5EC;"
|
||
"border-color:#263B32;}"
|
||
)
|
||
output_layout.addWidget(self.output)
|
||
layout.addWidget(output_card)
|
||
layout.addStretch(1)
|
||
|
||
self.runtimeStatusReady.connect(self._apply_runtime_status)
|
||
self.modelSyncReady.connect(self._apply_model_sync_result)
|
||
self.installProgress.connect(self._apply_install_progress)
|
||
self.installFinished.connect(self._install_finished)
|
||
self.diagnosticFinished.connect(self._diagnostic_finished)
|
||
self.taskPreparationFinished.connect(self._task_preparation_finished)
|
||
QTimer.singleShot(120, self.sync_model_configuration)
|
||
|
||
@staticmethod
|
||
def _style_status(label: QLabel, ok: bool, warning: bool = False) -> None:
|
||
label.setObjectName("WarningText" if warning else ("SuccessText" if ok else "DangerText"))
|
||
label.style().unpolish(label)
|
||
label.style().polish(label)
|
||
|
||
def choose_workspace(self) -> None:
|
||
selected = QFileDialog.getExistingDirectory(
|
||
self,
|
||
"选择 Grok Build 工作目录",
|
||
self.workspace.text().strip() or str(SCRIPT_DIR),
|
||
)
|
||
if selected:
|
||
self.workspace.setText(selected)
|
||
self.manager.save_integration_settings({"default_workspace": selected})
|
||
|
||
def sync_model_configuration(self) -> None:
|
||
if self._sync_running:
|
||
self._sync_pending = True
|
||
return
|
||
self._sync_running = True
|
||
self.sync_model_button.setEnabled(False)
|
||
self.model_status.setText("正在同步后台 Agent 自有模型…")
|
||
self._style_status(self.model_status, False, warning=True)
|
||
|
||
def worker() -> None:
|
||
try:
|
||
result = self.manager.sync_model_configuration()
|
||
except Exception as exc:
|
||
result = exc
|
||
self.modelSyncReady.emit(result)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _apply_model_sync_result(self, result) -> None:
|
||
self._sync_running = False
|
||
self.sync_model_button.setEnabled(True)
|
||
if isinstance(result, Exception):
|
||
exc = result
|
||
self.model_status.setText(f"后台模型同步失败:{exc}")
|
||
self.model_meta.setText("")
|
||
self._style_status(self.model_status, False)
|
||
self.logMessage.emit(f"Grok Build 模型同步失败:{exc}", "err")
|
||
else:
|
||
self.model_status.setText(result.message)
|
||
source_route = (
|
||
str(getattr(result, "source_base_url", "") or result.base_url)
|
||
)
|
||
effective_route = str(
|
||
getattr(result, "effective_base_url", "") or ""
|
||
)
|
||
route_text = source_route or "未生成模型端点"
|
||
if effective_route and effective_route != source_route:
|
||
route_text = f"{route_text} → {effective_route}"
|
||
self.model_meta.setText(
|
||
f"{route_text} · 配置 {result.config_path}"
|
||
)
|
||
self._style_status(
|
||
self.model_status,
|
||
result.compatible,
|
||
warning=not result.compatible,
|
||
)
|
||
if result.compatible:
|
||
self.model.setText(result.profile)
|
||
self.manager.save_integration_settings(
|
||
{"default_model": result.profile, "sync_backend_model": True}
|
||
)
|
||
self.logMessage.emit(result.message, "ok")
|
||
else:
|
||
self.model.setText(result.profile)
|
||
self.logMessage.emit(result.message, "warn")
|
||
self.refresh_status()
|
||
if self._sync_pending:
|
||
self._sync_pending = False
|
||
QTimer.singleShot(0, self.sync_model_configuration)
|
||
|
||
def refresh_status(self) -> None:
|
||
if self._refresh_running:
|
||
return
|
||
self._refresh_running = True
|
||
self.refresh_button.setEnabled(False)
|
||
self.runtime_status.setText("正在检测 Grok Build 运行时…")
|
||
self._style_status(self.runtime_status, False, warning=True)
|
||
|
||
def worker() -> None:
|
||
try:
|
||
status = self.manager.status()
|
||
except Exception as exc:
|
||
status = exc
|
||
self.runtimeStatusReady.emit(status)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _apply_runtime_status(self, status) -> None:
|
||
self._refresh_running = False
|
||
self.refresh_button.setEnabled(True)
|
||
if isinstance(status, Exception):
|
||
self.runtime_status.setText(f"运行时检测失败:{status}")
|
||
self.runtime_meta.setText("")
|
||
self._style_status(self.runtime_status, False)
|
||
return
|
||
if status.installed:
|
||
version = status.version or "版本未知"
|
||
self.runtime_status.setText(f"Grok Build 已就绪 · {version}")
|
||
meta = (
|
||
f"{status.binary_path} · 仅后台自有模型"
|
||
f" · 状态目录 {status.runtime_home}"
|
||
)
|
||
if status.warnings:
|
||
meta += " · " + ";".join(status.warnings)
|
||
self.runtime_meta.setText(meta)
|
||
self._style_status(self.runtime_status, True)
|
||
self.install_button.setText("安装 / 更新")
|
||
else:
|
||
self.runtime_status.setText("尚未安装 Grok Build 官方运行时")
|
||
self.runtime_meta.setText(f"将安装到 {status.binary_path}")
|
||
self._style_status(self.runtime_status, False, warning=True)
|
||
self.install_button.setText("安装运行时")
|
||
|
||
def install_runtime(self) -> None:
|
||
if self._install_running:
|
||
return
|
||
self._install_running = True
|
||
self.install_button.setEnabled(False)
|
||
self.install_bar.setRange(0, 0)
|
||
self.install_bar.setValue(0)
|
||
self.install_bar.setFormat("正在获取官方版本…")
|
||
self.install_bar.show()
|
||
self.runtime_status.setText("正在安装 Grok Build 官方运行时…")
|
||
self._style_status(self.runtime_status, False, warning=True)
|
||
|
||
def progress(received: int, total: int) -> None:
|
||
if total > 0:
|
||
percent = min(100, int(received * 100 / total))
|
||
message = f"{received / 1024 / 1024:.1f} / {total / 1024 / 1024:.1f} MB"
|
||
else:
|
||
percent = -1
|
||
message = f"已下载 {received / 1024 / 1024:.1f} MB"
|
||
self.installProgress.emit(percent, message)
|
||
|
||
def worker() -> None:
|
||
try:
|
||
status = self.manager.install_official_release(progress=progress)
|
||
except Exception as exc:
|
||
self.installFinished.emit(False, str(exc))
|
||
else:
|
||
self.installFinished.emit(True, status.version or "安装完成")
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _apply_install_progress(self, percent: int, message: str) -> None:
|
||
if percent < 0:
|
||
self.install_bar.setRange(0, 0)
|
||
else:
|
||
self.install_bar.setRange(0, 100)
|
||
self.install_bar.setValue(percent)
|
||
self.install_bar.setFormat(message)
|
||
|
||
def _install_finished(self, ok: bool, message: str) -> None:
|
||
self._install_running = False
|
||
self.install_button.setEnabled(True)
|
||
self.install_bar.setRange(0, 100)
|
||
self.install_bar.setValue(100 if ok else 0)
|
||
self.install_bar.setFormat("安装完成" if ok else "安装失败")
|
||
if ok:
|
||
self.runtime_status.setText(f"Grok Build 已安装 · {message}")
|
||
self._style_status(self.runtime_status, True)
|
||
self.logMessage.emit(f"Grok Build 运行时安装完成:{message}", "ok")
|
||
else:
|
||
self.runtime_status.setText(f"Grok Build 安装失败:{message}")
|
||
self._style_status(self.runtime_status, False)
|
||
self.logMessage.emit(f"Grok Build 安装失败:{message}", "err")
|
||
self.refresh_status()
|
||
|
||
def _workspace_path(self) -> str:
|
||
value = self.workspace.text().strip() or str(SCRIPT_DIR)
|
||
path = Path(value).expanduser().resolve()
|
||
if not path.is_dir():
|
||
raise ValueError(f"工作目录不存在:{path}")
|
||
self.manager.save_integration_settings({"default_workspace": str(path)})
|
||
return str(path)
|
||
|
||
def open_tui(self, initial_prompt: str) -> None:
|
||
try:
|
||
workspace = self._workspace_path()
|
||
process = self.manager.open_tui(
|
||
workspace=workspace,
|
||
initial_prompt=initial_prompt,
|
||
model=self.model.text().strip(),
|
||
)
|
||
if self.manager.model_profile().api_backend == "dify":
|
||
self._host_bound_tui_processes = [
|
||
item
|
||
for item in self._host_bound_tui_processes
|
||
if item.poll() is None
|
||
]
|
||
self._host_bound_tui_processes.append(process)
|
||
except Exception as exc:
|
||
QMessageBox.warning(self, "无法打开 Grok Build", str(exc))
|
||
return
|
||
self.task_status.setText("已在独立控制台打开完整 TUI")
|
||
self.logMessage.emit("已打开 Grok Build 完整 TUI", "notify")
|
||
|
||
def open_login(self) -> None:
|
||
QMessageBox.information(
|
||
self,
|
||
"不需要 Grok 登录",
|
||
"本项目只使用 Grok Build Agent 调度能力,所有模型调用均走后台自有模型。",
|
||
)
|
||
self.task_status.setText("Grok/xAI 登录已禁用;当前仅使用后台自有模型")
|
||
|
||
def run_diagnostic(self, args: list[str], title: str) -> None:
|
||
try:
|
||
workspace = self._workspace_path()
|
||
except Exception as exc:
|
||
QMessageBox.warning(self, f"无法执行{title}", str(exc))
|
||
return
|
||
self.task_status.setText(f"正在执行:{title}…")
|
||
|
||
def worker() -> None:
|
||
try:
|
||
completed = self.manager.run_capture(
|
||
args,
|
||
workspace=workspace,
|
||
timeout=180,
|
||
)
|
||
ok = completed.returncode == 0
|
||
content = completed.stdout or f"命令退出码:{completed.returncode}"
|
||
except Exception as exc:
|
||
ok = False
|
||
content = str(exc)
|
||
self.diagnosticFinished.emit(title, ok, content)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _diagnostic_finished(self, title: str, ok: bool, content: str) -> None:
|
||
self.output.append(f"\n===== {title} =====\n{content.rstrip()}\n")
|
||
self.task_status.setText(f"{title}{'完成' if ok else '失败'}")
|
||
self.logMessage.emit(
|
||
f"Grok Build {title}{'完成' if ok else '失败'}",
|
||
"ok" if ok else "err",
|
||
)
|
||
|
||
def clear_output(self) -> None:
|
||
self.output.clear()
|
||
|
||
def start_task(self) -> None:
|
||
from grok_build_bridge import MODEL_PROFILE
|
||
|
||
if self._task_preparing:
|
||
QMessageBox.information(self, "正在准备任务", "请等待安全检查完成。")
|
||
return
|
||
if self.agent_process is not None and self.agent_process.state() != QProcess.NotRunning:
|
||
QMessageBox.information(self, "任务运行中", "请先等待当前任务结束或点击取消。")
|
||
return
|
||
try:
|
||
workspace = self._workspace_path()
|
||
prompt = self.prompt.toPlainText().strip()
|
||
args = self.manager.build_headless_args(
|
||
prompt,
|
||
workspace=workspace,
|
||
model=MODEL_PROFILE,
|
||
effort=self.effort.currentText(),
|
||
max_turns=self.max_turns.value(),
|
||
auto_approve=self.auto_approve.isChecked(),
|
||
read_only=self.read_only.isChecked(),
|
||
continue_session=False,
|
||
resume_session="",
|
||
)
|
||
except Exception as exc:
|
||
QMessageBox.warning(self, "无法开始任务", str(exc))
|
||
return
|
||
if self.auto_approve.isChecked():
|
||
answer = QMessageBox.warning(
|
||
self,
|
||
"确认无人值守模式",
|
||
"此模式会自动批准命令执行和文件修改。Windows 当前没有 Grok Build 的系统级沙箱,"
|
||
"请确认工作目录和任务内容均可信。\n\n确定继续吗?",
|
||
QMessageBox.Yes | QMessageBox.No,
|
||
QMessageBox.No,
|
||
)
|
||
if answer != QMessageBox.Yes:
|
||
return
|
||
|
||
read_only = self.read_only.isChecked()
|
||
self._task_preparing = True
|
||
self.output.clear()
|
||
self.task_status.setText("正在执行运行时与凭据安全检查…")
|
||
self.run_button.setEnabled(False)
|
||
self.cancel_button.setEnabled(False)
|
||
|
||
def worker() -> None:
|
||
try:
|
||
binary = self.manager.require_binary()
|
||
self.manager.prepare_agent_configuration()
|
||
if read_only:
|
||
self.manager.verify_read_only_environment(workspace)
|
||
environment = self.manager.runtime_environment(
|
||
include_model_key=True,
|
||
include_mcp_secrets=not read_only,
|
||
workspace=workspace,
|
||
custom_model_only=True,
|
||
)
|
||
result = (binary, args, workspace, environment)
|
||
except Exception as exc:
|
||
result = exc
|
||
self.taskPreparationFinished.emit(result)
|
||
|
||
threading.Thread(target=worker, daemon=True).start()
|
||
|
||
def _task_preparation_finished(self, result) -> None:
|
||
self._task_preparing = False
|
||
if isinstance(result, Exception):
|
||
self.run_button.setEnabled(True)
|
||
self.cancel_button.setEnabled(False)
|
||
self.task_status.setText("任务安全检查失败")
|
||
self.logMessage.emit(f"Grok Build 任务未启动:{result}", "err")
|
||
QMessageBox.warning(self, "无法开始任务", str(result))
|
||
return
|
||
|
||
binary, args, workspace, values = result
|
||
process = QProcess(self)
|
||
process.setProcessChannelMode(QProcess.MergedChannels)
|
||
process.setWorkingDirectory(workspace)
|
||
# Rebuild from the bridge's sanitized copy. Starting from
|
||
# systemEnvironment() would retain bridge-owned stale credentials that
|
||
# runtime_environment() deliberately removed from its returned dict.
|
||
environment = QProcessEnvironment()
|
||
for key, value in values.items():
|
||
environment.insert(str(key), str(value))
|
||
process.setProcessEnvironment(environment)
|
||
process.readyReadStandardOutput.connect(self._read_process_output)
|
||
process.started.connect(self._task_started)
|
||
process.errorOccurred.connect(self._task_error)
|
||
process.finished.connect(self._task_finished)
|
||
self.agent_process = process
|
||
self._process_buffer = ""
|
||
self._decoder = codecs.getincrementaldecoder("utf-8")("replace")
|
||
self.task_status.setText("正在启动 Grok Build…")
|
||
self.cancel_button.setEnabled(True)
|
||
process.start(str(binary), args)
|
||
|
||
def _task_started(self) -> None:
|
||
self.task_status.setText("Grok Build 正在执行任务")
|
||
self.logMessage.emit("Grok Build 无头任务已启动", "notify")
|
||
|
||
def _read_process_output(self) -> None:
|
||
if self.agent_process is None:
|
||
return
|
||
raw = bytes(self.agent_process.readAllStandardOutput())
|
||
self._process_buffer += self._decoder.decode(raw)
|
||
while "\n" in self._process_buffer:
|
||
line, self._process_buffer = self._process_buffer.split("\n", 1)
|
||
self._consume_event_line(line)
|
||
|
||
def _append_output_text(self, value: str) -> None:
|
||
self.output.moveCursor(QTextCursor.End)
|
||
self.output.insertPlainText(value)
|
||
self.output.moveCursor(QTextCursor.End)
|
||
self.output.ensureCursorVisible()
|
||
|
||
def _consume_event_line(self, line: str) -> None:
|
||
from grok_build_bridge import parse_streaming_event
|
||
|
||
event_type, content = parse_streaming_event(line)
|
||
if event_type == "empty":
|
||
return
|
||
if event_type == "text":
|
||
self._append_output_text(content)
|
||
elif event_type == "thought":
|
||
self._append_output_text(f"\n[思考] {content}\n")
|
||
elif event_type == "end":
|
||
self._append_output_text(f"\n\n[{content}]\n")
|
||
elif event_type == "error":
|
||
self._append_output_text(f"\n[错误] {content}\n")
|
||
else:
|
||
self._append_output_text(f"\n[{event_type}] {content}\n")
|
||
|
||
def _task_error(self, error) -> None:
|
||
if self.agent_process is None:
|
||
return
|
||
self.task_status.setText(f"任务进程错误:{self.agent_process.errorString()}")
|
||
if error == QProcess.ProcessError.FailedToStart:
|
||
process = self.agent_process
|
||
self.run_button.setEnabled(True)
|
||
self.cancel_button.setEnabled(False)
|
||
self.logMessage.emit(
|
||
f"Grok Build 任务启动失败:{process.errorString()}",
|
||
"err",
|
||
)
|
||
self.agent_process = None
|
||
process.deleteLater()
|
||
|
||
def _task_finished(self, exit_code: int, _exit_status) -> None:
|
||
if self.agent_process is not None:
|
||
raw = bytes(self.agent_process.readAllStandardOutput())
|
||
self._process_buffer += self._decoder.decode(raw, final=True)
|
||
if self._process_buffer.strip():
|
||
self._consume_event_line(self._process_buffer)
|
||
self._process_buffer = ""
|
||
self.run_button.setEnabled(True)
|
||
self.cancel_button.setEnabled(False)
|
||
if exit_code == 0:
|
||
self.task_status.setText("任务已完成")
|
||
self.logMessage.emit("Grok Build 任务已完成", "ok")
|
||
else:
|
||
self.task_status.setText(f"任务已结束 · 退出码 {exit_code}")
|
||
self.logMessage.emit(f"Grok Build 任务失败,退出码 {exit_code}", "err")
|
||
if self.agent_process is not None:
|
||
self.agent_process.deleteLater()
|
||
self.agent_process = None
|
||
|
||
def cancel_task(self) -> None:
|
||
if self.agent_process is None or self.agent_process.state() == QProcess.NotRunning:
|
||
return
|
||
process = self.agent_process
|
||
self.task_status.setText("正在取消任务…")
|
||
process.terminate()
|
||
|
||
def force_kill(target=process) -> None:
|
||
if self.agent_process is target and target.state() != QProcess.NotRunning:
|
||
target.kill()
|
||
|
||
QTimer.singleShot(1800, force_kill)
|
||
|
||
def shutdown(self) -> None:
|
||
if self.agent_process is not None and self.agent_process.state() != QProcess.NotRunning:
|
||
self.agent_process.kill()
|
||
self.agent_process.waitForFinished(1200)
|
||
for process in self._host_bound_tui_processes:
|
||
if process.poll() is not None:
|
||
continue
|
||
try:
|
||
process.terminate()
|
||
process.wait(timeout=2)
|
||
except (OSError, subprocess.SubprocessError):
|
||
try:
|
||
process.kill()
|
||
except OSError:
|
||
pass
|
||
self._host_bound_tui_processes.clear()
|
||
|
||
|
||
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("07", "运行日志", "实时查看连接、识别、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 人格",
|
||
"Grok Build",
|
||
"运行日志",
|
||
)
|
||
|
||
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.customer_service_page = CustomerServicePage()
|
||
self.dashboard_page = DashboardPage()
|
||
self.settings_page = SettingsPage(self.runtime_settings)
|
||
self.business_page = BusinessPage()
|
||
self.persona_page = PersonaPage()
|
||
self.grok_page = GrokBuildPage()
|
||
self.log_page = LogPage()
|
||
for page in (
|
||
self.customer_service_page,
|
||
self.dashboard_page,
|
||
self.settings_page,
|
||
self.business_page,
|
||
self.persona_page,
|
||
self.grok_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.persona_page.backendSyncFinished.connect(self._backend_model_sync_finished)
|
||
self.grok_page.logMessage.connect(self.append_log)
|
||
self.grok_page.modelSyncReady.connect(
|
||
lambda _result: self.persona_page.refresh_grok_customer_status()
|
||
)
|
||
self.grok_page.modelSyncReady.connect(
|
||
lambda _result: self.customer_service_page.refresh_status(deep=True)
|
||
)
|
||
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:
|
||
self.customer_service_page.refresh_status(deep=False)
|
||
|
||
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(6)
|
||
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")
|
||
if ok:
|
||
self.customer_service_page.refresh_status(deep=False)
|
||
self.grok_page.sync_model_configuration()
|
||
|
||
def _backend_model_sync_finished(self, ok: bool, message: str) -> None:
|
||
if ok:
|
||
self.append_log(message, "ok")
|
||
self.customer_service_page.refresh_status(deep=False)
|
||
self.grok_page.sync_model_configuration()
|
||
|
||
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.customer_service_page.shutdown()
|
||
self.grok_page.shutdown()
|
||
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()
|