306 lines
15 KiB
Python
306 lines
15 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""P3 可行性研究引擎：FP 估算（规则引擎禁 LLM 报数）→ 八节可行性报告 → 评分 → 立项。
|
||
|
||
分工铁律（function-point-counting 技能）：LLM 只做语义（功能识别/DET/FTR 提取），
|
||
复杂度判定与 FP 加权必须走规则引擎 fp_calc.py（skills/global 部署位，进程内 importlib 加载，
|
||
加载失败如实报错，绝不 LLM 心算 FP）。
|
||
|
||
八节可行性模板（report_type=feasibility）：
|
||
## 市场需求 / ## 共性需求 / ## 需求规格 / ## 架构与可行性 /
|
||
## FP与成本 / ## 商业价值 / ## 风险 / ## 结论建议
|
||
"""
|
||
import importlib.util
|
||
import json
|
||
import logging
|
||
import os
|
||
|
||
from .opp_common import get_db, new_id, rows_to_dicts, get_param
|
||
from .opp_normalize import coverage_report, _llm
|
||
|
||
logger = logging.getLogger("pipeline.opp_feasibility")
|
||
|
||
# fp_calc.py 部署位（build.sh 6b 段复制 skills_library/all → skills/global）
|
||
FP_CALC_CANDIDATES = [
|
||
"/d/pipeline/pipeline-app/skills/global/function-point-counting/scripts/fp_calc.py",
|
||
os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||
"../../skills/global/function-point-counting/scripts/fp_calc.py"),
|
||
]
|
||
|
||
# 工作量/成本规则（人月/单价，params 可配；IFPUG FP→人月经验系数）
|
||
P = {
|
||
"opp_fp_pm_per_fp": "0.05", # 人月/FP（经验系数，可调）
|
||
"opp_cost_wan_per_pm": "3.0", # 万元/人月
|
||
}
|
||
|
||
|
||
def _load_fp_calc():
|
||
for path in FP_CALC_CANDIDATES:
|
||
p = os.path.abspath(path)
|
||
if os.path.isfile(p):
|
||
spec = importlib.util.spec_from_file_location("fp_calc_mod", p)
|
||
if not spec or not spec.loader:
|
||
continue
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
return None
|
||
|
||
|
||
async def get_fp_params(sor):
|
||
cfg = {}
|
||
for k, dv in P.items():
|
||
cfg[k] = await get_param(sor, k, dv)
|
||
for k in P:
|
||
try:
|
||
cfg[k] = float(cfg[k])
|
||
except (TypeError, ValueError):
|
||
cfg[k] = float(P[k])
|
||
return cfg
|
||
|
||
|
||
async def fp_estimate(sor, functions_json, project_name=""):
|
||
"""functions_json: [{name,type(EI/EO/EQ/ILF/EIF),desc,evidence,det,ret,ftr,confidence}]
|
||
返回 (ok, dict|err)。复杂度与 FP 全部由 fp_calc 规则引擎算，LLM 只提供结构化输入。"""
|
||
try:
|
||
funcs = json.loads(functions_json) if isinstance(functions_json, str) else functions_json
|
||
except Exception:
|
||
return False, "functions_json 解析失败"
|
||
if not isinstance(funcs, list) or not funcs:
|
||
return False, "functions 必须是非空数组"
|
||
data = {"project": project_name or "可行性估算", "boundary": {"internal": [], "external": []},
|
||
"functions": [], "pending": [], "assumptions": []}
|
||
for f in funcs:
|
||
if not isinstance(f, dict):
|
||
continue
|
||
ft = str(f.get("type") or "").upper()
|
||
if ft not in ("EI", "EO", "EQ", "ILF", "EIF"):
|
||
data["pending"].append({"item": str(f.get("name") or "?"),
|
||
"reason": "类型非法: %s" % ft, "ask": "确认功能类型"})
|
||
continue
|
||
data["functions"].append({
|
||
"name": str(f.get("name") or "")[:80], "type": ft,
|
||
"desc": str(f.get("desc") or "")[:200],
|
||
"evidence": str(f.get("evidence") or "")[:200],
|
||
"det": int(f.get("det") or 0), "ret": int(f.get("ret") or (1 if ft in ("ILF", "EIF") else 0)),
|
||
"ftr": int(f.get("ftr") or 0),
|
||
"confidence": str(f.get("confidence") or "medium"),
|
||
"assumption": str(f.get("assumption") or "")[:200]})
|
||
mod = _load_fp_calc()
|
||
if not mod:
|
||
return False, "fp_calc.py 规则引擎未找到（skills/global 未部署 function-point-counting）"
|
||
try:
|
||
res = mod.calc(data)
|
||
except Exception as e:
|
||
return False, "fp_calc 计算失败: %s" % str(e)[:200]
|
||
# fp_calc.calc() 实际返回键（2026-09-14 核对源码）：
|
||
# functions / counts / total_fp / errors / pending / assumptions / boundary
|
||
return True, {
|
||
"total_fp": res.get("total_fp") or 0,
|
||
"by_type": res.get("counts") or {},
|
||
"functions": res.get("functions") or data["functions"],
|
||
"pending": data["pending"],
|
||
"errors": res.get("errors") or [],
|
||
"markdown": mod.to_markdown(res, project_name) if hasattr(mod, "to_markdown") else "",
|
||
}
|
||
|
||
|
||
async def _cluster_ctx(sor, cluster_id, ctx):
|
||
"""取类别+批次+覆盖率，校验 org/项目可见性。返回 (ok, dict|err)。
|
||
|
||
项目隔离口径与 opp_mining_flow._check_batch_visible 一致：批次挂了项目的，
|
||
仅同项目会话可见；平台级批次（project_id 空）本机构可见。
|
||
"""
|
||
recs = await sor.sqlExe(
|
||
"SELECT c.*, b.org_id AS b_org, b.scope, b.project_id AS b_project "
|
||
"FROM opp_clusters c "
|
||
"LEFT JOIN opp_mining_batches b ON b.id=c.batch_id WHERE c.id=${c}$",
|
||
{"c": cluster_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return False, "类别不存在"
|
||
cl = rows_to_dicts(recs, limit=1)[0]
|
||
org_id = str(cl.get("org_id") or cl.get("b_org") or "")
|
||
if org_id and org_id != (ctx.get("org_id") or ""):
|
||
return False, "类别不属于当前机构"
|
||
bproj = str(cl.get("b_project") or cl.get("project_id") or "")
|
||
if bproj and bproj != str(ctx.get("project_id") or ""):
|
||
return False, "批次属于其他项目，当前会话不可见"
|
||
ok, cov = await coverage_report(sor, ctx, cluster_id)
|
||
if not ok:
|
||
return False, cov
|
||
return True, {"cluster": cl, "coverage": cov}
|
||
|
||
|
||
async def draft_feasibility(sor, cluster_id, ctx):
|
||
"""起草八节可行性研究报告（report_type=feasibility）。返回 (ok, msg|report_id)。"""
|
||
ok, info = await _cluster_ctx(sor, cluster_id, ctx)
|
||
if not ok:
|
||
return False, info
|
||
cl = info["cluster"]
|
||
cov = info["coverage"]
|
||
if cov["cluster"].get("normalize_status") != "done":
|
||
return False, "类别未完成共性提取，先 opp_normalize_cluster"
|
||
from .opp_report_capability import create_report
|
||
stds = cov["std_demands"]
|
||
core = [s for s in stds if s["tier"] == "core"]
|
||
ext = [s for s in stds if s["tier"] == "ext"]
|
||
if not core and not ext:
|
||
return False, ("类别「%s」无核心/扩展共性（全部标准需求覆盖率<40%%），"
|
||
"可行性样本不足——换一个共性更高的类别，"
|
||
"或改用 opp_create_report 写调研报告" % (cl.get("name") or ""))
|
||
|
||
# FP 输入：核心+扩展标准需求 → 规则映射功能点（每标准需求≈1 ILF + 2 EI + 1 EO/EQ，
|
||
# 保守经验映射；DET 按证据原子数粗估）——LLM 不参与 FP 数值
|
||
funcs = []
|
||
for s in core + ext:
|
||
ev = json.loads(s.get("evidence_json") or "{}")
|
||
n_atoms = len(ev.get("atoms") or []) or 1
|
||
funcs.append({"name": s["name"] + "档案", "type": "ILF", "desc": s["name"],
|
||
"evidence": "覆盖率%.0f%% 标准需求" % (float(s["coverage"] or 0) * 100),
|
||
"det": min(19, 5 + n_atoms), "ret": 1, "ftr": 0, "confidence": "medium"})
|
||
funcs.append({"name": s["name"] + "维护", "type": "EI", "desc": "新增/修改" + s["name"],
|
||
"evidence": "标准需求 CRUD", "det": min(15, 4 + n_atoms), "ret": 0,
|
||
"ftr": 1, "confidence": "medium"})
|
||
funcs.append({"name": s["name"] + "查询", "type": "EQ", "desc": "检索" + s["name"],
|
||
"evidence": "标准需求查询", "det": min(19, 4 + n_atoms), "ret": 0,
|
||
"ftr": 1, "confidence": "medium"})
|
||
ok_fp, fp = await fp_estimate(sor, funcs, project_name=cl.get("name") or "")
|
||
if not ok_fp:
|
||
return False, "FP 估算失败: %s" % fp
|
||
cfg = await get_fp_params(sor)
|
||
pm = round(float(fp["total_fp"]) * cfg["opp_fp_pm_per_fp"], 1)
|
||
cost = round(pm * cfg["opp_cost_wan_per_pm"], 1)
|
||
|
||
# LLM 写三节叙述（架构/商业价值/风险），其余节用真实数据拼装
|
||
arch = cov_txt = biz = risk = ""
|
||
try:
|
||
arch = await _llm(
|
||
"为「%s」软件写技术架构与可行性分析（≤200字）：推荐技术栈、架构分层、"
|
||
"关键技术风险点。核心功能：%s。只输出正文。" % (
|
||
cl.get("name"), "、".join(s["name"] for s in core[:6]) or "、".join(
|
||
s["name"] for s in stds[:6])), ctx, timeout=90)
|
||
except Exception as e:
|
||
arch = "（架构分析生成失败：%s）" % str(e)[:60]
|
||
try:
|
||
biz = await _llm(
|
||
"为「%s」软件写商业价值分析（≤150字）：目标客户、付费意愿依据（众包需求 %s 条、"
|
||
"核心共性 %d 项）、定价建议。只输出正文。" % (
|
||
cl.get("name"), cl.get("doc_count"), len(core)), ctx, timeout=90)
|
||
except Exception as e:
|
||
biz = "（商业价值分析生成失败：%s）" % str(e)[:60]
|
||
try:
|
||
risk = await _llm(
|
||
"为「%s」研发项目写风险清单（≤150字，3-5条，格式：- 风险：应对）。只输出正文。"
|
||
% cl.get("name"), ctx, timeout=90)
|
||
except Exception as e:
|
||
risk = "（风险分析生成失败：%s）" % str(e)[:60]
|
||
|
||
cov_txt = "\n".join("- %s（覆盖率 %.0f%%，%s，覆盖 %d 文档）" % (
|
||
s["name"], float(s["coverage"] or 0) * 100, s["tier"], s["doc_count"]) for s in stds[:20])
|
||
content = "\n\n".join([
|
||
"## 市场需求",
|
||
"类别「%s」来自需求挖掘批次 %s（scope=%s），类内众包需求 %s 条，占批次 %.1f%%。"
|
||
"数据来源：数据爬取平台众包需求（猪八戒/一品威客/Freelancer/PPH），明细可经"
|
||
" opp_cluster_detail 逐条查证。" % (
|
||
cl.get("name"), cl.get("batch_id"), cl.get("scope"), cl.get("doc_count"),
|
||
float(cl.get("share") or 0) * 100),
|
||
"## 共性需求",
|
||
cov_txt or "（无标准需求）",
|
||
"## 需求规格",
|
||
"核心共性（≥60%%）：%s\n扩展共性（≥40%%）：%s\n能力域：%s" % (
|
||
"、".join(s["name"] for s in core) or "无",
|
||
"、".join(s["name"] for s in ext) or "无",
|
||
"、".join(d["name"] for d in cov["domains"])),
|
||
"## 架构与可行性",
|
||
arch,
|
||
"## FP与成本",
|
||
"功能点合计 %s FP（规则引擎 fp_calc 计算，IFPUG 口径）；按 %.2f 人月/FP 估 %s 人月；"
|
||
"按 %.1f 万元/人月估研发成本 %s 万元。\n%s" % (
|
||
fp["total_fp"], cfg["opp_fp_pm_per_fp"], pm, cfg["opp_cost_wan_per_pm"], cost,
|
||
(fp.get("markdown") or "")[:1500]),
|
||
"## 商业价值",
|
||
biz,
|
||
"## 风险",
|
||
risk,
|
||
"## 结论建议",
|
||
"建议立项优先级见评分（opp_score_candidate）；本报告为草稿，需人工确认门禁后进入审批。",
|
||
])
|
||
ok_r, rid = await create_report(
|
||
ctx.get("project_id") or "", cl.get("name") or "未命名类别",
|
||
title="%s 研发可行性研究" % (cl.get("name") or ""), analysis=content,
|
||
created_by=ctx.get("user_id") or "agent.opportunity", report_type="feasibility")
|
||
if not ok_r:
|
||
return False, rid
|
||
# 评分落库到报告 content 前置 JSON？不——评分独立工具按需算。记录 FP 摘要进批次无关处：
|
||
return True, rid
|
||
|
||
|
||
async def score_candidate(sor, cluster_id, ctx):
|
||
"""规则评分：市场需求量×共性集中度×规模适配→优先级分。返回 (ok, dict|err)。"""
|
||
ok, info = await _cluster_ctx(sor, cluster_id, ctx)
|
||
if not ok:
|
||
return False, info
|
||
cl = info["cluster"]
|
||
cov = info["coverage"]
|
||
stds = cov["std_demands"]
|
||
n_docs = int(cl.get("doc_count") or 0)
|
||
core = [s for s in stds if s["tier"] == "core"]
|
||
ext = [s for s in stds if s["tier"] == "ext"]
|
||
# 需求热度分（log 规模，防大簇垄断）
|
||
import math
|
||
heat = min(100.0, 20 * math.log10(n_docs + 1))
|
||
# 共性集中度：核心+扩展覆盖文档占比
|
||
covered = sum(s["doc_count"] for s in core + ext)
|
||
conc = (covered / n_docs * 100) if n_docs else 0
|
||
# 规模适配：标准需求 8-25 个为最佳研发规模（过小不值得/过大需分期）
|
||
n_std = len(stds)
|
||
size_fit = 100.0 if 8 <= n_std <= 25 else (70.0 if 5 <= n_std < 8 or 25 < n_std <= 40 else 40.0)
|
||
score = round(heat * 0.4 + conc * 0.35 + size_fit * 0.25, 1)
|
||
stars = max(1, min(5, int(score // 20) + 1))
|
||
return True, {
|
||
"cluster": cl.get("name"), "score": score, "priority": "★" * stars,
|
||
"factors": {"需求热度(40%)": round(heat, 1), "共性集中度(35%)": round(conc, 1),
|
||
"规模适配(25%)": size_fit},
|
||
"detail": {"doc_count": n_docs, "core": len(core), "ext": len(ext), "std": n_std},
|
||
}
|
||
|
||
|
||
async def promote_to_project(sor, report_id, ctx):
|
||
"""审批通过的可行性报告 → 立项（sd_projects，pipeline=opportunity_general）。
|
||
门禁：报告必须 approved（人工审批回流），agent 不能自批自审。"""
|
||
recs = await sor.sqlExe(
|
||
"SELECT r.*, a.status AS appr_status FROM opp_reports r "
|
||
"LEFT JOIN opp_approvals a ON a.report_id=r.id "
|
||
"WHERE r.id=${r}$ ORDER BY a.created_at DESC LIMIT 1", {"r": report_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return False, "报告不存在"
|
||
rep = rows_to_dicts(recs, limit=1)[0]
|
||
if rep.get("report_type") != "feasibility":
|
||
return False, "仅可行性研究报告可立项"
|
||
# 门禁链：draft→confirmed（人工确认）→approval_initiated→approved（审批回流）。
|
||
# resolve_approval 会把报告 status 置 approved，所以 confirmed 只是中间态——
|
||
# 状态在 confirmed/approval_initiated/approved 都说明已过人工确认门禁。
|
||
if rep.get("status") not in ("confirmed", "approval_initiated", "approved"):
|
||
return False, "报告未过人工确认门禁（status=%s），不能立项" % (rep.get("status") or "draft")
|
||
if (rep.get("appr_status") or "") != "approved":
|
||
return False, "研发审批未通过（status=%s），不能立项" % (rep.get("appr_status") or "无审批单")
|
||
# 幂等守卫（2026-09-14 E2E 实测：重复调用会重复建项目）：description 带
|
||
# report_id 精确标记，立项前先查——已立过则返回既有项目，不再新建。
|
||
desc = "由需求挖掘可行性报告立项：%s [report_id=%s]" % (rep.get("title"), report_id)
|
||
exist = await sor.sqlExe(
|
||
"SELECT id, name FROM sd_projects WHERE description=${d}$ LIMIT 1", {"d": desc})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if exist:
|
||
return True, {"project_id": getattr(exist[0], "id", ""),
|
||
"project_name": getattr(exist[0], "name", ""), "reused": True}
|
||
from pipeline_service.project_capability import create_project
|
||
ok, pid = await create_project(
|
||
name="%s 研发项目" % (rep.get("software") or "未命名"),
|
||
project_type="demand_mining", description=desc,
|
||
org_id=ctx.get("org_id") or "0", created_by=ctx.get("user_id") or "",
|
||
pipeline_id="opportunity_general")
|
||
if not ok:
|
||
return False, "立项失败: %s" % pid
|
||
return True, {"project_id": pid, "reused": False}
|