sorrow404null
a7412824e0
Import the pre-repair source tree as the history baseline. Runtime data (data/), virtualenvs, bytecode caches and logs are gitignored so local secrets and user state stay out of the repo.
189 lines
8.7 KiB
Python
189 lines
8.7 KiB
Python
# 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)
|