pipeline-sdlc/wwwroot/api/workspace_view.dspy

110 lines
4.4 KiB
Plaintext
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# workspace_view.dspy - 查看文件txt/md/json 用 MdWidget 渲染,其他文档下载查看
import os
import json as _json
from urllib.parse import quote, unquote
def _decode_id(s):
# 前端 id 可能被多层编码Tree 懒加载一次 + encodeURIComponent 一次),
# 循环解码到无 % 为止兼容任意编码层数2026-09-03 中文文件名"文件不存在"根因)
for _ in range(4):
if '%' not in s:
break
s2 = unquote(s)
if s2 == s:
break
s = s2
return s
file_id = _decode_id((params_kw or {}).get('id', '').strip())
uid = await get_user()
if not uid:
uid = 'user-01'
session_id = (params_kw or {}).get('session_id', '') or ''
pipeline_id = (params_kw or {}).get('pipeline_id', '') or ''
dbname = get_module_dbname('pipeline-sdlc')
async with DBPools().sqlorContext(dbname) as sor:
# 产线隔离:跨产线项目视为无项目(与弹窗入口一致,防绕过)
project_dir, _ = await get_project_dir_pl(sor, uid, session_id, pipeline_id)
space_dir, _ = await get_space_dir(sor, uid, session_id)
# 通用会话pipeline_id=_generic锁用户专属目录 _general/{uid}
# 不进入任何项目的工作空间2026-09-08 用户要求)
if pipeline_id == '_generic':
_gd = generic_workspace_dir(uid)
os.makedirs(_gd, exist_ok=True)
project_dir = _gd
space_dir = _gd
if not file_id or file_id == '__root__':
return {"widgettype": "Message", "options": {"title": "查看失败", "message": "未指定文件"}}
full_path = resolve_workspace_path(project_dir, space_dir, file_id)
# 路径穿越校验
real_ws = os.path.realpath(space_dir)
real_full = os.path.realpath(full_path)
if not real_full.startswith(real_ws + os.sep):
return {"widgettype": "Message", "options": {"title": "错误", "message": "非法路径"}}
if not os.path.isfile(full_path):
return {"widgettype": "Message", "options": {"title": "查看失败", "message": "文件不存在: " + file_id}}
name = os.path.basename(full_path)
ext = os.path.splitext(name)[1].lower()
file_url = entire_url("/pipeline-sdlc/api/workspace_file.dspy") + "?id=" + quote(file_id) + (("&session_id=" + session_id) if session_id else "")
# txt/md/json → VScrollPanel 包裹 MdWidget 渲染可滚动长文档不截断json 直接显示正文不下载)
md_exts = ['.md', '.txt', '.json']
if ext in md_exts:
if ext == '.json':
# json 包代码块渲染marked 按 pre 显示正文)
try:
with open(full_path, 'r', encoding='utf-8', errors='replace') as fh:
body = fh.read(200000)
except Exception as e:
return {"widgettype": "Message", "options": {"title": "查看失败", "message": str(e)[:200]}}
return {
"widgettype": "PopupWindow",
"options": {"title": name + " (查看)", "width": "85%", "height": "85%", "auto_open": True, "resizable": True},
"subwidgets": [{
"widgettype": "VScrollPanel",
"options": {"css": "filler", "width": "100%", "height": "100%"},
"subwidgets": [{
"widgettype": "MdWidget",
"options": {"mdtext": "```json\n" + body + "\n```", "width": "100%"}
}]
}]
}
return {
"widgettype": "PopupWindow",
"options": {"title": name + " (查看)", "width": "85%", "height": "85%", "auto_open": True, "resizable": True},
"subwidgets": [{
"widgettype": "VScrollPanel",
"options": {"css": "filler", "width": "100%", "height": "100%"},
"subwidgets": [{
"widgettype": "MdWidget",
"options": {"md_url": file_url, "width": "100%"}
}]
}]
}
# 其他 → 下载查看
download_url = file_url + "&download=1"
download_script = (
"window.open(" + _json.dumps(download_url) + ",'_blank');"
)
return {
"widgettype": "PopupWindow",
"options": {"title": name, "cwidth": 34, "cheight": 12, "auto_open": True, "resizable": True},
"subwidgets": [
{"widgettype": "Text", "options": {"text": "该文件类型需下载后查看", "cfontsize": 1, "padding": "16px", "halign": "middle"}},
{"widgettype": "Button", "options": {"label": "下载文件", "css": "primary"},
"binds": [{"wid": "self", "event": "click", "actiontype": "script", "target": "self", "script": download_script}]}
]
}