diff --git a/pipeline_bidding/bid_flow.py b/pipeline_bidding/bid_flow.py index 3b493d7..c9db20f 100644 --- a/pipeline_bidding/bid_flow.py +++ b/pipeline_bidding/bid_flow.py @@ -280,34 +280,46 @@ async def _auto_close_file_upload_task(sor, project_id): # ══════════════ QC 门禁辅助 ══════════════ async def _qc_pending_types(sor, project_id): - """返回未放行的产出类型清单（按依赖序），每项 (qc_type, state, improvement, round)： + """返回未放行的产出类型清单（按依赖序），每项 (qc_type, state, improvement, round, out_newer)： state=unaudited：产出已存在但从未审核；state=rework：最近一轮审核未通过。 已通过（或人工在记录页强制放行）的类型不再出现。 + out_newer（2026-09-03）：产出表最新更新时间晚于最近一轮审核时间—— + 说明审核后产出被人工/修正任务更新过，即使轮次用尽也应再派一轮复审 + （闭环「修正落库→自动复审」；只有产出真更新才开轮，不会无限烧轮）。 """ pending = [] for t in QC_TYPES: tbl = QC_OUTPUT_TABLE[t] - n_out = await _count(sor, "SELECT COUNT(*) AS c FROM " + tbl + - " WHERE project_id=${p}$", {"p": project_id}) recs = await sor.sqlExe( - "SELECT round, passed, improvement FROM bid_qc_reviews " + "SELECT round, passed, improvement, updated_at FROM bid_qc_reviews " "WHERE project_id=${p}$ AND qc_type=${t}$ ORDER BY round DESC LIMIT 1", {"p": project_id, "t": t}) await sor.sqlExe("COMMIT", {}) if not recs: + n_out = await _count(sor, "SELECT COUNT(*) AS c FROM " + tbl + + " WHERE project_id=${p}$", {"p": project_id}) if n_out > 0: - pending.append((t, "unaudited", "", 0)) + pending.append((t, "unaudited", "", 0, False)) continue r = rec_to_dict(recs[0]) if str(r.get("passed")) == "1": continue rnd = to_int(r.get("round"), 1) + rev_t = str(r.get("updated_at") or r.get("created_at") or "") + out_rec = await sor.sqlExe( + "SELECT MAX(GREATEST(created_at, IFNULL(updated_at, created_at))) AS mt " + "FROM " + tbl + " WHERE project_id=${p}$", {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + out_t = str(getattr(out_rec[0], 'mt', '') or '') if out_rec else "" + out_newer = bool(out_t and rev_t and out_t > rev_t) + n_out = await _count(sor, "SELECT COUNT(*) AS c FROM " + tbl + + " WHERE project_id=${p}$", {"p": project_id}) if n_out > 0: - pending.append((t, "rework", r.get("improvement") or "", rnd)) + pending.append((t, "rework", r.get("improvement") or "", rnd, out_newer)) else: # 产出已被清空待重做：round 信息带回，供对账器判断是否已超轮次上限 - pending.append((t, "awaiting_redo", r.get("improvement") or "", rnd)) + pending.append((t, "awaiting_redo", r.get("improvement") or "", rnd, False)) return pending @@ -531,12 +543,14 @@ async def reconcile_project(sor, project_id, project_name=""): qc_pending = await _qc_pending_types(sor, project_id) over_types = set() if qc_pending: - # 轮次用尽守卫：最新审核已达上限仍未通过 → 不再空转派审核任务， + # 轮次用尽守卫：最新审核已达上限仍未通过且产出未更新 → 不再空转派审核任务， # 确保有阻塞人工任务（逃逸阀：人工修产出物或在记录页强制放行后流程自恢复）。 - over = [(t, s, imp, rnd) for t, s, imp, rnd in qc_pending - if s == "rework" and rnd >= th["qc_max_round"]] + # 2026-09-03：产出在审核后更新过（out_newer，人工/修正任务落库）→ 不算用尽， + # 闭环「修正落库→自动复审」；只有产出真更新才开轮，不会无限烧轮。 + over = [(t, s, imp, rnd, on) for t, s, imp, rnd, on in qc_pending + if s == "rework" and rnd >= th["qc_max_round"] and not on] if over: - over_types = set(t for t, _, _, _ in over) + over_types = set(t for t, _, _, _, _ in over) ht = await find_human_task(sor, project_id, HT_QC_ESCALATION, status="pending") if not ht: await create_human_task( @@ -547,13 +561,13 @@ async def reconcile_project(sor, project_id, project_name=""): "把对应记录 passed 改为 1 强制放行。\n处理完成本任务后流程自动恢复。" % (th["qc_max_round"], "\n".join("- %s（第%s轮）：%s" % (t, rnd, (imp or "")[:300]) - for t, s, imp, rnd in over))), + for t, s, imp, rnd, _on in over))), assignee_role="owner.superuser") await sor.sqlExe("COMMIT", {}) acts.append("escalated: QC 审核轮次用尽（%s），等人工介入；不依赖它的章节照常流转" % "/".join(sorted(over_types))) # awaiting_redo：产出已被 QC 清空 → 重派对应维度的重做任务，不能对空产出审核 - needs_redo = [(t, s, imp, rnd) for t, s, imp, rnd in qc_pending if s == "awaiting_redo"] + needs_redo = [(t, s, imp, rnd) for t, s, imp, rnd, _on in qc_pending if s == "awaiting_redo"] if needs_redo: redo_dims = [] for t, _, _, _ in needs_redo: @@ -588,7 +602,7 @@ async def reconcile_project(sor, project_id, project_name=""): else: acts.append("waiting: 分析重做任务在办（QC 退回产出已清空）") # 逐类型并行审核：每类一个审核任务（跳过：在办的类型 / 空产出待重做 / 轮次用尽等人工） - needs_qc = [(t, s, imp, rnd) for t, s, imp, rnd in qc_pending + needs_qc = [(t, s, imp, rnd) for t, s, imp, rnd, _on in qc_pending if s == "unaudited" or (s == "rework" and t not in over_types)] created_qc = [] for t, _, _, _ in needs_qc: diff --git a/wwwroot/api/bid_task_tree.dspy b/wwwroot/api/bid_task_tree.dspy index 042a1e6..a77db28 100644 --- a/wwwroot/api/bid_task_tree.dspy +++ b/wwwroot/api/bid_task_tree.dspy @@ -219,7 +219,9 @@ async with DBPools().sqlorContext(dbname) as sor: return json.dumps(nodes, ensure_ascii=False) if role == 'agent.qc': - # QC 按审核记录分组（同一任务可多轮复审）：一次过=叶子，多轮=组节点 + # QC 按审核记录分组（同一任务可多轮复审）：一次过=叶子，多轮=组节点。 + # 2026-09-03：正在执行/还没审核的任务也是任务——无审核记录的 QC 任务 + # 按 params.qc_types 归入对应类型组，组标签带在办数，展开可见。 revs = await sor.sqlExe( "SELECT id, task_id, qc_type, round, fit_score, passed FROM bid_qc_reviews " "WHERE project_id=${p}$ ORDER BY qc_type, round ASC", {"p": pid}) or [] @@ -227,9 +229,21 @@ async with DBPools().sqlorContext(dbname) as sor: for rv in revs: groups.setdefault(getattr(rv, 'qc_type', '') or '__none__', []).append(rv) reviewed_tids = set(getattr(rv, 'task_id', '') for rv in revs) - legacy = [t for t in role_tasks if getattr(t, 'id', '') not in reviewed_tids] - if legacy: - groups.setdefault('__legacy__', []) + open_tasks = {} + for t in role_tasks: + if getattr(t, 'id', '') in reviewed_tids: + continue + try: + qts = json.loads(getattr(t, 'params', '') or '{}').get('qc_types') or [] + except Exception: + qts = [] + for qt in qts: + open_tasks.setdefault(qt, []).append(t) + if not qts: + open_tasks.setdefault('__legacy__', []).append(t) + for qt in open_tasks: + groups.setdefault(qt, []) + legacy = open_tasks.get('__legacy__', []) qc_labels = { 'scoring_items': '评分项 QC', 'qualifications': '资质 QC', 'doc_requirements': '投标文件要求 QC', 'chapter_outline': '章节骨架 QC', @@ -244,6 +258,7 @@ async with DBPools().sqlorContext(dbname) as sor: continue rs = groups[key] label = qc_labels.get(key, key) + n_open = len(open_tasks.get(key, [])) if legacy and key == '__legacy__': if not rs and legacy: # 只有遗留任务无审核记录：多任务=组，单任务=叶 @@ -259,8 +274,8 @@ async with DBPools().sqlorContext(dbname) as sor: "label": "{}（{} 个早期任务）".format(label, len(legacy)), "is_leaf": False}) continue - if len(rs) == 1 and str(getattr(rs[0], 'passed', '0')) == '1': - # 一次过：叶子 + if len(rs) == 1 and str(getattr(rs[0], 'passed', '0')) == '1' and not n_open: + # 一次过且无在办：叶子 t_id = getattr(rs[0], 'task_id', '') sc = getattr(rs[0], 'fit_score', '') nodes.append({ @@ -271,10 +286,12 @@ async with DBPools().sqlorContext(dbname) as sor: else: n_rounds = len(rs) if rs else len(legacy) last_passed = str(getattr(rs[-1], 'passed', '0')) == '1' if rs else False + tail = '通过' if last_passed else '未通过' + if n_open: + tail += ' + %d在办' % n_open nodes.append({ "id": "grp:" + role + ":" + key, - "label": "{}（{} 轮）最新[{}]".format(label, n_rounds, - '通过' if last_passed else '未通过'), + "label": "{}（{} 轮）最新[{}]".format(label, n_rounds, tail), "is_leaf": False, }) return json.dumps(nodes, ensure_ascii=False) @@ -311,39 +328,51 @@ async with DBPools().sqlorContext(dbname) as sor: 'reqs_outline': 'doc_requirements', 'cost_benefit': 'cost_benefit'} if role == 'agent.qc': - # QC 组：按审核记录轮次升序展开（同一任务多轮复审各占一行） + # QC 组：按审核记录轮次升序展开（同一任务多轮复审各占一行）， + # 末尾追加该类型正在执行/还没审核的任务（也是任务，不能藏）。 revs = await sor.sqlExe( "SELECT id, task_id, qc_type, round, fit_score, passed, created_at " "FROM bid_qc_reviews WHERE project_id=${p}$ AND qc_type=${k}$ " "ORDER BY round ASC", {"p": pid, "k": key}) or [] - if revs: - nodes = [] - for rv in revs: - rnd = int(getattr(rv, 'round', 0) or 0) - passed = str(getattr(rv, 'passed', '0')) == '1' - sc = float(getattr(rv, 'fit_score', 0) or 0) - upd = str(getattr(rv, 'created_at', '') or '')[5:16] - icon = '\u2705' if passed else '\U0001f6ab' - nodes.append({ - "id": "rev:" + getattr(rv, 'id', ''), - "label": "第%d轮 %s [%s] %.1f 分 %s" % ( - rnd, icon, '通过' if passed else '未通过', sc, upd), - "is_leaf": True, - }) - return json.dumps(nodes, ensure_ascii=False) - # 无审核记录的早期 QC 任务：平铺 nodes = [] + reviewed_tids = set() + for rv in revs: + reviewed_tids.add(getattr(rv, 'task_id', '')) + rnd = int(getattr(rv, 'round', 0) or 0) + passed = str(getattr(rv, 'passed', '0')) == '1' + sc = float(getattr(rv, 'fit_score', 0) or 0) + upd = str(getattr(rv, 'created_at', '') or '')[5:16] + icon = '\u2705' if passed else '\U0001f6ab' + nodes.append({ + "id": "rev:" + getattr(rv, 'id', ''), + "label": "第%d轮 %s [%s] %.1f 分 %s" % ( + rnd, icon, '通过' if passed else '未通过', sc, upd), + "is_leaf": True, + }) for t in all_tasks: if (getattr(t, 'role', '') or '').strip() != role: continue if (getattr(t, 'parent_id', '') or '').strip(): continue + tid = getattr(t, 'id', '') + if tid in reviewed_tids: + continue + try: + qts = json.loads(getattr(t, 'params', '') or '{}').get('qc_types') or [] + except Exception: + qts = [] + if key == '__legacy__' and qts: + continue + if key != '__legacy__' and key not in qts: + continue st = getattr(t, 'state', '') or '' - nodes.append({"id": getattr(t, 'id', ''), - "label": "{} [{}] {}".format(_state_icons.get(st, '\u2b1c'), - _state_zh.get(st, st), - str(getattr(t, 'updated_at', '') or '')[5:16]), - "is_leaf": not (del_cnt.get(getattr(t, 'id', ''), 0) > 0)}) + nodes.append({ + "id": tid, + "label": "%s [%s] %s" % (_state_icons.get(st, '\u2b1c'), + _state_zh.get(st, st), + (getattr(t, 'title', '') or '')[:40]), + "is_leaf": not (del_cnt.get(tid, 0) > 0), + }) return json.dumps(nodes, ensure_ascii=False) # analyst 组：按任务创建序展开，轮次号与QC结果取自审核记录