249 lines
10 KiB
Plaintext
249 lines
10 KiB
Plaintext
# todo_form_submit.dspy - 待办动态表单统一提交端点（问题通道 + 人类任务通道通用）
|
||
#
|
||
# 校验必填 → 文件统一先落项目根 → 按 schema 字段的 target 移到目标位置
|
||
# → 把「文本字段 + 已上传文件路径」写入完成结果，关闭待办、任务恢复执行。
|
||
#
|
||
# 入参（JSON body 或 params_kw）：
|
||
# question_id / task_id 待办 id（二选一，按 kind 决定）
|
||
# kind question | human_task（默认 question）
|
||
# fields JSON 对象：{字段名: 值}。文件字段值 = base64 字符串（或 data: 前缀）；
|
||
# 文本字段值 = 字符串。文件名随附 {字段名}_filename。
|
||
#
|
||
# 文件处理约定（用户确认）：
|
||
# 统一先落到项目根目录，再按 form_schema 里该字段的 target 移动：
|
||
# - target 形如 "env/test.json"（带文件名）→ 移到 {root}/env/test.json
|
||
# - target 形如 "env/"（目录，/ 结尾）→ 移到 {root}/env/{原文件名}
|
||
# - target 为空 → 留在项目根
|
||
# 目标路径相对项目根解析，禁绝对路径与 .. 穿越（服务端二次校验）。
|
||
#
|
||
# 安全：
|
||
# - question：沿用 question_answer（回答即恢复任务）
|
||
# - human_task：走 complete_human_task（内含同机构 + assignee 匹配校验 + 触发 QC）
|
||
|
||
import os
|
||
import json as _json
|
||
import base64 as _b64
|
||
|
||
user_id = await get_user()
|
||
if not user_id:
|
||
return _json.dumps({"success": False, "error": "未登录"}, ensure_ascii=False)
|
||
|
||
kind = ((params_kw or {}).get('kind') or 'question').strip()
|
||
oid = ''
|
||
fields_raw = (params_kw or {}).get('fields') or ''
|
||
if isinstance(fields_raw, str):
|
||
try:
|
||
fields = _json.loads(fields_raw) if fields_raw.strip() else {}
|
||
except Exception:
|
||
fields = {}
|
||
else:
|
||
fields = fields_raw or {}
|
||
if not isinstance(fields, dict):
|
||
fields = {}
|
||
|
||
if kind == 'human_task':
|
||
oid = ((params_kw or {}).get('task_id') or '').strip()
|
||
else:
|
||
oid = ((params_kw or {}).get('question_id') or '').strip()
|
||
if not oid:
|
||
return _json.dumps({"success": False, "error": "缺少待办 id"}, ensure_ascii=False)
|
||
|
||
dbname = get_module_dbname('pipeline-sdlc')
|
||
|
||
# ── 读取待办记录 + form_schema + 项目上下文 ──
|
||
schema_fields = []
|
||
project_id = ''
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
if kind == 'human_task':
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, project_id, form_schema, status FROM pipeline_human_tasks WHERE id=${i}$",
|
||
{"i": oid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return _json.dumps({"success": False, "error": "待办不存在"}, ensure_ascii=False)
|
||
r = recs[0]
|
||
project_id = str(getattr(r, 'project_id', '') or '')
|
||
if str(getattr(r, 'status', '') or '') != 'pending':
|
||
return _json.dumps({"success": False, "error": "该待办已处理，无需重复提交"}, ensure_ascii=False)
|
||
try:
|
||
_fs_raw = str(getattr(r, 'form_schema', '') or '')
|
||
if _fs_raw:
|
||
schema_fields = (_json.loads(_fs_raw) or {}).get('fields') or []
|
||
except Exception:
|
||
schema_fields = []
|
||
else:
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, tenant_id, context, status FROM pipeline_agent_questions WHERE id=${i}$",
|
||
{"i": oid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return _json.dumps({"success": False, "error": "待办不存在或已处理"}, ensure_ascii=False)
|
||
r = recs[0]
|
||
project_id = str(getattr(r, 'tenant_id', '') or '')
|
||
if str(getattr(r, 'status', '') or '') not in ('', 'pending', 'waiting'):
|
||
return _json.dumps({"success": False, "error": "该待办已处理，无需重复提交"}, ensure_ascii=False)
|
||
try:
|
||
_ctx_raw = str(getattr(r, 'context', '') or '')
|
||
if _ctx_raw:
|
||
schema_fields = (_json.loads(_ctx_raw) or {}).get('form_schema', {}).get('fields') or []
|
||
except Exception:
|
||
schema_fields = []
|
||
|
||
if not schema_fields:
|
||
return _json.dumps({"success": False, "error": "该待办没有动态表单定义"}, ensure_ascii=False)
|
||
if not project_id:
|
||
return _json.dumps({"success": False, "error": "待办缺少项目上下文，无法定位工作空间"}, ensure_ascii=False)
|
||
|
||
# ── 项目根目录（与角色 agent 读的位置一致） ──
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
project_dir, _base = await get_project_dir_by_id(sor, project_id)
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not project_dir or not os.path.isdir(project_dir):
|
||
return _json.dumps({"success": False, "error": "项目工作空间不可用"}, ensure_ascii=False)
|
||
|
||
|
||
def _safe_rel(target):
|
||
"""校验相对路径：禁绝对路径、禁 .. 穿越。返回规整后的相对路径或 ''。
|
||
|
||
保留尾部 '/'——它是「目录 target」（保留原文件名）与「文件 target」
|
||
（直接用 target 作路径）的区分标志，归一化时不能丢。
|
||
"""
|
||
t = (target or '').strip()
|
||
if not t:
|
||
return ''
|
||
if t.startswith('/'):
|
||
return ''
|
||
is_dir = t.endswith('/')
|
||
parts = [p for p in t.split('/') if p not in ('', '.')]
|
||
if '..' in parts:
|
||
return ''
|
||
return '/'.join(parts) + ('/' if is_dir else '')
|
||
|
||
|
||
uploaded = [] # [(name, rel_path)]
|
||
text_entries = [] # [(label, value)]
|
||
fatal = [] # 致命错误：必填缺失/解码失败/写盘失败 → 整体拒绝提交
|
||
warnings = [] # 非致命：文件已落项目根但移动到 target 失败
|
||
|
||
for sf in schema_fields:
|
||
name = str(sf.get('name') or '')
|
||
ftype = str(sf.get('type') or 'text').lower()
|
||
required = bool(sf.get('required'))
|
||
val = fields.get(name)
|
||
label = str(sf.get('label') or name)
|
||
|
||
if ftype == 'file':
|
||
if not val:
|
||
if required:
|
||
fatal.append(label + '：必须上传文件')
|
||
continue
|
||
# 解码 base64
|
||
try:
|
||
if isinstance(val, str) and val.startswith('data:'):
|
||
_, b64 = val.split(',', 1)
|
||
else:
|
||
b64 = val
|
||
data = _b64.b64decode(b64)
|
||
except Exception as e:
|
||
fatal.append(label + '：文件解码失败 ' + str(e)[:80])
|
||
continue
|
||
# 文件名：优先取随字段附带的 {name}_filename，否则用字段名
|
||
fname = str(fields.get(name + '_filename') or '').strip()
|
||
fname = os.path.basename(fname) if fname else ''
|
||
if not fname:
|
||
fname = name
|
||
# 1) 先落项目根
|
||
root_path = os.path.join(project_dir, fname)
|
||
try:
|
||
with open(root_path, 'wb') as f:
|
||
f.write(data)
|
||
except Exception as e:
|
||
fatal.append(label + '：写入失败 ' + str(e)[:80])
|
||
continue
|
||
# 2) 按 target 移动（失败不致命——文件已在项目根，agent 仍可读到）
|
||
rel = _safe_rel(sf.get('target'))
|
||
final_rel = fname
|
||
try:
|
||
if rel:
|
||
if '/' in rel and not rel.endswith('/'):
|
||
# target 含文件名（如 env/test.json）→ 直接用
|
||
dst_rel = rel
|
||
else:
|
||
# target 是目录（如 env/）→ 保留原文件名
|
||
dst_rel = rel.rstrip('/') + '/' + fname
|
||
dst_abs = os.path.join(project_dir, dst_rel)
|
||
os.makedirs(os.path.dirname(dst_abs), exist_ok=True)
|
||
if os.path.exists(dst_abs):
|
||
os.remove(dst_abs) # 覆盖（补齐/替换场景）
|
||
os.rename(root_path, dst_abs)
|
||
final_rel = dst_rel
|
||
except Exception as e:
|
||
warnings.append(label + '：已保存到项目根，移动到 ' + rel + ' 失败 ' + str(e)[:80])
|
||
uploaded.append((name, final_rel))
|
||
else:
|
||
# text / textarea
|
||
sv = '' if val is None else str(val).strip()
|
||
if not sv and required:
|
||
fatal.append(label + '：不能为空')
|
||
elif sv:
|
||
text_entries.append((label, sv))
|
||
|
||
# 致命错误 → 拒绝提交（回答/结果不写，待办不关闭，让用户补齐重来）
|
||
if fatal:
|
||
return _json.dumps({"success": False, "error": "；".join(fatal)}, ensure_ascii=False)
|
||
|
||
# ── 构造完成结果：文本字段 + 已上传文件路径 ──
|
||
result_files = [rel for _, rel in uploaded]
|
||
free_text = str(fields.get('_note') or '').strip()
|
||
|
||
if kind == 'human_task':
|
||
result_data = {
|
||
'files': result_files,
|
||
'fields': {label: v for label, v in text_entries},
|
||
}
|
||
if free_text:
|
||
result_data['note'] = free_text
|
||
ok, msg = await complete_human_task(oid, result_data, operator_id=user_id)
|
||
if ok:
|
||
summary = '提交成功，进入 QC 检查。'
|
||
if uploaded:
|
||
summary += '已上传：' + '、'.join(result_files) + '。'
|
||
if warnings:
|
||
summary += '（部分提示：' + '；'.join(warnings) + '）'
|
||
return _json.dumps({"success": True, "message": summary}, ensure_ascii=False)
|
||
return _json.dumps({"success": False, "error": str(msg) or "提交失败"}, ensure_ascii=False)
|
||
|
||
# ── question 通道：回答文本回流给 _build_qna_section ──
|
||
answer_lines = [label + '：' + v for label, v in text_entries]
|
||
if uploaded:
|
||
answer_lines.append('已上传文件（位于项目工作空间，可直接 read_file）：')
|
||
for name, rel in uploaded:
|
||
answer_lines.append(' · ' + rel)
|
||
if free_text:
|
||
answer_lines.append('补充说明：' + free_text)
|
||
answer = '\n'.join(answer_lines) or '（已按要求上传/填写）'
|
||
|
||
res = await question_answer(oid, answer, answered_by=user_id, answer_source='owner.superuser')
|
||
|
||
ok = False
|
||
msg = ''
|
||
if isinstance(res, tuple) or isinstance(res, list):
|
||
ok = bool(res[0]) if len(res) > 0 else False
|
||
msg = str(res[1]) if len(res) > 1 else ''
|
||
elif isinstance(res, dict):
|
||
ok = bool(res.get('success', res.get('ok', True)))
|
||
msg = str(res.get('message') or res.get('error') or '')
|
||
else:
|
||
ok = bool(res)
|
||
|
||
if ok:
|
||
summary = '提交成功。'
|
||
if uploaded:
|
||
summary += '已上传 ' + '、'.join(result_files) + '，任务恢复执行。'
|
||
else:
|
||
summary += '任务恢复执行。'
|
||
if warnings:
|
||
summary += '（部分提示：' + '；'.join(warnings) + '）'
|
||
return _json.dumps({"success": True, "message": summary}, ensure_ascii=False)
|
||
return _json.dumps({"success": False, "error": msg or "提交失败"}, ensure_ascii=False)
|