pipeline-sdlc/wwwroot/api/task_tree.dspy

# 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:
    # 当前项目
    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)