224 lines
9.1 KiB
Plaintext
224 lines
9.1 KiB
Plaintext
# task_tree.dspy - 任务树（树状结构：项目 → 迭代 → 阶段 → 任务）
|
||
# 懒加载协议：
|
||
# 无 id → 返回项目根节点
|
||
# __root__ → 返回迭代列表（sd_iterations + 未分配迭代）
|
||
# iter:{iter_id} → 返回阶段列表（需求/设计/开发/测试/部署，只显示有任务的）
|
||
# iter:{iter_id}:phase:{role} → 返回该迭代该阶段的任务列表（叶子）
|
||
|
||
import json
|
||
|
||
uid = await get_user()
|
||
if not uid:
|
||
uid = 'user-01'
|
||
|
||
node_id = (params_kw or {}).get('id', '').strip()
|
||
|
||
dbname = get_module_dbname('pipeline-sdlc')
|
||
|
||
_state_icons = {
|
||
'completed': '✅', 'approved': '✅', 'running': '🟡', 'failed': '❌',
|
||
'waiting': '⚠️', 'submitted': '⬜', 'review': '👀', 'rejected': '🚫',
|
||
'paused': '⏸️', 'cancelled': '🛑', 'qc_review': '🔍',
|
||
}
|
||
_state_zh = {
|
||
'completed': '已完成', 'approved': '已批准', 'running': '运行中',
|
||
'failed': '失败', 'waiting': '等待中', 'submitted': '已提交',
|
||
'review': '审核中', 'rejected': '已驳回', 'paused': '已暂停', 'cancelled': '已取消',
|
||
}
|
||
# 阶段顺序：从产线角色集（SDL_ROLES）动态生成，用 next_role 链推导线性角色（自动排除 qc 门禁）。
|
||
def _get_phase_order():
|
||
try:
|
||
from pipeline_core import list_roles
|
||
roles = list_roles("sdlc_general") or []
|
||
role_map = {r.name: r for r in roles}
|
||
next_set = {getattr(r, 'next_role', '') for r in roles if getattr(r, 'next_role', '')}
|
||
heads = [r for r in roles if r.name not in next_set and getattr(r, 'next_role', '')]
|
||
order = []
|
||
seen = set()
|
||
cur = heads[0] if heads else None
|
||
while cur and cur.name not in seen:
|
||
seen.add(cur.name)
|
||
desc = getattr(cur, 'description', '') or cur.name
|
||
order.append((cur.name, desc))
|
||
cur = role_map.get(getattr(cur, 'next_role', ''))
|
||
return order
|
||
except Exception:
|
||
return [
|
||
('agent.requirement', '需求阶段'),
|
||
('agent.design', '设计阶段'),
|
||
('agent.develop', '开发阶段'),
|
||
('agent.deploy_test', '测试环境部署'),
|
||
('agent.test', '测试阶段'),
|
||
('agent.deploy_prod', '生产环境部署'),
|
||
]
|
||
|
||
|
||
def _iter_of(task):
|
||
"""从任务 params 里提取 iteration_id（存的是迭代名称，非 id）。"""
|
||
params_str = getattr(task, 'params', '') or ''
|
||
try:
|
||
p = json.loads(params_str) if params_str else {}
|
||
return (p.get('iteration_id') or '').strip()
|
||
except Exception:
|
||
return ''
|
||
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# 当前项目（2026-09-11 升级：会话级解析——session 优先→全局兜底，
|
||
# 含悬空自愈+产线隔离；旧实现只读全局 pipeline_agent_settings，多 tab 串项目）
|
||
_sid = (params_kw or {}).get('session_id', '') or ''
|
||
pid = ''
|
||
try:
|
||
from pipeline_service.workspace import get_session_project_id
|
||
pid = await get_session_project_id(sor, uid, _sid, 'sdlc_general') or ''
|
||
except Exception:
|
||
pid = ''
|
||
if not pid:
|
||
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 not pid:
|
||
if not node_id:
|
||
return json.dumps([{"id": "__root__", "label": "请先选择或新建项目", "is_leaf": True}], ensure_ascii=False)
|
||
return json.dumps([], ensure_ascii=False)
|
||
|
||
# 项目名
|
||
pname = ''
|
||
precs = await sor.sqlExe("SELECT name FROM sd_projects WHERE id=${p}$", {"p": pid})
|
||
if precs:
|
||
pname = getattr(precs[0], 'name', '') or ''
|
||
|
||
# 一次查所有任务 + 所有迭代（任务数不大，Python 分组避免 SQL JSON 函数坑）
|
||
all_tasks = await sor.sqlExe(
|
||
"SELECT id, title, role, state, params, parent_id FROM pipeline_tasks WHERE tenant_id=${p}$ ORDER BY created_at ASC",
|
||
{"p": pid}) or []
|
||
iters = await sor.sqlExe(
|
||
"SELECT id, iteration_name, status FROM sd_iterations WHERE project_id=${p}$ ORDER BY created_at ASC",
|
||
{"p": pid}) or []
|
||
|
||
if not node_id:
|
||
# 根节点
|
||
return json.dumps([{"id": "__root__", "label": pname or "项目", "is_leaf": False}], ensure_ascii=False)
|
||
|
||
if node_id == '__root__':
|
||
# 迭代列表
|
||
nodes = []
|
||
for it in iters:
|
||
iid = getattr(it, 'id', '')
|
||
iname = getattr(it, 'iteration_name', '') or ''
|
||
istatus = getattr(it, 'status', '') or ''
|
||
if not iid:
|
||
continue
|
||
cnt = sum(1 for t in all_tasks if _iter_of(t) == iname)
|
||
nodes.append({
|
||
"id": "iter:" + iid,
|
||
"label": "🗂 {} [{}] ({})".format(iname, istatus, cnt),
|
||
"is_leaf": False,
|
||
})
|
||
# 未分配迭代（无 iteration_id 的任务）—— 有未分配任务才显示
|
||
none_cnt = sum(1 for t in all_tasks if not _iter_of(t))
|
||
if none_cnt > 0:
|
||
nodes.append({
|
||
"id": "iter:__none__",
|
||
"label": "🗂 未分配迭代 ({})".format(none_cnt),
|
||
"is_leaf": False,
|
||
})
|
||
return json.dumps(nodes, ensure_ascii=False)
|
||
|
||
if node_id.startswith('iter:'):
|
||
rest = node_id[5:]
|
||
parts = rest.split(':phase:')
|
||
iter_id = parts[0]
|
||
phase_role = parts[1] if len(parts) > 1 else ''
|
||
|
||
# 该迭代的迭代名
|
||
iter_name = ''
|
||
if iter_id != '__none__':
|
||
for it in iters:
|
||
if getattr(it, 'id', '') == iter_id:
|
||
iter_name = getattr(it, 'iteration_name', '') or ''
|
||
break
|
||
|
||
# 该迭代的任务
|
||
def _in_iter(t):
|
||
if iter_id == '__none__':
|
||
return not _iter_of(t)
|
||
return _iter_of(t) == iter_name
|
||
|
||
iter_tasks = [t for t in all_tasks if _in_iter(t)]
|
||
|
||
if not phase_role:
|
||
# 阶段列表
|
||
role_cnt = {}
|
||
for t in iter_tasks:
|
||
# 子任务（有 parent_id）挂在父任务下展示，不计入阶段计数，
|
||
# 否则阶段节点计数与点开后的平铺任务数不一致（计数含子任务、平铺只列顶层任务）。
|
||
if (getattr(t, 'parent_id', '') or '').strip():
|
||
continue
|
||
r = (getattr(t, 'role', '') or '').strip()
|
||
role_cnt[r] = role_cnt.get(r, 0) + 1
|
||
nodes = []
|
||
for rn, rzh in _get_phase_order():
|
||
if rn in role_cnt:
|
||
nodes.append({
|
||
"id": "iter:{}:phase:{}".format(iter_id, rn),
|
||
"label": "📌 {} ({})".format(rzh, role_cnt[rn]),
|
||
"is_leaf": False,
|
||
})
|
||
if '' in role_cnt:
|
||
nodes.append({
|
||
"id": "iter:{}:phase:__none__".format(iter_id),
|
||
"label": "📌 其他 ({})".format(role_cnt['']),
|
||
"is_leaf": False,
|
||
})
|
||
return json.dumps(nodes, ensure_ascii=False)
|
||
|
||
# 任务列表（只列顶层任务：parent_id 为空的；子任务挂在父任务下展开，避免平铺重复）
|
||
nodes = []
|
||
for t in iter_tasks:
|
||
r = (getattr(t, 'role', '') or '').strip()
|
||
if phase_role == '__none__':
|
||
if r:
|
||
continue
|
||
elif r != phase_role:
|
||
continue
|
||
tid = getattr(t, 'id', '')
|
||
title = getattr(t, 'title', '') or ''
|
||
state = getattr(t, 'state', '') or ''
|
||
parent = (getattr(t, 'parent_id', '') or '').strip()
|
||
if parent:
|
||
continue # 子任务不在阶段平铺，挂在父任务下展示
|
||
if not tid or not title:
|
||
continue
|
||
icon = _state_icons.get(state, '⬜')
|
||
zh = _state_zh.get(state, state)
|
||
has_child = any((getattr(c, 'parent_id', '') or '').strip() == tid for c in all_tasks)
|
||
nodes.append({
|
||
"id": tid,
|
||
"label": "{} [{}] {}".format(icon, zh, title),
|
||
"is_leaf": not has_child,
|
||
})
|
||
return json.dumps(nodes, ensure_ascii=False)
|
||
|
||
# 子任务层：裸任务ID → 返回该任务的子任务（parent_id == 该ID），递归支持多级分解
|
||
nodes = []
|
||
for t in all_tasks:
|
||
if (getattr(t, 'parent_id', '') or '').strip() != node_id:
|
||
continue
|
||
tid = getattr(t, 'id', '')
|
||
title = getattr(t, 'title', '') or ''
|
||
state = getattr(t, 'state', '') or ''
|
||
if not tid or not title:
|
||
continue
|
||
icon = _state_icons.get(state, '⬜')
|
||
zh = _state_zh.get(state, state)
|
||
has_child = any((getattr(c, 'parent_id', '') or '').strip() == tid for c in all_tasks)
|
||
nodes.append({
|
||
"id": tid,
|
||
"label": "{} [{}] {}".format(icon, zh, title),
|
||
"is_leaf": not has_child,
|
||
})
|
||
return json.dumps(nodes, ensure_ascii=False)
|