+
@@ -1261,6 +1635,10 @@ def validate_config_form(form: dict[str, str], current: dict[str, Any]) -> dict[
config = {key: current.get(key) for key in CONFIG_KEYS}
for key in BOOL_KEYS:
config[key] = form.get(key) == "1"
+ provider = form.get("AI_PROVIDER_TYPE", "").strip().lower()
+ if provider not in PROVIDER_TYPES:
+ raise ValueError("服务类型无效")
+ config["AI_PROVIDER_TYPE"] = provider
for key in ("AI_API_BASE", "AI_MODEL", "AI_AGENT_NAME", "AI_HOSPITAL_NAME"):
config[key] = form.get(key, "").strip()
api_key = form.get("AI_API_KEY", "").strip()
diff --git a/wechat_rpa/ai_chat.py b/wechat_rpa/ai_chat.py
index 69a74e5..ef35cb5 100644
--- a/wechat_rpa/ai_chat.py
+++ b/wechat_rpa/ai_chat.py
@@ -16,6 +16,16 @@ import ai_config
# 这样在 GUI「AI 高级配置」中修改保存后，下一次请求立即生效，无需重启。
+def _provider_type() -> str:
+ value = str(getattr(ai_config, "AI_PROVIDER_TYPE", "auto") or "auto").lower()
+ if value in {"openai", "dify", "comfyui"}:
+ return value
+ path = (urlparse((ai_config.AI_API_BASE or "").rstrip("/")).path or "").lower()
+ if "chat-messages" in path or "completion-messages" in path:
+ return "dify"
+ return "openai"
+
+
def _completions_url() -> str:
"""
组装实际请求地址：
@@ -24,6 +34,15 @@ def _completions_url() -> str:
"""
base = (ai_config.AI_API_BASE or "").rstrip("/")
path = urlparse(base).path or ""
+ if _provider_type() == "dify":
+ lower_path = path.lower().rstrip("/")
+ if lower_path.endswith(("/chat-messages", "/completion-messages")):
+ return base
+ if lower_path.endswith("/v1"):
+ return f"{base}/chat-messages"
+ return f"{base}/v1/chat-messages"
+ if path.lower().rstrip("/").endswith("/chat/completions"):
+ return base
if re.match(r"^/v1/.+", path):
return base
return f"{base}/chat/completions"
@@ -31,10 +50,60 @@ def _completions_url() -> str:
def _is_dify_endpoint() -> bool:
"""AI_API_BASE 指向 Dify 的 chat-messages / completion-messages 时走 Dify 协议。"""
+ if _provider_type() == "dify":
+ return True
+ if _provider_type() in {"openai", "comfyui"}:
+ return False
path = (urlparse((ai_config.AI_API_BASE or "").rstrip("/")).path or "").lower()
return "chat-messages" in path or "completion-messages" in path
+def _development_mode_enabled() -> bool:
+ value = getattr(ai_config, "AI_DEVELOPMENT_MODE", False)
+ if isinstance(value, str):
+ return value.strip().lower() in {"1", "true", "yes", "on"}
+ return bool(value)
+
+
+def _log_request_diagnostics(url: str, protocol: str) -> None:
+ """Log request routing/configuration without exposing prompts or credentials."""
+ if not _development_mode_enabled():
+ return
+ try:
+ from backend_client import diagnostic_url
+
+ safe_url = diagnostic_url(url)
+ safe_base = diagnostic_url(getattr(ai_config, "AI_API_BASE", ""))
+ except Exception:
+ safe_url = str(url or "")
+ safe_base = str(getattr(ai_config, "AI_API_BASE", "") or "")
+ details = {
+ "服务类型": _provider_type(),
+ "调用协议": protocol,
+ "API 基础地址": safe_base,
+ "实际请求地址": safe_url,
+ "模型名称": str(getattr(ai_config, "AI_MODEL", "") or ""),
+ "API Key": (
+ "[已配置，值已隐藏]"
+ if str(getattr(ai_config, "AI_API_KEY", "") or "").strip()
+ else "[未配置]"
+ ),
+ "请求超时（秒）": getattr(ai_config, "AI_TIMEOUT", None),
+ "最大回复 tokens": getattr(ai_config, "AI_MAX_TOKENS", None),
+ "温度": getattr(ai_config, "AI_TEMPERATURE", None),
+ "始终视觉模式": bool(getattr(ai_config, "AI_USE_VISION", False)),
+ "媒体消息自动视觉": True,
+ "AI 页面守护": bool(getattr(ai_config, "AI_UI_GUARD_ENABLED", True)),
+ "上下文": bool(getattr(ai_config, "AI_CONTEXT_ENABLED", False)),
+ "MCP 工具": bool(getattr(ai_config, "AI_MCP_ENABLED", False)),
+ }
+ print(f"[开发模式] 模型请求地址: {safe_url}")
+ print(
+ "[开发模式] 本次模型配置（聊天内容与敏感值未输出）: "
+ + json.dumps(details, ensure_ascii=False, separators=(",", ":"))
+ )
+
+
def _system_prompt() -> str:
"""
动态构建系统提示词：基础人设（用当前昵称/医院名实时渲染）
@@ -48,10 +117,12 @@ def _system_prompt() -> str:
return prompt
-_CHAT_HEADER_RE = re.compile(
- r"^.+?\s+(?:(?:\d{4}[/-])?\d{1,2}[/-]\d{1,2}\s+)?"
+_CHAT_HEADER_PARTS_RE = re.compile(
+ r"^(?P
.+?)\s+"
+ r"(?:(?:\d{4}[/-])?\d{1,2}[/-]\d{1,2}\s+)?"
r"\d{1,2}:\d{2}(?::\d{2})?$"
)
+_CHAT_HEADER_RE = _CHAT_HEADER_PARTS_RE
_CASUAL_RE = re.compile(
r"(?:好困|困死|想睡|好累|累死|无聊|好烦|烦死|好饿|饿死|"
r"在干嘛|干什么呢|多大了|几岁|哪里人|叫什么|吃饭了吗|"
@@ -66,6 +137,84 @@ _MEDICAL_DRIFT_RE = re.compile(
r"调药|调整方案|治疗方案)"
)
_CLARIFY_RE = re.compile(r"^(?:什么|啥|什么意思|没懂|没看懂|没明白)[？?。！!]*$")
+_LOW_INFORMATION_RE = re.compile(
+ r"^[\s!！?？。,.，…~～啊呀哦噢嗯唔诶欸哎]+$"
+)
+_LOW_INFORMATION_GUESS_RE = re.compile(
+ r"(?:不小心|误触|碰到手机|按到手机|是不是.{0,12}(?:生气|不舒服|出事|发生什么))"
+)
+
+# 企业微信会把部分非文字气泡复制成占位符。标准 Unicode emoji 仍是普通
+# 文字，不在这里识别；只有图片、贴纸、语音等真正需要额外能力的气泡才触发。
+_MEDIA_PATTERNS = {
+ "image": re.compile(
+ r"[\[【]\s*(?:图片|照片|相片|图像|image|photo)\s*[\]】]",
+ re.IGNORECASE,
+ ),
+ "sticker": re.compile(
+ r"[\[【]\s*(?:动画表情|表情包|表情|贴纸|sticker|emoticon)\s*[\]】]",
+ re.IGNORECASE,
+ ),
+ "voice": re.compile(
+ r"[\[【]\s*(?:语音|语音消息|音频|voice|audio)"
+ r"(?:\s*[:：]?\s*(?:\d+(?:\.\d+)?\s*(?:秒|s|″|”|′|＇|'|\")|\d{1,2}:\d{2}))?\s*[\]】]"
+ r"|(?m:^[ \t]*(?:语音消息|语音|音频|voice|audio)[ \t:：]+(?:\d{1,2}:\d{2}|"
+ r"\d+(?:\.\d+)?[ \t]*(?:秒|s|″|”|′|＇|'|\"))[ \t]*$)",
+ re.IGNORECASE,
+ ),
+ "video": re.compile(
+ r"[\[【]\s*(?:视频|小视频|video)\s*[\]】]",
+ re.IGNORECASE,
+ ),
+ "file": re.compile(
+ r"[\[【]\s*(?:文件|文档|file)\s*[\]】]",
+ re.IGNORECASE,
+ ),
+}
+_MEDIA_PLACEHOLDER_RE = re.compile(
+ "|".join(f"(?:{pattern.pattern})" for pattern in _MEDIA_PATTERNS.values()),
+ re.IGNORECASE,
+)
+_VOICE_DURATION_ONLY_RE = re.compile(
+ r"^[\s，,。:：-]*(?:\d{1,2}:\d{2}|\d+(?:\.\d+)?\s*(?:秒|s|″|”|′|＇|'|\")+)"
+ r"[\s，,。]*$",
+ re.IGNORECASE,
+)
+_VOICE_CONTROL_ONLY_RE = re.compile(
+ r"^(?:转(?:成|为)?文字|语音转文字|重新转文字|播放|暂停|"
+ r"(?:正在)?(?:识别|转写|转文字|转换(?:成|为)?文字)中?(?:…|\.\.\.)?|"
+ r"(?:暂时)?无法识别(?:该)?语音|未识别出文字|(?:语音识别|转文字)失败)[。！!…\s]*$"
+)
+
+VISION_NO_INCOMING = "__NO_INCOMING_MESSAGE__"
+VISION_VOICE_NEEDS_TEXT = "__VOICE_NOT_TRANSCRIBED__"
+
+
+class VisionReply(str):
+ """String-compatible reply carrying media facts proved by the vision response."""
+
+ def __new__(cls, value: str, media_types=None, voice_transcribed: bool = False):
+ obj = str.__new__(cls, value or "")
+ obj.media_types = frozenset(media_types or ())
+ obj.voice_transcribed = bool(voice_transcribed)
+ return obj
+
+
+def detect_media_types(chat_text: str) -> set[str]:
+ """Return media bubble types explicitly present in copied WeCom text."""
+ text = str(chat_text or "")
+ return {
+ media_type
+ for media_type, pattern in _MEDIA_PATTERNS.items()
+ if pattern.search(text)
+ }
+
+
+def _chat_header_match(line: str):
+ """Do not mistake media durations such as '[语音] 00:05' for speaker headers."""
+ if detect_media_types(line):
+ return None
+ return _CHAT_HEADER_PARTS_RE.match(line)
def latest_customer_message(chat_text: str) -> str:
@@ -75,7 +224,7 @@ def latest_customer_message(chat_text: str) -> str:
return ""
last_header = -1
for index, line in enumerate(lines):
- if _CHAT_HEADER_RE.match(line):
+ if _chat_header_match(line):
last_header = index
if 0 <= last_header < len(lines) - 1:
return "\n".join(lines[last_header + 1:]).strip()
@@ -83,8 +232,122 @@ def latest_customer_message(chat_text: str) -> str:
return "\n".join(lines).strip()
+def latest_customer_turn(chat_text: str) -> str:
+ """合并聊天末尾由同一位客户连续发送的多个消息气泡。"""
+ lines = [line.strip() for line in str(chat_text or "").splitlines() if line.strip()]
+ if not lines:
+ return ""
+
+ blocks = []
+ current = None
+ for line in lines:
+ match = _chat_header_match(line)
+ if match:
+ if current is not None:
+ blocks.append(current)
+ current = {
+ "speaker": match.group("speaker").strip(),
+ "content": [],
+ }
+ elif current is not None:
+ current["content"].append(line)
+
+ if current is not None:
+ blocks.append(current)
+ if not blocks:
+ return "\n".join(lines).strip()
+
+ tail_speaker = blocks[-1]["speaker"]
+ merged = []
+ for block in reversed(blocks):
+ if block["speaker"] != tail_speaker:
+ break
+ content = "\n".join(block["content"]).strip()
+ if content:
+ merged.append(content)
+ return "\n".join(reversed(merged)).strip()
+
+
+def media_text_content(chat_text: str) -> str:
+ """Return the current customer turn with media placeholders removed."""
+ current = latest_customer_turn(chat_text)
+ if not current:
+ return ""
+ cleaned = _MEDIA_PLACEHOLDER_RE.sub("", current).strip()
+ if "voice" in detect_media_types(current):
+ # 企业微信有的版本复制成“[语音] 00:05”，也可能把时长单独放一行。
+ # 时长不是转写，必须删除；真正显示的转写文字仍会保留。
+ cleaned = "\n".join(
+ line
+ for line in cleaned.splitlines()
+ if line.strip()
+ and not _VOICE_DURATION_ONLY_RE.fullmatch(line.strip())
+ and not _VOICE_CONTROL_ONLY_RE.fullmatch(line.strip())
+ ).strip()
+ if not cleaned:
+ return ""
+ return cleaned.strip(" \t\r\n，,。；;：:")
+
+
+def media_archive_text(
+ chat_text: str,
+ media_types=None,
+ *,
+ voice_transcribed: bool = False,
+) -> str:
+ """Build a non-sensitive archive entry without storing screenshot contents."""
+ kinds = set(
+ detect_media_types(latest_customer_turn(chat_text))
+ if media_types is None
+ else media_types
+ )
+ visible_text = media_text_content(chat_text)
+ labels = {
+ "image": "（客户发来图片）",
+ "sticker": "（客户发来表情）",
+ "voice": (
+ "（客户发来语音及可见文字/转写）"
+ if visible_text
+ else (
+ "（客户发来语音，视觉确认已有转写）"
+ if voice_transcribed
+ else "（客户发来语音，未取得转写）"
+ )
+ ),
+ "video": "（客户发来视频）",
+ "file": "（客户发来文件）",
+ }
+ parts = [labels[kind] for kind in ("image", "sticker", "voice", "video", "file") if kind in kinds]
+ if visible_text:
+ parts.append(visible_text)
+ if not parts:
+ parts.append("（客户发来非文字消息，内容未识别）")
+ return "\n".join(parts)
+
+
+def safe_media_reply(media_types=None) -> str:
+ """Conservative fallback used when the configured model cannot read media."""
+ kinds = set(media_types or ())
+ if "voice" in kinds:
+ return "这条语音我这边暂时没法准确听清，麻烦您把重点打成文字发我一下。"
+ if "image" in kinds:
+ return "图片我收到了，这边暂时没看清具体内容，您想让我重点看哪一处？"
+ if "sticker" in kinds:
+ return "看到您发的表情啦，您接着说，我在看。"
+ if "video" in kinds:
+ return "视频我收到了，这边暂时没法准确读取内容，麻烦您把重点打字说一下。"
+ if "file" in kinds:
+ return "文件我收到了，麻烦您说下需要我重点看什么内容。"
+ return "收到您刚才的消息了，这边暂时没能完整识别，麻烦您把重点打字说一下。"
+
+
def _conversation_mode_instruction(latest: str) -> str:
text = str(latest or "").strip()
+ if _LOW_INFORMATION_RE.fullmatch(text):
+ return (
+ "【本轮信息很少】客户只发了语气词或标点。不要猜测误触手机、情绪、病情或任何原因；"
+ "自然表示自己在听，再用至多一个问题请对方继续说。"
+ )
if _CLARIFY_RE.fullmatch(text):
return (
"【本轮是追问澄清】客户是在说没听懂你上一句。"
@@ -154,6 +417,7 @@ def _call_dify(query: str, user: str = "wechat-rpa", conversation_id: str = "")
payload["conversation_id"] = conversation_id
url = _completions_url()
+ _log_request_diagnostics(url, "Dify chat-messages")
resp = requests.post(
url, headers=_headers(), json=payload, timeout=ai_config.AI_TIMEOUT
)
@@ -191,7 +455,7 @@ def _dify_query_from_chat(chat_text: str, history: list = None) -> str:
except Exception:
hosp = "甄养堂互联网医院"
agent = "客服"
- latest = latest_customer_message(chat_text)
+ latest = latest_customer_turn(chat_text)
mode_instruction = _conversation_mode_instruction(latest)
rules = (
"【事实铁律｜必须遵守】\n"
@@ -206,7 +470,7 @@ def _dify_query_from_chat(chat_text: str, history: list = None) -> str:
"可轻提一句需要可来本院挂号；【禁止】直接说已帮您预约。\n"
f"6. 仅当客户明确说要挂号/预约/面诊/帮我约 时，才说「已帮您预约了，稍后预约上了再联系您」，"
f"医院是{hosp}。客户说不需要/挂啥号时绝不能预约。\n"
- "7. 只回复客户最后一条需要处理的问题。语气像干了十几年的老客服：口语、沉稳、"
+ "7. 把客户本轮连续发送的多段话作为一个整体理解并统一回复。语气像干了十几年的老客服：口语、沉稳、"
"不急不躁，一次只说一件事，最多顺带问一个问题，不要复述对方原话，不要一次抛一大段方案。\n"
"8. 默认只写1～2句、20～60个汉字；先用一句自然的话接住对方的担心或不舒服，再回答重点。"
"不要标题、列表、客套收尾，不说「希望能帮到您」「请您放心」等套话。\n"
@@ -218,6 +482,8 @@ def _dify_query_from_chat(chat_text: str, history: list = None) -> str:
"12. 示例：客户说「好困啊」，可回「困了就先眯一会儿，别硬撑着」；"
"客户问「你多大了」，可回「四十来岁啦，怎么突然问这个？」。"
"示例只说明说话方式，不要机械重复。\n"
+ "13. 客户只发「啊」「嗯」「！」「？」等低信息内容时，严禁猜测对方误触手机、"
+ "生气或身体不适；只自然接住并请对方继续说。\n"
)
hist = _history_messages(history)
if hist:
@@ -229,13 +495,13 @@ def _dify_query_from_chat(chat_text: str, history: list = None) -> str:
"【近期对话｜仅供参考，其中客服所述业务数据可能不实】\n"
+ "\n".join(lines)
+ f"\n\n{mode_instruction}\n"
- + f"【客户最后一句｜唯一回答对象】\n{latest}\n\n"
+ + f"【客户本轮连续消息｜统一回答对象】\n{latest}\n\n"
+ f"【本次原始聊天片段｜仅用于理解上下文】\n{chat_text}\n\n请生成回复："
)
else:
body = (
f"{mode_instruction}\n"
- f"【客户最后一句｜唯一回答对象】\n{latest}\n\n"
+ f"【客户本轮连续消息｜统一回答对象】\n{latest}\n\n"
f"【本次原始聊天片段｜仅用于理解上下文】\n{chat_text}\n\n请生成回复："
)
return rules + "\n" + body
@@ -333,6 +599,12 @@ def _repair_obvious_mismatch(reply: str, chat_text: str, history: list = None) -
if not latest or not cleaned:
return cleaned
+ if (
+ _LOW_INFORMATION_RE.fullmatch(latest)
+ and _LOW_INFORMATION_GUESS_RE.search(cleaned)
+ ):
+ return "我在呢，您慢慢说，怎么啦？"
+
if _CLARIFY_RE.fullmatch(latest):
previous = ""
for item in reversed(_history_messages(history or [])):
@@ -396,6 +668,7 @@ def _chat_completion(messages: list, tools: list = None) -> dict:
payload["tools"] = tools
payload["tool_choice"] = "auto"
url = _completions_url()
+ _log_request_diagnostics(url, "OpenAI 兼容 chat/completions")
resp = requests.post(
url, headers=_headers(), json=payload, timeout=ai_config.AI_TIMEOUT
)
@@ -406,13 +679,13 @@ def _chat_completion(messages: list, tools: list = None) -> dict:
def _user_turn(chat_text: str) -> dict:
- latest = latest_customer_message(chat_text)
+ latest = latest_customer_turn(chat_text)
mode_instruction = _conversation_mode_instruction(latest)
return {
"role": "user",
"content": (
f"{mode_instruction}\n"
- f"【客户最后一句｜唯一回答对象】\n{latest}\n\n"
+ f"【客户本轮连续消息｜统一回答对象】\n{latest}\n\n"
f"【原始聊天片段｜仅用于理解上下文】\n{chat_text}\n\n"
"像真人微信聊天；普通闲聊直接接话，只有对方明确担心或难受时才先关心。默认1～2句、20～60字，"
"最多问一个问题，不要列表、标题和客套收尾。"
@@ -443,7 +716,7 @@ def call_ai_text(chat_text: str, history: list = None) -> str:
history,
)
except Exception as e:
- print(f" [MCP] ⚠ 工具增强失败，回退普通回复: {e}")
+ print(f" [MCP] [!] 工具增强失败，回退普通回复: {e}")
messages = [{"role": "system", "content": _system_prompt()}]
messages += _history_messages(history)
@@ -513,12 +786,462 @@ async def _call_ai_text_with_mcp(chat_text: str, history: list = None) -> str:
return _strip_thinking(msg.get("content") or "")
-def call_ai_vision(image_bytes: bytes, history: list = None) -> str:
- """
- 视觉模式：将聊天区域截图发给多模态 AI，让 AI 直接阅读并回复。
- image_bytes 为 PNG 图片的 bytes。
- history 为该会话的历史上下文（多轮记忆），可为 None。
- """
+_UI_GUARD_STATES = {
+ "chat_ready",
+ "blocking_modal",
+ "non_message_page",
+ "security_verification",
+ "unknown",
+}
+_UI_GUARD_ACTIONS = {"none", "escape", "close_modal", "open_messages"}
+
+
+def _dify_api_root() -> str:
+ """返回 Dify App API 根地址（通常以 /v1 结尾）。"""
+ endpoint = _completions_url().rstrip("/")
+ for suffix in ("/chat-messages", "/completion-messages"):
+ if endpoint.lower().endswith(suffix):
+ return endpoint[: -len(suffix)]
+ return endpoint
+
+
+def _call_dify_with_image(
+ query: str,
+ image_bytes: bytes,
+ *,
+ user: str = "wechat-rpa-vision",
+ timeout: float | None = None,
+) -> str:
+ """按 Dify App API 的“上传文件 → chat-messages 引用文件”流程调用视觉应用。"""
+ timeout = timeout or ai_config.AI_TIMEOUT
+ api_root = _dify_api_root()
+ upload_url = f"{api_root}/files/upload"
+ headers = {
+ "Authorization": f"Bearer {(ai_config.AI_API_KEY or '').strip()}",
+ "Accept": "application/json",
+ }
+ _log_request_diagnostics(upload_url, "Dify 文件上传（AI 页面守护）")
+ upload = requests.post(
+ upload_url,
+ headers=headers,
+ data={"user": user},
+ files={"file": ("wecom-ui.png", image_bytes, "image/png")},
+ timeout=timeout,
+ )
+ if not upload.ok:
+ raise RuntimeError(
+ f"Dify 截图上传失败 {upload.status_code}: {(upload.text or '')[:300]}"
+ )
+ upload_id = str((upload.json() or {}).get("id") or "").strip()
+ if not upload_id:
+ raise RuntimeError("Dify 截图上传成功，但响应中没有文件 ID")
+
+ payload = {
+ "inputs": {},
+ "query": query,
+ "response_mode": "blocking",
+ "user": user,
+ "files": [
+ {
+ "type": "image",
+ "transfer_method": "local_file",
+ "upload_file_id": upload_id,
+ }
+ ],
+ }
+ chat_url = f"{api_root}/chat-messages"
+ _log_request_diagnostics(chat_url, "Dify 视觉 chat-messages")
+ response = requests.post(
+ chat_url,
+ headers=_headers(),
+ json=payload,
+ timeout=timeout,
+ )
+ if not response.ok:
+ raise RuntimeError(
+ f"Dify 视觉请求失败 {response.status_code}: {(response.text or '')[:300]}"
+ )
+ data = response.json() or {}
+ answer = data.get("answer") or (data.get("data") or {}).get("answer")
+ if not answer:
+ raise RuntimeError("Dify 视觉请求未返回 answer")
+ return str(answer)
+
+
+def _parse_ui_guard_decision(raw: str) -> dict:
+ """从模型文本中提取并收紧为页面守护允许的结构和动作。"""
+ text = _strip_thinking(str(raw or "")).strip()
+ candidates = []
+ decoder = json.JSONDecoder()
+ for index, char in enumerate(text):
+ if char != "{":
+ continue
+ try:
+ value, _ = decoder.raw_decode(text[index:])
+ except json.JSONDecodeError:
+ continue
+ # 模型有时会先输出调试/计量 JSON，随后才给页面守护协议对象。
+ # 只有同时显式包含 state 与 action 的对象才有执行资格，不能让第一个
+ # 无关 dict 抢占解析结果。
+ if isinstance(value, dict) and "state" in value and "action" in value:
+ candidates.append(value)
+ if not candidates:
+ return {
+ "state": "unknown",
+ "action": "none",
+ "confidence": 0.0,
+ "reason": "模型未返回有效 JSON",
+ }
+
+ # 同一次响应出现两个互相矛盾的协议对象时，无法证明哪个才是最终判断。
+ # 失败关闭比任意选择一个并执行 Esc/点击更安全；完全重复的对象不算冲突。
+ decisions = {
+ (
+ str(item.get("state") or "unknown").strip().lower(),
+ str(item.get("action") or "none").strip().lower(),
+ )
+ for item in candidates
+ }
+ if len(decisions) != 1:
+ return {
+ "state": "unknown",
+ "action": "none",
+ "confidence": 0.0,
+ "reason": "模型返回冲突 JSON",
+ }
+ data = candidates[-1]
+
+ state = str(data.get("state") or "unknown").strip().lower()
+ action = str(data.get("action") or "none").strip().lower()
+ if state not in _UI_GUARD_STATES:
+ state = "unknown"
+ if action not in _UI_GUARD_ACTIONS:
+ action = "none"
+ try:
+ confidence = float(data.get("confidence", 0.0))
+ except (TypeError, ValueError):
+ confidence = 0.0
+ confidence = max(0.0, min(1.0, confidence))
+
+ # 状态与动作必须匹配。即使模型返回了额外动作，也不能越过本地白名单。
+ allowed_by_state = {
+ "chat_ready": {"none"},
+ "blocking_modal": {"none", "escape", "close_modal"},
+ "non_message_page": {"none", "escape", "open_messages"},
+ "security_verification": {"none"},
+ "unknown": {"none"},
+ }
+ if action not in allowed_by_state[state]:
+ action = "none"
+ return {
+ "state": state,
+ "action": action,
+ "confidence": confidence,
+ "reason": str(data.get("reason") or "")[:120],
+ }
+
+
+def classify_wecom_ui(image_bytes: bytes, trigger: str = "页面异常") -> dict:
+ """用视觉模型判断企业微信当前页面；只返回受限动作，不返回点击坐标。"""
+ if not image_bytes:
+ return _parse_ui_guard_decision("")
+ if _provider_type() == "comfyui":
+ return {
+ "state": "unknown",
+ "action": "none",
+ "confidence": 0.0,
+ "reason": "ComfyUI 不用于界面判断",
+ }
+
+ prompt = (
+ "你是企业微信桌面端的页面安全分类器。判断截图是否妨碍客服自动回复。"
+ f"触发环节：{trigger}。只输出一个 JSON 对象，禁止输出解释、Markdown 或点击坐标。\n"
+ "state 只能是：chat_ready（聊天页和输入框可用）、blocking_modal（弹窗/浮层遮挡）、"
+ "non_message_page（文档、微盘、工作台、会议等非消息页面）、"
+ "security_verification（登录、扫码、安全验证）、unknown。\n"
+ "action 只能是：none、escape、close_modal、open_messages。"
+ "安全验证必须 none；不确定必须 unknown+none；普通聊天页必须 chat_ready+none。\n"
+ "格式：{\"state\":\"...\",\"action\":\"...\","
+ "\"confidence\":0.0,\"reason\":\"不超过30字\"}"
+ )
+ timeout = min(float(getattr(ai_config, "AI_TIMEOUT", 120) or 120), 30.0)
+ if _is_dify_endpoint():
+ raw = _call_dify_with_image(
+ prompt,
+ image_bytes,
+ user="wechat-rpa-ui-guard",
+ timeout=timeout,
+ )
+ return _parse_ui_guard_decision(raw)
+
+ encoded = base64.b64encode(image_bytes).decode("ascii")
+ payload = {
+ "model": ai_config.AI_MODEL,
+ "messages": [
+ {
+ "role": "user",
+ "content": [
+ {"type": "text", "text": prompt},
+ {
+ "type": "image_url",
+ "image_url": {"url": f"data:image/png;base64,{encoded}"},
+ },
+ ],
+ }
+ ],
+ "max_tokens": 180,
+ "temperature": 0,
+ }
+ url = _completions_url()
+ _log_request_diagnostics(url, "OpenAI 兼容 AI 页面守护")
+ response = requests.post(
+ url,
+ headers=_headers(),
+ json=payload,
+ timeout=timeout,
+ )
+ if not response.ok:
+ raise RuntimeError(
+ f"AI 页面判断失败 {response.status_code}: {(response.text or '')[:300]}"
+ )
+ raw = response.json()["choices"][0]["message"]["content"]
+ return _parse_ui_guard_decision(raw)
+
+
+def _vision_chat_prompt(chat_text: str = "", media_types=None) -> str:
+ kinds = set(
+ detect_media_types(latest_customer_turn(chat_text))
+ if media_types is None
+ else media_types
+ )
+ kind_names = {
+ "image": "图片",
+ "sticker": "表情/贴纸",
+ "voice": "语音",
+ "video": "视频",
+ "file": "文件",
+ }
+ detected = "、".join(
+ kind_names[kind]
+ for kind in ("image", "sticker", "voice", "video", "file")
+ if kind in kinds
+ ) or "剪贴板未识别具体类型"
+ current = latest_customer_turn(chat_text) or "（剪贴板未提取到本轮文字，请只依据截图中可见的新内容判断）"
+ return (
+ "这是企业微信聊天消息区域的局部截图。通常左侧气泡是客户，右侧气泡是我方；"
+ "请同时按气泡左右位置、时间顺序和本轮文字判断，不要把我方旧回复当成客户消息。\n"
+ f"【剪贴板检测到的媒体】{detected}\n"
+ f"【本轮可提取文字】\n{current}\n\n"
+ "把客户在本轮约20秒内连续发送的文字、图片和表情作为一个整体理解后统一回复。"
+ "动画表情内部换帧不代表又来了一条消息；只有最末端新出现的客户侧气泡，"
+ "或位于最近一条我方气泡之后的客户连续气泡，才算本轮新消息。"
+ "图片或贴纸只能描述截图里确实可见的信息；看不清就自然追问，不得编造病情、处方、"
+ "订单、物流或其他业务事实，也不得仅凭图片声称已经修改、提交、预约或执行任何操作。\n"
+ "重要：截图不含语音声音。绝对不要根据波形或时长猜语音内容。\n"
+ "只输出一个 JSON 对象，不要 Markdown 或解释，字段固定为："
+ '{"has_new_customer_message":true,"media_type":"text|image|sticker|voice|video|file|mixed|unknown",'
+ '"media_types":["image"],"contains_voice":false,"voice_transcribed":false,'
+ '"reply":"可直接发送的回复"}。media_types 只能包含 image、sticker、voice、video、file；'
+ "mixed 时必须列出全部可确认类型。\n"
+ "若最新内容其实是我方发送的或没有新客户消息，has_new_customer_message=false 且 reply 为空；"
+ "若 mixed 中包含任何语音，contains_voice 必须为 true；不含语音则必须为 false。"
+ "若是语音且截图/本轮文字里没有可见转写，media_type=voice、contains_voice=true、"
+ "voice_transcribed=false、reply 为空。"
+ "其他情况 reply 默认1～2句、20～60字，最多问一个问题，不要标题或列表。"
+ )
+
+
+def _finalize_vision_reply(raw: str, chat_text: str = "", media_types=None) -> str:
+ cleaned = _strip_thinking(str(raw or "")).strip()
+ # Some otherwise capable models prepend a small analysis JSON object before
+ # the requested protocol object. Taking the first arbitrary dict turns a
+ # valid result into ``has_new=False`` because the field is absent. Collect
+ # only protocol-shaped objects and use the last one (the final answer).
+ protocol_objects = []
+ decoder = json.JSONDecoder()
+ index = 0
+ while index < len(cleaned):
+ object_start = cleaned.find("{", index)
+ if object_start < 0:
+ break
+ try:
+ value, consumed = decoder.raw_decode(cleaned[object_start:])
+ except json.JSONDecodeError:
+ index = object_start + 1
+ continue
+ if isinstance(value, dict) and "has_new_customer_message" in value:
+ protocol_objects.append(value)
+ # Skip the whole decoded object. This prevents a nested diagnostic dict
+ # from overriding the enclosing protocol result when we choose the last
+ # top-level answer.
+ index = object_start + max(1, consumed)
+
+ data = protocol_objects[-1] if protocol_objects else None
+
+ # Keep compatibility with the two internal sentinels, but do not search for
+ # them as substrings inside a JSON reply: customer-visible text could contain
+ # those words and must not silently override a valid structured result.
+ if data is None:
+ standalone = cleaned.strip().strip("`").strip().upper()
+ if standalone == VISION_VOICE_NEEDS_TEXT.upper():
+ return VISION_VOICE_NEEDS_TEXT
+ if standalone == VISION_NO_INCOMING.upper():
+ return VISION_NO_INCOMING
+ return ""
+
+ def protocol_bool(value):
+ if isinstance(value, bool):
+ return value
+ if isinstance(value, int) and value in (0, 1):
+ return bool(value)
+ if isinstance(value, str):
+ normalized = value.strip().lower()
+ if normalized in {"1", "true", "yes"}:
+ return True
+ if normalized in {"0", "false", "no"}:
+ return False
+ return None
+
+ has_new = protocol_bool(data.get("has_new_customer_message"))
+ if has_new is None:
+ return ""
+ if not has_new:
+ return VISION_NO_INCOMING
+
+ allowed_kinds = {"image", "sticker", "voice", "video", "file"}
+ media_type_value = data.get("media_type", "unknown")
+ if not isinstance(media_type_value, str):
+ return ""
+ media_type = media_type_value.strip().lower() or "unknown"
+ if media_type not in allowed_kinds | {"text", "mixed", "unknown"}:
+ return ""
+
+ kinds = set(
+ detect_media_types(latest_customer_turn(chat_text))
+ if media_types is None
+ else media_types
+ )
+ kinds.intersection_update(allowed_kinds)
+
+ transcribed = protocol_bool(data.get("voice_transcribed", False))
+ contains_voice = protocol_bool(data.get("contains_voice", False))
+ if transcribed is None or contains_voice is None:
+ if "voice" in kinds:
+ return VisionReply(
+ VISION_VOICE_NEEDS_TEXT,
+ media_types=kinds,
+ voice_transcribed=False,
+ )
+ return ""
+
+ response_kinds = set()
+ raw_response_kinds = data.get("media_types")
+ if raw_response_kinds is not None:
+ if not isinstance(raw_response_kinds, (list, tuple, set)):
+ if "voice" in kinds:
+ return VisionReply(
+ VISION_VOICE_NEEDS_TEXT,
+ media_types=kinds,
+ voice_transcribed=bool(
+ transcribed and media_text_content(chat_text)
+ ),
+ )
+ return ""
+ normalized_response_kinds = [
+ str(item).strip().lower() for item in raw_response_kinds
+ ]
+ if any(item not in allowed_kinds for item in normalized_response_kinds):
+ if "voice" in kinds:
+ return VisionReply(
+ VISION_VOICE_NEEDS_TEXT,
+ media_types=kinds,
+ voice_transcribed=bool(
+ transcribed and media_text_content(chat_text)
+ ),
+ )
+ return ""
+ response_kinds.update(normalized_response_kinds)
+
+ claimed_kinds = set(response_kinds)
+ if media_type in allowed_kinds:
+ claimed_kinds.add(media_type)
+ if contains_voice:
+ claimed_kinds.add("voice")
+ # Clipboard evidence or the model's own declaration of a voice bubble is
+ # stronger than any remaining schema inconsistency. A screenshot can never
+ # reveal audio, so always collapse to the non-guessing voice path while still
+ # preserving other media types for the archive.
+ if "voice" in kinds | claimed_kinds:
+ return VisionReply(
+ VISION_VOICE_NEEDS_TEXT,
+ media_types=(kinds | claimed_kinds | {"voice"}),
+ voice_transcribed=bool(transcribed and media_text_content(chat_text)),
+ )
+
+ # media_type and media_types are two views of the same model decision. A
+ # primitive type cannot disagree with, or omit a different type from, its
+ # list; multiple media must be declared as mixed.
+ if media_type in allowed_kinds:
+ if response_kinds and response_kinds != {media_type}:
+ return ""
+ response_kinds.add(media_type)
+ elif media_type == "mixed":
+ if "contains_voice" not in data or not response_kinds:
+ return ""
+ elif response_kinds:
+ # text/unknown with a non-empty media list is internally inconsistent.
+ return ""
+
+ if contains_voice:
+ response_kinds.add("voice")
+
+ all_kinds = kinds | response_kinds
+ if transcribed:
+ # A transcription flag without a voice is impossible under the contract.
+ return ""
+ if kinds:
+ if media_type in {"text", "unknown"}:
+ return ""
+ if media_type in allowed_kinds and kinds != {media_type}:
+ return ""
+ if media_type == "mixed" and not kinds.issubset(response_kinds):
+ return ""
+
+ # 对“剪贴板完全为空 + 模型也无法判断媒体类型”的结果不采纳正文，
+ # 避免把未转写语音误当成图片并生成臆测回复。
+ if media_type == "unknown" and not str(chat_text or "").strip():
+ return ""
+ reply = data.get("reply", "")
+ if not isinstance(reply, str):
+ return ""
+ return VisionReply(
+ _humanize(reply),
+ media_types=all_kinds,
+ voice_transcribed=bool(transcribed),
+ )
+
+
+
+def call_ai_vision(
+ image_bytes: bytes,
+ history: list = None,
+ chat_text: str = "",
+ media_types=None,
+) -> str:
+ """Read a chat-area screenshot together with copied text and archive history."""
+ prompt = _vision_chat_prompt(chat_text, media_types)
+ if _is_dify_endpoint():
+ # Dify's image request is a single query, so include the same conversation
+ # rules and current copied text in that query before attaching the screenshot.
+ query_text = chat_text or "（客户本轮发来一条无法复制为文字的消息）"
+ query = _dify_query_from_chat(query_text, history) + "\n\n【媒体识别规则】\n" + prompt
+ raw = _call_dify_with_image(
+ query,
+ image_bytes,
+ user="wechat-rpa-chat-vision",
+ )
+ return _finalize_vision_reply(raw, chat_text, media_types)
+
b64 = base64.b64encode(image_bytes).decode("utf-8")
simple_headers = {
"Authorization": f"Bearer {ai_config.AI_API_KEY}",
@@ -530,16 +1253,7 @@ def call_ai_vision(image_bytes: bytes, history: list = None) -> str:
"model": ai_config.AI_MODEL,
"messages": messages + [
{"role": "user", "content": [
- {
- "type": "text",
- "text": (
- "这是一个聊天对话窗口的截图。"
- "左边的灰色气泡是对方（客户）发的消息，右边的蓝色气泡是我方之前的回复。"
- "请只关注对方（客户）发的最后一条消息，针对那条消息直接回复。"
- "像真人微信聊天，先关心一句，再说重点；默认1～2句、20～60字，最多问一个问题。"
- "只输出回复内容，不要描述图片，不要解释，不要加引号、标题或列表。"
- ),
- },
+ {"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{b64}"},
@@ -550,22 +1264,34 @@ def call_ai_vision(image_bytes: bytes, history: list = None) -> str:
"temperature": ai_config.AI_TEMPERATURE,
}
url = _completions_url()
+ _log_request_diagnostics(url, "OpenAI 兼容视觉请求")
resp = requests.post(url, headers=simple_headers, json=payload, timeout=ai_config.AI_TIMEOUT)
resp.raise_for_status()
content = resp.json()["choices"][0]["message"]["content"]
- return _humanize(_strip_thinking(content))
+ return _finalize_vision_reply(content, chat_text, media_types)
-def get_ai_reply(chat_text: str = None, image_bytes: bytes = None, history: list = None) -> str:
- """
- 统一入口：根据 AI_USE_VISION 配置自动选择模式。
- history 为该会话的历史上下文（多轮记忆），可为 None。
- 返回 AI 生成的回复文本。
- """
+def get_ai_reply(
+ chat_text: str = None,
+ image_bytes: bytes = None,
+ history: list = None,
+ *,
+ force_vision: bool = False,
+ media_types=None,
+) -> str:
+ """Unified text/vision entry; media fallback may explicitly force vision."""
try:
- if ai_config.AI_USE_VISION and image_bytes:
+ if _provider_type() == "comfyui":
+ print(" [AI] [!] 当前配置为 ComfyUI 文生图服务，不能用于企微文本自动回复")
+ return ""
+ if image_bytes and (force_vision or ai_config.AI_USE_VISION):
print(" [AI] 使用视觉模式分析聊天截图...")
- return call_ai_vision(image_bytes, history=history)
+ return call_ai_vision(
+ image_bytes,
+ history=history,
+ chat_text=chat_text or "",
+ media_types=media_types,
+ )
elif chat_text:
if getattr(ai_config, "AI_MCP_ENABLED", False):
print(" [AI] 文本模式 + MCP 工具增强...")
@@ -575,11 +1301,11 @@ def get_ai_reply(chat_text: str = None, image_bytes: bytes = None, history: list
else:
return ""
except requests.exceptions.Timeout:
- print(" [AI] ⚠ API 请求超时")
+ print(" [AI] [!] API 请求超时")
return ""
except requests.exceptions.RequestException as e:
- print(f" [AI] ⚠ API 请求失败: {e}")
+ print(f" [AI] [!] API 请求失败: {e}")
return ""
except (KeyError, IndexError, json.JSONDecodeError) as e:
- print(f" [AI] ⚠ 解析响应失败: {e}")
+ print(f" [AI] [!] 解析响应失败: {e}")
return ""
diff --git a/wechat_rpa/ai_config.py b/wechat_rpa/ai_config.py
index e775f93..d54c360 100644
--- a/wechat_rpa/ai_config.py
+++ b/wechat_rpa/ai_config.py
@@ -13,11 +13,13 @@ from runtime_paths import application_data_dir
# ── 是否启用 AI 回复（False 时使用固定文本回复）──
AI_ENABLED = True
+AI_DEVELOPMENT_MODE = False
# ── API 配置 ──
# 可为根地址（如 https://api.deepseek.com）或完整端点（如 .../v1/chat-messages）；
# /v1/ 后已有路径时不再自动拼接 /chat/completions
# Dify：填 .../v1/chat-messages，AI_API_KEY 用应用「访问 API」里的 Key（通常 app- 开头）
+AI_PROVIDER_TYPE = "auto" # auto / openai / dify / comfyui
AI_API_BASE = "https://api.deepseek.com"
AI_API_KEY = "sk-992b66aec315400d92848a676acb0e99" # 你的 API Key
AI_MODEL = "deepseek-chat" # 模型名称（Dify 应用侧选模型时此项可忽略）
@@ -34,6 +36,11 @@ AI_CONTEXT_MAX_ROUNDS = 5 # 每个会话最多记忆的历史轮数（1 轮
# False → 尝试通过框选复制识别聊天文字，再发给文本 AI
AI_USE_VISION = False
+# ── AI 页面守护 ──────────────────────────────────────────────────────────────
+# 只在本地页面识别无法确定、会话切换被拦截、消息提取失败或发送前校验异常时，
+# 才把当前企业微信窗口交给视觉模型判断。模型仅能建议安全动作，不能自由点击。
+AI_UI_GUARD_ENABLED = True
+
# ── 客服身份（出现在提示词里，可按需修改为你的真实昵称/工号）──
AI_AGENT_NAME = "高兴亮"
AI_CLOUD_AGENT_NAME = AI_AGENT_NAME
@@ -224,8 +231,8 @@ _AGENT_OVERRIDE_FILE = str(_DATA_DIR / "ai_agent_override.json")
# 允许通过 GUI 修改并持久化的配置项
CONFIGURABLE_KEYS = [
- "AI_ENABLED", "AI_API_BASE", "AI_API_KEY", "AI_MODEL",
- "AI_USE_VISION", "AI_CONTEXT_ENABLED", "AI_CONTEXT_MAX_ROUNDS",
+ "AI_ENABLED", "AI_DEVELOPMENT_MODE", "AI_PROVIDER_TYPE", "AI_API_BASE", "AI_API_KEY", "AI_MODEL",
+ "AI_USE_VISION", "AI_UI_GUARD_ENABLED", "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",
diff --git a/wechat_rpa/ai_settings.json b/wechat_rpa/ai_settings.json
index b31d542..1e7cf64 100644
--- a/wechat_rpa/ai_settings.json
+++ b/wechat_rpa/ai_settings.json
@@ -1,9 +1,12 @@
{
"AI_ENABLED": true,
- "AI_API_BASE": "http://chat2.zhenyangtang.com.cn:8088/v1",
+ "AI_DEVELOPMENT_MODE": true,
+ "AI_PROVIDER_TYPE": "dify",
+ "AI_API_BASE": "http://ai.zhenyangtang.com.cn/v1",
"AI_API_KEY": "app-TMCZfuo5Jj8lxgbL6shMaDCc",
"AI_MODEL": "gpt-5.6-sol",
- "AI_USE_VISION": false,
+ "AI_USE_VISION": true,
+ "AI_UI_GUARD_ENABLED": true,
"AI_CONTEXT_ENABLED": true,
"AI_CONTEXT_MAX_ROUNDS": 5,
"AI_COUNTER_INSULT_ENABLED": false,
diff --git a/wechat_rpa/app_settings.json b/wechat_rpa/app_settings.json
index 1e96043..72062ab 100644
--- a/wechat_rpa/app_settings.json
+++ b/wechat_rpa/app_settings.json
@@ -1,6 +1,7 @@
{
- "auto_reply_text": "在的，您慢慢说，我这边看着呢。",
+ "auto_reply_text": "在的",
"poll_interval": 2.0,
- "mouse_idle_enabled": false,
- "mouse_idle_seconds": 10.0
+ "mouse_idle_enabled": true,
+ "mouse_idle_seconds": 5.0,
+ "message_batch_window_seconds": 2.0
}
\ No newline at end of file
diff --git a/wechat_rpa/app_version.py b/wechat_rpa/app_version.py
index fca64d9..dceb901 100644
--- a/wechat_rpa/app_version.py
+++ b/wechat_rpa/app_version.py
@@ -7,7 +7,7 @@ import re
from typing import Any
-APP_VERSION = "1.0.0"
+APP_VERSION = "1.1.0"
_VERSION_PATTERN = re.compile(
r"^v?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
r"(?:[-+][0-9A-Za-z.-]+)?$"
diff --git a/wechat_rpa/backend_client.py b/wechat_rpa/backend_client.py
index 2537ae0..acd7d40 100644
--- a/wechat_rpa/backend_client.py
+++ b/wechat_rpa/backend_client.py
@@ -5,6 +5,7 @@ from __future__ import annotations
import json
import os
+import re
import socket
import threading
import time
@@ -25,6 +26,23 @@ DEFAULT_SERVER_URL = "http://xchat.zhenyangtang.com.cn"
DESKTOP_SYNC_KEY = "wcrpa-v1-H3q9mT7xK2pN8cR5vL4sF6dB1yG0uJ"
DESKTOP_CONFIG_PATH = "/api/v1/desktop/config"
_LOCK = threading.RLock()
+_SENSITIVE_NAME_PARTS = (
+ "api_key",
+ "apikey",
+ "key",
+ "token",
+ "secret",
+ "password",
+ "passwd",
+ "authorization",
+ "cookie",
+ "credential",
+ "signature",
+ "auth",
+)
+_INLINE_SECRET_RE = re.compile(
+ r"(?i)(--?(?:api[-_]?key|token|secret|password|passwd|authorization|auth)\s*[=:]\s*)([^\s,;]+)"
+)
class BackendError(RuntimeError):
@@ -197,6 +215,96 @@ def normalize_server_url(value: Any) -> str:
return url
+def _development_mode_enabled(config: dict[str, Any]) -> bool:
+ value = config.get("AI_DEVELOPMENT_MODE", False)
+ if isinstance(value, str):
+ return value.strip().lower() in {"1", "true", "yes", "on"}
+ return bool(value)
+
+
+def _sensitive_name(name: Any) -> bool:
+ normalized = str(name or "").strip().lower().replace("-", "_")
+ return any(part in normalized for part in _SENSITIVE_NAME_PARTS)
+
+
+def redact_diagnostic_value(value: Any, *, field_name: str = "") -> Any:
+ """Return a JSON-safe diagnostic copy with credentials removed."""
+ if _sensitive_name(field_name):
+ return "[已配置，值已隐藏]" if value not in (None, "", [], {}) else "[未配置]"
+ if isinstance(value, dict):
+ return {
+ str(key): redact_diagnostic_value(item, field_name=str(key))
+ for key, item in value.items()
+ }
+ if isinstance(value, list):
+ result = []
+ hide_next = False
+ for item in value:
+ if hide_next:
+ result.append("[值已隐藏]")
+ hide_next = False
+ continue
+ result.append(redact_diagnostic_value(item))
+ if isinstance(item, str) and _sensitive_name(item.lstrip("-")):
+ hide_next = True
+ return result
+ if isinstance(value, str):
+ text = value
+ if re.match(r"(?i)^bearer\s+\S+", text.strip()):
+ return "[认证值已隐藏]"
+ if re.match(r"(?i)^(?:sk|app)-[A-Za-z0-9_.-]{8,}$", text.strip()):
+ return "[密钥值已隐藏]"
+ text = _INLINE_SECRET_RE.sub(r"\1[值已隐藏]", text)
+ if text.startswith(("http://", "https://")):
+ return diagnostic_url(text)
+ return text
+ return value
+
+
+def diagnostic_url(value: Any) -> str:
+ """Keep the request destination visible while hiding credentials in its query."""
+ url = str(value or "")
+ try:
+ parsed = urllib.parse.urlsplit(url)
+ if not parsed.scheme or not parsed.netloc:
+ return url
+ hostname = parsed.hostname or ""
+ if ":" in hostname and not hostname.startswith("["):
+ hostname = f"[{hostname}]"
+ netloc = hostname
+ if parsed.port:
+ netloc += f":{parsed.port}"
+ query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
+ safe_query = urllib.parse.urlencode(
+ [
+ (name, "[值已隐藏]" if _sensitive_name(name) else item)
+ for name, item in query
+ ]
+ )
+ return urllib.parse.urlunsplit(
+ (parsed.scheme, netloc, parsed.path, safe_query, "")
+ )
+ except (TypeError, ValueError):
+ return url
+
+
+def _config_diagnostics(
+ config: dict[str, Any], response: dict[str, Any], request_url: str
+) -> list[str]:
+ if not _development_mode_enabled(config):
+ return []
+ safe_config = redact_diagnostic_value(config)
+ config_json = json.dumps(safe_config, ensure_ascii=False, indent=2, sort_keys=True)
+ return [
+ f"[开发模式] 云端配置请求地址: {diagnostic_url(request_url)}",
+ (
+ f"[开发模式] 云端配置版本: v{int(response.get('version', 0))}"
+ f"；更新时间: {response.get('updated_at') or '未提供'}"
+ ),
+ f"[开发模式] 获取到的云端配置（敏感值已隐藏）:\n{config_json}",
+ ]
+
+
def is_configured(settings: dict[str, Any] | None = None) -> bool:
current = settings or load_settings()
return bool(
@@ -331,7 +439,7 @@ def logout(*, revoke_remote: bool = True) -> None:
def _apply_config_response(
- response: dict[str, Any], settings: dict[str, Any]
+ response: dict[str, Any], settings: dict[str, Any], *, request_url: str = ""
) -> dict[str, Any]:
config = response.get("config")
if not isinstance(config, dict):
@@ -370,6 +478,7 @@ def _apply_config_response(
"local_app_version": APP_VERSION,
"update_available": release["update_available"],
"force_upgrade": release["force_upgrade"],
+ "diagnostics": _config_diagnostics(config, response, request_url),
}
@@ -396,7 +505,11 @@ def sync_config(*, force: bool = False, timeout: float = 10.0) -> dict[str, Any]
),
timeout=timeout,
)
- return _apply_config_response(response, settings)
+ return _apply_config_response(
+ response,
+ settings,
+ request_url=normalize_server_url(settings["server_url"]) + "/api/v1/config",
+ )
except Exception as exc:
settings["last_error"] = str(exc)
save_settings(settings)
@@ -422,7 +535,11 @@ def sync_cloud_config(*, timeout: float = 10.0) -> dict[str, Any]:
desktop_sync_key=DESKTOP_SYNC_KEY,
timeout=timeout,
)
- return _apply_config_response(response, settings)
+ return _apply_config_response(
+ response,
+ settings,
+ request_url=normalize_server_url(DEFAULT_SERVER_URL) + DESKTOP_CONFIG_PATH,
+ )
except Exception as exc:
settings["last_error"] = str(exc)
save_settings(settings)
diff --git a/wechat_rpa/backend_deploy/wechat-backend-baota/.env.example b/wechat_rpa/backend_deploy/wechat-backend-baota/.env.example
new file mode 100644
index 0000000..a9949fe
--- /dev/null
+++ b/wechat_rpa/backend_deploy/wechat-backend-baota/.env.example
@@ -0,0 +1,3 @@
+# 至少 10 位，且同时包含字母和数字。首次登录后仍会要求修改密码。
+WECOM_ADMIN_INITIAL_PASSWORD=ChangeThisPassword123
+BACKEND_PORT=8765
diff --git a/wechat_rpa/backend_deploy/wechat-backend-baota/Dockerfile b/wechat_rpa/backend_deploy/wechat-backend-baota/Dockerfile
new file mode 100644
index 0000000..a09530a
--- /dev/null
+++ b/wechat_rpa/backend_deploy/wechat-backend-baota/Dockerfile
@@ -0,0 +1,20 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+ PYTHONUNBUFFERED=1
+
+RUN useradd --create-home --uid 10001 backend \
+ && mkdir -p /data \
+ && chown backend:backend /data
+
+WORKDIR /app
+COPY --chown=backend:backend admin_backend.py /app/admin_backend.py
+
+USER backend
+EXPOSE 8765
+VOLUME ["/data"]
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+ CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8765/health', timeout=3)"
+
+CMD ["python", "/app/admin_backend.py", "--host", "0.0.0.0", "--port", "8765", "--db", "/data/backend.db", "--runtime-file", "/tmp/backend_runtime.json"]
diff --git a/wechat_rpa/backend_deploy/wechat-backend-baota/README.md b/wechat_rpa/backend_deploy/wechat-backend-baota/README.md
new file mode 100644
index 0000000..e9437dc
--- /dev/null
+++ b/wechat_rpa/backend_deploy/wechat-backend-baota/README.md
@@ -0,0 +1,18 @@
+# 企微客服配置后台 · 宝塔部署包
+
+1. 将整个目录上传并解压到 `/www/wwwroot/wechat-backend`。
+2. 把 `.env.example` 复制为 `.env`，设置初始管理员强密码。
+3. 在宝塔终端进入该目录，执行：
+
+```bash
+docker compose --env-file .env up -d --build
+curl http://127.0.0.1:8765/health
+```
+
+4. 在宝塔创建带 SSL 的网站，将域名反向代理到 `http://127.0.0.1:8765`。
+5. 登录后台，在“桌面端版本升级”中填写最新版本号、下载地址和更新说明。
+6. 开启“强制升级”后，版本不一致的桌面端只能下载新版或退出；未开启时允许稍后继续使用。
+
+首次启用强制升级前，请务必确认下载地址能从客户电脑正常打开。
+
+完整步骤见项目中的 `BAOTA_DEPLOY.md`。
diff --git a/wechat_rpa/backend_deploy/wechat-backend-baota/admin_backend.py b/wechat_rpa/backend_deploy/wechat-backend-baota/admin_backend.py
new file mode 100644
index 0000000..7471ceb
--- /dev/null
+++ b/wechat_rpa/backend_deploy/wechat-backend-baota/admin_backend.py
@@ -0,0 +1,1782 @@
+# -*- 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 http.client
+import ipaddress
+import json
+import os
+import re
+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"
+DEFAULT_APP_VERSION = "1.0.0"
+DEFAULT_DESKTOP_SYNC_KEY = "wcrpa-v1-H3q9mT7xK2pN8cR5vL4sF6dB1yG0uJ"
+DESKTOP_SYNC_KEY = os.environ.get(
+ "WECOM_DESKTOP_SYNC_KEY", DEFAULT_DESKTOP_SYNC_KEY
+).strip()
+ROLES = ("admin", "operator", "viewer")
+ROLE_LABELS = {"admin": "管理员", "operator": "配置员", "viewer": "只读用户"}
+APP_VERSION_PATTERN = re.compile(
+ r"^v?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
+ r"(?:[-+][0-9A-Za-z.-]+)?$"
+)
+PBKDF2_ITERATIONS = 310_000
+CONFIG_KEYS = (
+ "AI_ENABLED",
+ "AI_PROVIDER_TYPE",
+ "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",
+}
+PROVIDER_TYPES = {
+ "openai": "OpenAI 兼容（GPT / DeepSeek / vLLM / SGLang 等）",
+ "dify": "Dify 应用（chat-messages 接口）",
+ "comfyui": "ComfyUI 文生图",
+}
+
+
+def pbkdf2_sha256(
+ password: bytes, salt: bytes, iterations: int = PBKDF2_ITERATIONS
+) -> bytes:
+ """PBKDF2-HMAC-SHA256，兼容未编译 OpenSSL PBKDF2 的 Python。"""
+ native_pbkdf2 = getattr(hashlib, "pbkdf2_hmac", None)
+ if callable(native_pbkdf2):
+ return native_pbkdf2("sha256", password, salt, iterations)
+
+ # SHA-256 的输出正好是本项目需要的 32 字节，因此只需计算一个块。
+ current = hmac.new(password, salt + b"\x00\x00\x00\x01", hashlib.sha256).digest()
+ derived = int.from_bytes(current, "big")
+ for _ in range(1, iterations):
+ current = hmac.new(password, current, hashlib.sha256).digest()
+ derived ^= int.from_bytes(current, "big")
+ return derived.to_bytes(32, "big")
+
+
+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 = pbkdf2_sha256(password.encode("utf-8"), salt)
+ 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 = pbkdf2_sha256(password.encode("utf-8"), salt)
+ return hmac.compare_digest(actual, expected)
+
+
+def token_hash(token: str) -> str:
+ return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+def provider_type(config: dict[str, Any]) -> str:
+ value = str(config.get("AI_PROVIDER_TYPE") or "").strip().lower()
+ if value in PROVIDER_TYPES:
+ return value
+ base = str(config.get("AI_API_BASE") or "").lower()
+ if "chat-messages" in base or "completion-messages" in base:
+ return "dify"
+ if "system_stats" in base or "comfyui" in base:
+ return "comfyui"
+ try:
+ if urllib.parse.urlparse(base).port == 8188:
+ return "comfyui"
+ except ValueError:
+ pass
+ return "openai"
+
+
+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_PROVIDER_TYPE": "openai",
+ "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})
+ defaults["AI_PROVIDER_TYPE"] = provider_type(defaults)
+ 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 app_release (
+ id INTEGER PRIMARY KEY CHECK(id = 1),
+ latest_version TEXT NOT NULL,
+ download_url TEXT NOT NULL DEFAULT '',
+ release_notes TEXT NOT NULL DEFAULT '',
+ force_upgrade INTEGER NOT NULL DEFAULT 0,
+ 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,
+ ),
+ )
+ if db.execute("SELECT COUNT(*) FROM app_release").fetchone()[0] == 0:
+ admin = db.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
+ db.execute(
+ """INSERT INTO app_release
+ (id,latest_version,download_url,release_notes,force_upgrade,updated_at,updated_by)
+ VALUES (1,?,?,?,0,?,?)""",
+ (
+ DEFAULT_APP_VERSION,
+ "",
+ "",
+ 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
+
+ def release(self) -> sqlite3.Row:
+ with self.connect() as db:
+ return db.execute(
+ """SELECT r.*, u.username AS updated_by_name
+ FROM app_release r LEFT JOIN users u ON u.id=r.updated_by WHERE r.id=1"""
+ ).fetchone()
+
+ def save_release(
+ self,
+ latest_version: str,
+ download_url: str,
+ release_notes: str,
+ force_upgrade: bool,
+ user_id: int,
+ ip: str,
+ ) -> None:
+ with self.connect() as db:
+ db.execute(
+ """UPDATE app_release SET latest_version=?,download_url=?,release_notes=?,
+ force_upgrade=?,updated_at=?,updated_by=? WHERE id=1""",
+ (
+ latest_version,
+ download_url,
+ release_notes,
+ int(force_upgrade),
+ now_text(),
+ user_id,
+ ),
+ )
+ self._audit(
+ db,
+ user_id,
+ "release.update",
+ f"version={latest_version}, force={int(force_upgrade)}",
+ ip,
+ )
+ db.commit()
+
+ @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()
+
+
+def _model_endpoint(api_base: str, provider: str) -> str:
+ base = str(api_base or "").strip().rstrip("/")
+ parsed = urllib.parse.urlparse(base)
+ if parsed.scheme not in ("http", "https") or not parsed.netloc:
+ raise ValueError("API 地址必须是完整的 http 或 https 地址")
+ if parsed.query or parsed.fragment:
+ raise ValueError("API 地址不能包含查询参数或片段")
+ try:
+ parsed.port
+ except ValueError as exc:
+ raise ValueError("API 地址中的端口无效") from exc
+ path = (parsed.path or "").rstrip("/")
+ lower_path = path.lower()
+ if provider == "dify":
+ if lower_path.endswith(("/chat-messages", "/completion-messages")):
+ return base
+ if lower_path.endswith("/v1"):
+ return f"{base}/chat-messages"
+ return f"{base}/v1/chat-messages"
+ if provider == "comfyui":
+ if lower_path.endswith("/system_stats"):
+ return base
+ return f"{base}/system_stats"
+ if lower_path.endswith("/chat/completions"):
+ return base
+ if re.match(r"^/v1/.+", path):
+ return base
+ return f"{base}/chat/completions"
+
+
+def model_test_config(values: dict[str, Any], current: dict[str, Any]) -> dict[str, Any]:
+ """合并页面临时值与已保存密钥，不写入数据库。"""
+ api_base_value = (
+ values.get("AI_API_BASE") if "AI_API_BASE" in values else current.get("AI_API_BASE")
+ )
+ model_value = values.get("AI_MODEL") if "AI_MODEL" in values else current.get("AI_MODEL")
+ api_base = str(api_base_value or "").strip()
+ model = str(model_value or "").strip()
+ merged_provider = {
+ "AI_PROVIDER_TYPE": values.get(
+ "AI_PROVIDER_TYPE", current.get("AI_PROVIDER_TYPE")
+ ),
+ "AI_API_BASE": api_base,
+ }
+ provider = provider_type(merged_provider)
+ if "AI_PROVIDER_TYPE" in values:
+ requested_provider = str(values.get("AI_PROVIDER_TYPE") or "").strip().lower()
+ if requested_provider not in PROVIDER_TYPES:
+ raise ValueError("服务类型无效")
+ provider = requested_provider
+ supplied_key = str(values.get("AI_API_KEY") or "").strip()
+ api_key = supplied_key or str(current.get("AI_API_KEY") or "").strip()
+ raw_timeout = values.get("AI_TIMEOUT", current.get("AI_TIMEOUT", 30))
+ try:
+ timeout = int(raw_timeout)
+ except (TypeError, ValueError) as exc:
+ raise ValueError("请求超时必须是整数") from exc
+ timeout = min(60, max(5, timeout))
+ endpoint = _model_endpoint(api_base, provider)
+ if provider == "dify" and not api_key:
+ raise ValueError("Dify 连通性测试需要 API Key")
+ if provider == "openai" and not model:
+ raise ValueError("OpenAI 兼容接口的模型名称不能为空")
+ return {
+ "api_base": api_base,
+ "api_key": api_key,
+ "model": model,
+ "timeout": timeout,
+ "endpoint": endpoint,
+ "provider_type": provider,
+ }
+
+
+def _safe_endpoint_label(endpoint: str) -> str:
+ parsed = urllib.parse.urlparse(endpoint)
+ host = parsed.hostname or ""
+ if parsed.port:
+ host = f"{host}:{parsed.port}"
+ return urllib.parse.urlunparse((parsed.scheme, host, parsed.path, "", "", ""))
+
+
+def _remote_error_detail(raw: bytes, api_key: str) -> str:
+ text = raw.decode("utf-8", errors="replace")[:1000].strip()
+ try:
+ data = json.loads(text)
+ error = data.get("error") if isinstance(data, dict) else None
+ if isinstance(error, dict):
+ text = str(error.get("message") or error.get("code") or text)
+ elif isinstance(data, dict):
+ text = str(data.get("message") or data.get("detail") or text)
+ except (TypeError, ValueError):
+ pass
+ if api_key:
+ text = text.replace(api_key, "[已隐藏]")
+ return " ".join(text.split())[:300]
+
+
+def _model_answer(data: dict[str, Any], provider: str) -> str:
+ if provider == "dify":
+ answer = data.get("answer")
+ if answer is None and isinstance(data.get("data"), dict):
+ answer = data["data"].get("answer")
+ return str(answer or "").strip()
+ if provider == "comfyui":
+ return "ComfyUI system_stats 可用" if data else ""
+ choices = data.get("choices")
+ if not isinstance(choices, list) or not choices:
+ return ""
+ message = choices[0].get("message") if isinstance(choices[0], dict) else None
+ content = message.get("content") if isinstance(message, dict) else ""
+ if isinstance(content, list):
+ content = " ".join(
+ str(item.get("text") or "") for item in content if isinstance(item, dict)
+ )
+ return str(content or "").strip()
+
+
+def _perform_http_request(
+ endpoint: str,
+ *,
+ method: str,
+ headers: dict[str, str],
+ payload: dict[str, Any] | None,
+ timeout: int,
+) -> tuple[int, bytes]:
+ """直接使用 http.client，避免部分精简 Python 缺少 urllib HTTPSHandler。"""
+ parsed = urllib.parse.urlparse(endpoint)
+ host = parsed.hostname
+ if not host:
+ raise ValueError("API 地址缺少主机名")
+ path = urllib.parse.urlunparse(("", "", parsed.path or "/", "", parsed.query, ""))
+ if parsed.scheme == "https":
+ connection_class = getattr(http.client, "HTTPSConnection", None)
+ if connection_class is None:
+ raise OSError("当前后端 Python 环境缺少 HTTPS/SSL 支持")
+ elif parsed.scheme == "http":
+ connection_class = http.client.HTTPConnection
+ else:
+ raise ValueError(f"不支持的 URL 协议：{parsed.scheme or '空'}")
+ connection = connection_class(host, parsed.port, timeout=timeout)
+ body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
+ try:
+ connection.request(method, path, body=body, headers=headers)
+ response = connection.getresponse()
+ raw = response.read(1_000_001)
+ return int(response.status), raw
+ finally:
+ connection.close()
+
+
+def test_model_connection(config: dict[str, Any]) -> dict[str, Any]:
+ """向模型发出一个最小请求，返回不含密钥的诊断结果。"""
+ endpoint = str(config["endpoint"])
+ api_key = str(config.get("api_key") or "")
+ provider_type_value = str(config.get("provider_type") or "openai")
+ provider = PROVIDER_TYPES.get(provider_type_value, provider_type_value)
+ if provider_type_value == "dify":
+ payload = {
+ "inputs": {},
+ "query": "连通性测试：请只回复 OK。",
+ "response_mode": "blocking",
+ "user": "zhen-ai-backend-test",
+ }
+ method = "POST"
+ elif provider_type_value == "comfyui":
+ payload = None
+ method = "GET"
+ else:
+ payload = {
+ "model": config["model"],
+ "messages": [{"role": "user", "content": "连通性测试：请只回复 OK。"}],
+ "stream": False,
+ }
+ method = "POST"
+ headers = {
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ "User-Agent": "ZhenAI-Backend-Connectivity-Test/1.0",
+ }
+ if api_key:
+ headers["Authorization"] = f"Bearer {api_key}"
+ started = time.perf_counter()
+ try:
+ status, raw = _perform_http_request(
+ endpoint,
+ method=method,
+ headers=headers,
+ payload=payload,
+ timeout=int(config["timeout"]),
+ )
+ latency_ms = round((time.perf_counter() - started) * 1000)
+ if len(raw) > 1_000_000:
+ raise ValueError("模型响应过大，已停止读取")
+ if not 200 <= status < 300:
+ detail = _remote_error_detail(raw, api_key)
+ labels = {
+ 400: "请求参数不被模型服务接受",
+ 401: "API Key 无效或缺少鉴权",
+ 403: "当前 API Key 没有访问权限",
+ 404: "接口地址或模型名称不存在",
+ 429: "请求受限、余额不足或调用频率过高",
+ }
+ message = labels.get(status, f"模型服务返回 HTTP {status}")
+ if detail:
+ message += f"：{detail}"
+ return {
+ "ok": False,
+ "provider": provider,
+ "model": str(config.get("model") or "-"),
+ "endpoint": _safe_endpoint_label(endpoint),
+ "http_status": status,
+ "latency_ms": latency_ms,
+ "message": message,
+ }
+ try:
+ data = json.loads(raw.decode("utf-8"))
+ except (UnicodeDecodeError, ValueError) as exc:
+ raise ValueError("模型已响应，但返回内容不是有效 JSON") from exc
+ if not isinstance(data, dict):
+ raise ValueError("模型已响应，但返回 JSON 不是对象")
+ answer = _model_answer(data, provider_type_value)
+ if not answer:
+ raise ValueError("模型已响应，但未返回可识别的回复内容")
+ if api_key:
+ answer = answer.replace(api_key, "[已隐藏]")
+ return {
+ "ok": True,
+ "provider": provider,
+ "model": str(config.get("model") or "-"),
+ "endpoint": _safe_endpoint_label(endpoint),
+ "http_status": status,
+ "latency_ms": latency_ms,
+ "message": f"连接成功，模型回复：{answer[:120]}",
+ }
+ except (TimeoutError, OSError, http.client.HTTPException, ValueError) as exc:
+ latency_ms = round((time.perf_counter() - started) * 1000)
+ reason = getattr(exc, "reason", exc)
+ message = "连接超时" if isinstance(reason, TimeoutError) else str(reason)
+ if api_key:
+ message = message.replace(api_key, "[已隐藏]")
+ return {
+ "ok": False,
+ "provider": provider,
+ "model": str(config.get("model") or "-"),
+ "endpoint": _safe_endpoint_label(endpoint),
+ "http_status": None,
+ "latency_ms": latency_ms,
+ "message": f"连接失败：{' '.join(message.split())[:300]}",
+ }
+
+
+def model_test_result_page(result: dict[str, Any]) -> str:
+ ok = bool(result.get("ok"))
+ title = "模型连接成功" if ok else "模型连接失败"
+ tone = "flash" if ok else "flash error"
+ http_status = result.get("http_status")
+ status_text = str(http_status) if http_status is not None else "未建立 HTTP 响应"
+ body = f"""
+
+ ZHEN AI ADMIN
+ {title}
+ {html.escape(str(result.get('message') or ''))}
+
+
+ 测试不会保存页面配置，也不会显示 API Key。
+ """
+ return page(title, body)
+
+
+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, *, roles: tuple[str, ...] | None = None
+ ) -> 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
+ if roles and user["role"] not in roles:
+ 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)
+
+ def desktop_sync_authorized(self) -> bool:
+ supplied = self.headers.get("X-Desktop-Sync-Key", "")
+ return bool(
+ supplied
+ and DESKTOP_SYNC_KEY
+ and hmac.compare_digest(supplied, DESKTOP_SYNC_KEY)
+ )
+
+ def config_payload(self) -> dict[str, Any]:
+ row = self.db.config()
+ release = self.db.release()
+ return {
+ "version": row["version"],
+ "updated_at": row["updated_at"],
+ "updated_by": row["updated_by_name"] or "system",
+ "config": json.loads(row["config_json"]),
+ "release": {
+ "latest_version": release["latest_version"],
+ "download_url": release["download_url"],
+ "release_notes": release["release_notes"],
+ "force_upgrade": bool(release["force_upgrade"]),
+ "updated_at": release["updated_at"],
+ },
+ }
+
+ @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:
+ self.json_response(HTTPStatus.OK, self.config_payload())
+ elif path == "/api/v1/desktop/config":
+ if self.desktop_sync_authorized():
+ self.json_response(HTTPStatus.OK, self.config_payload())
+ else:
+ self.json_response(HTTPStatus.UNAUTHORIZED, {"error": "桌面端同步凭证无效"})
+ 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/model/test":
+ self.web_test_model()
+ elif path == "/admin/release":
+ self.web_save_release()
+ 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()
+ elif path == "/api/v1/model/test":
+ self.api_test_model()
+ 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_test_model(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"])
+ try:
+ result = test_model_connection(model_test_config(form, current))
+ except ValueError as exc:
+ result = {
+ "ok": False,
+ "provider": "-",
+ "model": form.get("AI_MODEL", ""),
+ "endpoint": "",
+ "http_status": None,
+ "latency_ms": 0,
+ "message": str(exc),
+ }
+ self.db.audit(
+ user["id"],
+ "model.test",
+ f"ok={int(bool(result['ok']))}, model={str(result.get('model') or '')[:80]}, "
+ f"endpoint={str(result.get('endpoint') or '')[:200]}, http={result.get('http_status')}",
+ self.client_ip,
+ )
+ self.html_response(HTTPStatus.OK, model_test_result_page(result))
+
+ def api_test_model(self) -> None:
+ user = self.require_api_auth(roles=("admin", "operator"))
+ if not user:
+ return
+ current = json.loads(self.db.config()["config_json"])
+ result = test_model_connection(model_test_config(self.json_body(), current))
+ self.db.audit(
+ user["id"],
+ "model.test.api",
+ f"ok={int(bool(result['ok']))}, model={str(result.get('model') or '')[:80]}, "
+ f"endpoint={str(result.get('endpoint') or '')[:200]}, http={result.get('http_status')}",
+ self.client_ip,
+ )
+ status = HTTPStatus.OK if result["ok"] else HTTPStatus.BAD_GATEWAY
+ self.json_response(status, result)
+
+ def web_save_release(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("页面已过期，请刷新后重试")
+ release = validate_release_form(form)
+ self.db.save_release(
+ release["latest_version"],
+ release["download_url"],
+ release["release_notes"],
+ release["force_upgrade"],
+ user["id"],
+ self.client_ip,
+ )
+ self.redirect(
+ "/?message="
+ + urllib.parse.quote(f"桌面端版本策略已更新为 v{release['latest_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)}
+ 模型配置中心
保存后，已登录桌面端会在启动或定时同步时自动应用。
+
应用 v{html.escape(release['latest_version'])} · 配置 v{int(row['version'])}
{release_card}{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 ""
+ selected_provider = provider_type(config)
+ provider_options = "".join(
+ f""
+ for key, label in PROVIDER_TYPES.items()
+ )
+ 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')}
"""
+
+ @staticmethod
+ def release_card(user: sqlite3.Row, csrf: str, release: sqlite3.Row) -> str:
+ can_edit = user["role"] in ("admin", "operator")
+ disabled = " disabled" if not can_edit else ""
+ checked = " checked" if release["force_upgrade"] else ""
+ latest = html.escape(release["latest_version"], quote=True)
+ download_url = html.escape(release["download_url"], quote=True)
+ notes = html.escape(release["release_notes"])
+ submit = (
+ ""
+ if can_edit
+ else "当前为只读角色，不能修改版本策略。
"
+ )
+ return f"""
+ 桌面端版本升级
+
软件每次打开都会检查这里的版本策略。
最后更新：{html.escape(release['updated_at'])} · {html.escape(release['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 '
暂无记录
'}