test(feasibility): P3可行性研究E2E脚本——FP规则引擎独立复算/draft正负链(全personal拒绝+跨项目+伪org+未归一化)/八节报告落库+成本系数核对/评分/promote门禁(draft拒→confirmed拒审批→approved放行)/八节PPT断言(python-pptx解析章节页)/重复立项行为记录/清理立项项目
This commit is contained in:
parent
4092304726
commit
5686b255e9
304
scripts/p3_feasibility_e2e.py
Normal file
304
scripts/p3_feasibility_e2e.py
Normal file
@ -0,0 +1,304 @@
|
||||
# -*- 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 {}
|
||||
check("A4 非法type进pending不炸", db_.get("总FP") == 7 and len(db_.get("待确认") or []) == 1,
|
||||
bad[:150])
|
||||
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", "")))
|
||||
# 重复立项行为记录（当前实现无幂等守卫，重复调用会再建项目——如实记录，清理兜底）
|
||||
l2 = await call("opp_promote_to_project", {"report_id": RID})
|
||||
DUP = l2.startswith("OK:") and "project_id=" in l2
|
||||
if DUP:
|
||||
DUPPID = l2.split("project_id=")[1].split("（")[0].strip()
|
||||
print("-- FINDING: promote 重复调用再建项目 %s（无幂等守卫）--" % DUPPID, flush=True)
|
||||
check("L3 重复立项行为已记录", l2.startswith("OK:") or l2.startswith("ERROR:"), l2[:120])
|
||||
|
||||
# ── 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())
|
||||
Loading…
x
Reference in New Issue
Block a user