diff --git a/pipeline_bidding/bid_flow.py b/pipeline_bidding/bid_flow.py index c9db20f..ddc9ec5 100644 --- a/pipeline_bidding/bid_flow.py +++ b/pipeline_bidding/bid_flow.py @@ -88,6 +88,14 @@ ANALYSIS_DIMS = ( DIM_QC_TYPES = {d: ts for d, _, ts in ANALYSIS_DIMS} QC_TYPE_TO_DIM = {q: d for d, _, ts in ANALYSIS_DIMS for q in ts} +# 维度 → 落库表(落库核验用;reqs_outline 维度同时写两张表) +DIM_TABLES = { + "scoring": (QC_OUTPUT_TABLE[QC_TYPE_SCORING],), + "quals": (QC_OUTPUT_TABLE[QC_TYPE_QUALS],), + "reqs_outline": (QC_OUTPUT_TABLE[QC_TYPE_REQS], QC_OUTPUT_TABLE[QC_TYPE_OUTLINE]), + "cost_benefit": (QC_OUTPUT_TABLE[QC_TYPE_COST],), +} + # 章节级上游依赖矩阵(2026-09-03):每种章节 section 只依赖自己必需的分析产出 # 通过 QC——某一维度 QC 卡住只阻塞依赖它的章节,其余章节照常流转。 # 公共事实:章节骨架(bid_chapters)本身就是章节,OUTLINE 过了章节才可信; @@ -710,6 +718,61 @@ async def reconcile_project(sor, project_id, project_name=""): if created_r: await sor.sqlExe("COMMIT", {}) + # ── D2. 落库核验(2026-09-03,机制硬门禁)── + # 实测根因:修正任务 approved 但产出只写文件没落库(bid_chapters 停在旧时间), + # 机制对旧数据不复审、流程静默卡死。核验:approved 的分析/修正任务, + # 对应产出表 max 更新时间必须 ≥ 任务创建时间,否则抛人工补落库/重跑。 + verify_bad = [] + _vtasks = await sor.sqlExe( + "SELECT id, title, params, created_at FROM pipeline_tasks WHERE tenant_id=${p}$ " + "AND role=${r}$ AND state='approved'", {"p": project_id, "r": R_ANALYST}) or [] + for t in _vtasks: + try: + p = json.loads(getattr(t, 'params', '') or '{}') + except Exception: + p = {} + dim = (p.get('analysis_dim') or '').strip() + title = getattr(t, 'title', '') or '' + is_manual = ('修正' in title or '补录' in title) and not p.get('qc_redo') + if not dim and not is_manual: + continue + if dim: + tbls = DIM_TABLES.get(dim) or (QC_OUTPUT_TABLE[QC_TYPE_REQS], QC_OUTPUT_TABLE[QC_TYPE_OUTLINE]) + else: + # 人工修正任务按标题关键词定目标表,避免跨维度误判 + if 'chapter_outline' in title or '骨架' in title: + tbls = (QC_OUTPUT_TABLE[QC_TYPE_OUTLINE],) + elif 'doc_requirements' in title or '文件要求' in title: + tbls = (QC_OUTPUT_TABLE[QC_TYPE_REQS],) + else: + tbls = (QC_OUTPUT_TABLE[QC_TYPE_REQS], QC_OUTPUT_TABLE[QC_TYPE_OUTLINE]) + t_created = str(getattr(t, 'created_at', '') or '') + fresh = False + for tbl in tbls: + 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}) + mt = str(getattr(rec[0], 'mt', '') or '') if rec else '' + if mt and mt >= t_created: + fresh = True + break + if not fresh: + verify_bad.append((getattr(t, 'id', ''), title[:50], dim or 'reqs_outline')) + if verify_bad: + ht = await find_human_task(sor, project_id, "output_verify", status="pending") + if not ht: + await create_human_task( + sor, project_id, "output_verify", + "分析/修正任务声称完成但产出未落库", + ("以下任务已 approved,但对应产出表的最大更新时间早于任务创建时间" + "(只写了文件或根本没写库):\n%s\n\n" + "请处理:在对应管理页补录落库,或重跑该任务;落库后流程自动复审。" + "处理完完成本任务。" + % "\n".join("- %s [%s] %s" % (tid, dm, ti) for tid, ti, dm in verify_bad)), + assignee_role="owner.superuser") + await sor.sqlExe("COMMIT", {}) + acts.append("escalated: 产出落库核验不通过(%d 个任务),抛人工" % len(verify_bad)) + if created_w or created_r: return acts