diff --git a/wechat_rpa/.gitignore b/wechat_rpa/.gitignore
new file mode 100644
index 0000000..7acaffd
--- /dev/null
+++ b/wechat_rpa/.gitignore
@@ -0,0 +1,11 @@
+.edge-headless/
+kf-preview.png
+__pycache__/
+*.pyc
+backend.db
+backend.db-shm
+backend.db-wal
+backend_connection.json
+backend_connection.json.tmp
+backend_runtime.json
+backend_runtime.json.tmp
diff --git a/wechat_rpa/BACKEND.md b/wechat_rpa/BACKEND.md
new file mode 100644
index 0000000..0d3e5c1
--- /dev/null
+++ b/wechat_rpa/BACKEND.md
@@ -0,0 +1,75 @@
+# 配置管理后台
+
+项目内置了一个零额外依赖的配置后台，用于统一管理登录账号、角色、AI 模型和 MCP 配置。
+
+## 1. 启动后台
+
+在项目目录运行：
+
+```powershell
+python admin_backend.py
+```
+
+后台会优先使用 `8765`。如果该端口已被其他程序占用，会自动尝试 `8766`、`8767`……，并在控制台打印最终地址。桌面端会读取 `backend_runtime.json` 自动跟随实际端口，无需手工修改。
+
+浏览器打开控制台打印的地址。首次创建数据库时使用：
+
+- 用户名：`admin`
+- 初始密码：`Admin@123456`
+
+首次登录必须修改密码。也可以在第一次启动前通过环境变量 `WECOM_ADMIN_INITIAL_PASSWORD` 设置不同的初始密码。
+
+数据库默认保存在 `backend.db`，已加入 `.gitignore`。首次建库会读取当前 `ai_settings.json` 作为第一版模型配置。
+
+## 2. 用户与角色
+
+- 管理员（admin）：管理用户、角色并发布模型配置。
+- 配置员（operator）：查看和发布模型配置，不能管理用户。
+- 只读用户（viewer）：查看后台并登录桌面端同步配置，不能修改配置。
+
+管理员创建的新用户第一次登录网页时也必须修改初始密码，之后才能从桌面端登录。
+
+## 3. 桌面端自动配置
+
+启动 `wechat_gui.py`，进入“AI 人格与能力”页：
+
+1. 点击“登录后台”。
+2. 填写后台地址、用户名和密码。
+3. 登录成功后，后台配置立即写入本机 `ai_settings.json` 并生效。
+4. 桌面端以后会在启动时及每 5 分钟自动同步，也可以点击“立即同步”。
+
+本机只保存 30 天有效的访问令牌，不保存后台密码。登录后台并启用自动同步后，本地 AI 配置字段会变成只读，以后台配置为准；退出后台后可恢复本地编辑。
+
+后台发生端口冲突并自动更换端口时，同一项目目录中的桌面端会自动发现新地址；已有登录令牌可以继续使用。
+
+同一台电脑上启动后台与桌面端时，后台会在 `backend_runtime.json` 发布一个仅限回环地址使用的临时只读同步凭证。桌面软件启动时会先检测该服务并拉取最新配置，所以即使没有保存后台账号登录，也会立即刷新“能力开关”“模型与身份”和“MCP 服务器”中的内容。该凭证不能管理用户或修改后台配置，后台停止后即失效。
+
+## 4. 局域网部署
+
+如需让其他电脑连接，可在后台所在电脑运行：
+
+```powershell
+python admin_backend.py --host 0.0.0.0 --port 8765
+```
+
+桌面端后台地址填写服务器的局域网地址，例如 `http://192.168.1.20:8765`。需要在 Windows 防火墙中仅对可信局域网放行该端口。
+
+跨公网使用时不要直接暴露此 HTTP 服务，应使用 Nginx、Caddy 等反向代理配置 HTTPS，再把桌面端地址改为 HTTPS 地址。
+
+## 5. 管理员密码恢复
+
+停止后台后运行：
+
+```powershell
+python admin_backend.py --reset-admin-password
+```
+
+按提示输入新密码。重置后已有 admin 登录令牌会失效，下次登录仍需再修改一次密码。
+
+## API
+
+- `POST /api/v1/auth/login`：桌面端登录并取得令牌。
+- `POST /api/v1/auth/logout`：注销令牌。
+- `GET /api/v1/me`：读取当前用户及角色。
+- `GET /api/v1/config`：读取当前版本的模型配置。
+- `GET /health`：健康检查。
diff --git a/wechat_rpa/__pycache__/ai_chat.cpython-311.pyc b/wechat_rpa/__pycache__/ai_chat.cpython-311.pyc
index 184b49a..416f21d 100644
Binary files a/wechat_rpa/__pycache__/ai_chat.cpython-311.pyc and b/wechat_rpa/__pycache__/ai_chat.cpython-311.pyc differ
diff --git a/wechat_rpa/__pycache__/ai_config.cpython-311.pyc b/wechat_rpa/__pycache__/ai_config.cpython-311.pyc
index f0e3ee7..11ce606 100644
Binary files a/wechat_rpa/__pycache__/ai_config.cpython-311.pyc and b/wechat_rpa/__pycache__/ai_config.cpython-311.pyc differ
diff --git a/wechat_rpa/__pycache__/conversation_store.cpython-311.pyc b/wechat_rpa/__pycache__/conversation_store.cpython-311.pyc
index f60c13c..de9cd57 100644
Binary files a/wechat_rpa/__pycache__/conversation_store.cpython-311.pyc and b/wechat_rpa/__pycache__/conversation_store.cpython-311.pyc differ
diff --git a/wechat_rpa/__pycache__/registration_store.cpython-311.pyc b/wechat_rpa/__pycache__/registration_store.cpython-311.pyc
index de23151..6a55567 100644
Binary files a/wechat_rpa/__pycache__/registration_store.cpython-311.pyc and b/wechat_rpa/__pycache__/registration_store.cpython-311.pyc differ
diff --git a/wechat_rpa/__pycache__/wechat_bot.cpython-311.pyc b/wechat_rpa/__pycache__/wechat_bot.cpython-311.pyc
index 31e3968..e768116 100644
Binary files a/wechat_rpa/__pycache__/wechat_bot.cpython-311.pyc and b/wechat_rpa/__pycache__/wechat_bot.cpython-311.pyc differ
diff --git a/wechat_rpa/__pycache__/wechat_gui.cpython-311.pyc b/wechat_rpa/__pycache__/wechat_gui.cpython-311.pyc
index 7922708..f4234d4 100644
Binary files a/wechat_rpa/__pycache__/wechat_gui.cpython-311.pyc and b/wechat_rpa/__pycache__/wechat_gui.cpython-311.pyc differ
diff --git a/wechat_rpa/admin_backend.py b/wechat_rpa/admin_backend.py
new file mode 100644
index 0000000..7ca2db0
--- /dev/null
+++ b/wechat_rpa/admin_backend.py
@@ -0,0 +1,1228 @@
+# -*- coding: utf-8 -*-
+"""企微客服助手的内置配置后台。
+
+仅依赖 Python 标准库，提供：
+- 用户名/密码登录和强制首次改密
+- admin / operator / viewer 三种角色
+- AI 与 MCP 配置的网页管理
+- 供桌面端认证并拉取配置的 JSON API
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import getpass
+import hashlib
+import hmac
+import html
+import ipaddress
+import json
+import os
+import secrets
+import sqlite3
+import sys
+import threading
+import time
+import urllib.parse
+from collections import defaultdict, deque
+from datetime import datetime
+from http import HTTPStatus
+from http.cookies import SimpleCookie
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+DEFAULT_DB = SCRIPT_DIR / "backend.db"
+DEFAULT_RUNTIME_FILE = SCRIPT_DIR / "backend_runtime.json"
+DEFAULT_HOST = "127.0.0.1"
+DEFAULT_PORT = 8765
+DEFAULT_PORT_ATTEMPTS = 100
+DEFAULT_ADMIN_PASSWORD = "Admin@123456"
+ROLES = ("admin", "operator", "viewer")
+ROLE_LABELS = {"admin": "管理员", "operator": "配置员", "viewer": "只读用户"}
+PBKDF2_ITERATIONS = 310_000
+CONFIG_KEYS = (
+ "AI_ENABLED",
+ "AI_API_BASE",
+ "AI_API_KEY",
+ "AI_MODEL",
+ "AI_USE_VISION",
+ "AI_CONTEXT_ENABLED",
+ "AI_CONTEXT_MAX_ROUNDS",
+ "AI_COUNTER_INSULT_ENABLED",
+ "AI_AGENT_NAME",
+ "AI_HOSPITAL_NAME",
+ "AI_MAX_TOKENS",
+ "AI_TEMPERATURE",
+ "AI_TIMEOUT",
+ "AI_MCP_ENABLED",
+ "AI_MCP_MAX_ROUNDS",
+ "AI_MCP_SERVERS",
+)
+BOOL_KEYS = {
+ "AI_ENABLED",
+ "AI_USE_VISION",
+ "AI_CONTEXT_ENABLED",
+ "AI_COUNTER_INSULT_ENABLED",
+ "AI_MCP_ENABLED",
+}
+
+
+def now_text() -> str:
+ return datetime.now().astimezone().isoformat(timespec="seconds")
+
+
+def password_hash(password: str, salt: bytes | None = None) -> tuple[str, str]:
+ salt = salt or secrets.token_bytes(16)
+ digest = hashlib.pbkdf2_hmac(
+ "sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
+ )
+ return base64.b64encode(salt).decode("ascii"), base64.b64encode(digest).decode("ascii")
+
+
+def verify_password(password: str, salt_text: str, digest_text: str) -> bool:
+ try:
+ salt = base64.b64decode(salt_text)
+ expected = base64.b64decode(digest_text)
+ except Exception:
+ return False
+ actual = hashlib.pbkdf2_hmac(
+ "sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
+ )
+ return hmac.compare_digest(actual, expected)
+
+
+def token_hash(token: str) -> str:
+ return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+def load_initial_config() -> dict[str, Any]:
+ path = SCRIPT_DIR / "ai_settings.json"
+ try:
+ saved = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, ValueError, TypeError):
+ saved = {}
+ defaults: dict[str, Any] = {
+ "AI_ENABLED": True,
+ "AI_API_BASE": "",
+ "AI_API_KEY": "",
+ "AI_MODEL": "",
+ "AI_USE_VISION": False,
+ "AI_CONTEXT_ENABLED": True,
+ "AI_CONTEXT_MAX_ROUNDS": 5,
+ "AI_COUNTER_INSULT_ENABLED": False,
+ "AI_AGENT_NAME": "客服",
+ "AI_HOSPITAL_NAME": "",
+ "AI_MAX_TOKENS": 500,
+ "AI_TEMPERATURE": 0.35,
+ "AI_TIMEOUT": 120,
+ "AI_MCP_ENABLED": False,
+ "AI_MCP_MAX_ROUNDS": 5,
+ "AI_MCP_SERVERS": [],
+ }
+ if isinstance(saved, dict):
+ defaults.update({key: saved[key] for key in CONFIG_KEYS if key in saved})
+ return defaults
+
+
+class ClosingConnection(sqlite3.Connection):
+ """提交/回滚后立即关闭，避免 Windows 上数据库文件长期被占用。"""
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ try:
+ return super().__exit__(exc_type, exc_value, traceback)
+ finally:
+ self.close()
+
+
+class Database:
+ def __init__(self, path: Path):
+ self.path = path
+ self.path.parent.mkdir(parents=True, exist_ok=True)
+
+ def connect(self) -> sqlite3.Connection:
+ connection = sqlite3.connect(
+ self.path, timeout=10, factory=ClosingConnection
+ )
+ connection.row_factory = sqlite3.Row
+ connection.execute("PRAGMA foreign_keys = ON")
+ return connection
+
+ def initialize(self, initial_password: str) -> bool:
+ with self.connect() as db:
+ db.executescript(
+ """
+ PRAGMA journal_mode = WAL;
+ CREATE TABLE IF NOT EXISTS users (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ username TEXT NOT NULL UNIQUE COLLATE NOCASE,
+ password_salt TEXT NOT NULL,
+ password_digest TEXT NOT NULL,
+ role TEXT NOT NULL CHECK(role IN ('admin','operator','viewer')),
+ active INTEGER NOT NULL DEFAULT 1,
+ must_change_password INTEGER NOT NULL DEFAULT 1,
+ created_at TEXT NOT NULL,
+ updated_at TEXT NOT NULL
+ );
+ CREATE TABLE IF NOT EXISTS auth_tokens (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ token_digest TEXT NOT NULL UNIQUE,
+ kind TEXT NOT NULL,
+ csrf_token TEXT NOT NULL,
+ device_name TEXT NOT NULL DEFAULT '',
+ created_at TEXT NOT NULL,
+ expires_at INTEGER NOT NULL,
+ last_used_at TEXT NOT NULL
+ );
+ CREATE TABLE IF NOT EXISTS model_config (
+ id INTEGER PRIMARY KEY CHECK(id = 1),
+ config_json TEXT NOT NULL,
+ version INTEGER NOT NULL DEFAULT 1,
+ updated_at TEXT NOT NULL,
+ updated_by INTEGER REFERENCES users(id)
+ );
+ CREATE TABLE IF NOT EXISTS audit_log (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_id INTEGER REFERENCES users(id),
+ action TEXT NOT NULL,
+ detail TEXT NOT NULL DEFAULT '',
+ ip_address TEXT NOT NULL DEFAULT '',
+ created_at TEXT NOT NULL
+ );
+ CREATE INDEX IF NOT EXISTS idx_tokens_digest ON auth_tokens(token_digest);
+ CREATE INDEX IF NOT EXISTS idx_audit_created ON audit_log(created_at DESC);
+ """
+ )
+ created = db.execute("SELECT COUNT(*) FROM users").fetchone()[0] == 0
+ if created:
+ salt, digest = password_hash(initial_password)
+ stamp = now_text()
+ db.execute(
+ """INSERT INTO users
+ (username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
+ VALUES (?,?,?,?,1,1,?,?)""",
+ ("admin", salt, digest, "admin", stamp, stamp),
+ )
+ if db.execute("SELECT COUNT(*) FROM model_config").fetchone()[0] == 0:
+ admin = db.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
+ db.execute(
+ """INSERT INTO model_config
+ (id,config_json,version,updated_at,updated_by) VALUES (1,?,1,?,?)""",
+ (
+ json.dumps(load_initial_config(), ensure_ascii=False),
+ now_text(),
+ admin["id"] if admin else None,
+ ),
+ )
+ db.commit()
+ return created
+
+ def authenticate(self, username: str, password: str) -> sqlite3.Row | None:
+ with self.connect() as db:
+ user = db.execute(
+ "SELECT * FROM users WHERE username = ? AND active = 1", (username,)
+ ).fetchone()
+ if user and verify_password(password, user["password_salt"], user["password_digest"]):
+ return user
+ return None
+
+ def create_token(
+ self, user_id: int, kind: str, device_name: str, lifetime: int
+ ) -> tuple[str, str]:
+ token = secrets.token_urlsafe(36)
+ csrf = secrets.token_urlsafe(24)
+ stamp = now_text()
+ with self.connect() as db:
+ db.execute(
+ """INSERT INTO auth_tokens
+ (user_id,token_digest,kind,csrf_token,device_name,created_at,expires_at,last_used_at)
+ VALUES (?,?,?,?,?,?,?,?)""",
+ (
+ user_id,
+ token_hash(token),
+ kind,
+ csrf,
+ str(device_name or "")[:120],
+ stamp,
+ int(time.time()) + lifetime,
+ stamp,
+ ),
+ )
+ db.execute("DELETE FROM auth_tokens WHERE expires_at < ?", (int(time.time()),))
+ db.commit()
+ return token, csrf
+
+ def session(self, token: str) -> sqlite3.Row | None:
+ if not token:
+ return None
+ with self.connect() as db:
+ row = db.execute(
+ """SELECT u.*, t.id AS token_id, t.kind AS token_kind,
+ t.csrf_token, t.expires_at
+ FROM auth_tokens t JOIN users u ON u.id=t.user_id
+ WHERE t.token_digest=? AND t.expires_at>=? AND u.active=1""",
+ (token_hash(token), int(time.time())),
+ ).fetchone()
+ if row:
+ db.execute(
+ "UPDATE auth_tokens SET last_used_at=? WHERE id=?",
+ (now_text(), row["token_id"]),
+ )
+ db.commit()
+ return row
+
+ def revoke(self, token: str) -> None:
+ if not token:
+ return
+ with self.connect() as db:
+ db.execute("DELETE FROM auth_tokens WHERE token_digest=?", (token_hash(token),))
+ db.commit()
+
+ def config(self) -> sqlite3.Row:
+ with self.connect() as db:
+ return db.execute(
+ """SELECT c.*, u.username AS updated_by_name
+ FROM model_config c LEFT JOIN users u ON u.id=c.updated_by WHERE c.id=1"""
+ ).fetchone()
+
+ def save_config(self, config: dict[str, Any], user_id: int, ip: str) -> int:
+ with self.connect() as db:
+ db.execute("BEGIN IMMEDIATE")
+ current = db.execute("SELECT version FROM model_config WHERE id=1").fetchone()
+ version = int(current["version"]) + 1
+ db.execute(
+ """UPDATE model_config SET config_json=?,version=?,updated_at=?,updated_by=?
+ WHERE id=1""",
+ (json.dumps(config, ensure_ascii=False), version, now_text(), user_id),
+ )
+ self._audit(db, user_id, "config.update", f"version={version}", ip)
+ db.commit()
+ return version
+
+ @staticmethod
+ def _audit(
+ db: sqlite3.Connection, user_id: int | None, action: str, detail: str, ip: str
+ ) -> None:
+ db.execute(
+ "INSERT INTO audit_log(user_id,action,detail,ip_address,created_at) VALUES(?,?,?,?,?)",
+ (user_id, action, detail[:500], ip[:80], now_text()),
+ )
+
+ def audit(self, user_id: int | None, action: str, detail: str, ip: str) -> None:
+ with self.connect() as db:
+ self._audit(db, user_id, action, detail, ip)
+ db.commit()
+
+ def list_users(self) -> list[sqlite3.Row]:
+ with self.connect() as db:
+ return list(db.execute("SELECT * FROM users ORDER BY id"))
+
+ def recent_audit(self, limit: int = 12) -> list[sqlite3.Row]:
+ with self.connect() as db:
+ return list(
+ db.execute(
+ """SELECT a.*, u.username FROM audit_log a
+ LEFT JOIN users u ON u.id=a.user_id ORDER BY a.id DESC LIMIT ?""",
+ (limit,),
+ )
+ )
+
+ def create_user(
+ self, username: str, password: str, role: str, actor_id: int, ip: str
+ ) -> None:
+ salt, digest = password_hash(password)
+ stamp = now_text()
+ with self.connect() as db:
+ db.execute(
+ """INSERT INTO users
+ (username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
+ VALUES(?,?,?,?,1,1,?,?)""",
+ (username, salt, digest, role, stamp, stamp),
+ )
+ self._audit(db, actor_id, "user.create", f"username={username}, role={role}", ip)
+ db.commit()
+
+ def update_user(
+ self,
+ user_id: int,
+ role: str,
+ active: bool,
+ new_password: str,
+ actor_id: int,
+ ip: str,
+ ) -> None:
+ with self.connect() as db:
+ target = db.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
+ if not target:
+ raise ValueError("用户不存在")
+ if user_id == actor_id and not active:
+ raise ValueError("不能停用当前登录账号")
+ if target["role"] == "admin" and (role != "admin" or not active):
+ count = db.execute(
+ "SELECT COUNT(*) FROM users WHERE role='admin' AND active=1"
+ ).fetchone()[0]
+ if count <= 1:
+ raise ValueError("系统至少需要保留一个启用的管理员")
+ values: list[Any] = [role, int(active), now_text()]
+ sql = "UPDATE users SET role=?,active=?,updated_at=?"
+ if new_password:
+ salt, digest = password_hash(new_password)
+ sql += ",password_salt=?,password_digest=?,must_change_password=1"
+ values.extend([salt, digest])
+ sql += " WHERE id=?"
+ values.append(user_id)
+ db.execute(sql, values)
+ if not active or new_password:
+ db.execute("DELETE FROM auth_tokens WHERE user_id=?", (user_id,))
+ self._audit(
+ db,
+ actor_id,
+ "user.update",
+ f"username={target['username']}, role={role}, active={int(active)}, password_reset={bool(new_password)}",
+ ip,
+ )
+ db.commit()
+
+ def change_password(self, user_id: int, current: str, new_password: str, ip: str) -> None:
+ with self.connect() as db:
+ user = db.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
+ if not user or not verify_password(
+ current, user["password_salt"], user["password_digest"]
+ ):
+ raise ValueError("当前密码不正确")
+ salt, digest = password_hash(new_password)
+ db.execute(
+ """UPDATE users SET password_salt=?,password_digest=?,must_change_password=0,
+ updated_at=? WHERE id=?""",
+ (salt, digest, now_text(), user_id),
+ )
+ self._audit(db, user_id, "password.change", "", ip)
+ db.commit()
+
+ def reset_admin_password(self, password: str) -> None:
+ salt, digest = password_hash(password)
+ with self.connect() as db:
+ result = db.execute(
+ """UPDATE users SET password_salt=?,password_digest=?,active=1,
+ must_change_password=1,updated_at=? WHERE username='admin'""",
+ (salt, digest, now_text()),
+ )
+ if result.rowcount == 0:
+ stamp = now_text()
+ db.execute(
+ """INSERT INTO users
+ (username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
+ VALUES('admin',?,?,'admin',1,1,?,?)""",
+ (salt, digest, stamp, stamp),
+ )
+ db.execute(
+ "DELETE FROM auth_tokens WHERE user_id=(SELECT id FROM users WHERE username='admin')"
+ )
+ db.commit()
+
+
+class LoginLimiter:
+ def __init__(self):
+ self._attempts: dict[str, deque[float]] = defaultdict(deque)
+ self._lock = threading.Lock()
+
+ def allowed(self, key: str) -> bool:
+ cutoff = time.time() - 300
+ with self._lock:
+ items = self._attempts[key]
+ while items and items[0] < cutoff:
+ items.popleft()
+ return len(items) < 8
+
+ def failure(self, key: str) -> None:
+ with self._lock:
+ self._attempts[key].append(time.time())
+
+ def success(self, key: str) -> None:
+ with self._lock:
+ self._attempts.pop(key, None)
+
+
+LOGIN_LIMITER = LoginLimiter()
+
+
+BASE_CSS = """
+:root{--bg:#f3f7f5;--surface:#fff;--ink:#17251f;--muted:#67786f;--line:#dce7e1;
+--accent:#0d9871;--deep:#102b21;--danger:#c94352;--soft:#e1f4ec}
+*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--ink);font-family:"Microsoft YaHei UI","Segoe UI",sans-serif}
+a{color:var(--accent);text-decoration:none}.shell{min-height:100vh;display:grid;grid-template-columns:250px 1fr}
+aside{background:var(--deep);color:white;padding:30px 24px;position:sticky;top:0;height:100vh}.brand{font-size:22px;font-weight:800;letter-spacing:.04em}
+.brand small{display:block;margin-top:8px;color:#9fc0b3;font-size:12px;font-weight:500}.userbox{margin-top:44px;padding:16px;background:#17392c;border:1px solid #2b5143;border-radius:14px}
+.role{display:inline-block;margin-top:8px;padding:4px 9px;background:#275643;border-radius:999px;color:#cce9de;font-size:12px}
+nav{margin-top:30px;display:grid;gap:8px}nav a{color:#c6d8d0;padding:10px 12px;border-radius:9px}nav a:hover{background:#1c4435;color:white}
+main{padding:34px;max-width:1450px;width:100%;margin:0 auto}.top{display:flex;justify-content:space-between;align-items:end;margin-bottom:24px}.top h1{font-size:30px;margin:0 0 7px}.muted{color:var(--muted);font-size:13px}
+.grid{display:grid;grid-template-columns:1.35fr .9fr;gap:20px;align-items:start}.card{background:var(--surface);border:1px solid var(--line);border-radius:17px;padding:22px;margin-bottom:20px;box-shadow:0 10px 30px rgba(17,51,38,.035)}
+.card h2{font-size:18px;margin:0 0 5px}.cardhead{display:flex;justify-content:space-between;gap:15px;align-items:start;margin-bottom:18px}.version{font:600 12px Consolas,monospace;color:var(--accent);background:var(--soft);padding:6px 9px;border-radius:8px}
+.formgrid{display:grid;grid-template-columns:1fr 1fr;gap:14px}.full{grid-column:1/-1}label{display:block;color:#53675d;font-size:13px;margin-bottom:6px}input,select,textarea{width:100%;border:1px solid #ceddd5;background:#fbfcfb;border-radius:9px;padding:10px 12px;color:var(--ink);font:inherit;outline:none}
+input:focus,select:focus,textarea:focus{border-color:var(--accent);box-shadow:0 0 0 3px #dff3eb}textarea{min-height:150px;font:13px Consolas,monospace;resize:vertical}.switches{display:grid;grid-template-columns:1fr 1fr;gap:10px 18px}.check{display:flex;align-items:center;gap:9px;color:var(--ink)}.check input{width:18px;height:18px;accent-color:var(--accent)}
+.actions{display:flex;justify-content:flex-end;margin-top:18px;gap:10px}button,.button{border:0;border-radius:9px;padding:10px 17px;font:600 14px inherit;cursor:pointer;background:var(--accent);color:white}button:hover{filter:brightness(.93)}button.secondary{background:#edf3f0;color:#28483b}.danger{color:var(--danger)}
+.flash{padding:12px 15px;border-radius:10px;margin-bottom:18px;background:var(--soft);color:#087656;border:1px solid #c9eadc}.flash.error{background:#fbeaec;color:var(--danger);border-color:#f2cbd1}
+.user{display:grid;grid-template-columns:1.1fr .8fr .6fr 1.1fr auto;gap:8px;align-items:end;padding:12px 0;border-top:1px solid #edf2ef}.user:first-of-type{border-top:0}.username{font-weight:700;padding:11px 0}.user input,.user select{padding:8px 9px}.tiny{font-size:11px;color:#829189}
+.audit{display:grid;gap:12px}.event{border-left:3px solid #b8ddce;padding-left:11px}.event b{font-size:13px}.event div{font-size:11px;color:#829189;margin-top:3px}
+.loginwrap{min-height:100vh;display:grid;place-items:center;padding:24px}.login{width:min(430px,100%);background:white;border:1px solid var(--line);border-radius:22px;padding:34px;box-shadow:0 25px 70px rgba(12,48,34,.12)}.login h1{margin:0 0 8px;font-size:27px}.login form{display:grid;gap:15px;margin-top:26px}.login button{margin-top:5px;padding:12px}.notice{background:#fff7e8;border:1px solid #f0d6a5;color:#8a5a0a;padding:13px;border-radius:10px;font-size:13px;margin-bottom:18px}
+@media(max-width:980px){.shell{grid-template-columns:1fr}aside{height:auto;position:static}.grid{grid-template-columns:1fr}.user{grid-template-columns:1fr 1fr}.user .actions{grid-column:1/-1}.formgrid{grid-template-columns:1fr}.full{grid-column:auto}main{padding:20px}}
+"""
+
+
+def page(title: str, body: str) -> str:
+ return (
+ ""
+ ""
+ f"{html.escape(title)}{body}"
+ )
+
+
+class AdminServer(ThreadingHTTPServer):
+ daemon_threads = True
+
+ def __init__(self, address: tuple[str, int], database: Database):
+ self.local_sync_token = secrets.token_urlsafe(36)
+ super().__init__(address, AdminHandler)
+ self.database = database
+
+
+def _address_in_use(exc: OSError) -> bool:
+ """兼容 Windows、Linux 和 macOS 的端口占用错误码。"""
+ # Windows 在独占地址被另一个进程监听时，可能返回 10013 而不是 10048。
+ return exc.errno in (13, 48, 98, 10013, 10048) or getattr(
+ exc, "winerror", None
+ ) in (10013, 10048)
+
+
+def create_server(
+ host: str,
+ preferred_port: int,
+ database: Database,
+ *,
+ max_attempts: int = DEFAULT_PORT_ATTEMPTS,
+) -> tuple[AdminServer, int]:
+ """创建服务；首选端口被占用时自动向后尝试可用端口。"""
+ if not 0 <= preferred_port <= 65535:
+ raise ValueError("端口必须在 0-65535 之间")
+ if max_attempts < 1:
+ raise ValueError("端口尝试次数必须大于 0")
+ if preferred_port == 0:
+ server = AdminServer((host, 0), database)
+ return server, int(server.server_address[1])
+
+ last_error: OSError | None = None
+ for port in range(preferred_port, min(65535, preferred_port + max_attempts - 1) + 1):
+ try:
+ return AdminServer((host, port), database), port
+ except OSError as exc:
+ if not _address_in_use(exc):
+ raise
+ last_error = exc
+ raise OSError(
+ f"端口 {preferred_port}-{min(65535, preferred_port + max_attempts - 1)} 均被占用"
+ ) from last_error
+
+
+def public_server_url(host: str, port: int) -> str:
+ display_host = "127.0.0.1" if host in ("0.0.0.0", "::", "") else host
+ if ":" in display_host and not display_host.startswith("["):
+ display_host = f"[{display_host}]"
+ return f"http://{display_host}:{port}"
+
+
+def write_runtime_info(
+ path: Path,
+ host: str,
+ port: int,
+ *,
+ local_sync_token: str = "",
+) -> dict[str, Any]:
+ """发布实际监听端口，供同项目桌面端自动发现。"""
+ info = {
+ "pid": os.getpid(),
+ "host": host,
+ "port": int(port),
+ "server_url": public_server_url(host, port),
+ "local_sync_token": local_sync_token,
+ "started_at": now_text(),
+ }
+ path = path.resolve()
+ path.parent.mkdir(parents=True, exist_ok=True)
+ temporary = path.with_suffix(path.suffix + ".tmp")
+ temporary.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8")
+ os.replace(temporary, path)
+ return info
+
+
+def clear_runtime_info(path: Path, port: int) -> None:
+ """只清理由当前进程写入的发现文件，避免影响另一个后台实例。"""
+ path = path.resolve()
+ try:
+ info = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, ValueError, TypeError):
+ return
+ if info.get("pid") == os.getpid() and info.get("port") == int(port):
+ try:
+ path.unlink()
+ except FileNotFoundError:
+ pass
+
+
+class AdminHandler(BaseHTTPRequestHandler):
+ server_version = "WeComConfig/1.0"
+
+ @property
+ def db(self) -> Database:
+ return self.server.database # type: ignore[attr-defined]
+
+ @property
+ def client_ip(self) -> str:
+ forwarded = self.headers.get("X-Forwarded-For", "").split(",", 1)[0].strip()
+ return forwarded or self.client_address[0]
+
+ def log_message(self, format_string: str, *args: Any) -> None:
+ sys.stdout.write(
+ f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {self.client_ip} {format_string % args}\n"
+ )
+
+ def _security_headers(self) -> None:
+ self.send_header("X-Content-Type-Options", "nosniff")
+ self.send_header("X-Frame-Options", "DENY")
+ self.send_header("Referrer-Policy", "no-referrer")
+ self.send_header("Cache-Control", "no-store")
+ self.send_header(
+ "Content-Security-Policy",
+ "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; base-uri 'none'; frame-ancestors 'none'",
+ )
+
+ def _send(self, status: int, content: str, content_type: str) -> None:
+ data = content.encode("utf-8")
+ self.send_response(status)
+ self._security_headers()
+ self.send_header("Content-Type", content_type)
+ self.send_header("Content-Length", str(len(data)))
+ self.end_headers()
+ self.wfile.write(data)
+
+ def json_response(self, status: int, data: dict[str, Any]) -> None:
+ self._send(
+ status,
+ json.dumps(data, ensure_ascii=False),
+ "application/json; charset=utf-8",
+ )
+
+ def html_response(self, status: int, content: str) -> None:
+ self._send(status, content, "text/html; charset=utf-8")
+
+ def redirect(self, location: str, *, cookie: str = "") -> None:
+ self.send_response(HTTPStatus.SEE_OTHER)
+ self._security_headers()
+ self.send_header("Location", location)
+ if cookie:
+ self.send_header("Set-Cookie", cookie)
+ self.send_header("Content-Length", "0")
+ self.end_headers()
+
+ def body_bytes(self, limit: int = 512_000) -> bytes:
+ try:
+ length = int(self.headers.get("Content-Length", "0"))
+ except ValueError:
+ raise ValueError("Content-Length 无效")
+ if length < 0 or length > limit:
+ raise ValueError("请求内容过大")
+ return self.rfile.read(length)
+
+ def json_body(self) -> dict[str, Any]:
+ try:
+ data = json.loads(self.body_bytes().decode("utf-8") or "{}")
+ except (UnicodeDecodeError, ValueError) as exc:
+ raise ValueError("请求 JSON 无效") from exc
+ if not isinstance(data, dict):
+ raise ValueError("请求 JSON 必须是对象")
+ return data
+
+ def form_body(self) -> dict[str, str]:
+ try:
+ parsed = urllib.parse.parse_qs(
+ self.body_bytes().decode("utf-8"), keep_blank_values=True
+ )
+ except UnicodeDecodeError as exc:
+ raise ValueError("表单编码无效") from exc
+ return {key: values[-1] for key, values in parsed.items()}
+
+ def bearer_token(self) -> str:
+ authorization = self.headers.get("Authorization", "")
+ if authorization.startswith("Bearer "):
+ return authorization[7:].strip()
+ return ""
+
+ def cookie_token(self) -> str:
+ cookie = SimpleCookie()
+ try:
+ cookie.load(self.headers.get("Cookie", ""))
+ except Exception:
+ return ""
+ return cookie.get("session_token").value if cookie.get("session_token") else ""
+
+ def auth(self, api: bool = False) -> tuple[sqlite3.Row | None, str]:
+ token = self.bearer_token() if api else self.cookie_token()
+ return self.db.session(token), token
+
+ @staticmethod
+ def valid_password(value: str) -> bool:
+ return len(value) >= 10 and any(c.isalpha() for c in value) and any(
+ c.isdigit() for c in value
+ )
+
+ def require_web_auth(
+ self, *, roles: tuple[str, ...] | None = None, allow_password_change: bool = False
+ ) -> tuple[sqlite3.Row, str] | None:
+ user, token = self.auth(False)
+ if not user:
+ self.redirect("/login")
+ return None
+ if user["must_change_password"] and not allow_password_change:
+ self.redirect("/?error=" + urllib.parse.quote("首次登录请先修改密码"))
+ return None
+ if roles and user["role"] not in roles:
+ self.redirect("/?error=" + urllib.parse.quote("当前角色没有执行此操作的权限"))
+ return None
+ return user, token
+
+ def require_api_auth(self) -> sqlite3.Row | None:
+ user, _ = self.auth(True)
+ if not user:
+ self.json_response(HTTPStatus.UNAUTHORIZED, {"error": "登录已失效，请重新登录"})
+ return None
+ if user["must_change_password"]:
+ self.json_response(
+ HTTPStatus.FORBIDDEN, {"error": "请先在后台网页修改初始密码"}
+ )
+ return None
+ return user
+
+ def local_sync_authorized(self) -> bool:
+ """允许同机桌面端使用运行时随机凭证只读模型配置。"""
+ supplied = self.headers.get("X-Desktop-Sync-Token", "")
+ expected = getattr(self.server, "local_sync_token", "")
+ if not supplied or not expected:
+ return False
+ try:
+ is_loopback = ipaddress.ip_address(self.client_address[0]).is_loopback
+ except ValueError:
+ return False
+ return is_loopback and hmac.compare_digest(supplied, expected)
+
+ @staticmethod
+ def csrf_ok(user: sqlite3.Row, form: dict[str, str]) -> bool:
+ return hmac.compare_digest(str(user["csrf_token"]), str(form.get("csrf", "")))
+
+ def do_GET(self) -> None:
+ path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+ if path == "/health":
+ self.json_response(HTTPStatus.OK, {"status": "ok", "time": now_text()})
+ elif path == "/login":
+ user, _ = self.auth(False)
+ if user:
+ self.redirect("/")
+ else:
+ self.render_login()
+ elif path == "/":
+ auth = self.require_web_auth(allow_password_change=True)
+ if auth:
+ self.render_dashboard(auth[0])
+ elif path == "/api/v1/me":
+ user = self.require_api_auth()
+ if user:
+ self.json_response(HTTPStatus.OK, {"user": self.public_user(user)})
+ elif path == "/api/v1/config":
+ local_sync = self.local_sync_authorized()
+ user = None if local_sync else self.require_api_auth()
+ if local_sync or user:
+ row = self.db.config()
+ self.json_response(
+ HTTPStatus.OK,
+ {
+ "version": row["version"],
+ "updated_at": row["updated_at"],
+ "updated_by": row["updated_by_name"] or "system",
+ "config": json.loads(row["config_json"]),
+ },
+ )
+ else:
+ self.json_response(HTTPStatus.NOT_FOUND, {"error": "页面不存在"})
+
+ def do_POST(self) -> None:
+ path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+ try:
+ if path == "/login":
+ self.web_login()
+ elif path == "/logout":
+ self.web_logout()
+ elif path == "/admin/config":
+ self.web_save_config()
+ elif path == "/admin/users/create":
+ self.web_create_user()
+ elif path == "/admin/users/update":
+ self.web_update_user()
+ elif path == "/admin/password":
+ self.web_change_password()
+ elif path == "/api/v1/auth/login":
+ self.api_login()
+ elif path == "/api/v1/auth/logout":
+ self.api_logout()
+ else:
+ self.json_response(HTTPStatus.NOT_FOUND, {"error": "接口不存在"})
+ except ValueError as exc:
+ if path.startswith("/api/"):
+ self.json_response(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ else:
+ self.redirect("/?error=" + urllib.parse.quote(str(exc)))
+ except sqlite3.IntegrityError:
+ if path.startswith("/api/"):
+ self.json_response(HTTPStatus.CONFLICT, {"error": "用户名已存在"})
+ else:
+ self.redirect("/?error=" + urllib.parse.quote("用户名已存在"))
+
+ @staticmethod
+ def public_user(user: sqlite3.Row) -> dict[str, Any]:
+ return {
+ "id": user["id"],
+ "username": user["username"],
+ "role": user["role"],
+ "role_name": ROLE_LABELS[user["role"]],
+ }
+
+ def render_login(self, error: str = "") -> None:
+ message = (
+ f"{html.escape(error)}
" if error else ""
+ )
+ body = f"""
+
+ ZHEN AI ADMIN
+ 配置管理后台
统一管理桌面端账号、角色和 AI 模型参数。
+ {message}
+
+ """
+ self.html_response(HTTPStatus.OK, page("登录 · 配置管理后台", body))
+
+ def web_login(self) -> None:
+ form = self.form_body()
+ username = form.get("username", "").strip()
+ key = f"{self.client_ip}:{username.lower()}"
+ if not LOGIN_LIMITER.allowed(key):
+ self.render_login("登录失败次数过多，请 5 分钟后再试")
+ return
+ user = self.db.authenticate(username, form.get("password", ""))
+ if not user:
+ LOGIN_LIMITER.failure(key)
+ self.db.audit(None, "login.failed", f"username={username}", self.client_ip)
+ self.render_login("用户名或密码不正确")
+ return
+ LOGIN_LIMITER.success(key)
+ token, _ = self.db.create_token(user["id"], "web", "browser", 12 * 3600)
+ self.db.audit(user["id"], "login.web", "", self.client_ip)
+ cookie = f"session_token={token}; Path=/; HttpOnly; SameSite=Strict; Max-Age=43200"
+ self.redirect("/", cookie=cookie)
+
+ def web_logout(self) -> None:
+ user, token = self.auth(False)
+ if user:
+ try:
+ form = self.form_body()
+ except ValueError:
+ form = {}
+ if self.csrf_ok(user, form):
+ self.db.revoke(token)
+ self.redirect(
+ "/login", cookie="session_token=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0"
+ )
+
+ def api_login(self) -> None:
+ data = self.json_body()
+ username = str(data.get("username") or "").strip()
+ key = f"{self.client_ip}:{username.lower()}"
+ if not LOGIN_LIMITER.allowed(key):
+ self.json_response(
+ HTTPStatus.TOO_MANY_REQUESTS, {"error": "登录失败次数过多，请稍后再试"}
+ )
+ return
+ user = self.db.authenticate(username, str(data.get("password") or ""))
+ if not user:
+ LOGIN_LIMITER.failure(key)
+ self.db.audit(None, "login.failed", f"username={username}", self.client_ip)
+ self.json_response(HTTPStatus.UNAUTHORIZED, {"error": "用户名或密码不正确"})
+ return
+ if user["must_change_password"]:
+ self.json_response(
+ HTTPStatus.FORBIDDEN, {"error": "首次登录请先在后台网页修改密码"}
+ )
+ return
+ LOGIN_LIMITER.success(key)
+ token, _ = self.db.create_token(
+ user["id"], "api", str(data.get("device_name") or "desktop"), 30 * 86400
+ )
+ self.db.audit(user["id"], "login.api", str(data.get("device_name") or ""), self.client_ip)
+ self.json_response(
+ HTTPStatus.OK,
+ {"access_token": token, "expires_in": 30 * 86400, "user": self.public_user(user)},
+ )
+
+ def api_logout(self) -> None:
+ user, token = self.auth(True)
+ if user:
+ self.db.revoke(token)
+ self.db.audit(user["id"], "logout.api", "", self.client_ip)
+ self.json_response(HTTPStatus.OK, {"ok": True})
+
+ def web_save_config(self) -> None:
+ auth = self.require_web_auth(roles=("admin", "operator"))
+ if not auth:
+ return
+ user, _ = auth
+ form = self.form_body()
+ if not self.csrf_ok(user, form):
+ raise ValueError("页面已过期，请刷新后重试")
+ current = json.loads(self.db.config()["config_json"])
+ config = validate_config_form(form, current)
+ version = self.db.save_config(config, user["id"], self.client_ip)
+ self.redirect("/?message=" + urllib.parse.quote(f"配置已发布为 v{version}"))
+
+ def web_create_user(self) -> None:
+ auth = self.require_web_auth(roles=("admin",))
+ if not auth:
+ return
+ user, _ = auth
+ form = self.form_body()
+ if not self.csrf_ok(user, form):
+ raise ValueError("页面已过期，请刷新后重试")
+ username = form.get("username", "").strip()
+ password = form.get("password", "")
+ role = form.get("role", "viewer")
+ if not (3 <= len(username) <= 40) or not all(
+ char.isalnum() or char in "_.-" for char in username
+ ):
+ raise ValueError("用户名需为 3-40 位字母、数字、点、下划线或短横线")
+ if role not in ROLES:
+ raise ValueError("角色无效")
+ if not self.valid_password(password):
+ raise ValueError("初始密码至少 10 位，并同时包含字母和数字")
+ self.db.create_user(username, password, role, user["id"], self.client_ip)
+ self.redirect("/?message=" + urllib.parse.quote(f"用户 {username} 已创建"))
+
+ def web_update_user(self) -> None:
+ auth = self.require_web_auth(roles=("admin",))
+ if not auth:
+ return
+ user, _ = auth
+ form = self.form_body()
+ if not self.csrf_ok(user, form):
+ raise ValueError("页面已过期，请刷新后重试")
+ try:
+ user_id = int(form.get("user_id", "0"))
+ except ValueError as exc:
+ raise ValueError("用户编号无效") from exc
+ role = form.get("role", "viewer")
+ password = form.get("new_password", "")
+ if role not in ROLES:
+ raise ValueError("角色无效")
+ if password and not self.valid_password(password):
+ raise ValueError("重置密码至少 10 位，并同时包含字母和数字")
+ self.db.update_user(
+ user_id,
+ role,
+ form.get("active") == "1",
+ password,
+ user["id"],
+ self.client_ip,
+ )
+ self.redirect("/?message=" + urllib.parse.quote("用户资料已更新"))
+
+ def web_change_password(self) -> None:
+ auth = self.require_web_auth(allow_password_change=True)
+ if not auth:
+ return
+ user, _ = auth
+ form = self.form_body()
+ if not self.csrf_ok(user, form):
+ raise ValueError("页面已过期，请刷新后重试")
+ new_password = form.get("new_password", "")
+ if new_password != form.get("confirm_password", ""):
+ raise ValueError("两次输入的新密码不一致")
+ if not self.valid_password(new_password):
+ raise ValueError("新密码至少 10 位，并同时包含字母和数字")
+ self.db.change_password(
+ user["id"], form.get("current_password", ""), new_password, self.client_ip
+ )
+ self.redirect("/?message=" + urllib.parse.quote("密码已修改"))
+
+ def render_dashboard(self, user: sqlite3.Row) -> None:
+ query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
+ message = query.get("message", [""])[-1]
+ error = query.get("error", [""])[-1]
+ flash = ""
+ if message:
+ flash = f"{html.escape(message)}
"
+ elif error:
+ flash = f"{html.escape(error)}
"
+ csrf = html.escape(user["csrf_token"], quote=True)
+ password_card = self.password_card(user, csrf)
+ if user["must_change_password"]:
+ content = f"""
+ {self.sidebar(user, csrf)}
+
+ {flash}{password_card}
+ {self.sidebar(user, csrf)}
+ 模型配置中心
保存后，已登录桌面端会在启动或定时同步时自动应用。
+
CURRENT · v{int(row['version'])}
{users_card}{password_card}{audit_card}
+ 当前仍在使用初始密码，必须修改后才能让桌面端登录。
"
+ if user["must_change_password"]
+ else ""
+ )
+ return f"""
+ """
+
+ @staticmethod
+ def config_card(
+ user: sqlite3.Row, csrf: str, row: sqlite3.Row, config: dict[str, Any]
+ ) -> str:
+ can_edit = user["role"] in ("admin", "operator")
+ esc = lambda key: html.escape(str(config.get(key, "")), quote=True)
+ checked = lambda key: " checked" if config.get(key) else ""
+ disabled = " disabled" if not can_edit else ""
+ mcp = html.escape(
+ json.dumps(config.get("AI_MCP_SERVERS", []), ensure_ascii=False, indent=2)
+ )
+ submit = (
+ ""
+ if can_edit
+ else "当前为只读角色，可查看配置但不能修改。
"
+ )
+ return f"""
+
+ 最后更新：{html.escape(row['updated_at'])} · {html.escape(row['updated_by_name'] or 'system')}
"""
+
+ def users_card(self, user: sqlite3.Row, csrf: str) -> str:
+ if user["role"] != "admin":
+ return f""""""
+ items = []
+ role_options = lambda current: "".join(
+ f""
+ for role in ROLES
+ )
+ for item in self.db.list_users():
+ active = " checked" if item["active"] else ""
+ items.append(
+ f""""""
+ )
+ return f"""
+ 用户与角色
管理员管理全部功能，配置员可发布配置，只读用户仅同步。
新建用户
+ """
+
+ def audit_card(self) -> str:
+ events = []
+ for item in self.db.recent_audit():
+ events.append(
+ f"{html.escape(item['action'])} · {html.escape(item['username'] or 'anonymous')}"
+ f"
{html.escape(item['created_at'])} · {html.escape(item['detail'])}
"
+ )
+ return f"""{''.join(events) or '
暂无记录
'}
"
+ "AI 客服网页在测试模式下未加载
"
+ )
+
+ def _install_light_theme(self) -> None:
+ css_path = SCRIPT_DIR / "edge_light_theme" / "light-theme.css"
+ try:
+ css = css_path.read_text(encoding="utf-8")
+ except OSError:
+ css = "html,body,#app{background:#f8faf9!important;color:#17251f!important}"
+ source = f"""
+ (() => {{
+ const css = {json.dumps(css)};
+ const applyTheme = () => {{
+ let style = document.getElementById('zhen-qt-light-theme');
+ if (!style) {{
+ style = document.createElement('style');
+ style.id = 'zhen-qt-light-theme';
+ (document.head || document.documentElement).appendChild(style);
+ }}
+ style.textContent = css;
+ }};
+ applyTheme();
+ new MutationObserver(applyTheme).observe(document.documentElement, {{childList:true}});
+ }})();
+ """
+ script = QWebEngineScript()
+ script.setName("zhen-light-theme")
+ script.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentReady)
+ script.setWorldId(QWebEngineScript.ScriptWorldId.MainWorld)
+ script.setRunsOnSubFrames(True)
+ script.setSourceCode(source)
+ self.profile.scripts().insert(script)
+
+ def _load_started(self) -> None:
+ self.progress.setValue(2)
+ self.progress.show()
+
+ def _load_finished(self, _ok: bool) -> None:
+ self.progress.hide()
+ self.view.setFocus(Qt.OtherFocusReason)
+
+ def reload(self) -> None:
+ self.view.reload()
+
+ def open_external(self) -> None:
+ QDesktopServices.openUrl(self.view.url() or QUrl(CUSTOMER_SERVICE_URL))
+
+
+class MetricCard(QFrame):
+ def __init__(self, label: str, value: str, meta: str):
+ super().__init__()
+ self.setObjectName("MetricCard")
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(19, 17, 19, 17)
+ layout.setSpacing(6)
+ layout.addWidget(_label(label, "MetricLabel"))
+ self.value = _label(value, "MetricValue")
+ layout.addWidget(self.value)
+ layout.addWidget(_label(meta, "MetricMeta"))
+
+
+class DashboardPage(QScrollArea):
+ startRequested = Signal()
+ stopRequested = Signal()
+ pageRequested = Signal(int)
+
+ def __init__(self):
+ super().__init__()
+ self.setObjectName("PageScroll")
+ self.setWidgetResizable(True)
+ self.setFrameShape(QFrame.NoFrame)
+ root = QWidget()
+ root.setObjectName("PageRoot")
+ self.setWidget(root)
+ layout = QVBoxLayout(root)
+ layout.setContentsMargins(28, 25, 28, 28)
+ layout.setSpacing(18)
+ layout.addLayout(
+ _page_header(
+ "02",
+ "自动回复控制台",
+ "监控企业微信窗口、AI 决策和客户沉淀状态。",
+ )
+ )
+
+ hero = QFrame()
+ hero.setObjectName("HeroCard")
+ hero_layout = QHBoxLayout(hero)
+ hero_layout.setContentsMargins(25, 22, 25, 22)
+ hero_layout.setSpacing(20)
+ status_box = QVBoxLayout()
+ status_box.setSpacing(5)
+ status_box.addWidget(_label("WECOM LIVE MONITOR", "Eyebrow"))
+ self.status = _label("待命", "HeroStatus")
+ status_box.addWidget(self.status)
+ self.status_hint = _label("确认企业微信已登录后即可启动。", "HeroHint")
+ status_box.addWidget(self.status_hint)
+ self.timer = _label("运行时长 --:--:--", "CardSubtitle")
+ status_box.addWidget(self.timer)
+ hero_layout.addLayout(status_box, 1)
+ self.hero_button = _button("开始监听", "primary")
+ self.hero_button.setMinimumWidth(128)
+ self.hero_button.clicked.connect(self.startRequested)
+ hero_layout.addWidget(self.hero_button, 0, Qt.AlignVCenter)
+ layout.addWidget(hero)
+
+ metrics = QGridLayout()
+ metrics.setSpacing(12)
+ self.replied = MetricCard("本次已回复", "0", "自动回复数量")
+ self.false_pos = MetricCard("识别误判", "0", "已跳过的会话")
+ self.registration = MetricCard("待处理登记", "0", "挂号与回访线索")
+ self.sessions = MetricCard("会话档案", "0", "已保存上下文")
+ for column, card in enumerate(
+ (self.replied, self.false_pos, self.registration, self.sessions)
+ ):
+ metrics.addWidget(card, 0, column)
+ metrics.setColumnStretch(column, 1)
+ layout.addLayout(metrics)
+
+ workflow, workflow_layout = _card(
+ "实际工作流",
+ "每一步都映射到当前软件已经具备的执行能力。",
+ )
+ steps = (
+ ("A1", "消息监听", "未读识别 → 会话提取 → 自动回复", 2),
+ ("A2", "AI 决策", "上下文记忆 → 医疗客服人格 → MCP 工具", 4),
+ ("A3", "业务沉淀", "挂号登记 → 会话档案 → 运行记录", 3),
+ )
+ for code, title, text, page_index in steps:
+ row = QFrame()
+ row.setStyleSheet(
+ "QFrame{background:#F8FAF9;border:1px solid #E4ECE8;border-radius:11px;}"
+ )
+ row_layout = QHBoxLayout(row)
+ row_layout.setContentsMargins(15, 12, 15, 12)
+ badge = _label(code, "Eyebrow")
+ badge.setFixedWidth(38)
+ row_layout.addWidget(badge)
+ copy = QVBoxLayout()
+ copy.setSpacing(2)
+ copy.addWidget(_label(title, "CardTitle"))
+ copy.addWidget(_label(text, "CardSubtitle"))
+ row_layout.addLayout(copy, 1)
+ action = _button("打开")
+ action.clicked.connect(lambda _checked=False, i=page_index: self.pageRequested.emit(i))
+ row_layout.addWidget(action)
+ workflow_layout.addWidget(row)
+ layout.addWidget(workflow)
+ layout.addStretch(1)
+
+ def set_status(self, state: str, text: str, hint: str) -> None:
+ colors = {
+ "running": COLORS["success"],
+ "waiting": COLORS["warning"],
+ "connecting": COLORS["warning"],
+ "error": COLORS["danger"],
+ "stopping": COLORS["warning"],
+ "stopped": COLORS["ink"],
+ }
+ self.status.setText(text)
+ self.status.setStyleSheet(
+ f"color:{colors.get(state, COLORS['ink'])};font-size:30px;font-weight:700;"
+ )
+ self.status_hint.setText(hint)
+ active = state in {"running", "waiting", "connecting", "stopping"}
+ self.hero_button.setText("监听运行中" if active else "开始监听")
+ self.hero_button.setEnabled(not active)
+
+
+class SettingsPage(QScrollArea):
+ saved = Signal(dict)
+
+ def __init__(self, settings: dict):
+ super().__init__()
+ self.setObjectName("PageScroll")
+ self.setWidgetResizable(True)
+ self.setFrameShape(QFrame.NoFrame)
+ root = QWidget()
+ root.setObjectName("PageRoot")
+ self.setWidget(root)
+ layout = QVBoxLayout(root)
+ layout.setContentsMargins(28, 25, 28, 28)
+ layout.setSpacing(18)
+ layout.addLayout(_page_header("03", "通用设置", "修改后自动保存，并在下次监听时生效。"))
+
+ runtime, runtime_layout = _card(
+ "运行参数", "参数直接参与企业微信监听，不是静态展示。"
+ )
+ runtime_grid = QGridLayout()
+ runtime_grid.setHorizontalSpacing(18)
+ runtime_grid.setVerticalSpacing(8)
+ runtime_grid.addWidget(_label("固定回复（关闭 AI 时使用）", "CardSubtitle"), 0, 0, 1, 2)
+ self.reply = QLineEdit(str(settings["auto_reply_text"]))
+ self.reply.setPlaceholderText("请输入固定回复")
+ runtime_grid.addWidget(self.reply, 1, 0, 1, 2)
+ runtime_grid.addWidget(_label("轮询间隔（秒）", "CardSubtitle"), 2, 0)
+ runtime_grid.addWidget(_label("鼠标静止等待（秒）", "CardSubtitle"), 2, 1)
+ self.poll = QDoubleSpinBox()
+ self.poll.setRange(0.2, 300.0)
+ self.poll.setDecimals(1)
+ self.poll.setValue(float(settings["poll_interval"]))
+ self.idle_seconds = QDoubleSpinBox()
+ self.idle_seconds.setRange(0.0, 3600.0)
+ self.idle_seconds.setDecimals(1)
+ self.idle_seconds.setValue(float(settings["mouse_idle_seconds"]))
+ runtime_grid.addWidget(self.poll, 3, 0)
+ runtime_grid.addWidget(self.idle_seconds, 3, 1)
+ runtime_grid.setColumnStretch(0, 1)
+ runtime_grid.setColumnStretch(1, 1)
+ runtime_layout.addLayout(runtime_grid)
+ self.save_status = _label("已加载保存配置", "SuccessText")
+ self.save_status.setAlignment(Qt.AlignRight)
+ runtime_layout.addWidget(self.save_status)
+ layout.addWidget(runtime)
+
+ safety, safety_layout = _card(
+ "人机共存",
+ "检测到用户正在操作时自动等待，避免与人工客服争抢鼠标。",
+ )
+ self.mouse_idle = QCheckBox("启用鼠标空闲检测")
+ self.mouse_idle.setChecked(bool(settings["mouse_idle_enabled"]))
+ safety_layout.addWidget(self.mouse_idle)
+ safety_layout.addWidget(
+ _label(
+ "企业微信未显示或最小化时，机器人会尝试还原到前台；本轮未就绪会暂停并自动重试。",
+ "CardSubtitle",
+ )
+ )
+ layout.addWidget(safety)
+
+ checks, checks_layout = _card("启动前检查")
+ for index, text in enumerate(
+ (
+ "企业微信 PC 端已登录，且账号状态正常。",
+ "如启用 AI，API 地址、密钥与模型配置已经保存。",
+ "双屏环境可以使用不同缩放比例，Qt 会跟随当前屏幕自动缩放。",
+ ),
+ start=1,
+ ):
+ row = QHBoxLayout()
+ badge = _label(f"{index:02d}", "Eyebrow")
+ badge.setFixedWidth(32)
+ row.addWidget(badge)
+ row.addWidget(_label(text, "CardSubtitle"), 1)
+ checks_layout.addLayout(row)
+ layout.addWidget(checks)
+ layout.addStretch(1)
+
+ self.save_timer = QTimer(self)
+ self.save_timer.setSingleShot(True)
+ self.save_timer.setInterval(500)
+ self.save_timer.timeout.connect(self._emit_save)
+ self.reply.textChanged.connect(self._schedule_save)
+ self.poll.valueChanged.connect(self._schedule_save)
+ self.idle_seconds.valueChanged.connect(self._schedule_save)
+ self.mouse_idle.toggled.connect(self._schedule_save)
+
+ def values(self) -> dict:
+ return {
+ "auto_reply_text": self.reply.text().strip() or "你好",
+ "poll_interval": self.poll.value(),
+ "mouse_idle_enabled": self.mouse_idle.isChecked(),
+ "mouse_idle_seconds": self.idle_seconds.value(),
+ }
+
+ def _schedule_save(self, *_args) -> None:
+ self.save_status.setText("正在保存…")
+ self.save_status.setObjectName("WarningText")
+ self.save_status.style().unpolish(self.save_status)
+ self.save_status.style().polish(self.save_status)
+ self.save_timer.start()
+
+ def _emit_save(self) -> None:
+ self.saved.emit(self.values())
+
+ def mark_saved(self, ok: bool, message: str) -> None:
+ self.save_status.setText(message)
+ self.save_status.setObjectName("SuccessText" if ok else "DangerText")
+ self.save_status.style().unpolish(self.save_status)
+ self.save_status.style().polish(self.save_status)
+
+
+class SessionDetailDialog(QDialog):
+ def __init__(
+ self,
+ session_id: str,
+ history: list[dict],
+ last_lines: list[str],
+ parent: QWidget | None = None,
+ ):
+ super().__init__(parent)
+ self.setObjectName("SessionDetailDialog")
+ self.setWindowTitle("客户会话记录")
+ self.setModal(True)
+ self.resize(920, 720)
+ self.setMinimumSize(700, 520)
+ self.session_id = session_id
+ self.history = history
+ self.last_lines = last_lines
+
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(24, 22, 24, 20)
+ layout.setSpacing(16)
+
+ header = QHBoxLayout()
+ title_box = QVBoxLayout()
+ title_box.setSpacing(4)
+ title_box.addWidget(_label("客户会话记录", "PageTitle"))
+ title_box.addWidget(
+ _label(
+ f"会话 ID {session_id} · 共 {len(history)} 条消息",
+ "PageSubtitle",
+ )
+ )
+ header.addLayout(title_box, 1)
+ copy_button = _button("复制全部记录")
+ copy_button.clicked.connect(self._copy_all)
+ header.addWidget(copy_button, 0, Qt.AlignBottom)
+ layout.addLayout(header)
+
+ scroll = QScrollArea()
+ scroll.setWidgetResizable(True)
+ scroll.setFrameShape(QFrame.NoFrame)
+ scroll.setStyleSheet(
+ "QScrollArea{background:#EEF3F0;border:1px solid #DDE7E1;border-radius:15px;}"
+ "QScrollArea>QWidget>QWidget{background:#EEF3F0;}"
+ )
+ message_root = QWidget()
+ message_layout = QVBoxLayout(message_root)
+ message_layout.setContentsMargins(20, 20, 20, 20)
+ message_layout.setSpacing(13)
+ scroll.setWidget(message_root)
+
+ if history:
+ for message in history:
+ self._add_message(message_layout, message)
+ elif last_lines:
+ self._add_message(
+ message_layout,
+ {
+ "role": "system",
+ "content": "仅保留了最近画面快照：\n\n" + "\n".join(last_lines),
+ "ts": None,
+ },
+ )
+ else:
+ empty = _label("这份会话档案暂时没有可显示的消息。", "PageSubtitle")
+ empty.setAlignment(Qt.AlignCenter)
+ message_layout.addWidget(empty)
+ message_layout.addStretch(1)
+ layout.addWidget(scroll, 1)
+
+ footer = QHBoxLayout()
+ footer.addWidget(_label("按时间顺序展示已保存的客户与客服消息。", "CardSubtitle"), 1)
+ close_button = _button("关闭")
+ close_button.clicked.connect(self.accept)
+ footer.addWidget(close_button)
+ layout.addLayout(footer)
+ QTimer.singleShot(0, lambda: scroll.verticalScrollBar().setValue(0))
+
+ @staticmethod
+ def _message_time(value) -> str:
+ try:
+ return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(value)))
+ except (TypeError, ValueError, OSError):
+ return ""
+
+ def _add_message(self, parent: QVBoxLayout, message: dict) -> None:
+ role = str(message.get("role") or "system").lower()
+ role_name = {"user": "客户", "assistant": "客服", "system": "系统记录"}.get(
+ role, role
+ )
+ bubble_name = {
+ "user": "CustomerBubble",
+ "assistant": "AssistantBubble",
+ }.get(role, "SystemBubble")
+ bubble = QFrame()
+ bubble.setObjectName(bubble_name)
+ bubble.setMaximumWidth(690)
+ bubble_layout = QVBoxLayout(bubble)
+ bubble_layout.setContentsMargins(15, 12, 15, 13)
+ bubble_layout.setSpacing(6)
+ meta = QHBoxLayout()
+ meta.setSpacing(10)
+ meta.addWidget(_label(role_name, "MessageRole"))
+ stamp = self._message_time(message.get("ts"))
+ if stamp:
+ meta.addWidget(_label(stamp, "MessageTime"))
+ meta.addStretch(1)
+ bubble_layout.addLayout(meta)
+ content = _label(str(message.get("content") or ""), "MessageContent")
+ content.setTextInteractionFlags(Qt.TextSelectableByMouse)
+ content.setMinimumWidth(260)
+ bubble_layout.addWidget(content)
+
+ row = QHBoxLayout()
+ row.setContentsMargins(0, 0, 0, 0)
+ if role == "assistant":
+ row.addStretch(1)
+ row.addWidget(bubble)
+ elif role == "system":
+ row.addStretch(1)
+ row.addWidget(bubble)
+ row.addStretch(1)
+ else:
+ row.addWidget(bubble)
+ row.addStretch(1)
+ parent.addLayout(row)
+
+ def _copy_all(self) -> None:
+ lines = [f"客户会话记录\n会话 ID：{self.session_id}\n"]
+ role_names = {"user": "客户", "assistant": "客服", "system": "系统记录"}
+ if self.history:
+ for message in self.history:
+ role = str(message.get("role") or "system").lower()
+ name = role_names.get(role, role)
+ stamp = self._message_time(message.get("ts"))
+ heading = f"[{stamp}] {name}" if stamp else name
+ lines.append(f"{heading}\n{message.get('content') or ''}")
+ elif self.last_lines:
+ lines.append("系统记录\n" + "\n".join(self.last_lines))
+ QApplication.clipboard().setText("\n\n".join(lines))
+
+
+class BusinessPage(QScrollArea):
+ logMessage = Signal(str, str)
+
+ def __init__(self):
+ super().__init__()
+ self.setObjectName("PageScroll")
+ self.setWidgetResizable(True)
+ self.setFrameShape(QFrame.NoFrame)
+ root = QWidget()
+ root.setObjectName("PageRoot")
+ self.setWidget(root)
+ layout = QVBoxLayout(root)
+ layout.setContentsMargins(28, 25, 28, 28)
+ layout.setSpacing(18)
+
+ header = QHBoxLayout()
+ header.addLayout(_page_header("04", "业务数据", "管理挂号线索与客户会话档案。"), 1)
+ refresh = _button("刷新数据")
+ refresh.clicked.connect(self.refresh_data)
+ header.addWidget(refresh, 0, Qt.AlignBottom)
+ layout.addLayout(header)
+
+ summary = QGridLayout()
+ summary.setSpacing(12)
+ self.registration_metric = MetricCard("待处理登记", "0", "需要人工跟进")
+ self.session_metric = MetricCard("会话档案", "0", "长期上下文记录")
+ summary.addWidget(self.registration_metric, 0, 0)
+ summary.addWidget(self.session_metric, 0, 1)
+ summary.setColumnStretch(0, 1)
+ summary.setColumnStretch(1, 1)
+ layout.addLayout(summary)
+
+ registrations, registrations_layout = _card(
+ "挂号与回访登记", "AI 识别出的明确挂号意向会沉淀在这里。"
+ )
+ self.registration_table = self._table(
+ ["最近更新", "微信客户", "症状 / 诉求", "状态"]
+ )
+ registrations_layout.addWidget(self.registration_table)
+ registration_actions = QHBoxLayout()
+ registration_actions.addStretch(1)
+ done = _button("标记已联系", "primary")
+ done.clicked.connect(self._mark_registration_done)
+ delete = _button("删除所选", "danger")
+ delete.clicked.connect(self._delete_registrations)
+ registration_actions.addWidget(done)
+ registration_actions.addWidget(delete)
+ registrations_layout.addLayout(registration_actions)
+ layout.addWidget(registrations)
+
+ sessions, sessions_layout = _card(
+ "客户会话档案",
+ "机器人重启后仍可继续使用已保存的上下文；双击任意一行查看完整记录。",
+ )
+ self.session_table = self._table(["最近更新", "会话 ID", "消息数", "内容预览"])
+ self.session_table.setToolTip("双击任意会话查看完整记录")
+ self.session_table.cellDoubleClicked.connect(self._open_session_detail)
+ sessions_layout.addWidget(self.session_table)
+ session_actions = QHBoxLayout()
+ session_actions.addStretch(1)
+ view_session = _button("查看记录", "primary")
+ view_session.clicked.connect(self._open_selected_session)
+ delete_session = _button("删除所选", "danger")
+ delete_session.clicked.connect(self._delete_sessions)
+ session_actions.addWidget(view_session)
+ session_actions.addWidget(delete_session)
+ sessions_layout.addLayout(session_actions)
+ layout.addWidget(sessions)
+ layout.addStretch(1)
+ self.refresh_data()
+
+ @staticmethod
+ def _table(headers: list[str]) -> QTableWidget:
+ table = QTableWidget(0, len(headers))
+ table.setHorizontalHeaderLabels(headers)
+ table.setSelectionBehavior(QAbstractItemView.SelectRows)
+ table.setSelectionMode(QAbstractItemView.ExtendedSelection)
+ table.setEditTriggers(QAbstractItemView.NoEditTriggers)
+ table.setAlternatingRowColors(True)
+ table.verticalHeader().hide()
+ table.verticalHeader().setDefaultSectionSize(42)
+ table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeToContents)
+ table.horizontalHeader().setStretchLastSection(True)
+ table.setMinimumHeight(230)
+ return table
+
+ @staticmethod
+ def _time_text(value) -> str:
+ try:
+ return time.strftime("%Y-%m-%d %H:%M", time.localtime(float(value)))
+ except (TypeError, ValueError, OSError):
+ return "--"
+
+ @staticmethod
+ def _selected_ids(table: QTableWidget) -> list[str]:
+ rows = sorted({index.row() for index in table.selectionModel().selectedRows()})
+ ids = []
+ for row in rows:
+ item = table.item(row, 0)
+ if item:
+ ids.append(str(item.data(Qt.UserRole) or ""))
+ return [item for item in ids if item]
+
+ def refresh_data(self) -> None:
+ try:
+ from registration_store import RegistrationStore
+
+ self.registration_store = RegistrationStore()
+ leads = self.registration_store.list_leads(include_done=True)
+ except Exception as exc:
+ leads = []
+ self.logMessage.emit(f"读取挂号登记失败：{exc}", "err")
+ status_map = {"pending_symptom": "待补症状", "booked": "待联系", "done": "已联系"}
+ self.registration_table.setRowCount(len(leads))
+ for row, lead in enumerate(leads):
+ values = (
+ self._time_text(lead.get("updated")),
+ lead.get("contact") or "未知客户",
+ (lead.get("symptom") or "待补充").replace("\n", " "),
+ status_map.get(lead.get("status"), lead.get("status") or "--"),
+ )
+ for column, value in enumerate(values):
+ item = QTableWidgetItem(str(value))
+ if column == 0:
+ item.setData(Qt.UserRole, lead.get("id"))
+ self.registration_table.setItem(row, column, item)
+ pending = sum(1 for item in leads if item.get("status") in {"pending_symptom", "booked"})
+ self.registration_metric.value.setText(str(pending))
+
+ try:
+ from conversation_store import ConversationStore
+
+ self.conversation_store = ConversationStore(str(SCRIPT_DIR / "conversations.json"))
+ sessions = self.conversation_store.list_sessions(limit=500)
+ except Exception as exc:
+ sessions = []
+ self.logMessage.emit(f"读取会话档案失败：{exc}", "err")
+ self.session_table.setRowCount(len(sessions))
+ for row, session in enumerate(sessions):
+ values = (
+ self._time_text(session.get("updated")),
+ session.get("session_id") or "--",
+ session.get("message_count") or 0,
+ session.get("preview") or "--",
+ )
+ for column, value in enumerate(values):
+ item = QTableWidgetItem(str(value))
+ if column == 0:
+ item.setData(Qt.UserRole, session.get("session_id"))
+ self.session_table.setItem(row, column, item)
+ self.session_metric.value.setText(str(len(sessions)))
+
+ def _mark_registration_done(self) -> None:
+ ids = self._selected_ids(self.registration_table)
+ if not ids:
+ QMessageBox.information(self, "提示", "请先选择登记记录。")
+ return
+ count = self.registration_store.set_status_many(ids, "done")
+ self.logMessage.emit(f"已将 {count} 条登记标记为已联系", "ok")
+ self.refresh_data()
+
+ def _delete_registrations(self) -> None:
+ ids = self._selected_ids(self.registration_table)
+ if not ids:
+ QMessageBox.information(self, "提示", "请先选择要删除的登记。")
+ return
+ if QMessageBox.question(self, "确认删除", f"确定删除选中的 {len(ids)} 条登记吗？") != QMessageBox.Yes:
+ return
+ count = self.registration_store.delete_many(ids)
+ self.logMessage.emit(f"已删除 {count} 条挂号登记", "warn")
+ self.refresh_data()
+
+ def _delete_sessions(self) -> None:
+ ids = self._selected_ids(self.session_table)
+ if not ids:
+ QMessageBox.information(self, "提示", "请先选择要删除的会话档案。")
+ return
+ if QMessageBox.question(self, "确认删除", f"确定删除选中的 {len(ids)} 份档案吗？") != QMessageBox.Yes:
+ return
+ count = sum(1 for session_id in ids if self.conversation_store.delete(session_id))
+ self.logMessage.emit(f"已删除 {count} 份会话档案", "warn")
+ self.refresh_data()
+
+ def _open_selected_session(self) -> None:
+ rows = self.session_table.selectionModel().selectedRows()
+ if not rows:
+ QMessageBox.information(self, "提示", "请先选择要查看的会话档案。")
+ return
+ self._open_session_detail(rows[0].row(), 0)
+
+ def _open_session_detail(self, row: int, _column: int) -> None:
+ id_item = self.session_table.item(row, 0)
+ session_id = str(id_item.data(Qt.UserRole) or "") if id_item else ""
+ if not session_id:
+ QMessageBox.warning(self, "无法打开", "未找到这条会话的 ID。")
+ return
+ try:
+ history = list(self.conversation_store.history(session_id))
+ last_lines = list(self.conversation_store.last_lines(session_id))
+ except Exception as exc:
+ QMessageBox.critical(self, "无法读取会话", str(exc))
+ return
+ SessionDetailDialog(session_id, history, last_lines, self).exec()
+
+
+class BackendLoginDialog(QDialog):
+ completed = Signal(bool, str)
+
+ def __init__(self, parent: QWidget | None = None):
+ super().__init__(parent)
+ self.setWindowTitle("登录配置后台")
+ self.setObjectName("SessionDetailDialog")
+ self.setModal(True)
+ self.setMinimumWidth(480)
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(26, 24, 26, 24)
+ layout.setSpacing(14)
+ layout.addWidget(_label("登录配置后台", "CardTitle"))
+ layout.addWidget(
+ _label("登录成功后，模型与 MCP 参数会自动同步到当前电脑。", "CardSubtitle")
+ )
+
+ import backend_client
+
+ settings = backend_client.load_settings()
+ layout.addWidget(_label("后台地址", "CardSubtitle"))
+ self.server_url = QLineEdit(settings.get("server_url", backend_client.DEFAULT_SERVER_URL))
+ self.server_url.setPlaceholderText("http://127.0.0.1:8765")
+ layout.addWidget(self.server_url)
+ layout.addWidget(_label("用户名", "CardSubtitle"))
+ self.username = QLineEdit(settings.get("username", ""))
+ self.username.setPlaceholderText("请输入后台用户名")
+ layout.addWidget(self.username)
+ layout.addWidget(_label("密码", "CardSubtitle"))
+ self.password = QLineEdit()
+ self.password.setEchoMode(QLineEdit.Password)
+ self.password.setPlaceholderText("密码不会保存在本地")
+ self.password.returnPressed.connect(self.submit)
+ layout.addWidget(self.password)
+ self.auto_sync = QCheckBox("启动及运行期间自动同步")
+ self.auto_sync.setChecked(bool(settings.get("auto_sync", True)))
+ layout.addWidget(self.auto_sync)
+ self.status = _label("", "DangerText")
+ self.status.setWordWrap(True)
+ layout.addWidget(self.status)
+ actions = QHBoxLayout()
+ actions.addStretch(1)
+ cancel = _button("取消")
+ cancel.clicked.connect(self.reject)
+ actions.addWidget(cancel)
+ self.login_button = _button("登录并同步", "primary")
+ self.login_button.clicked.connect(self.submit)
+ actions.addWidget(self.login_button)
+ layout.addLayout(actions)
+ self.completed.connect(self._finished)
+
+ def submit(self) -> None:
+ if not self.server_url.text().strip() or not self.username.text().strip() or not self.password.text():
+ self.status.setText("请填写后台地址、用户名和密码")
+ return
+ self.login_button.setEnabled(False)
+ self.status.setObjectName("WarningText")
+ self.status.setText("正在登录并获取模型配置…")
+ self.status.style().unpolish(self.status)
+ self.status.style().polish(self.status)
+ server_url = self.server_url.text().strip()
+ username = self.username.text().strip()
+ password = self.password.text()
+ auto_sync = self.auto_sync.isChecked()
+
+ def worker() -> None:
+ try:
+ import backend_client
+
+ backend_client.login(server_url, username, password, auto_sync=auto_sync)
+ result = backend_client.sync_config(force=True)
+ message = str(result.get("message") or "登录并同步成功")
+ except Exception as exc:
+ self.completed.emit(False, str(exc))
+ else:
+ self.completed.emit(True, message)
+
+ threading.Thread(target=worker, daemon=True).start()
+
+ def _finished(self, ok: bool, message: str) -> None:
+ self.login_button.setEnabled(True)
+ self.status.setObjectName("SuccessText" if ok else "DangerText")
+ self.status.setText(message)
+ self.status.style().unpolish(self.status)
+ self.status.style().polish(self.status)
+ if ok:
+ QTimer.singleShot(250, self.accept)
+
+
+class PersonaPage(QScrollArea):
+ saved = Signal(bool, str)
+ backendSyncFinished = Signal(bool, str)
+
+ def __init__(self):
+ super().__init__()
+ self.setObjectName("PageScroll")
+ self.setWidgetResizable(True)
+ self.setFrameShape(QFrame.NoFrame)
+ root = QWidget()
+ root.setObjectName("PageRoot")
+ self.setWidget(root)
+ layout = QVBoxLayout(root)
+ layout.setContentsMargins(28, 25, 28, 28)
+ layout.setSpacing(18)
+ layout.addLayout(_page_header("05", "AI 人格与能力", "统一管理模型、客服身份、上下文和 MCP 工具。"))
+
+ import ai_config
+
+ remote, remote_layout = _card(
+ "后台同步",
+ "在管理后台配置模型后，本机启动及运行期间会自动获取并应用。",
+ )
+ remote_row = QHBoxLayout()
+ remote_copy = QVBoxLayout()
+ self.remote_status = _label("正在读取后台连接…", "WarningText")
+ self.remote_meta = _label("", "CardSubtitle")
+ remote_copy.addWidget(self.remote_status)
+ remote_copy.addWidget(self.remote_meta)
+ remote_row.addLayout(remote_copy, 1)
+ self.logout_backend_button = _button("退出后台")
+ self.logout_backend_button.clicked.connect(self.logout_backend)
+ remote_row.addWidget(self.logout_backend_button)
+ self.manage_backend_button = _button("打开管理后台")
+ self.manage_backend_button.clicked.connect(self.open_backend_page)
+ remote_row.addWidget(self.manage_backend_button)
+ self.sync_backend_button = _button("立即同步")
+ self.sync_backend_button.clicked.connect(lambda: self.sync_from_backend(silent=False))
+ remote_row.addWidget(self.sync_backend_button)
+ self.login_backend_button = _button("登录后台", "primary")
+ self.login_backend_button.clicked.connect(self.open_backend_login)
+ remote_row.addWidget(self.login_backend_button)
+ remote_layout.addLayout(remote_row)
+ layout.addWidget(remote)
+
+ overview, overview_layout = _card("能力开关", "关闭 AI 后将使用通用设置中的固定回复。")
+ switch_grid = QGridLayout()
+ self.ai_enabled = QCheckBox("启用 AI 回复")
+ self.ai_enabled.setChecked(bool(ai_config.AI_ENABLED))
+ self.context_enabled = QCheckBox("启用会话上下文")
+ self.context_enabled.setChecked(bool(ai_config.AI_CONTEXT_ENABLED))
+ self.counter_enabled = QCheckBox("启用反辱骂策略")
+ self.counter_enabled.setChecked(bool(ai_config.AI_COUNTER_INSULT_ENABLED))
+ self.mcp_enabled = QCheckBox("启用 MCP 工具")
+ self.mcp_enabled.setChecked(bool(ai_config.AI_MCP_ENABLED))
+ for index, switch in enumerate(
+ (self.ai_enabled, self.context_enabled, self.counter_enabled, self.mcp_enabled)
+ ):
+ switch_grid.addWidget(switch, index // 2, index % 2)
+ overview_layout.addLayout(switch_grid)
+ layout.addWidget(overview)
+
+ identity, identity_layout = _card("模型与身份", "保存后立即写入 ai_settings.json。")
+ form = QGridLayout()
+ form.setHorizontalSpacing(18)
+ form.setVerticalSpacing(8)
+ self.api_base = self._field(form, 0, 0, "API 地址", ai_config.AI_API_BASE)
+ self.model = self._field(form, 0, 1, "模型名称", ai_config.AI_MODEL)
+ self.api_key = self._field(form, 2, 0, "API Key", ai_config.AI_API_KEY)
+ self.api_key.setEchoMode(QLineEdit.Password)
+ self.agent_name = self._field(form, 2, 1, "客服名称", ai_config.AI_AGENT_NAME)
+ self.hospital = self._field(form, 4, 0, "机构名称", ai_config.AI_HOSPITAL_NAME, 1, 2)
+ self.rounds = QSpinBox()
+ self.rounds.setRange(1, 50)
+ self.rounds.setValue(int(ai_config.AI_CONTEXT_MAX_ROUNDS))
+ self.max_tokens = QSpinBox()
+ self.max_tokens.setRange(50, 32000)
+ self.max_tokens.setValue(int(ai_config.AI_MAX_TOKENS))
+ self.temperature = QDoubleSpinBox()
+ self.temperature.setRange(0.0, 2.0)
+ self.temperature.setSingleStep(0.05)
+ self.temperature.setValue(float(ai_config.AI_TEMPERATURE))
+ form.addWidget(_label("上下文轮数", "CardSubtitle"), 6, 0)
+ form.addWidget(_label("最大回复 tokens", "CardSubtitle"), 6, 1)
+ form.addWidget(self.rounds, 7, 0)
+ form.addWidget(self.max_tokens, 7, 1)
+ form.addWidget(_label("温度", "CardSubtitle"), 8, 0)
+ form.addWidget(self.temperature, 9, 0)
+ form.setColumnStretch(0, 1)
+ form.setColumnStretch(1, 1)
+ identity_layout.addLayout(form)
+ layout.addWidget(identity)
+
+ mcp, mcp_layout = _card("MCP 服务器", "填写服务器 JSON 数组，配置将与 AI 能力一起保存。")
+ self.mcp_json = QPlainTextEdit()
+ self.mcp_json.setMinimumHeight(160)
+ self.mcp_json.setPlainText(
+ json.dumps(getattr(ai_config, "AI_MCP_SERVERS", []) or [], ensure_ascii=False, indent=2)
+ )
+ mcp_layout.addWidget(self.mcp_json)
+ mcp_row = QHBoxLayout()
+ mcp_row.addWidget(_label("单次回复最多工具轮数", "CardSubtitle"))
+ self.mcp_rounds = QSpinBox()
+ self.mcp_rounds.setRange(1, 20)
+ self.mcp_rounds.setValue(int(getattr(ai_config, "AI_MCP_MAX_ROUNDS", 5)))
+ mcp_row.addWidget(self.mcp_rounds)
+ mcp_row.addStretch(1)
+ mcp_layout.addLayout(mcp_row)
+ layout.addWidget(mcp)
+
+ actions = QHBoxLayout()
+ self.save_status = _label("", "SuccessText")
+ actions.addWidget(self.save_status, 1)
+ self.save_button = _button("保存 AI 配置", "primary")
+ self.save_button.clicked.connect(self.save_config)
+ actions.addWidget(self.save_button)
+ layout.addLayout(actions)
+ layout.addStretch(1)
+ self._backend_sync_running = False
+ self._backend_sync_silent = False
+ self.backendSyncFinished.connect(self._backend_sync_finished)
+ self._managed_widgets = (
+ self.ai_enabled,
+ self.context_enabled,
+ self.counter_enabled,
+ self.mcp_enabled,
+ self.api_base,
+ self.model,
+ self.api_key,
+ self.agent_name,
+ self.hospital,
+ self.rounds,
+ self.max_tokens,
+ self.temperature,
+ self.mcp_json,
+ self.mcp_rounds,
+ self.save_button,
+ )
+ self.refresh_backend_status()
+
+ @staticmethod
+ def _field(
+ layout: QGridLayout,
+ row: int,
+ column: int,
+ title: str,
+ value,
+ row_span: int = 1,
+ column_span: int = 1,
+ ) -> QLineEdit:
+ layout.addWidget(_label(title, "CardSubtitle"), row, column, 1, column_span)
+ field = QLineEdit(str(value or ""))
+ layout.addWidget(field, row + 1, column, row_span, column_span)
+ return field
+
+ def save_config(self) -> None:
+ try:
+ import backend_client
+
+ remote_managed = backend_client.is_configured() and bool(
+ backend_client.load_settings().get("auto_sync", True)
+ )
+ except Exception:
+ remote_managed = False
+ if remote_managed:
+ self.save_status.setText("当前由后台统一管理，请在后台修改后点击“立即同步”")
+ self.save_status.setObjectName("WarningText")
+ self.save_status.style().unpolish(self.save_status)
+ self.save_status.style().polish(self.save_status)
+ return
+ try:
+ servers = json.loads(self.mcp_json.toPlainText().strip() or "[]")
+ if not isinstance(servers, list):
+ raise ValueError("MCP JSON 根节点必须是数组")
+ import ai_config
+
+ ai_config.AI_ENABLED = self.ai_enabled.isChecked()
+ ai_config.AI_CONTEXT_ENABLED = self.context_enabled.isChecked()
+ ai_config.AI_COUNTER_INSULT_ENABLED = self.counter_enabled.isChecked()
+ ai_config.AI_MCP_ENABLED = self.mcp_enabled.isChecked()
+ ai_config.AI_API_BASE = self.api_base.text().strip()
+ ai_config.AI_API_KEY = self.api_key.text().strip()
+ ai_config.AI_MODEL = self.model.text().strip()
+ ai_config.AI_AGENT_NAME = self.agent_name.text().strip()
+ ai_config.AI_HOSPITAL_NAME = self.hospital.text().strip()
+ ai_config.AI_CONTEXT_MAX_ROUNDS = self.rounds.value()
+ ai_config.AI_MAX_TOKENS = self.max_tokens.value()
+ ai_config.AI_TEMPERATURE = self.temperature.value()
+ ai_config.AI_MCP_SERVERS = servers
+ ai_config.AI_MCP_MAX_ROUNDS = self.mcp_rounds.value()
+ ai_config.AI_SYSTEM_PROMPT = ai_config.build_system_prompt()
+ ai_config.save_settings()
+ except Exception as exc:
+ self.save_status.setText(f"保存失败：{exc}")
+ self.save_status.setObjectName("DangerText")
+ self.saved.emit(False, str(exc))
+ else:
+ self.save_status.setText("配置已保存并立即生效")
+ self.save_status.setObjectName("SuccessText")
+ self.saved.emit(True, "AI 配置已保存并生效")
+ self.save_status.style().unpolish(self.save_status)
+ self.save_status.style().polish(self.save_status)
+
+ def open_backend_login(self) -> None:
+ dialog = BackendLoginDialog(self)
+ if dialog.exec() == QDialog.Accepted:
+ self.reload_from_ai_config()
+ self.refresh_backend_status()
+ self.saved.emit(True, "后台已登录，模型配置已自动同步")
+
+ def open_backend_page(self) -> None:
+ try:
+ import backend_client
+
+ url = backend_client.load_settings().get(
+ "server_url", backend_client.DEFAULT_SERVER_URL
+ )
+ QDesktopServices.openUrl(QUrl(str(url)))
+ except Exception as exc:
+ self.saved.emit(False, f"无法打开管理后台：{exc}")
+
+ def logout_backend(self) -> None:
+ answer = QMessageBox.question(
+ self,
+ "退出配置后台",
+ "退出后将停止自动同步，但已下发到本机的配置会继续保留。确定退出吗？",
+ )
+ if answer != QMessageBox.Yes:
+ return
+ try:
+ import backend_client
+
+ backend_client.logout()
+ except Exception as exc:
+ self.saved.emit(False, f"退出后台失败：{exc}")
+ return
+ self.refresh_backend_status()
+ self.saved.emit(True, "已退出配置后台，当前配置保留在本机")
+
+ def refresh_backend_status(self) -> None:
+ try:
+ import backend_client
+
+ summary = backend_client.connection_summary()
+ except Exception as exc:
+ self.remote_status.setObjectName("DangerText")
+ self.remote_status.setText(f"后台连接配置读取失败：{exc}")
+ return
+ configured = bool(summary["configured"])
+ authenticated = bool(summary.get("authenticated"))
+ managed = configured and bool(summary["auto_sync"])
+ if configured:
+ last_error = str(summary.get("last_error") or "")
+ self.remote_status.setObjectName("DangerText" if last_error else "SuccessText")
+ self.remote_status.setText(
+ f"最近同步失败 · {last_error}"
+ if last_error
+ else (
+ f"已连接 · {summary['username']} · 配置 v{summary['last_version']}"
+ if authenticated
+ else f"已发现本机后台 · 配置 v{summary['last_version']}"
+ )
+ )
+ sync_at = summary.get("last_sync_at") or "尚未同步"
+ self.remote_meta.setText(f"{summary['server_url']} · 最后同步 {sync_at}")
+ else:
+ self.remote_status.setObjectName("WarningText")
+ self.remote_status.setText("尚未登录配置后台")
+ self.remote_meta.setText(f"默认地址 {summary['server_url']} · 当前使用本机配置")
+ self.remote_status.style().unpolish(self.remote_status)
+ self.remote_status.style().polish(self.remote_status)
+ self.login_backend_button.setVisible(not authenticated)
+ self.login_backend_button.setText("账号登录" if configured else "登录后台")
+ self.logout_backend_button.setVisible(authenticated)
+ self.sync_backend_button.setVisible(configured)
+ for widget in self._managed_widgets:
+ widget.setEnabled(not managed)
+ self.save_status.setText("后台统一管理中，本地字段为只读" if managed else "")
+ self.save_status.setObjectName("SuccessText" if managed else "CardSubtitle")
+ self.save_status.style().unpolish(self.save_status)
+ self.save_status.style().polish(self.save_status)
+
+ def sync_from_backend(self, *, silent: bool = True) -> None:
+ if self._backend_sync_running:
+ return
+ try:
+ import backend_client
+
+ if not backend_client.is_configured():
+ return
+ except Exception as exc:
+ if not silent:
+ self.saved.emit(False, f"后台连接配置读取失败：{exc}")
+ return
+ self._backend_sync_running = True
+ self._backend_sync_silent = silent
+ self.sync_backend_button.setEnabled(False)
+ self.remote_status.setObjectName("WarningText")
+ self.remote_status.setText("正在从后台同步模型配置…")
+ self.remote_status.style().unpolish(self.remote_status)
+ self.remote_status.style().polish(self.remote_status)
+
+ def worker() -> None:
+ try:
+ result = backend_client.sync_config(force=not silent)
+ message = str(result.get("message") or "后台配置同步完成")
+ except Exception as exc:
+ self.backendSyncFinished.emit(False, str(exc))
+ else:
+ self.backendSyncFinished.emit(True, message)
+
+ threading.Thread(target=worker, daemon=True).start()
+
+ def _backend_sync_finished(self, ok: bool, message: str) -> None:
+ silent = self._backend_sync_silent
+ self._backend_sync_running = False
+ self.sync_backend_button.setEnabled(True)
+ if ok:
+ self.reload_from_ai_config()
+ self.refresh_backend_status()
+ if not silent or not ok:
+ self.saved.emit(ok, message if ok else f"后台同步失败：{message}")
+
+ def reload_from_ai_config(self) -> None:
+ import ai_config
+
+ self.ai_enabled.setChecked(bool(ai_config.AI_ENABLED))
+ self.context_enabled.setChecked(bool(ai_config.AI_CONTEXT_ENABLED))
+ self.counter_enabled.setChecked(bool(ai_config.AI_COUNTER_INSULT_ENABLED))
+ self.mcp_enabled.setChecked(bool(ai_config.AI_MCP_ENABLED))
+ self.api_base.setText(str(ai_config.AI_API_BASE or ""))
+ self.model.setText(str(ai_config.AI_MODEL or ""))
+ self.api_key.setText(str(ai_config.AI_API_KEY or ""))
+ self.agent_name.setText(str(ai_config.AI_AGENT_NAME or ""))
+ self.hospital.setText(str(ai_config.AI_HOSPITAL_NAME or ""))
+ self.rounds.setValue(int(ai_config.AI_CONTEXT_MAX_ROUNDS))
+ self.max_tokens.setValue(int(ai_config.AI_MAX_TOKENS))
+ self.temperature.setValue(float(ai_config.AI_TEMPERATURE))
+ self.mcp_json.setPlainText(
+ json.dumps(getattr(ai_config, "AI_MCP_SERVERS", []) or [], ensure_ascii=False, indent=2)
+ )
+ self.mcp_rounds.setValue(int(getattr(ai_config, "AI_MCP_MAX_ROUNDS", 5)))
+
+
+class LogPage(QWidget):
+ def __init__(self):
+ super().__init__()
+ self.setObjectName("PageRoot")
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(28, 25, 28, 28)
+ layout.setSpacing(18)
+ header = QHBoxLayout()
+ header.addLayout(_page_header("06", "运行日志", "实时查看连接、识别、AI 与业务沉淀事件。"), 1)
+ clear = _button("清空日志")
+ clear.clicked.connect(self.clear)
+ header.addWidget(clear, 0, Qt.AlignBottom)
+ layout.addLayout(header)
+ card, card_layout = _card()
+ self.editor = QTextEdit()
+ self.editor.setReadOnly(True)
+ self.editor.setAcceptRichText(True)
+ self.editor.setStyleSheet(
+ "QTextEdit{font-family:'Cascadia Mono','Microsoft YaHei UI';font-size:13px;line-height:1.5;}"
+ )
+ card_layout.addWidget(self.editor)
+ layout.addWidget(card, 1)
+
+ def append(self, message: str, tag: str = "") -> None:
+ colors = {
+ "ok": COLORS["success"],
+ "warn": COLORS["warning"],
+ "err": COLORS["danger"],
+ "notify": COLORS["accent_dark"],
+ }
+ color = colors.get(tag, COLORS["ink"])
+ stamp = time.strftime("%H:%M:%S")
+ safe = (
+ str(message)
+ .replace("&", "&")
+ .replace("<", "<")
+ .replace(">", ">")
+ .replace("\n", "
")
+ )
+ self.editor.append(
+ f'[{stamp}] '
+ f'{safe}'
+ )
+
+ def clear(self) -> None:
+ self.editor.clear()
+
+
+class CapsuleWindow(QWidget):
+ expandRequested = Signal()
+ stopRequested = Signal()
+
+ def __init__(self):
+ super().__init__(None)
+ self.setWindowFlags(
+ Qt.Tool
+ | Qt.FramelessWindowHint
+ | Qt.WindowStaysOnTopHint
+ | Qt.WindowDoesNotAcceptFocus
+ )
+ self.setAttribute(Qt.WA_TranslucentBackground)
+ self.setAttribute(Qt.WA_ShowWithoutActivating)
+ self.setFixedSize(430, 82)
+ self._drag_origin: QPoint | None = None
+ self._window_origin = QPoint()
+
+ outer = QVBoxLayout(self)
+ outer.setContentsMargins(12, 8, 12, 14)
+ panel = QFrame()
+ panel.setObjectName("CapsulePanel")
+ panel.setStyleSheet(
+ "QFrame#CapsulePanel{background:#FFFFFF;border:1px solid #CFE0D8;border-radius:16px;}"
+ "QFrame#CapsulePanel QLabel{background:transparent;border:none;}"
+ )
+ _shadow(panel, 28, 7, 38)
+ row = QHBoxLayout(panel)
+ row.setContentsMargins(17, 10, 10, 10)
+ row.setSpacing(11)
+ grip = QLabel("•••")
+ grip.setAlignment(Qt.AlignCenter)
+ grip.setFixedSize(38, 38)
+ grip.setStyleSheet(
+ "QLabel{background:#F0F6F3;border:1px solid #D8E6DF;border-radius:19px;"
+ "color:#789087;font-family:'Cascadia Mono';font-weight:700;}"
+ )
+ row.addWidget(grip)
+ self.dot = QLabel("●")
+ self.dot.setStyleSheet("QLabel{color:#10A57A;font-size:17px;}")
+ row.addWidget(self.dot)
+ copy = QVBoxLayout()
+ copy.setSpacing(1)
+ self.status = QLabel("监听中")
+ self.status.setStyleSheet(
+ "QLabel{color:#087A59;font-size:14px;font-weight:700;}"
+ )
+ self.hint = QLabel("安全运行中")
+ self.hint.setStyleSheet("QLabel{color:#7B8D84;font-size:11px;}")
+ copy.addWidget(self.status)
+ copy.addWidget(self.hint)
+ row.addLayout(copy, 1)
+ self.timer = QLabel("00:00:00")
+ self.timer.setAlignment(Qt.AlignCenter)
+ self.timer.setMinimumWidth(82)
+ self.timer.setStyleSheet(
+ "QLabel{background:#F2F7F4;border:1px solid #DCE8E2;border-radius:9px;"
+ "padding:7px 9px;font-family:'Cascadia Mono';font-size:12px;color:#30473D;}"
+ )
+ row.addWidget(self.timer)
+ expand = QPushButton("↗")
+ expand.setToolTip("展开控制台")
+ expand.setCursor(Qt.PointingHandCursor)
+ expand.setFixedSize(34, 34)
+ expand.setStyleSheet(
+ "QPushButton{background:#E7F4EE;color:#087A59;border:1px solid #CFE8DD;"
+ "border-radius:8px;font-size:17px;font-weight:700;}"
+ "QPushButton:hover{background:#D4EFE4;border-color:#AEDCC9;}"
+ "QPushButton:pressed{background:#C5E8D9;}"
+ )
+ expand.clicked.connect(self.expandRequested)
+ row.addWidget(expand)
+ stop = QPushButton("×")
+ stop.setToolTip("停止监听")
+ stop.setCursor(Qt.PointingHandCursor)
+ stop.setFixedSize(34, 34)
+ stop.setStyleSheet(
+ "QPushButton{background:#FBEAEC;color:#C74452;border:1px solid #F2CDD2;"
+ "border-radius:8px;font-size:18px;}"
+ "QPushButton:hover{background:#D85260;color:#FFFFFF;border-color:#D85260;}"
+ )
+ stop.clicked.connect(self.stopRequested)
+ row.addWidget(stop)
+ outer.addWidget(panel)
+
+ def show_near(self, window: QWidget) -> None:
+ screen = window.screen() or QApplication.primaryScreen()
+ area = screen.availableGeometry()
+ x = area.x() + (area.width() - self.width()) // 2
+ y = area.y() + 28
+ self.move(x, y)
+ self.show()
+
+ def mousePressEvent(self, event) -> None:
+ if event.button() == Qt.LeftButton:
+ self._drag_origin = event.globalPosition().toPoint()
+ self._window_origin = self.pos()
+ event.accept()
+
+ def mouseMoveEvent(self, event) -> None:
+ if self._drag_origin is not None and event.buttons() & Qt.LeftButton:
+ self.move(self._window_origin + event.globalPosition().toPoint() - self._drag_origin)
+ event.accept()
+
+ def mouseReleaseEvent(self, event) -> None:
+ self._drag_origin = None
+ super().mouseReleaseEvent(event)
+
+ def set_status(self, state: str, text: str) -> None:
+ colors = {
+ "running": "#10A57A",
+ "waiting": "#B67820",
+ "connecting": "#B67820",
+ "stopping": "#B67820",
+ "error": "#D85260",
+ }
+ hints = {
+ "running": "安全运行中",
+ "waiting": "等待企业微信",
+ "connecting": "正在连接窗口",
+ "stopping": "正在结束任务",
+ "error": "请查看运行日志",
+ "stopped": "监听已停止",
+ }
+ color = colors.get(state, "#7B8D84")
+ self.dot.setStyleSheet(f"QLabel{{color:{color};font-size:17px;}}")
+ self.status.setText(text)
+ self.status.setStyleSheet(
+ f"QLabel{{color:{color};font-size:14px;font-weight:700;}}"
+ )
+ self.hint.setText(hints.get(state, text))
+
+
+class Sidebar(QFrame):
+ pageRequested = Signal(int)
+ startRequested = Signal()
+ stopRequested = Signal()
+
+ PAGE_NAMES = ("AI 客服", "自动回复", "通用设置", "业务数据", "AI 人格", "运行日志")
+
+ def __init__(self):
+ super().__init__()
+ self.setObjectName("Sidebar")
+ self.setFixedWidth(258)
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(18, 24, 18, 18)
+ layout.setSpacing(8)
+ layout.addWidget(_label("ZHEN AI DESK", "Brand"))
+ layout.addWidget(_label("AI 客服 · 企业微信自动回复", "BrandSub"))
+ layout.addSpacing(13)
+
+ status_card = QFrame()
+ status_card.setObjectName("SideStatus")
+ status_layout = QVBoxLayout(status_card)
+ status_layout.setContentsMargins(15, 13, 15, 13)
+ status_layout.setSpacing(3)
+ self.status_title = _label("● 已停止", "SideStatusTitle")
+ self.status_hint = _label("等待连接企业微信", "SideStatusHint")
+ status_layout.addWidget(self.status_title)
+ status_layout.addWidget(self.status_hint)
+ layout.addWidget(status_card)
+ layout.addSpacing(9)
+ layout.addWidget(_label("CONTROL CENTER", "SectionLabel"))
+
+ self.nav_buttons: list[QPushButton] = []
+ for index, name in enumerate(self.PAGE_NAMES, start=1):
+ button = QPushButton(f"{index:02d} {name}")
+ button.setObjectName("NavButton")
+ button.setCheckable(True)
+ button.setAutoExclusive(True)
+ button.setCursor(Qt.PointingHandCursor)
+ button.clicked.connect(lambda _checked=False, i=index - 1: self.pageRequested.emit(i))
+ layout.addWidget(button)
+ self.nav_buttons.append(button)
+ self.nav_buttons[0].setChecked(True)
+ layout.addStretch(1)
+ self.monitor_button = _button("开始监听", "side")
+ self.monitor_button.clicked.connect(self.startRequested)
+ layout.addWidget(self.monitor_button)
+ layout.addWidget(_label("RPA DESKTOP / LOCAL", "SectionLabel"))
+
+ def select(self, index: int) -> None:
+ if 0 <= index < len(self.nav_buttons):
+ self.nav_buttons[index].setChecked(True)
+
+ def set_status(self, state: str, text: str, hint: str) -> None:
+ colors = {
+ "running": "#43E1AE",
+ "waiting": "#F2BC64",
+ "connecting": "#F2BC64",
+ "stopping": "#F2BC64",
+ "error": "#FF818E",
+ "stopped": "#A5B9B0",
+ }
+ self.status_title.setText(f"● {text}")
+ self.status_title.setStyleSheet(
+ f"color:{colors.get(state, '#A5B9B0')};font-size:15px;font-weight:700;"
+ )
+ self.status_hint.setText(hint)
+ active = state in {"running", "waiting", "connecting", "stopping"}
+ self.monitor_button.setText("停止监听" if active else "开始监听")
+ try:
+ self.monitor_button.clicked.disconnect()
+ except RuntimeError:
+ pass
+ self.monitor_button.clicked.connect(self.stopRequested if active else self.startRequested)
+ self.monitor_button.setObjectName("DangerButton" if active else "SidePrimary")
+ self.monitor_button.style().unpolish(self.monitor_button)
+ self.monitor_button.style().polish(self.monitor_button)
+
+
+class MainWindow(QMainWindow):
+ def __init__(self):
+ super().__init__()
+ self.setWindowTitle(WIN_TITLE)
+ self.setMinimumSize(1120, 720)
+ self._set_initial_geometry()
+ self._queue: queue.Queue = queue.Queue()
+ self._thread = None
+ self._running = False
+ self._start_time: float | None = None
+ self._status_key = "stopped"
+ self._original_stdout = sys.stdout
+ self._stdout_proxy = None
+ self._saved_geometry = None
+ self._was_maximized = False
+ self.runtime_settings = self._load_runtime_settings()
+
+ root = QWidget()
+ root.setObjectName("AppRoot")
+ self.setCentralWidget(root)
+ root_layout = QHBoxLayout(root)
+ root_layout.setContentsMargins(0, 0, 0, 0)
+ root_layout.setSpacing(0)
+
+ self.sidebar = Sidebar()
+ root_layout.addWidget(self.sidebar)
+ self.stack = FadingStack()
+ root_layout.addWidget(self.stack, 1)
+
+ self.portal_page = PortalPage()
+ self.dashboard_page = DashboardPage()
+ self.settings_page = SettingsPage(self.runtime_settings)
+ self.business_page = BusinessPage()
+ self.persona_page = PersonaPage()
+ self.log_page = LogPage()
+ for page in (
+ self.portal_page,
+ self.dashboard_page,
+ self.settings_page,
+ self.business_page,
+ self.persona_page,
+ self.log_page,
+ ):
+ self.stack.addWidget(page)
+
+ self.capsule = CapsuleWindow()
+ self._connect_signals()
+ self.business_refresh_timer = QTimer(self)
+ self.business_refresh_timer.setSingleShot(True)
+ self.business_refresh_timer.setInterval(220)
+ self.business_refresh_timer.timeout.connect(self._refresh_counts)
+ self.set_status("stopped", "已停止")
+ self._refresh_counts()
+ self.append_log("新控制台已就绪。请确认企业微信已登录，再开始监听。", "notify")
+
+ self.queue_timer = QTimer(self)
+ self.queue_timer.setInterval(140)
+ self.queue_timer.timeout.connect(self._process_queue)
+ self.queue_timer.start()
+
+ try:
+ import backend_client
+
+ sync_seconds = int(
+ backend_client.load_settings().get("sync_interval_seconds", 300)
+ )
+ except Exception:
+ sync_seconds = 300
+ self.backend_sync_timer = QTimer(self)
+ self.backend_sync_timer.setInterval(max(60, sync_seconds) * 1000)
+ self.backend_sync_timer.timeout.connect(
+ lambda: self.persona_page.sync_from_backend(silent=True)
+ )
+ self.backend_sync_timer.start()
+ QTimer.singleShot(350, lambda: self.persona_page.sync_from_backend(silent=True))
+
+ def _set_initial_geometry(self) -> None:
+ screen = QApplication.primaryScreen()
+ area = screen.availableGeometry() if screen else QRect(0, 0, 1440, 900)
+ width = min(1480, max(1120, area.width() - 72))
+ height = min(940, max(720, area.height() - 72))
+ self.resize(width, height)
+ self.move(area.x() + (area.width() - width) // 2, area.y() + (area.height() - height) // 2)
+
+ def _connect_signals(self) -> None:
+ self.sidebar.pageRequested.connect(self.show_page)
+ self.sidebar.startRequested.connect(self.start_monitoring)
+ self.sidebar.stopRequested.connect(self.stop_monitoring)
+ self.dashboard_page.startRequested.connect(self.start_monitoring)
+ self.dashboard_page.stopRequested.connect(self.stop_monitoring)
+ self.dashboard_page.pageRequested.connect(self.show_page)
+ self.settings_page.saved.connect(self.save_runtime_settings)
+ self.business_page.logMessage.connect(self.append_log)
+ self.persona_page.saved.connect(self._persona_saved)
+ self.capsule.expandRequested.connect(self.expand_console)
+ self.capsule.stopRequested.connect(self.stop_monitoring)
+
+ @staticmethod
+ def _runtime_defaults() -> dict:
+ return {
+ "auto_reply_text": "你好",
+ "poll_interval": 2.0,
+ "mouse_idle_enabled": True,
+ "mouse_idle_seconds": 20.0,
+ }
+
+ def _load_runtime_settings(self) -> dict:
+ settings = self._runtime_defaults()
+ try:
+ saved = json.loads(APP_SETTINGS_FILE.read_text(encoding="utf-8"))
+ if isinstance(saved, dict):
+ settings.update({key: saved[key] for key in settings if key in saved})
+ except (OSError, ValueError, TypeError):
+ pass
+ try:
+ settings["poll_interval"] = max(0.2, float(settings["poll_interval"]))
+ settings["mouse_idle_seconds"] = max(0.0, float(settings["mouse_idle_seconds"]))
+ settings["mouse_idle_enabled"] = bool(settings["mouse_idle_enabled"])
+ settings["auto_reply_text"] = str(settings["auto_reply_text"]).strip() or "你好"
+ except (TypeError, ValueError):
+ return self._runtime_defaults()
+ return settings
+
+ def save_runtime_settings(self, settings: dict | None = None) -> None:
+ self.runtime_settings = dict(settings or self.settings_page.values())
+ try:
+ temporary = APP_SETTINGS_FILE.with_suffix(".json.tmp")
+ temporary.write_text(
+ json.dumps(self.runtime_settings, ensure_ascii=False, indent=2),
+ encoding="utf-8",
+ )
+ os.replace(temporary, APP_SETTINGS_FILE)
+ except OSError as exc:
+ self.settings_page.mark_saved(False, f"保存失败：{exc}")
+ self.append_log(f"运行参数保存失败：{exc}", "err")
+ else:
+ self.settings_page.mark_saved(True, "已自动保存")
+
+ def show_page(self, index: int) -> None:
+ self.sidebar.select(index)
+ self.stack.setCurrentIndexAnimated(index)
+ if index == 3:
+ self.business_page.refresh_data()
+ if index == 0:
+ QTimer.singleShot(80, lambda: self.portal_page.view.setFocus(Qt.OtherFocusReason))
+
+ def start_monitoring(self) -> None:
+ if self._thread is not None and self._thread.is_alive():
+ self.append_log("监听线程已经在运行", "warn")
+ return
+ self.save_runtime_settings()
+ self.persona_page.save_config()
+ self._stdout_proxy = LogQueue(self._queue)
+ sys.stdout = self._stdout_proxy
+ self._thread = BotThread(
+ self._queue,
+ self.runtime_settings["auto_reply_text"],
+ float(self.runtime_settings["poll_interval"]),
+ mouse_idle_enabled=bool(self.runtime_settings["mouse_idle_enabled"]),
+ mouse_idle_seconds=float(self.runtime_settings["mouse_idle_seconds"]),
+ )
+ self._running = True
+ self._start_time = time.time()
+ self.set_status("connecting", "连接中")
+ self.append_log("正在连接企业微信窗口…", "notify")
+ self._thread.start()
+
+ def stop_monitoring(self) -> None:
+ if self._thread is None or not self._thread.is_alive():
+ self._finish_thread("stopped")
+ return
+ self._running = False
+ self._thread.stop()
+ self.set_status("stopping", "正在停止")
+ self.append_log("正在停止监听…", "warn")
+
+ def set_status(self, state: str, text: str) -> None:
+ self._status_key = state
+ hints = {
+ "connecting": "正在查找企业微信窗口",
+ "waiting": WECOM_WAITING_MESSAGE,
+ "running": "安全运行中，检测到未读消息后自动处理",
+ "stopping": "等待后台任务退出",
+ "error": "连接失败，请查看运行日志",
+ "stopped": "等待连接企业微信",
+ }
+ hint = hints.get(state, text)
+ self.sidebar.set_status(state, text, hint)
+ self.dashboard_page.set_status(state, text, hint)
+ self.capsule.set_status(state, text)
+
+ def enter_capsule_mode(self) -> None:
+ if self.capsule.isVisible():
+ return
+ self._saved_geometry = self.saveGeometry()
+ self._was_maximized = self.isMaximized()
+ self.capsule.show_near(self)
+ self.hide()
+
+ def expand_console(self) -> None:
+ self.capsule.hide()
+ if self._saved_geometry is not None:
+ self.restoreGeometry(self._saved_geometry)
+ if self._was_maximized:
+ self.showMaximized()
+ else:
+ self.showNormal()
+ self.raise_()
+ self.activateWindow()
+
+ def _finish_thread(self, final_state: str) -> None:
+ self._running = False
+ if self.capsule.isVisible():
+ self.expand_console()
+ if sys.stdout is self._stdout_proxy:
+ sys.stdout = self._original_stdout
+ self._stdout_proxy = None
+ if final_state == "error":
+ self.set_status("error", "连接失败")
+ self.show_page(5)
+ else:
+ self.set_status("stopped", "已停止")
+ self.dashboard_page.timer.setText("运行时长 --:--:--")
+
+ def append_log(self, message: str, tag: str = "") -> None:
+ self.log_page.append(message, tag)
+
+ def _persona_saved(self, ok: bool, message: str) -> None:
+ self.append_log(message, "ok" if ok else "err")
+
+ def _refresh_counts(self) -> None:
+ try:
+ self.business_page.refresh_data()
+ self.dashboard_page.registration.value.setText(
+ self.business_page.registration_metric.value.text()
+ )
+ self.dashboard_page.sessions.value.setText(
+ self.business_page.session_metric.value.text()
+ )
+ except Exception as exc:
+ self.append_log(f"刷新业务数据失败：{exc}", "err")
+
+ def _schedule_business_refresh(self) -> None:
+ """合并短时间内的多条挂号日志，避免在队列循环中反复重建表格。"""
+ self.business_refresh_timer.start()
+
+ def _process_queue(self) -> None:
+ try:
+ while True:
+ kind, data = self._queue.get_nowait()
+ if kind == "log":
+ lowered = str(data).lower()
+ if any(token in str(data) for token in ("[+]", "回复完成", "已保存")):
+ tag = "ok"
+ elif any(token in str(data) for token in ("[!]", "误判", "停止")):
+ tag = "warn"
+ elif any(token in lowered for token in ("错误", "[-]", "失败", "exception", "traceback")):
+ tag = "err"
+ elif any(token in str(data) for token in ("[MCP]", "[挂号]", "启动", "连接")):
+ tag = "notify"
+ else:
+ tag = ""
+ self.append_log(str(data), tag)
+ if "[挂号]" in str(data):
+ self._schedule_business_refresh()
+ elif kind == "status":
+ if data == "running":
+ self.set_status("running", "监听中")
+ self.enter_capsule_mode()
+ elif data == "waiting":
+ self.set_status("waiting", "等待企业微信")
+ elif data == "error":
+ self._finish_thread("error")
+ elif data == "stopped":
+ self._finish_thread("stopped")
+ elif kind == "stats":
+ self.dashboard_page.replied.value.setText(str(data.get("replied", 0)))
+ self.dashboard_page.false_pos.value.setText(str(data.get("false_pos", 0)))
+ except KeyboardInterrupt:
+ # Qt 默认会把 Ctrl+C 打断留在当前槽函数中并打印堆栈；转为正常关闭。
+ QTimer.singleShot(0, self.close)
+ return
+ except queue.Empty:
+ pass
+
+ if self._start_time and self._running:
+ elapsed = int(time.time() - self._start_time)
+ runtime = f"{elapsed // 3600:02d}:{(elapsed % 3600) // 60:02d}:{elapsed % 60:02d}"
+ self.dashboard_page.timer.setText(f"运行时长 {runtime}")
+ self.capsule.timer.setText(runtime)
+
+ def closeEvent(self, event) -> None:
+ self.save_runtime_settings()
+ self._running = False
+ if self._thread is not None and self._thread.is_alive():
+ self._thread.stop()
+ if sys.stdout is self._stdout_proxy:
+ sys.stdout = self._original_stdout
+ self.capsule.close()
+ event.accept()
+
+
+def _app_icon() -> QIcon:
+ pixmap = QPixmap(64, 64)
+ pixmap.fill(Qt.transparent)
+ painter = QPainter(pixmap)
+ painter.setRenderHint(QPainter.Antialiasing)
+ painter.setBrush(QColor(COLORS["accent"]))
+ painter.setPen(Qt.NoPen)
+ painter.drawRoundedRect(5, 5, 54, 54, 17, 17)
+ painter.setBrush(QColor("#FFFFFF"))
+ painter.drawEllipse(17, 18, 30, 24)
+ painter.drawPolygon([QPoint(25, 39), QPoint(22, 49), QPoint(34, 41)])
+ painter.end()
+ return QIcon(pixmap)
+
+
+def install_console_shutdown_handler(app: QApplication, window: MainWindow):
+ """将控制台 Ctrl+C / Ctrl+Break 转换为 Qt 的正常关闭流程。"""
+ shutdown_requested = False
+
+ def request_shutdown(_signum=None, _frame=None) -> None:
+ nonlocal shutdown_requested
+ if shutdown_requested:
+ return
+ shutdown_requested = True
+ QTimer.singleShot(0, window.close)
+ QTimer.singleShot(0, app.quit)
+
+ handled = [signal.SIGINT]
+ if hasattr(signal, "SIGBREAK"):
+ handled.append(signal.SIGBREAK)
+ for signal_number in handled:
+ try:
+ signal.signal(signal_number, request_shutdown)
+ except (OSError, ValueError):
+ pass
+ return request_shutdown
+
+
+def main() -> None:
+ QApplication.setHighDpiScaleFactorRoundingPolicy(
+ Qt.HighDpiScaleFactorRoundingPolicy.PassThrough
+ )
+ app = QApplication.instance() or QApplication(sys.argv)
+ app.setApplicationName(WIN_TITLE)
+ app.setApplicationDisplayName(WIN_TITLE)
+ app.setWindowIcon(_app_icon())
+ app.setStyle("Fusion")
+ app.setStyleSheet(APP_QSS)
+ font = QFont("Microsoft YaHei UI", 10)
+ font.setHintingPreference(QFont.PreferFullHinting)
+ app.setFont(font)
+
+ try:
+ import backend_client
+
+ backend_client.startup_sync_config(timeout=3.0)
+ except Exception:
+ # 启动检测失败不阻止软件打开，具体原因会显示在后台同步卡片中。
+ pass
+
+ window = MainWindow()
+ console_shutdown = install_console_shutdown_handler(app, window)
+ window.show()
+
+ if "--qt-smoke-test" in sys.argv:
+ for index in range(window.stack.count()):
+ def show_test_page(i=index) -> None:
+ window.show_page(i)
+
+ def capture(i=index) -> None:
+ window.grab().save(str(SCRIPT_DIR / f"qt-ui-smoke-{i + 1}.png"))
+
+ QTimer.singleShot(650 + index * 650, show_test_page)
+ QTimer.singleShot(950 + index * 650, capture)
+ QTimer.singleShot(1100 + window.stack.count() * 650, app.quit)
+ try:
+ exit_code = app.exec()
+ except KeyboardInterrupt:
+ console_shutdown()
+ exit_code = 0
+ raise SystemExit(exit_code)
+
+
+if __name__ == "__main__":
+ main()