# -*- coding:utf-8 -*- """P3 可行性研究 E2E（测试机真机）： fp_estimate 规则引擎独立复算 → draft_feasibility 正/负链 → score_candidate → promote 门禁（draft/confirmed 拒绝→审批 approved 放行→立项落库核对）→ 八节 PPT 生成 → 项目/org 隔离。钉钉审批通道 monkeypatch 为失败→走降级人工（不真发）。 用法（测试机 /d/pipeline/pipeline-app 下）： setsid nohup ./py3/bin/python pkgs/pipeline-opportunity/scripts/p3_feasibility_e2e.py \ > /tmp/p3_e2e.log 2>&1 & 环境变量： P3_PROJECT_ID 正链项目（默认 L2NJCn7XuOnzVV89lpXF1，owner=user-01） P3_CLUSTER 正链类别（默认 AAbym8nFt1fLH1DayfDec 公众号 39条，须属于 P3_PROJECT_ID） P3_CLUSTER_PERSONAL 全personal类别（默认 d6D5BAoEHgBn7KeiNbuE1，normalize done 无 core/ext） P3_CLUSTER_OTHER 跨项目隔离用类别（默认 3kPqI-COvrU66QQhbEDp_，批次平台级） P3_KEEP =1 保留 E2E 建的可行性报告与项目（默认报告保留、立项项目删除） """ import asyncio import importlib.util import json import math import os import sys import time W = "/d/pipeline/pipeline-app" os.chdir(W) sys.path.insert(0, W) from appPublic.folderUtils import ProgramPath from appPublic.jsonConfig import getConfig from appPublic.event_dispatcher import EventDispatcher from sqlor.dbpools import DBPools from ahserver.serverenv import ServerEnv from ahserver.globalEnv import initEnv p = ProgramPath() c = getConfig(W, {"workdir": W, "ProgramPath": p}) DBPools(c.databases) se = ServerEnv() se.event_dispatcher = EventDispatcher() se.get_module_dbname = lambda m: "pipeline" initEnv() PID = os.environ.get("P3_PROJECT_ID", "L2NJCn7XuOnzVV89lpXF1") CL_OK = os.environ.get("P3_CLUSTER", "AAbym8nFt1fLH1DayfDec") # 公众号 39条 init CL_PERSONAL = os.environ.get("P3_CLUSTER_PERSONAL", "d6D5BAoEHgBn7KeiNbuE1") # done 全personal CL_PLATFORM = os.environ.get("P3_CLUSTER_OTHER", "3kPqI-COvrU66QQhbEDp_") # 平台级批次类别 KEEP = os.environ.get("P3_KEEP", "") == "1" FAIL = [] def check(n, cond, d=""): print("[%s] %s %s" % ("OK " if cond else "FAIL", n, str(d)[:220]), flush=True) if not cond: FAIL.append(n) # fp_calc 规则引擎（skills/global 部署位），用于独立复算 FP_CALC = "/d/pipeline/pipeline-app/skills/global/function-point-counting/scripts/fp_calc.py" _spec = importlib.util.spec_from_file_location("fp_calc_ref", FP_CALC) fp_ref = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(fp_ref) CTX = {"project_id": PID, "user_id": "user-01", "org_id": "0", "pipeline_id": "opportunity_general", "space": "opportunity_general", "session_id": "e2e-p3", "workspace_dir": "/tmp"} async def sql(q, args=None, one=False): async with DBPools().sqlorContext("pipeline") as sor: r = await sor.sqlExe(q, args or {}) await sor.sqlExe("COMMIT", {}) return r[0] if (one and r) else r async def m(): from pipeline_opportunity import opp_ability ab = opp_ability.register_opp_ability() async def call(t, prm, ctx=None): async with DBPools().sqlorContext("pipeline") as sor: return await ab.handlers[t](sor, prm, ctx or CTX) print("== P3 E2E start %s project=%s ==" % (time.strftime("%F %T"), PID), flush=True) # ── A. fp_estimate 规则引擎（独立复算，不经 LLM）── funcs = [ {"name": "用户档案", "type": "ILF", "det": 12, "ret": 2, "ftr": 0, "confidence": "high"}, {"name": "订单录入", "type": "EI", "det": 8, "ret": 0, "ftr": 1, "confidence": "high"}, {"name": "报表导出", "type": "EO", "det": 9, "ret": 0, "ftr": 2, "confidence": "medium"}, {"name": "余额查询", "type": "EQ", "det": 4, "ret": 0, "ftr": 1, "confidence": "medium"}, {"name": "外部支付接口", "type": "EIF", "det": 6, "ret": 1, "ftr": 0, "confidence": "low"}, ] out = await call("opp_fp_estimate", {"functions_json": json.dumps(funcs), "project_name": "E2E样例"}) d = json.loads(out) if out.startswith("{") else {} ref = fp_ref.calc({"project": "E2E样例", "functions": [ dict(f, ret=f.get("ret") or (1 if f["type"] in ("ILF", "EIF") else 0)) for f in funcs]}) check("A1 fp_total=独立复算", d.get("总FP") == ref["total_fp"], "tool=%s ref=%s" % (d.get("总FP"), ref["total_fp"])) check("A2 分类型计数一致", d.get("分类型") == ref["counts"], "%s vs %s" % (d.get("分类型"), ref["counts"])) check("A3 四件套markdown非空", bool((d.get("四件套") or "").strip()), len(d.get("四件套") or "")) bad = await call("opp_fp_estimate", {"functions_json": json.dumps( [{"name": "x", "type": "XXX", "det": 5}, {"name": "y", "type": "EI", "det": 6, "ftr": 1}])}) db_ = json.loads(bad) if bad.startswith("{") else {} # EI det=6(档5-15) ftr=1(档0-1) → 低复杂度 → 3 FP（独立复算核对） ref_bad = fp_ref.calc({"project": "", "functions": [ {"name": "y", "type": "EI", "det": 6, "ftr": 1}]}) check("A4 非法type进pending不炸", db_.get("总FP") == ref_bad["total_fp"] == 3 and len(db_.get("待确认") or []) == 1, "fp=%s ref=%s pending=%d" % (db_.get("总FP"), ref_bad["total_fp"], len(db_.get("待确认") or []))) e1 = await call("opp_fp_estimate", {"functions_json": "[]"}) check("A5 空数组拒绝", e1.startswith("ERROR:"), e1[:80]) # ── B. draft_feasibility 负链 ── b1 = await call("opp_draft_feasibility", {"cluster_id": CL_PERSONAL}, CTX) check("B1 全personal类别拒绝(样本不足)", b1.startswith("ERROR:") and "样本不足" in b1, b1[:160]) ctxB = dict(CTX); ctxB["project_id"] = "FAKE_PROJECT_X" b2 = await call("opp_draft_feasibility", {"cluster_id": CL_OK}, ctxB) check("B2 跨项目类别拒绝", b2.startswith("ERROR:") and "其他项目" in b2, b2[:160]) ctxC = dict(CTX); ctxC["org_id"] = "org-evil" b3 = await call("opp_draft_feasibility", {"cluster_id": CL_PLATFORM}, ctxC) check("B3 伪org拒绝", b3.startswith("ERROR:") and "机构" in b3, b3[:160]) b4 = await call("opp_draft_feasibility", {"cluster_id": "NO_SUCH_CLUSTER"}) check("B4 类别不存在拒绝", b4.startswith("ERROR:") and "不存在" in b4, b4[:120]) # 未归一化的其它 init 类别（限当前会话可见：平台级批次或本项目批次） r = await sql( "SELECT c.id FROM opp_clusters c LEFT JOIN opp_mining_batches b ON b.id=c.batch_id " "WHERE c.normalize_status='init' AND c.id<>${a}$ AND (b.project_id='' OR b.project_id=${p}$) " "LIMIT 1", {"a": CL_OK, "p": PID}, one=True) if r: b5 = await call("opp_draft_feasibility", {"cluster_id": r.id}) check("B5 未归一化拒绝(引导先normalize)", b5.startswith("ERROR:") and "共性提取" in b5, b5[:150]) # ── C. normalize 正链类别（真实 LLM+VDB 路径，E2E 前置）── st = await sql("SELECT normalize_status FROM opp_clusters WHERE id=${c}$", {"c": CL_OK}, one=True) if getattr(st, "normalize_status", "") != "done": print("-- normalize %s (may take minutes) --" % CL_OK, flush=True) t0 = time.time() nout = await call("opp_normalize_cluster", {"cluster_id": CL_OK}) check("C1 normalize完成", nout.startswith("OK:"), "%s (%.0fs)" % (nout[:150], time.time() - t0)) else: check("C1 normalize完成(已done)", True, "skip") cov = await call("opp_coverage_report", {"cluster_id": CL_OK}) covd = json.loads(cov) if cov.startswith("{") else {} tiers = covd.get("层级统计") or {} n_core, n_ext = tiers.get("核心(≥60%)", 0), tiers.get("扩展(≥40%)", 0) check("C2 有core/ext共性(正链可走)", (n_core + n_ext) > 0, "core=%s ext=%s" % (n_core, n_ext)) # ── D. draft_feasibility 正链 ── t0 = time.time() dout = await call("opp_draft_feasibility", {"cluster_id": CL_OK}) check("D1 draft_feasibility成功", dout.startswith("OK:") and "report_id=" in dout, "%s (%.0fs)" % (dout[:150], time.time() - t0)) RID = "" if "report_id=" in dout: RID = dout.split("report_id=")[1].split("（")[0].split("）")[0].strip() rep = await sql("SELECT * FROM opp_reports WHERE id=${r}$", {"r": RID}, one=True) if RID else None check("D2 落库report_type=feasibility", rep is not None and rep.report_type == "feasibility", getattr(rep, "report_type", None)) content = getattr(rep, "content", "") or "" SECS = ("## 市场需求", "## 共性需求", "## 需求规格", "## 架构与可行性", "## FP与成本", "## 商业价值", "## 风险", "## 结论建议") miss = [s for s in SECS if s not in content] check("D3 八节齐全", not miss, "缺:%s" % miss if miss else "8/8") check("D4 挂项目%s" % PID[:8], getattr(rep, "project_id", "") == PID, rep.project_id if rep else "") check("D5 status=draft", getattr(rep, "status", "") == "draft", getattr(rep, "status", "")) # FP/成本与 params 系数一致性 import re as _re mm = _re.search(r"功能点合计 (\d+(?:\.\d+)?) FP.*?按 ([\d.]+) 人月/FP 估 ([\d.]+) 人月；" r"按 ([\d.]+) 万元/人月估研发成本 ([\d.]+) 万元", content) if mm: fp_v, pm_per, pm_v, wan_per, cost_v = (float(x) for x in mm.groups()) check("D6 成本=FP×系数(params一致)", abs(pm_v - round(fp_v * pm_per, 1)) < 0.05 and abs(cost_v - round(pm_v * wan_per, 1)) < 0.05 and pm_per == 0.05 and wan_per == 3.0, "FP=%s pm=%s cost=%s" % (fp_v, pm_v, cost_v)) check("D7 FP>0", fp_v > 0, fp_v) else: check("D6 成本=FP×系数", False, "FP与成本节格式不符: %s" % content[content.find("## FP"):content.find("## FP") + 200]) check("D7 FP>0", False, "") check("D8 架构节非LLM失败占位", "架构分析生成失败" not in content, "") # ── E. score_candidate ── sout = await call("opp_score_candidate", {"cluster_id": CL_OK}) sd = json.loads(sout) if sout.startswith("{") else {} sc = sd.get("score") check("E1 评分∈[0,100]", isinstance(sc, (int, float)) and 0 <= sc <= 100, sout[:150]) check("E2 三因子齐全", len(sd.get("factors") or {}) == 3, list((sd.get("factors") or {}).keys())) check("E3 优先级星", bool(sd.get("priority", "").startswith("★")), sd.get("priority")) # ── F. promote 门禁负链 ── f1 = await call("opp_promote_to_project", {"report_id": RID}) check("F1 draft立项拒绝(未过确认)", f1.startswith("ERROR:") and "确认" in f1, f1[:150]) # research 报告立项拒绝（report_type 门禁在 status 门禁之前） rr = await sql("SELECT id FROM opp_reports WHERE report_type='research' AND id<>${r}$ LIMIT 1", {"r": RID}, one=True) if rr: f2 = await call("opp_promote_to_project", {"report_id": rr.id}) check("F2 research报告立项拒绝", f2.startswith("ERROR:") and "仅可行性" in f2, f2[:120]) # ── G. 提交人工确认 + PPT ── from pipeline_opportunity.opp_report_capability import submit_for_confirmation, confirm_report, \ initiate_approval, resolve_approval g1ok, g1msg = await submit_for_confirmation(RID, note="P3 E2E") check("G1 submit_for_confirmation", g1ok, g1msg[:120]) rep2 = await sql("SELECT confirm_task_id,ppt_path,status FROM opp_reports WHERE id=${r}$", {"r": RID}, one=True) ht = getattr(rep2, "confirm_task_id", "") or "" check("G2 确认任务绑定", bool(ht), ht) htr = await sql("SELECT task_type,status FROM pipeline_human_tasks WHERE id=${i}$", {"i": ht}, one=True) if ht else None check("G3 人工任务pending", htr is not None and htr.status == "pending" and htr.task_type == "opp_report_confirm", "%s/%s" % (getattr(htr, "task_type", ""), getattr(htr, "status", ""))) ppt = getattr(rep2, "ppt_path", "") or "" ok_ppt = ppt and os.path.isfile(ppt) and os.path.getsize(ppt) > 5000 check("G4 PPT已生成", bool(ok_ppt), "%s %sB" % (ppt, os.path.getsize(ppt) if ppt and os.path.isfile(ppt) else -1)) if ok_ppt: from pptx import Presentation prs = Presentation(ppt) all_txt = [] for sl in prs.slides: for sh in sl.shapes: if sh.has_text_frame: all_txt.append(sh.text_frame.text) joined = "\n".join(all_txt) sec_hits = [s for s in ("市场需求", "共性需求", "需求规格", "架构与可行性", "FP与成本", "商业价值", "风险", "结论建议") if s in joined] check("G5 PPT八节章节页", len(sec_hits) >= 8, "%d/8 %s" % (len(sec_hits), sec_hits)) n_slides = len(prs.slides._sldIdLst) check("G6 PPT页数>10", n_slides > 10, n_slides) check("G7 PPT文件名feasibility前缀", os.path.basename(ppt).startswith("feasibility_report_"), os.path.basename(ppt)) # ── H. 人工确认回流 ── h1ok, h1msg = await confirm_report(RID, True, operator="user-01") stt = await sql("SELECT status FROM opp_reports WHERE id=${r}$", {"r": RID}, one=True) check("H1 confirmed", h1ok and stt.status == "confirmed", stt.status) ht2 = await sql("SELECT status FROM pipeline_human_tasks WHERE id=${i}$", {"i": ht}, one=True) check("H2 确认任务done", ht2 and ht2.status == "done", getattr(ht2, "status", "")) # ── I. confirmed 但未审批 → promote 拒绝 ── i1 = await call("opp_promote_to_project", {"report_id": RID}) check("I1 confirmed未审批拒绝", i1.startswith("ERROR:") and "审批" in i1, i1[:150]) # ── J. 发起审批（钉钉 monkeypatch 失败 → 降级人工兜底，不真发）── async def _fake_submit(biz, aid, title, applicant, org): return {"success": False, "message": "E2E monkeypatch: 跳过钉钉"} se.submit_approval = _fake_submit jok, aid = await initiate_approval(RID, note="P3 E2E 审批", created_by="user-01", applicant_id="user-01") check("J1 审批单发起", jok, str(aid)[:80]) stj = await sql("SELECT status FROM opp_reports WHERE id=${r}$", {"r": RID}, one=True) apr = await sql("SELECT status,report_id FROM opp_approvals WHERE id=${a}$", {"a": aid}, one=True) check("J2 报告approval_initiated+审批单initiated", stj.status == "approval_initiated" and apr and apr.status == "initiated", "%s/%s" % (stj.status, getattr(apr, "status", ""))) dht = await sql("SELECT id,status FROM pipeline_human_tasks WHERE project_id=${p}$ AND task_type='opp_dev_approval' AND status='pending' LIMIT 1", {"p": PID}, one=True) check("J3 降级人工审批任务兜底", dht is not None, getattr(dht, "id", "")) # ── K. 审批回流 approved ── kok, kmsg = await resolve_approval(aid, "approved", operator="user-01") stk = await sql("SELECT status FROM opp_reports WHERE id=${r}$", {"r": RID}, one=True) apk = await sql("SELECT status FROM opp_approvals WHERE id=${a}$", {"a": aid}, one=True) check("K1 报告approved+审批单approved", kok and stk.status == "approved" and apk.status == "approved", "%s/%s" % (stk.status, apk.status)) # ── L. 立项 ── lout = await call("opp_promote_to_project", {"report_id": RID}) check("L1 promote成功", lout.startswith("OK:") and "project_id=" in lout, lout[:160]) NEWPID = lout.split("project_id=")[1].split("（")[0].strip() if "project_id=" in lout else "" if NEWPID: np = await sql("SELECT name,project_type,pipeline_id,org_id,created_by,status FROM sd_projects WHERE id=${i}$", {"i": NEWPID}, one=True) check("L2 项目落库(demand_mining/opportunity_general)", np is not None and np.project_type == "demand_mining" and np.pipeline_id == "opportunity_general" and np.org_id == "0", "%s/%s/%s" % (getattr(np, "project_type", ""), getattr(np, "pipeline_id", ""), getattr(np, "org_id", ""))) # 幂等守卫：重复 promote 必须复用既有项目，不再新建 l2 = await call("opp_promote_to_project", {"report_id": RID}) DUPPID = "" if l2.startswith("OK:") and "project_id=" in l2: cand = l2.split("project_id=")[1].split("（")[0].split(" ")[0].strip() if cand and cand != NEWPID: DUPPID = cand n_prj = await sql("SELECT COUNT(*) n FROM sd_projects WHERE description LIKE ${p}$", {"p": "由需求挖掘可行性报告立项%%[report_id=%s]%%" % RID}, one=True) check("L3 重复立项幂等(不重复建)", l2.startswith("OK:") and "幂等" in l2 and not DUPPID and getattr(n_prj, "n", 99) == 1, "%s | dup=%s | n=%s" % (l2[:120], DUPPID or "-", getattr(n_prj, "n", "?"))) # ── M. 清理（报告+PPT保留供浏览器验收；立项项目删除）── PROMOTED = [x for x in (NEWPID, locals().get("DUPPID", "")) if x] if not KEEP: for ppid in PROMOTED: wdir = await sql("SELECT workspace_dir FROM sd_projects WHERE id=${i}$", {"i": ppid}, one=True) await sql("DELETE FROM sd_projects WHERE id=${i}$ AND project_type='demand_mining'", {"i": ppid}) wd = getattr(wdir, "workspace_dir", "") or "" if wd and os.path.isdir(wd): import shutil shutil.rmtree(wd, ignore_errors=True) print("-- cleaned promoted project %s dir=%s" % (ppid, wd), flush=True) print("\n== SUMMARY: %d FAIL %s ==" % (len(FAIL), FAIL if FAIL else "(ALL PASS)"), flush=True) print("RID=%s PPT=%s" % (RID, ppt), flush=True) asyncio.run(m())