591 lines
23 KiB
Python
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.

"""
pipeline_service/workspace.py - 工作空间文件管理
提供项目工作空间的文件浏览、读取、保存功能。
"""
import os
import json
import logging
logger = logging.getLogger(__name__)
WORKSPACE_BASE = '/d/pipeline/workspaces'
# 通用助手项目空间键(保留值,不属于任何专业产线;专业产线用 pipelines.id)
GENERAL_SPACE = 'general'
def build_workspace_path(workspace_base, org_id, space, name, project_id=''):
"""构建项目 workspace 路径:{base}/{org_id}/{space}/{name}。
space = 项目空间键(pipeline_id;通用助手为 'general'),
通用助手与专业产线、不同产线之间由此隔离,不共享目录。
同空间重名时追加短 ID 保证唯一,避免多项目共享同一目录互相覆盖。
"""
space = (space or GENERAL_SPACE).strip() or GENERAL_SPACE
name = (name or 'unnamed').strip() or 'unnamed'
base_dir = os.path.join(workspace_base, str(org_id or '0'), space)
path = os.path.join(base_dir, name)
if project_id and os.path.isdir(path):
short = project_id[-8:] if len(project_id) >= 8 else project_id
path = os.path.join(base_dir, f"{name}_{short}")
return path
def build_space_path(workspace_base, org_id, space):
"""构建产线工作空间路径(机构工作空间层):{base}/{org_id}/{space}。
开发产线的机构工作空间(projects/apps/modules 三目录所在层)。
角色 agent 的工具路径基准用这个,从而能访问 projects/、apps/、modules/。
"""
space = (space or GENERAL_SPACE).strip() or GENERAL_SPACE
return os.path.join(workspace_base, str(org_id or '0'), space)
# 可编辑的文本文件扩展名
TEXT_EXTENSIONS = {
'.md', '.py', '.js', '.html', '.css', '.json', '.txt',
'.xml', '.yaml', '.yml', '.toml', '.cfg', '.sh', '.sql',
'.dspy', '.ui', '.java', '.go', '.rs', '.ts', '.tsx', '.vue',
}
# 媒体文件扩展名
MEDIA_EXTENSIONS = {
'.png', '.jpg', '.jpeg', '.gif', '.svg', '.mp4', '.webm', '.mp3', '.wav',
}
def _get_file_icon(ext):
"""根据扩展名返回图标"""
ext = ext.lower()
if ext in TEXT_EXTENSIONS:
return "📄"
if ext in MEDIA_EXTENSIONS:
return "🎬"
return "📁"
async def get_param(sor, name, default=""):
"""读 appbase params 表配置(params_name → params_value),表不存在/无值时兑底 default。"""
try:
recs = await sor.sqlExe(
"SELECT params_value FROM params WHERE params_name=${n}$ LIMIT 1", {"n": name})
if recs and getattr(recs[0], 'params_value', '') not in (None, ''):
return getattr(recs[0], 'params_value', '')
except Exception:
pass
return default
async def get_workspace_base(sor):
"""读 workspace_base 参数(appbase params 表,表不存在时兜底 WORKSPACE_BASE)。"""
return await get_param(sor, 'workspace_base', WORKSPACE_BASE)
async def get_max_concurrent_agents(sor):
"""读全局并发 agent 数限制(appbase params 表 max_concurrent_agents,默认 3)。"""
val = await get_param(sor, 'max_concurrent_agents', '3')
try:
n = int(float(str(val)))
return max(1, n)
except (ValueError, TypeError):
return 3
async def get_max_task_retry(sor):
"""读任务最大重复数(appbase params 表 task_max_retry,默认 3)。超限即暂停任务链、抛故障给人工。"""
val = await get_param(sor, 'task_max_retry', '3')
try:
n = int(float(str(val)))
return max(1, n)
except (ValueError, TypeError):
return 3
async def get_session_project_id(sor, uid, session_id=''):
"""按会话解析当前项目 id。
session_id 非空时优先读 pipeline_session_settings(web 多 tab 各自项目上下文),
无记录或表不存在时回退全局 pipeline_agent_settings.current_project_id。
存在性校验(悬空引用防护):记录指向的项目若已从 sd_projects 删除,
视为无效——顺手清掉死记录(自愈,避免每次请求重复命中),继续回退全局;
全局也无效则返回 ''。若不清理,删除项目后该会话的所有项目相关入口
(工作空间/任务/菜单)都会被死引用遮蔽,报「请先在会话中切换项目」。
"""
async def _valid(pid):
"""pid 对应项目存在则原样返回;不存在返回 '';查询异常保守放行(不因校验误伤功能)。"""
if not pid:
return ''
try:
recs = await sor.sqlExe(
"SELECT 1 FROM sd_projects WHERE id=${p}$ LIMIT 1", {"p": pid})
return pid if recs else ''
except Exception:
return pid
if session_id:
try:
recs = await sor.sqlExe(
"SELECT current_project_id FROM pipeline_session_settings "
"WHERE user_id=${u}$ AND session_id=${s}$",
{"u": uid, "s": session_id})
if recs:
pid = getattr(recs[0], 'current_project_id', '') or ''
if pid:
vpid = await _valid(pid)
if vpid:
return vpid
# 悬空引用:清掉死记录,回退全局
try:
await sor.sqlExe(
"DELETE FROM pipeline_session_settings "
"WHERE user_id=${u}$ AND session_id=${s}$",
{"u": uid, "s": session_id})
await sor.sqlExe("COMMIT", {})
except Exception:
pass
except Exception:
pass
try:
recs = await sor.sqlExe(
"SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$",
{"u": uid})
pid = getattr(recs[0], 'current_project_id', '') if recs else ''
if pid:
vpid = await _valid(pid)
if vpid:
return vpid
try:
await sor.sqlExe(
"UPDATE pipeline_agent_settings SET current_project_id='' "
"WHERE user_id=${u}$", {"u": uid})
await sor.sqlExe("COMMIT", {})
except Exception:
pass
return ''
except Exception:
return ''
async def get_session_context(sor, uid, session_id=''):
"""按会话解析 (project_id, iteration_id)。per-session 优先,回退 per-user 全局。
项目 id 统一走 get_session_project_id(含悬空引用防护:指向已删除项目的
记录视为无效并自愈清理)。禁止在本函数里重复实现解析逻辑——两份实现
语义漂移正是「会话有项目、工作空间说没有」这类 bug 的根因。
iteration_id 只从提供有效 pid 的同一条记录取,避免 pid 与 iid 来自不同
记录造成错配。
"""
pid = await get_session_project_id(sor, uid, session_id)
if not pid:
return '', ''
iid = ''
if session_id:
try:
recs = await sor.sqlExe(
"SELECT current_project_id, current_iteration_id FROM pipeline_session_settings "
"WHERE user_id=${u}$ AND session_id=${s}$",
{"u": uid, "s": session_id})
if recs and (getattr(recs[0], 'current_project_id', '') or '') == pid:
iid = getattr(recs[0], 'current_iteration_id', '') or ''
except Exception:
pass
if not iid:
try:
recs = await sor.sqlExe(
"SELECT current_project_id, current_iteration_id FROM pipeline_agent_settings "
"WHERE user_id=${u}$",
{"u": uid})
if recs and (getattr(recs[0], 'current_project_id', '') or '') == pid:
iid = getattr(recs[0], 'current_iteration_id', '') or ''
except Exception:
pass
return pid, iid
async def get_workspace_dir(sor, uid, session_id=''):
"""读当前项目的 workspace 目录(统一 workspace_*.dspy 的重复逻辑)。
返回 (ws_dir, workspace_base)。ws_dir 优先用 sd_projects.workspace_dir(绝对路径),
否则 base/org/name;无当前项目时 ws_dir 返回空字符串。
session_id 非空时按会话隔离项目上下文(多 tab 各自项目,互不覆盖)。
"""
workspace_base = await get_workspace_base(sor)
pid = await get_session_project_id(sor, uid, session_id)
if not pid:
return '', workspace_base
proj = await sor.sqlExe(
"SELECT name, org_id, workspace_dir, pipeline_id FROM sd_projects WHERE id=${p}$",
{"p": pid})
if not proj:
return '', workspace_base
pname = getattr(proj[0], 'name', '')
org_id = getattr(proj[0], 'org_id', '0') or '0'
ws = getattr(proj[0], 'workspace_dir', '') or ''
space = getattr(proj[0], 'pipeline_id', '') or GENERAL_SPACE
if ws.startswith('/'):
return ws, workspace_base
return build_workspace_path(workspace_base, org_id, space, pname, pid), workspace_base
async def get_space_dir(sor, uid, session_id=''):
"""返回机构工作空间层 {space}/(projects/apps/modules 三目录所在)。
用 build_space_path(org_id + pipeline_id)构造,不依赖 workspace_dir 的路径结构
(workspace_dir 迁移到 {space}/projects/{项目名} 后 dirname 会错)。
"""
workspace_base = await get_workspace_base(sor)
pid = await get_session_project_id(sor, uid, session_id)
if not pid:
return '', workspace_base
try:
proj = await sor.sqlExe(
"SELECT org_id, pipeline_id FROM sd_projects WHERE id=${p}$ LIMIT 1",
{"p": pid})
if proj:
org_id = getattr(proj[0], 'org_id', '0') or '0'
space = getattr(proj[0], 'pipeline_id', '') or GENERAL_SPACE
return build_space_path(workspace_base, org_id, space), workspace_base
except Exception:
pass
return '', workspace_base
async def get_project_dir(sor, uid, session_id=''):
"""返回项目目录 {space}/projects/{项目名}/(新结构,docs/env/spec.json/deliverables 所在)。"""
space_dir, workspace_base = await get_space_dir(sor, uid, session_id)
if not space_dir:
return '', workspace_base
pid = await get_session_project_id(sor, uid, session_id)
dname = ''
if pid:
try:
r = await sor.sqlExe(
"SELECT directory_name, name FROM sd_projects WHERE id=${p}$ LIMIT 1",
{"p": pid})
if r:
dname = getattr(r[0], 'directory_name', '') or getattr(r[0], 'name', '') or ''
except Exception:
pass
if not dname:
ws_dir, _ = await get_workspace_dir(sor, uid, session_id)
dname = os.path.basename(ws_dir.rstrip('/'))
return os.path.join(space_dir, 'projects', dname), workspace_base
async def get_project_dir_pl(sor, uid, session_id='', pipeline_id=''):
"""产线隔离版:会话项目与指定产线不一致时视为无项目(返回空目录)。
商机/投标产线入口的工作空间不能显示开发产线项目的空间——
会话里残留的跨产线项目(如元景项目)不应劫持本产线的工作空间。
pipeline_id 为空时不做隔离(向后兼容旧调用)。
"""
if not pipeline_id:
return await get_project_dir(sor, uid, session_id)
pid = await get_session_project_id(sor, uid, session_id)
if pid:
try:
r = await sor.sqlExe(
"SELECT pipeline_id FROM sd_projects WHERE id=${p}$ LIMIT 1",
{"p": pid})
if r:
proj_pl = getattr(r[0], 'pipeline_id', '') or ''
if proj_pl != pipeline_id:
# 跨产线项目:视为本产线无项目(与 gateway 的隔离一致)
return '', await get_workspace_base(sor)
except Exception:
pass
return await get_project_dir(sor, uid, session_id)
async def get_project_apps_modules(sor, uid, session_id=''):
"""读项目目录下 {应用名}_spec.json,返回 (apps, modules)。
apps = 各 spec.json 的 app 字段;modules = referenced_modules + generated_modules 去重。
"""
import json as _json
project_dir, _ = await get_project_dir(sor, uid, session_id)
if not project_dir or not os.path.isdir(project_dir):
return [], []
apps = []
modules = set()
try:
for f in sorted(os.listdir(project_dir)):
if not f.endswith('_spec.json'):
continue
fp = os.path.join(project_dir, f)
try:
with open(fp, 'r', encoding='utf-8') as fh:
data = _json.loads(fh.read())
except Exception:
continue
if not isinstance(data, dict):
continue
app = (data.get('app') or '').strip()
if app:
apps.append(app)
for m in (data.get('referenced_modules') or []) + (data.get('generated_modules') or []):
if isinstance(m, str) and m.strip():
modules.add(m.strip())
except Exception:
pass
return sorted(set(apps)), sorted(modules)
async def get_project_dir_by_id(sor, project_id):
"""项目根目录(按项目 id 直接解析,不依赖会话上下文)。
解析顺序与 agent 工作目录完全一致,保证「上传放的位置 = agent 读的位置」:
1. sd_projects.workspace_dir 非空(新项目)→ 直接用({space}/projects/{slug})
2. 为空(存量项目)→ 兜底旧平铺结构 {base}/{org}/{space}/{项目名}
"""
if not project_id:
return '', ''
recs = await sor.sqlExe(
"SELECT name, org_id, pipeline_id, workspace_dir FROM sd_projects WHERE id=${p}$ LIMIT 1",
{"p": project_id})
if not recs:
return '', ''
r = recs[0]
workspace_base = await get_workspace_base(sor)
ws = (getattr(r, 'workspace_dir', '') or '').strip()
if ws.startswith('/'):
return ws, workspace_base
name = (getattr(r, 'name', '') or '').strip()
if not name:
return '', workspace_base
org_id = getattr(r, 'org_id', '0') or '0'
space = getattr(r, 'pipeline_id', '') or GENERAL_SPACE
# 不传 project_id:查询语义下禁用重名后缀逻辑,返回的就是平铺目录本身
return build_workspace_path(workspace_base, org_id, space, name), workspace_base
def copy_uploads_to_project(project_dir, uploads):
"""把已落盘的上传文件复制进项目根目录(会话上传 → 项目内可被 agent 找到)。
uploads: [(src_abs_path, filename)];同名文件自动加 _1/_2 后缀防覆盖。
返回 [(保存后文件名, 绝对路径)],与入参顺序对应。
"""
import shutil
saved = []
if not project_dir:
return saved
try:
os.makedirs(project_dir, exist_ok=True)
except Exception:
return saved
for src, name in uploads:
name = os.path.basename((name or '').strip())
if not name or not os.path.isfile(src):
continue
target = os.path.join(project_dir, name)
if os.path.exists(target):
stem, ext = os.path.splitext(name)
i = 1
while os.path.exists(os.path.join(project_dir, f"{stem}_{i}{ext}")):
i += 1
target = os.path.join(project_dir, f"{stem}_{i}{ext}")
try:
shutil.copyfile(src, target)
saved.append((os.path.basename(target), target))
except Exception as e:
logger.warning(f"copy upload failed: {src} -> {target} err={e}")
return saved
def resolve_workspace_path(project_dir, space_dir, node_id):
"""把工作空间节点 ID 解析成绝对路径。
- '@apps/xxx' → space_dir/apps/xxx(项目用到的应用)
- '@modules/xxx' → space_dir/modules/xxx(项目用到的模块)
- '__root__' / 空 → project_dir(项目目录根)
- 其他 → project_dir/xxx(项目目录内的相对路径)
"""
node_id = (node_id or '').strip()
if node_id.startswith('@apps/'):
return os.path.join(space_dir, 'apps', node_id[len('@apps/'):])
if node_id.startswith('@modules/'):
return os.path.join(space_dir, 'modules', node_id[len('@modules/'):])
if not node_id or node_id == '__root__':
return project_dir
return os.path.join(project_dir, node_id)
async def get_workspace_path(sor, user_id, session_id=''):
"""获取用户当前项目的工作空间路径(无项目时返回 None)。"""
ws, _ = await get_workspace_dir(sor, user_id, session_id)
return ws or None
def build_tree_items(ws_dir):
"""递归扫描目录,返回 Tree widget 需要的 {id, parentid, label} 列表"""
if not ws_dir or not os.path.isdir(ws_dir):
return [{"id": "__root__", "parentid": "", "label": "📁 工作空间(不可用)"}]
root_name = os.path.basename(ws_dir)
items = [{"id": "__root__", "parentid": "", "label": "📁 " + root_name}]
def scan(path, parent_id):
try:
entries = sorted(os.listdir(path))
except Exception:
return
for name in entries:
full = os.path.join(path, name)
if name.startswith('.') or name == '__pycache__':
continue
if os.path.isdir(full):
items.append({"id": full, "parentid": parent_id, "label": "📁 " + name})
scan(full, full)
scan(ws_dir, "__root__")
return items
def build_file_widgets(folder_path):
"""返回指定目录下的文件列表(Bricks widget 定义)"""
if not folder_path or not os.path.isdir(folder_path):
return {"widgettype": "Text", "options": {"text": "目录不可用", "cfontsize": 0.9}}
files = []
try:
entries = sorted(os.listdir(folder_path))
except Exception as e:
return {"widgettype": "Text", "options": {"text": f"读取错误: {e}", "cfontsize": 0.9, "color": "#f44336"}}
for name in entries:
full = os.path.join(folder_path, name)
if name.startswith('.'):
continue
if not os.path.isfile(full):
continue
size = os.path.getsize(full)
ext = os.path.splitext(name)[1].lower()
icon = _get_file_icon(ext)
is_text = ext in TEXT_EXTENSIONS
kb_size = f"{size / 1024:.1f}KB"
# 为安全起见,在 widget JSON 中不嵌入完整文件路径
# 文件路径通过 row_id 传递(base64 编码)
import base64
path_b64 = base64.b64encode(full.encode()).decode()
if is_text:
row = {
"widgettype": "HBox",
"options": {
"padding": "6px 10px", "alignItems": "center", "gap": "8px",
"borderRadius": "4px", "style": {"cursor": "pointer"},
},
"subwidgets": [
{"widgettype": "Text", "options": {"text": icon, "cfontsize": 1.2}},
{"widgettype": "Text", "options": {"text": name, "cfontsize": 0.9, "color": "#1e293b", "css": "filler"}},
{"widgettype": "Text", "options": {"text": kb_size, "cfontsize": 0.75, "color": "#94a3b8"}},
],
"binds": [{
"wid": "self", "event": "click", "actiontype": "script", "target": "self",
"script": _build_open_file_script(path_b64, name)
}],
}
else:
row = {
"widgettype": "HBox",
"options": {
"padding": "6px 10px", "alignItems": "center", "gap": "8px",
"borderRadius": "4px",
},
"subwidgets": [
{"widgettype": "Text", "options": {"text": icon, "cfontsize": 1.2}},
{"widgettype": "Text", "options": {"text": name, "cfontsize": 0.9, "color": "#94a3b8", "css": "filler"}},
{"widgettype": "Text", "options": {"text": kb_size, "cfontsize": 0.75, "color": "#94a3b8"}},
],
}
files.append(row)
if not files:
return {"widgettype": "Text", "options": {"text": "(空目录)", "cfontsize": 0.85, "color": "#94a3b8"}}
return {
"widgettype": "VScrollPanel",
"options": {"css": "filler", "padding": "8px", "gap": "4px"},
"subwidgets": files,
}
def _build_open_file_script(path_b64, name):
"""生成文件打开脚本(Wterm 编辑器 + 保存按钮)"""
return (
"var p=atob('" + path_b64 + "');"
"var pw=new bricks.PopupWindow({title:'" + name.replace("'", "\\'") + "',cwidth:70,cheight:32,auto_open:true});"
"var wt=new bricks.Wterm({css:'filler',padding:'0'});pw.content_w.add_widget(wt);"
"var sb=new bricks.Button({label:'保存'});pw.content_w.add_widget(sb);"
"fetch('/pipeline-sdlc/api/workspace_file.dspy?action=read&path='+encodeURIComponent(p))"
".then(function(r){return r.json()}).then(function(d){"
"if(d.error){wt.write('Error: '+d.error);return;}"
"wt.write(d.content||'');"
"sb.bind('click',function(){"
"var c=wt.get_value();"
"fetch('/pipeline-sdlc/api/workspace_file.dspy',{method:'POST',headers:{'Content-Type':'application/x-www-form-urlencoded'},"
"body:'action=save&path='+encodeURIComponent(p)+'&content='+encodeURIComponent(c)})"
".then(function(r){return r.json()}).then(function(d){"
"if(d.success){sb.set_text('已保存');setTimeout(function(){sb.set_text('保存')},2000);}"
"});"
"});"
"});"
)
def read_file_content(filepath):
"""读取文件内容"""
if not filepath or not os.path.isfile(filepath):
return None, "文件不存在"
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read(50000)
return content, None
except Exception as e:
return None, str(e)
def save_file_content(filepath, content):
"""保存文件内容"""
if not filepath:
return False, "路径为空"
try:
dirpath = os.path.dirname(filepath)
if dirpath:
os.makedirs(dirpath, exist_ok=True)
with open(filepath, 'w', encoding='utf-8') as f:
f.write(content)
return True, None
except Exception as e:
return False, str(e)
def load_workspace():
"""注册工作空间函数到 ServerEnv,供 DSPY 直接调用。"""
from ahserver.serverenv import ServerEnv
g = ServerEnv()
g.get_workspace_base = get_workspace_base
g.get_workspace_dir = get_workspace_dir
g.get_workspace_path = get_workspace_path
g.get_space_dir = get_space_dir
g.get_project_dir = get_project_dir
g.get_project_dir_pl = get_project_dir_pl
g.get_project_dir_by_id = get_project_dir_by_id
g.copy_uploads_to_project = copy_uploads_to_project
g.get_project_apps_modules = get_project_apps_modules
g.get_session_project_id = get_session_project_id
g.get_session_context = get_session_context
g.build_workspace_path = build_workspace_path
g.build_space_path = build_space_path
g.resolve_workspace_path = resolve_workspace_path
g.GENERAL_SPACE = GENERAL_SPACE
g.build_tree_items = build_tree_items
g.build_file_widgets = build_file_widgets
g.read_file_content = read_file_content
g.save_file_content = save_file_content