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': '已取消',
}
# 阶段顺序（role 规范名 → 中文名）
_phase_order = [
    ('agent.requirement', '需求阶段'),
    ('agent.design', '设计阶段'),
    ('agent.develop', '开发阶段'),
    ('agent.test', '测试阶段'),
    ('agent.deploy', '部署阶段'),
]


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 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))
        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:
                r = (getattr(t, 'role', '') or '').strip()
                role_cnt[r] = role_cnt.get(r, 0) + 1
            nodes = []
            for rn, rzh in _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)

        # 任务列表
        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 ''
            if not tid or not title:
                continue
            icon = _state_icons.get(state, '⬜')
            zh = _state_zh.get(state, state)
            nodes.append({
                "id": tid,
                "label": "{} [{}] {}".format(icon, zh, title),
                "is_leaf": True,
            })
        return json.dumps(nodes, ensure_ascii=False)

    return json.dumps([], ensure_ascii=False)