# bid_task_tree.dspy - 投标产线任务树（懒加载，与开发产线 /task 同构） # 树状结构：项目 → 阶段（角色分组） → 任务 → 步骤/交付件 # 协议：无 id → 根节点；__root__ → 阶段；role:xxx → 任务；裸任务ID → 步骤+交付件 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-bidding') _state_icons = { 'completed': '\u2705', 'approved': '\u2705', 'running': '\U0001f7e1', 'failed': '\u274c', 'waiting': '\u26a0\ufe0f', 'submitted': '\u2b1c', 'review': '\U0001f440', 'rejected': '\U0001f6ab', 'paused': '\u23f8\ufe0f', 'cancelled': '\U0001f6d1', 'qc_review': '\U0001f50d', } _state_zh = { 'completed': '已完成', 'approved': '已批准', 'running': '运行中', 'failed': '失败', 'waiting': '等待中', 'submitted': '已提交', 'review': '审核中', 'rejected': '已驳回', 'paused': '已暂停', 'cancelled': '已取消', 'qc_review': '审核中', } # 投标产线角色分组（按业务流转顺序） _ROLE_GROUPS = [ ('agent.tender_analyst', '\U0001f50d 招标解析'), ('agent.qc', '\U0001f4cb QC契合度审核'), ('agent.bid_prep', '\U0001f4c4 资料准备'), ('agent.bid_writer', '\u270d\ufe0f 技术章编写'), ('agent.bid_biz_writer', '\u270d\ufe0f 商务章编写'), ('agent.bid_reviewer', '\U0001f50e 章节评审'), ('agent.bid_compositor', '\U0001f4e6 标书合成'), ('agent.bid_scorer', '\U0001f3af 整书评分'), ('agent.pm', '\U0001f91d 项目管理'), ] _ROLE_SET = set(r for r, _ in _ROLE_GROUPS) async with DBPools().sqlorContext(dbname) as sor: # ── 解析当前投标项目（三级兜底） ── pid = '' # 1) 投标产线默认会话的项目绑定 recs = await sor.sqlExe( "SELECT current_project_id FROM pipeline_session_settings " "WHERE user_id=${u}$ AND session_id='default_bidding_general' LIMIT 1", {"u": uid}) if recs: pid = getattr(recs[0], 'current_project_id', '') or '' # 2) agent 全局设置 if not pid: recs = await sor.sqlExe( "SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$", {"u": uid}) if recs: pid = getattr(recs[0], 'current_project_id', '') or '' # 3) 最近一个进行中的投标项目 if not pid: recs = await sor.sqlExe( "SELECT id FROM sd_projects WHERE pipeline_id='bidding_general' " "AND status='in_progress' ORDER BY updated_at DESC LIMIT 1") if recs: pid = getattr(recs[0], 'id', '') or '' 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 '' if not node_id: # 根节点 return json.dumps([{"id": "__root__", "label": pname or "投标项目", "is_leaf": False}], ensure_ascii=False) # 一次查全部任务 + 步骤数 + 交付件数（项目内任务量不大，Python 分组） all_tasks = await sor.sqlExe( "SELECT id, title, role, state, params, parent_id, created_at, updated_at " "FROM pipeline_tasks WHERE tenant_id=${p}$ ORDER BY created_at ASC", {"p": pid}) or [] step_cnt = {} for r in await sor.sqlExe( "SELECT task_id, COUNT(*) AS c FROM pipeline_task_steps " "WHERE task_id IN (SELECT id FROM pipeline_tasks WHERE tenant_id=${p}$) GROUP BY task_id", {"p": pid}) or []: step_cnt[getattr(r, 'task_id', '')] = int(getattr(r, 'c', 0) or 0) del_cnt = {} for r in await sor.sqlExe( "SELECT task_id, COUNT(*) AS c FROM pipeline_deliverables WHERE project_id=${p}$ GROUP BY task_id", {"p": pid}) or []: del_cnt[getattr(r, 'task_id', '')] = int(getattr(r, 'c', 0) or 0) if node_id == '__root__': # 阶段列表（按角色分组，只列有任务的） role_cnt = {} for t in all_tasks: 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, label in _ROLE_GROUPS: if rn in role_cnt: nodes.append({"id": "role:" + rn, "label": "{} ({})".format(label, role_cnt[rn]), "is_leaf": False}) other_cnt = sum(v for k, v in role_cnt.items() if k not in _ROLE_SET) if other_cnt > 0: nodes.append({"id": "role:__other__", "label": "\U0001f4ad 其他 ({})".format(other_cnt), "is_leaf": False}) if not nodes: nodes = [{"id": "__empty__", "label": "暂无任务", "is_leaf": True}] return json.dumps(nodes, ensure_ascii=False) if node_id.startswith('role:'): # 该角色分组下的任务（只列顶层任务） role = node_id[5:] nodes = [] for t in all_tasks: r = (getattr(t, 'role', '') or '').strip() if role == '__other__': if r in _ROLE_SET: continue elif r != role: continue if (getattr(t, 'parent_id', '') or '').strip(): 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, '\u2b1c') zh = _state_zh.get(state, state) upd = str(getattr(t, 'updated_at', '') or '')[5:16] has_child = step_cnt.get(tid, 0) > 0 or del_cnt.get(tid, 0) > 0 or \ any((getattr(c, 'parent_id', '') or '').strip() == tid for c in all_tasks) nodes.append({ "id": tid, "label": "{} [{}] {}".format(icon, zh, title[:60]), "is_leaf": not has_child, "upd": upd, }) return json.dumps(nodes, ensure_ascii=False) # ── 裸任务 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, '\u2b1c') zh = _state_zh.get(state, state) nodes.append({"id": tid, "label": "{} [{}] {}".format(icon, zh, title[:60]), "is_leaf": not (step_cnt.get(tid, 0) > 0 or del_cnt.get(tid, 0) > 0)}) steps = await sor.sqlExe( "SELECT id, step_name, display_name, state FROM pipeline_task_steps " "WHERE task_id=${t}$ ORDER BY step_order ASC", {"t": node_id}) or [] for s in steps: sid = getattr(s, 'id', '') label = getattr(s, 'display_name', '') or getattr(s, 'step_name', '') or '' st = getattr(s, 'state', '') or '' if label: nodes.append({"id": "step:" + sid, "label": " {} [{}] \u2699\ufe0f {}".format(_state_icons.get(st, '\u2b1c'), _state_zh.get(st, st), label[:50]), "is_leaf": True}) dels = await sor.sqlExe( "SELECT id, deliverable_type, title FROM pipeline_deliverables " "WHERE task_id=${t}$ ORDER BY created_at DESC", {"t": node_id}) or [] for d in dels: did = getattr(d, 'id', '') dt = getattr(d, 'deliverable_type', '') or '' dtitle = getattr(d, 'title', '') or '' nodes.append({"id": "del:" + did, "label": " \U0001f4e4 {}: {}".format(dt, (dtitle or '')[:40]), "is_leaf": True}) return json.dumps(nodes, ensure_ascii=False)