73 lines
2.9 KiB
Python
73 lines
2.9 KiB
Python
# myapp/services/backboard_service.py
|
||
import os
|
||
from flask import current_app
|
||
from flask_socketio import emit
|
||
from backboardManager import Backboard # 你已有的类
|
||
from ..utils.sudoUtil import sudo_open
|
||
from ..extension_ase.ase_client import HSAEngineClient
|
||
from manager import ChatManager
|
||
|
||
def _to_wsl_path_if_needed(path: str) -> str:
|
||
cfg = current_app.config["VSCODE_WEB_PATH"]
|
||
if cfg.get("is_wsl") and cfg.get("windows_path") and cfg.get("wsl_path"):
|
||
return path.replace("\\", "/").replace(cfg["windows_path"], cfg["wsl_path"])
|
||
return path
|
||
|
||
def realtime_response(config: dict, realtime_action: dict) -> None:
|
||
"""
|
||
业务：根据 VSCode 上报的动作，更新 Backboard（黑板）状态。
|
||
"""
|
||
backboard_manager = current_app.extensions["backboard_manager"]
|
||
user_uuid = config["user_uuid"]
|
||
chatmanager = current_app.extensions["user_uuid2chatmanager"][user_uuid]
|
||
|
||
|
||
bb = backboard_manager.get_backboard(user_uuid)
|
||
if bb is None: return
|
||
assert isinstance(bb, Backboard)
|
||
|
||
# 统一记历史
|
||
print(f"backboard action {realtime_action}")
|
||
rtype = realtime_action.get("type")
|
||
|
||
if rtype == "workspaceFolders":
|
||
bb.file_tree = realtime_action.get("fileTree")
|
||
|
||
elif rtype == "activeFile":
|
||
file_path = realtime_action.get("filePath")
|
||
bb.active_file_path = file_path
|
||
assert isinstance(file_path, str)
|
||
with open(os.path.join(bb.root_path, bb.active_file_path), "r", encoding="utf-8") as f:
|
||
bb.active_file_content = f.read()
|
||
|
||
elif rtype == "paste":
|
||
file_path = realtime_action.get("filePath")
|
||
assert isinstance(file_path, str)
|
||
bb.pasted_file_path = file_path
|
||
bb.pasted_content = realtime_action.get("content")
|
||
with open(os.path.join(bb.root_path, bb.active_file_path), "r", encoding="utf-8") as f:
|
||
bb.active_file_content = f.read()
|
||
#处理粘贴事件
|
||
|
||
# paste_data = {
|
||
# 'pasted_file_path': file_path,
|
||
# 'pasted_content': bb.pasted_content,
|
||
# 'pasted_content_length': len(bb.pasted_content)
|
||
# }
|
||
# emit('paste_detected', paste_data)
|
||
|
||
#判断一下，如果粘帖内容过多就发送，后期可加入更多条件判断
|
||
if bb.pasted_content and len(bb.pasted_content) > 250:
|
||
chatmanager.ase_client.send("pasted_detected_in", {"file_path": file_path, "content": bb.pasted_content})
|
||
|
||
|
||
elif rtype == "fileEdit":
|
||
file_path = realtime_action.get("filePath")
|
||
assert isinstance(file_path, str)
|
||
bb.active_file_path = file_path
|
||
with open(os.path.join(bb.root_path, bb.active_file_path), "r", encoding="utf-8") as f:
|
||
bb.active_file_content = f.read()
|
||
action_without_config = realtime_action.copy()
|
||
action_without_config.pop('config', None) # 使用pop并设置默认值None，避免键不存在时出错
|
||
bb.add_history(action_without_config)
|
||
chatmanager.upload_bb() |