# chat_bridge.py import json from PyQt6.QtCore import QObject, pyqtSlot, pyqtSignal class ChatBridge(QObject): # 信号定义 regenerate_clicked = pyqtSignal(str) # 携带需要重新回答的助手消息ID branch_switch_clicked = pyqtSignal(str, int) # 分支切换信号 (msg_id, direction: -1 为上一条, 1 为下一条) delete_message_requested = pyqtSignal(str) # 删除消息信号 (msg_id) attachment_clicked = pyqtSignal(str) scroll_changed = pyqtSignal(float, float, float) # 页面 scrollY / scrollHeight / innerHeight def __init__(self, page, channel, js_runner=None): """ :param page: QWebEnginePage 实例（QtWebEngine 路径） :param channel: QWebChannel 实例（QtWebEngine 路径） :param js_runner: 可选 JS 执行函数（WebView2 路径：script -> None） """ super().__init__() self.page = page self._js_runner = js_runner if channel is not None: # 将自身注册到 channel 中，前端可通过 bridge 对象调用 channel.registerObject("bridge", self) print("ChatBridge被初始化") def run_js(self, script: str): if self._js_runner is not None: self._js_runner(script) elif self.page is not None: self.page.runJavaScript(script) else: print("[ChatBridge] run_js: 无可用执行通道") # ---------- 消息构造 (Python -> JS) ---------- def create_message(self, msg_id: str, role: str, text: str = "", sender_name: str = "", branch_info: dict = None): """ 在前端创建一条消息容器 :param branch_info: 格式如 {"current": 0, "total": 2}，若为 None 则不显示分支切换器 """ safe_text = json.dumps(text) safe_name = json.dumps(sender_name) safe_branch = json.dumps(branch_info) if branch_info else "null" self.run_js(f"createMessage('{msg_id}', '{role}', {safe_text}, {safe_name}, {safe_branch});") def create_long_message(self, msg_id: str, role: str, text: str, sender_name: str = ""): safe_text = json.dumps(text) safe_name = json.dumps(sender_name) size_kb = round(len(text.encode('utf-8')) / 1024, 2) self.run_js(f"createLongMessage('{msg_id}', '{role}', {safe_text}, {safe_name}, {size_kb});") def create_user_message_with_attachments(self, msg_id: str, text: str, attachments: list): safe_text = json.dumps(text) safe_attachments = json.dumps(attachments) self.run_js(f"createUserMessageWithAttachments('{msg_id}', {safe_text}, {safe_attachments});") # ---------- 流式输出 (Python -> JS) ---------- def append_token(self, msg_id: str, token: str): safe_token = json.dumps(token) self.run_js(f"appendToken('{msg_id}', {safe_token});") def append_reasoning(self, msg_id: str, token: str): safe_token = json.dumps(token) self.run_js(f"appendReasoning('{msg_id}', {safe_token});") def finish_message(self, msg_id: str): self.run_js(f"finishMessage('{msg_id}');") # ---------- 工具执行事件 (Python -> JS) —— pi tool_execution_* ---------- def tool_execution_started(self, msg_id: str, call_id: str, name: str, args: str): """工具开始执行：前端在时间线当前位置插入 chip（带 call_id 对号入座）""" safe_call = json.dumps(call_id) safe_name = json.dumps(name) safe_args = json.dumps(args) self.run_js(f"toolExecutionStarted('{msg_id}', {safe_call}, {safe_name}, {safe_args});") def tool_execution_updated(self, msg_id: str, call_id: str, text: str): """工具执行中的增量输出（bash stdout 等）""" safe_call = json.dumps(call_id) safe_text = json.dumps(text) self.run_js(f"toolExecutionUpdated('{msg_id}', {safe_call}, {safe_text});") def tool_execution_timed(self, msg_id: str, call_id: str, elapsed: int, timeout: int): """🆕 bash 运行中每秒读秒：气泡摘要行刷新 N/Ts""" safe_call = json.dumps(call_id) self.run_js(f"toolExecutionTimed('{msg_id}', {safe_call}, " f"{int(elapsed)}, {int(timeout)});") def tool_execution_finished(self, msg_id: str, call_id: str, name: str, ok: bool, text: str): """工具执行结束：chip 按 call_id 标记成功/失败并显示结果摘要""" safe_call = json.dumps(call_id) safe_name = json.dumps(name) safe_text = json.dumps(text) self.run_js(f"toolExecutionFinished('{msg_id}', {safe_call}, {safe_name}, {str(ok).lower()}, {safe_text});") # ---------- 时间线 (Python -> JS) ---------- def restore_streaming_timeline(self, msg_id: str, timeline_json: str): """切回进行中的会话：按时间线 JSON 恢复 思考/文本/工具 块（续流）""" safe_json = json.dumps(timeline_json) self.run_js(f"restoreStreamingTimeline('{msg_id}', {safe_json});") def render_timeline_history(self, msg_id: str, timeline_json: str): """历史消息：按时间线 JSON 静态渲染 思考/文本/工具 块""" safe_json = json.dumps(timeline_json) self.run_js(f"renderTimelineHistory('{msg_id}', {safe_json});") # ---------- 压缩可视化 (Python -> JS) ---------- def compaction_started(self, msg_id: str, path: str): """压缩开始 → 当前消息时间线内显示「上下文压缩」思考气泡（与深度思考同款）""" safe_id = json.dumps(msg_id or "") safe_path = json.dumps(path or "") self.run_js(f"compactionStarted({safe_id}, {safe_path});") def compaction_finished(self, msg_id: str, payload: dict): """压缩完成 → 同一气泡原地更新（前→后 token + 摘要全文）""" safe_id = json.dumps(msg_id or "") safe_payload = json.dumps(payload or {}, ensure_ascii=False) self.run_js(f"compactionFinished({safe_id}, {safe_payload});") # ---------- 系统提示 (Python -> JS) ---------- def show_note(self, text: str): """在聊天流中插入一条居中系统提示（如：已自动压缩上下文）""" safe_text = json.dumps(text) self.run_js(f"showSystemNote({safe_text});") # ---------- 错误与清理 (Python -> JS) ---------- def show_error(self, msg_id: str, error_text: str): safe_text = json.dumps(error_text) self.run_js(f"showError('{msg_id}', {safe_text});") def clear_chat(self): self.run_js("clearChat();") def delete_message(self, msg_id: str): """通知前端从 DOM 中移除特定消息""" self.run_js(f"deleteMessage('{msg_id}');") # ---------- 历史记录渲染 ---------- def render_history_message(self, msg_id: str, role: str, content: str, reasoning: str = "", branch_info: dict = None): self.create_message(msg_id, role, content, branch_info=branch_info) if reasoning: safe_reasoning = json.dumps(reasoning) self.run_js(f"insertThinkBlock('{msg_id}', {safe_reasoning});") self.finish_message(msg_id) def show_welcome(self): self.run_js("showWelcome();") # ---------- 会话加载界面 (Python -> JS) ---------- def show_loading(self): """显示统一加载界面（图标 + 从左到右扫描条）""" self.run_js("showLoadingOverlay();") def hide_loading(self): """隐藏加载界面（带渐变退场）""" self.run_js("hideLoadingOverlay();") # ---------- 交互 Slot (JS -> Python) ---------- @pyqtSlot(str) def onRegenerateClicked(self, msg_id): """前端点击“重新生成”按钮时触发""" print(f"Regenerate requested for: {msg_id}") self.regenerate_clicked.emit(msg_id) @pyqtSlot(str, int) def onBranchSwitch(self, msg_id, direction): """前端点击分支切换箭头时触发""" print(f"Branch switch requested for: {msg_id}, direction: {direction}") self.branch_switch_clicked.emit(msg_id, direction) @pyqtSlot(str) def onDeleteMessageClicked(self, msg_id): """接收 JS 发来的确认删除指令（🌟 已去重，原先定义了两次）""" print(f"Delete requested for: {msg_id}") self.delete_message_requested.emit(msg_id) @pyqtSlot(str) def onAttachmentClicked(self, meta_json_str): """接收 JS 发来的附件点击事件""" self.attachment_clicked.emit(meta_json_str) @pyqtSlot(float, float, float) def onScrollChanged(self, y, content_h, client_h): """接收前端上报的页面滚动信息，用于自定义滚动条位置/滑块长度镜像""" self.scroll_changed.emit(y, content_h, client_h)