643 lines
34 KiB
Python
643 lines
34 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""纯技术方案流程能力（bid_tech_proposal，2026-09-14 批2）。
|
||
|
||
用户 2026-09-11 定义的产线第二流程：输入=技术需求书（非招标文件），产出=技术方案书。
|
||
分析集中在技术层面六类评估项：功能清单/功能点/硬件配置/技术架构/功能架构/部署架构；
|
||
方案书编写根据预定模版（无则 RAG/网络检索），**模版骨架由用户确认后**按章节分段编写，
|
||
章节 QC 评审**以用户提供的需求书为唯一基准**（对标覆盖度，不碰评分项），最终合成技术方案书。
|
||
|
||
机制分工（沿用产线既有约定：流转归机制，操作原语归本模块）：
|
||
- bid_flow 对账器按已确认计划的 base_flow_key 走 tech 分支派发任务；
|
||
- 本模块固化操作原语：技术评估项抽取/补录/查询、模版候选生成、模版确认落骨架；
|
||
- QC 复用 bid_qc_reviews 机制（qc_type=tech_items，start_qc/finish_qc 原语不变）；
|
||
- 用户确认走平台待办（tech_template_confirm 类型 + 专用确认端点，agent 无确认工具）。
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
|
||
from .bid_common import (
|
||
get_db, new_id, rec_to_dict, rows_to_dicts, to_int, to_float, json_loads,
|
||
record, create_human_task, find_human_task, get_thresholds, get_project,
|
||
resolve_project_id, llm_json,
|
||
CH_PENDING, TECH_CATEGORIES, HT_TEMPLATE_CONFIRM, QC_TYPE_TECH,
|
||
)
|
||
|
||
logger = logging.getLogger("pipeline.bidding.tech")
|
||
|
||
TABLE = "bid_tech_items"
|
||
|
||
# 六类评估项中文名（抽取提示词/展示用）
|
||
TECH_CATEGORY_LABELS = {
|
||
"function_list": "功能清单",
|
||
"function_point": "功能点",
|
||
"hardware": "硬件配置",
|
||
"tech_arch": "技术架构",
|
||
"func_arch": "功能架构",
|
||
"deploy_arch": "部署架构",
|
||
}
|
||
|
||
PROMPT_TECH_ITEMS = """你是技术方案分析师。下面是一份【技术需求书】正文（可能被截断），请抽取六类技术评估项。
|
||
本任务只做技术层面分析，不涉及商务/评分/资质。
|
||
|
||
【六类评估项】
|
||
- function_list 功能清单：需求书要求的功能模块/子系统清单（一个模块一条）
|
||
- function_point 功能点：功能模块下的具体功能点（细化到可评估颗粒度）
|
||
- hardware 硬件配置：服务器/GPU/存储/网络等硬件要求（型号/数量/参数，需求书没写具体型号的如实标"未指定"）
|
||
- tech_arch 技术架构：需求书对技术栈/框架/协议/性能指标/安全合规的要求
|
||
- func_arch 功能架构：系统分层/模块划分/集成关系要求
|
||
- deploy_arch 部署架构：部署模式（私有化/云/混合）/环境/容灾/扩缩容要求
|
||
|
||
【技术需求书正文】
|
||
__CONTENT__
|
||
|
||
严格只输出一个 JSON 对象：
|
||
{
|
||
"items": [
|
||
{"category": "function_list|function_point|hardware|tech_arch|func_arch|deploy_arch",
|
||
"item_name": "条目名称（简短可索引）",
|
||
"requirement": "需求书原文要求（引用原文，可截断到要点）",
|
||
"qty_note": "数量/规模说明（功能点数/台数/容量等，没有就空）",
|
||
"mandatory": "1|0（需求书明确必须的为1）",
|
||
"source_ref": "出处（章节/条款号）"}
|
||
]
|
||
}
|
||
要求：
|
||
1. 只录需求书**明确提出或可直接推出**的条目，禁止编造、禁止把通用最佳实践当需求；
|
||
2. requirement 必须能回溯原文（source_ref 给章节号），抽取不到原文出处的标 source_ref="未明示"；
|
||
3. 同一功能模块的多个功能点分多条录（function_point），不要合并成大条；
|
||
4. 硬件参数逐项拆条（如"GPU服务器""存储阵列"各一条），参数写进 requirement。"""
|
||
|
||
|
||
async def _load_req_text(sor, project_id, file_id=""):
|
||
"""读技术需求书正文（复用招标文件读取——需求书也登记在 bid_tender_files）。"""
|
||
from .bid_analysis_capability import _load_tender_text
|
||
return await _load_tender_text(sor, project_id, file_id=file_id)
|
||
|
||
|
||
# ══════════════ 技术评估项：抽取 / 补录 / 查询 ══════════════
|
||
|
||
async def extract_tech_items(project_id, file_id="", model_name="", offset=0,
|
||
length=30000, who=None, agent_id=None):
|
||
"""分析维度 tech：LLM 抽取六类技术评估项落库 bid_tech_items（大文件分段调）。"""
|
||
try:
|
||
project_id = await resolve_project_id(project_id)
|
||
except ValueError as e:
|
||
return False, str(e)
|
||
db, dbname = get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
txt, fname, err = await _load_req_text(sor, project_id, file_id=file_id)
|
||
if err:
|
||
return False, err
|
||
o = to_int(offset, 0)
|
||
seg = txt[o:o + to_int(length, 30000)]
|
||
if not seg.strip():
|
||
return False, "正文分段为空（offset=%s 超出文件长度），无需再抽" % o
|
||
data, err = await llm_json(PROMPT_TECH_ITEMS.replace("__CONTENT__", seg),
|
||
model_name=model_name, retries=1, project_id=project_id)
|
||
if data is None:
|
||
return False, "LLM 抽取失败: %s" % err
|
||
items = data.get("items") or []
|
||
if not isinstance(items, list):
|
||
return False, "LLM 返回格式错误（items 非数组）"
|
||
cnt, skipped = 0, 0
|
||
async with db.sqlorContext(dbname) as sor:
|
||
# 已有条目（同段重复调用去重：同名同类不重复插）
|
||
recs = await sor.sqlExe(
|
||
"SELECT category, item_name FROM " + TABLE + " WHERE project_id=${p}$",
|
||
{"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
exist = {(getattr(r, "category", ""), getattr(r, "item_name", ""))
|
||
for r in (recs or [])}
|
||
order_base = await _max_order(sor, project_id)
|
||
for it in items:
|
||
if not isinstance(it, dict):
|
||
continue
|
||
cat = str(it.get("category") or "").strip()
|
||
name = str(it.get("item_name") or "").strip()[:240]
|
||
if cat not in TECH_CATEGORIES or not name:
|
||
skipped += 1
|
||
continue
|
||
if (cat, name) in exist:
|
||
skipped += 1
|
||
continue
|
||
order_base += 1
|
||
await sor.C(TABLE, {
|
||
"id": new_id(), "project_id": project_id, "kind": "item",
|
||
"category": cat, "item_name": name,
|
||
"requirement": str(it.get("requirement") or "")[:20000],
|
||
"qty_note": str(it.get("qty_note") or "")[:200],
|
||
"mandatory": "1" if str(it.get("mandatory") or "") in ("1", "true", "True") else "0",
|
||
"confidence": "medium",
|
||
"source_ref": str(it.get("source_ref") or "")[:480],
|
||
"order_no": order_base,
|
||
})
|
||
exist.add((cat, name))
|
||
cnt += 1
|
||
await sor.sqlExe("COMMIT", {})
|
||
await record(sor, project_id, TABLE, project_id, "extract_tech_items",
|
||
to_state="extracted", who=who, agent_id=agent_id,
|
||
detail="新增%d条(去重跳过%d) file=%s offset=%s" % (cnt, skipped, fname, o))
|
||
await sor.sqlExe("COMMIT", {})
|
||
return True, ("技术评估项抽取完成：新增 %d 条（重复/非法跳过 %d 条）。"
|
||
"六类分布用 list_tech_items 核对，漏项用 add_tech_item 补录。" % (cnt, skipped))
|
||
|
||
|
||
async def _max_order(sor, project_id):
|
||
recs = await sor.sqlExe(
|
||
"SELECT COALESCE(MAX(order_no),0) AS m FROM " + TABLE + " WHERE project_id=${p}$",
|
||
{"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
return to_int(getattr(recs[0], "m", 0) if recs else 0)
|
||
|
||
|
||
async def add_tech_item(project_id, category, item_name, requirement="",
|
||
qty_note="", mandatory="0", source_ref="",
|
||
who=None, agent_id=None):
|
||
"""补录一条技术评估项（抽取漏项或人工修正）。"""
|
||
try:
|
||
project_id = await resolve_project_id(project_id)
|
||
except ValueError as e:
|
||
return False, str(e)
|
||
cat = str(category or "").strip()
|
||
name = str(item_name or "").strip()[:240]
|
||
if cat not in TECH_CATEGORIES:
|
||
return False, "category 非法（应为 %s）" % "/".join(TECH_CATEGORIES)
|
||
if not name:
|
||
return False, "缺少 item_name"
|
||
db, dbname = get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
order_no = await _max_order(sor, project_id) + 1
|
||
tid = new_id()
|
||
await sor.C(TABLE, {
|
||
"id": tid, "project_id": project_id, "kind": "item",
|
||
"category": cat, "item_name": name,
|
||
"requirement": str(requirement or "")[:20000],
|
||
"qty_note": str(qty_note or "")[:200],
|
||
"mandatory": "1" if str(mandatory) in ("1", "true", "True") else "0",
|
||
"confidence": "high",
|
||
"source_ref": str(source_ref or "")[:480],
|
||
"order_no": order_no,
|
||
})
|
||
await sor.sqlExe("COMMIT", {})
|
||
await record(sor, project_id, TABLE, tid, "add_tech_item",
|
||
who=who, agent_id=agent_id, detail="%s/%s" % (cat, name))
|
||
await sor.sqlExe("COMMIT", {})
|
||
return True, "已补录技术评估项：%s（%s）" % (name, TECH_CATEGORY_LABELS.get(cat, cat))
|
||
|
||
|
||
async def list_tech_items(project_id, category="", kind="item", limit=300):
|
||
"""列出技术评估项（默认只列 item 行；含六类分布统计头部）。"""
|
||
try:
|
||
project_id = await resolve_project_id(project_id)
|
||
except ValueError as e:
|
||
return "ERROR: %s" % e
|
||
db, dbname = get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
sql = ("SELECT id, kind, category, item_name, LEFT(requirement, 300) AS requirement, "
|
||
"qty_note, mandatory, confidence, source_ref, order_no "
|
||
"FROM " + TABLE + " WHERE project_id=${p}$")
|
||
p = {"p": project_id}
|
||
if category:
|
||
sql += " AND category=${c}$"
|
||
p["c"] = category
|
||
if kind:
|
||
sql += " AND kind=${k}$"
|
||
p["k"] = kind
|
||
sql += " ORDER BY category, order_no LIMIT " + str(to_int(limit, 300))
|
||
recs = await sor.sqlExe(sql, p)
|
||
await sor.sqlExe("COMMIT", {})
|
||
rows = rows_to_dicts(recs, limit=to_int(limit, 300))
|
||
stat = await sor.sqlExe(
|
||
"SELECT category, COUNT(*) AS c FROM " + TABLE +
|
||
" WHERE project_id=${p}$ AND kind='item' GROUP BY category", {"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
dist = {getattr(r, "category", ""): to_int(getattr(r, "c", 0)) for r in (stat or [])}
|
||
head = "六类分布：" + "、".join(
|
||
"%s=%d" % (TECH_CATEGORY_LABELS.get(k, k), v) for k, v in
|
||
sorted(dist.items(), key=lambda kv: TECH_CATEGORIES.index(kv[0])
|
||
if kv[0] in TECH_CATEGORIES else 99))
|
||
missing = [TECH_CATEGORY_LABELS[c] for c in TECH_CATEGORIES if not dist.get(c)]
|
||
if missing and kind == "item":
|
||
head += "（缺类：%s——需求书确实没有则如实说明，有漏项用 add_tech_item 补录）" % "、".join(missing)
|
||
if not rows:
|
||
return head + "\n（无记录）"
|
||
return head + "\n" + "\n".join(
|
||
json.dumps(r, ensure_ascii=False, default=str) for r in rows)
|
||
|
||
|
||
# ══════════════ 模版确定 + 用户确认骨架（template_confirm 阶段） ══════════════
|
||
|
||
async def find_template_candidates(project_id, query="", model_name="",
|
||
who=None, agent_id=None):
|
||
"""检索技术方案模版候选：知识库(doc_type=tech_template/sample_bid) → RAG → 网络。
|
||
|
||
返回候选清单文本（PM/analyst 角色工具）。三级降级链每级如实标注来源，
|
||
全部落空时如实说明——由角色 agent 按行业惯例自拟骨架（也要经用户确认）。
|
||
"""
|
||
try:
|
||
project_id = await resolve_project_id(project_id)
|
||
except ValueError as e:
|
||
return "ERROR: %s" % e
|
||
q = (query or "技术方案书 目录 模版").strip()
|
||
lines = ["## 模版候选检索（%s）" % q, ""]
|
||
# 1. 公司知识库
|
||
try:
|
||
from .bid_kb_capability import search_bid_kb
|
||
kb = await search_bid_kb(query=q, doc_type="tech_template")
|
||
if kb and "无命中" not in str(kb):
|
||
lines += ["### ① 公司知识库（doc_type=tech_template，预定模版优先）", str(kb)[:3000], ""]
|
||
else:
|
||
kb2 = await search_bid_kb(query=q, doc_type="sample_bid")
|
||
if kb2 and "无命中" not in str(kb2):
|
||
lines += ["### ① 公司知识库（样板标书参照）", str(kb2)[:3000], ""]
|
||
else:
|
||
lines += ["### ① 公司知识库：无预定模版", ""]
|
||
except Exception as e:
|
||
lines += ["### ① 公司知识库：查询失败（%s）" % str(e)[:100], ""]
|
||
# 2. RAG 检索
|
||
try:
|
||
from pipeline_service.rag_client import rag_search
|
||
data, err = await rag_search(project_id, q, top_k=5)
|
||
if err:
|
||
lines += ["### ② RAG 检索：失败（%s）" % err[:150], ""]
|
||
else:
|
||
chunks = (data or {}).get("results") or (data or {}).get("chunks") or []
|
||
if chunks:
|
||
brief = "\n".join("- [%s] %s" % (
|
||
(c.get("doc_name") or c.get("kb_name") or "?"),
|
||
str(c.get("content") or c.get("text") or "")[:200])
|
||
for c in chunks[:5] if isinstance(c, dict))
|
||
lines += ["### ② RAG 知识库检索", brief, ""]
|
||
else:
|
||
lines += ["### ② RAG 检索：无命中", ""]
|
||
except Exception as e:
|
||
lines += ["### ② RAG 检索：不可用（%s）" % str(e)[:100], ""]
|
||
# 3. 网络检索
|
||
try:
|
||
from pipeline_service.web_tools import tool_web_search
|
||
web = await tool_web_search(q + " 目录结构", limit=5)
|
||
lines += ["### ③ 网络检索", str(web)[:2500], ""]
|
||
except Exception as e:
|
||
lines += ["### ③ 网络检索：不可用（%s）" % str(e)[:100], ""]
|
||
lines += ["---", "下一步：结合六类评估项（list_tech_items）与候选模版拟定章节骨架，"
|
||
"调 propose_tech_template 发用户确认待办（骨架确认前不得开写章节）。"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
PROMPT_TECH_OUTLINE = """你是技术方案架构师。根据【技术需求书评估项】和【模版参考】拟定技术方案书的章节骨架。
|
||
|
||
【六类技术评估项（需求书抽取结果）】
|
||
__ITEMS__
|
||
|
||
【模版参考（可能为空——为空则按行业惯例自拟）】
|
||
__TEMPLATE__
|
||
|
||
严格只输出一个 JSON 对象：
|
||
{"chapters": [
|
||
{"chapter_no": "1", "title": "章节标题", "outline": "本章要写什么要点（覆盖哪些评估项，写到操作层）",
|
||
"cover_categories": ["function_list","tech_arch"]}
|
||
]}
|
||
要求：
|
||
1. 章节结构完整覆盖六类评估项：每个 category 至少被一章的 cover_categories 覆盖，漏覆盖 = 骨架不合格；
|
||
2. 有模版参考时优先沿用它章节顺序与命名惯例，按本项目评估项裁剪/补充（模版是蓝本不是成品）；
|
||
3. 每章 outline 写清要响应的需求书条目（引用 item_name），不写空话；
|
||
4. 章节数控制在 6~15 章，粒度以"一章一任务可写完"为准；
|
||
5. chapter_no 用数字字符串（"1","2",...），子节写进 outline 不单独成章。"""
|
||
|
||
|
||
async def propose_tech_template(project_id, chapters="", template_source="",
|
||
template_ref="", note="", model_name="",
|
||
who=None, agent_id=None):
|
||
"""生成技术方案章节骨架并发「模版确认」人类待办（用户确认后才落 bid_chapters 开写）。
|
||
|
||
chapters：JSON 数组 [{chapter_no,title,outline,cover_categories?}]；
|
||
为空时自动按六类评估项 + 模版候选 LLM 生成。
|
||
template_source：kb/rag/web/manual（如实标注骨架依据来源）。
|
||
确认动作是用户专属（专用端点 tech_template_confirm.dspy），agent 无确认工具。
|
||
"""
|
||
try:
|
||
project_id = await resolve_project_id(project_id)
|
||
except ValueError as e:
|
||
return False, str(e)
|
||
items = json_loads(chapters, []) if isinstance(chapters, str) else (chapters or [])
|
||
db, dbname = get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
proj = await get_project(sor, project_id)
|
||
if not proj:
|
||
return False, "项目不存在: %s" % project_id
|
||
if not items:
|
||
# 自动生成：六类评估项 + 模版候选 → LLM 拟骨架
|
||
irecs = await sor.sqlExe(
|
||
"SELECT category, item_name, LEFT(requirement,200) AS requirement, mandatory "
|
||
"FROM " + TABLE + " WHERE project_id=${p}$ AND kind='item' "
|
||
"ORDER BY category, order_no LIMIT 400", {"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
irows = rows_to_dicts(irecs, limit=400)
|
||
if not irows:
|
||
return False, ("无技术评估项——先完成 tech_analysis 维度抽取"
|
||
"（extract_tech_items），骨架必须覆盖评估项，不能凭空拟")
|
||
items_txt = "\n".join("- [%s]%s %s（%s）" % (
|
||
TECH_CATEGORY_LABELS.get(r.get("category"), r.get("category")),
|
||
"★" if str(r.get("mandatory")) == "1" else "",
|
||
r.get("item_name"), str(r.get("requirement") or "")[:120]) for r in irows)
|
||
tpl_txt = str(template_ref or "")[:6000] or "（无模版参考，按行业惯例自拟）"
|
||
data, err = await llm_json(
|
||
PROMPT_TECH_OUTLINE.replace("__ITEMS__", items_txt).replace("__TEMPLATE__", tpl_txt),
|
||
model_name=model_name, retries=1, project_id=project_id)
|
||
if data is None:
|
||
return False, "LLM 生成骨架失败: %s" % err
|
||
items = data.get("chapters") or []
|
||
if not isinstance(items, list) or not items:
|
||
return False, "章节骨架为空（chapters 解析失败），请传 JSON 数组或修正后重试"
|
||
# 规范化 + 覆盖度校验（机制硬校验：六类评估项必须全覆盖，缺类拒绝——
|
||
# 骨架漏类 = 方案书必漏需求 = QC 对标必退，不如在确认门之前就挡住）
|
||
norm, covered = [], set()
|
||
for idx, ch in enumerate(items):
|
||
if not isinstance(ch, dict):
|
||
continue
|
||
no = str(ch.get("chapter_no") or (idx + 1)).strip()[:30]
|
||
title = str(ch.get("title") or "").strip()[:250]
|
||
if not title:
|
||
continue
|
||
cats = ch.get("cover_categories") or []
|
||
if isinstance(cats, str):
|
||
cats = [c.strip() for c in cats.replace("，", ",").split(",") if c.strip()]
|
||
cats = [c for c in cats if c in TECH_CATEGORIES]
|
||
covered.update(cats)
|
||
norm.append({"chapter_no": no, "title": title,
|
||
"outline": str(ch.get("outline") or "")[:4000],
|
||
"cover_categories": cats})
|
||
if not norm:
|
||
return False, "章节骨架全部非法（缺 title）"
|
||
# 项目实际存在的评估类别（需求书没提硬件就不强制覆盖硬件）
|
||
crecs = await sor.sqlExe(
|
||
"SELECT DISTINCT category FROM " + TABLE + " WHERE project_id=${p}$ AND kind='item'",
|
||
{"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
need = {getattr(r, "category", "") for r in (crecs or [])} & set(TECH_CATEGORIES)
|
||
miss = sorted(need - covered)
|
||
if miss:
|
||
return False, ("骨架未覆盖以下评估类别：%s。每类至少一章 cover_categories 覆盖，"
|
||
"请补章节或修正 cover_categories 后重提。"
|
||
% "、".join(TECH_CATEGORY_LABELS.get(m, m) for m in miss))
|
||
# 骨架预览 markdown（进待办正文）
|
||
md = ["## 技术方案书章节骨架（待确认）", ""]
|
||
if template_source:
|
||
md.append("**模版来源：** %s%s" % (
|
||
{"kb": "公司知识库预定模版", "rag": "RAG 知识库检索",
|
||
"web": "网络检索", "manual": "按行业惯例自拟"}.get(template_source, template_source),
|
||
"（%s）" % template_ref[:200] if template_ref else ""))
|
||
md += ["", "| # | 章节 | 要点 | 覆盖评估类 |", "|---|---|---|---|"]
|
||
for ch in norm:
|
||
md.append("| %s | %s | %s | %s |" % (
|
||
ch["chapter_no"], ch["title"],
|
||
(ch["outline"] or "").replace("\n", " ")[:120],
|
||
"、".join(TECH_CATEGORY_LABELS.get(c, c) for c in ch["cover_categories"]) or "-"))
|
||
md += ["", "---", "",
|
||
"确认后骨架落库并开始分章编写（每章对标需求书原文，QC 按需求覆盖度评审）；",
|
||
"驳回请附意见（换模版/调章节/补覆盖），助手按意见修订后重新提案。"]
|
||
if note:
|
||
md += ["", "**助手说明：**", note[:2000]]
|
||
hid = await create_human_task(
|
||
sor, project_id, HT_TEMPLATE_CONFIRM,
|
||
"技术方案书章节骨架确认：%s" % (proj.get("name") or project_id),
|
||
"\n".join(md),
|
||
form_schema={"chapters": norm, "template_source": template_source or "",
|
||
"template_ref": (template_ref or "")[:2000]})
|
||
await sor.sqlExe("COMMIT", {})
|
||
await record(sor, project_id, "pipeline_human_tasks", hid, "propose_tech_template",
|
||
to_state="pending_confirm", who=who, agent_id=agent_id,
|
||
detail="%d章 source=%s" % (len(norm), template_source or "-"))
|
||
await sor.sqlExe("COMMIT", {})
|
||
return True, ("已生成技术方案章节骨架（%d 章）并发「模版确认」待办（%s）。"
|
||
"用户确认前流程阻塞，你不能代替确认；确认后系统自动落骨架开写。"
|
||
% (len(norm), hid))
|
||
|
||
|
||
async def confirm_tech_template(human_task_id, decision, comment="",
|
||
operator_id=None):
|
||
"""用户确认/驳回章节骨架（专用端点调用，agent 无此工具）。
|
||
|
||
confirm：骨架落 bid_chapters(section=technical, status=pending) + 关待办 → 对账器开写；
|
||
reject：关待办（status=rejected，意见存 result_data）→ 对账器重派模版任务带意见。
|
||
"""
|
||
if not human_task_id:
|
||
return False, "缺少 human_task_id"
|
||
if not operator_id:
|
||
return False, "未登录"
|
||
decision = str(decision or "").strip().lower()
|
||
if decision not in ("confirm", "reject"):
|
||
return False, "decision 必须是 confirm 或 reject"
|
||
if decision == "reject" and not (comment or "").strip():
|
||
return False, "驳回必须填写意见（助手按意见修订骨架）"
|
||
db, dbname = get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT * FROM pipeline_human_tasks WHERE id=${h}$", {"h": human_task_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return False, "待办不存在: %s" % human_task_id
|
||
ht = rec_to_dict(recs[0])
|
||
if ht.get("task_type") != HT_TEMPLATE_CONFIRM:
|
||
return False, "该待办不是模版确认类型（%s）" % ht.get("task_type")
|
||
if ht.get("status") != "pending":
|
||
return False, "待办已处理（当前 %s）" % ht.get("status")
|
||
project_id = ht.get("project_id") or ""
|
||
# 确认权限：复用引擎 _check_confirm_operator（同机构 + owner.superuser/项目创建者）——
|
||
# 与 flow_plan_confirm 确认门语义完全一致，不自己另写一套校验（2026-09-15 E2E 实测
|
||
# 发现自写「仅同机构」把超管/创建者语义丢了）。
|
||
try:
|
||
from pipeline_service.flow_plan_capability import _check_confirm_operator
|
||
ok_op, err_op = await _check_confirm_operator(
|
||
sor, {"project_id": project_id}, operator_id)
|
||
if not ok_op:
|
||
return False, err_op or "无权确认该项目模版"
|
||
except ImportError:
|
||
# 引擎不可用（理论不发生）：兜底同机构校验
|
||
prec = await sor.sqlExe("SELECT org_id FROM sd_projects WHERE id=${p}$",
|
||
{"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
proj_org = getattr(prec[0], "org_id", "") if prec else ""
|
||
from pipeline_service.human_task_capability import _get_user_org
|
||
user_org = await _get_user_org(sor, operator_id)
|
||
if proj_org and user_org != proj_org:
|
||
return False, "仅同机构用户可确认该项目模版"
|
||
|
||
if decision == "confirm":
|
||
schema = json_loads(ht.get("form_schema"), {}) or {}
|
||
if not isinstance(schema, dict):
|
||
schema = {}
|
||
chapters = schema.get("chapters") or []
|
||
if not chapters:
|
||
return False, "待办里没有骨架数据（form_schema.chapters 空），无法落库"
|
||
# 标准/纯技术方案双模式分流（2026-09-17 裁剪链挂标准流程）：
|
||
# 项目存在非 technical 章节（商务/报价章）= 标准投标流程 → 只替换
|
||
# technical 章（新骨架落原位 order_no，商务/报价章原样保留）；
|
||
# 纯技术方案流程无其他章节 → 整表清落（原语义不变）。
|
||
_all = await sor.sqlExe(
|
||
"SELECT id, section, status, order_no FROM bid_chapters "
|
||
"WHERE project_id=${p}$", {"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
_rows = rows_to_dicts(_all, limit=500)
|
||
_tech_rows = [r for r in _rows
|
||
if (r.get("section") or "technical") == "technical"]
|
||
_other_rows = [r for r in _rows if r not in _tech_rows]
|
||
std_mode = bool(_other_rows)
|
||
# 幂等重落守卫：仅当待替换章节全部未开写（pending）时清旧骨架——
|
||
# 已有 writing/written/approved 章节说明流程已推进，重新确认会毁掉
|
||
# 已写内容，拒绝并提示走驳回/人工处置（2026-09-15 加固）。
|
||
_scope = _tech_rows if std_mode else _rows
|
||
if any(str(r.get("status")) != CH_PENDING for r in _scope):
|
||
return False, ("项目已有开写/已完成的%s章节，不能重新确认骨架（会毁掉已写内容）。"
|
||
"如需换骨架请先驳回或在章节管理页人工处置既有章节。"
|
||
% ("技术" if std_mode else ""))
|
||
# 幂等：清掉此前确认落库但尚未开写的旧骨架（驳回重提后再确认的场景）
|
||
if std_mode:
|
||
await sor.sqlExe(
|
||
"DELETE FROM bid_chapters WHERE project_id=${p}$ "
|
||
"AND (section='technical' OR section IS NULL OR section='')",
|
||
{"p": project_id})
|
||
else:
|
||
await sor.sqlExe(
|
||
"DELETE FROM bid_chapters WHERE project_id=${p}$", {"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
# 标准模式：新骨架插回原技术章位置，全序列重排 order_no=1..N——
|
||
# 不能只按 min(order_no)+idx 顺延：新骨架章数多于原技术章时会与排在
|
||
# 技术部分之后的商务/报价章撞号，合成 ORDER BY order_no 会把报价章
|
||
# 插进技术章中间（2026-09-17 状态机走查发现的边界缺陷）。
|
||
_seq = [] # [(row_or_None, new_order)]，None=新章占位
|
||
if std_mode:
|
||
_others = sorted(_other_rows, key=lambda r: to_int(r.get("order_no"), 0))
|
||
_tnums = [to_int(r.get("order_no"), 0) for r in _tech_rows
|
||
if to_int(r.get("order_no"), 0) > 0]
|
||
_ins = len(_others)
|
||
if _tnums:
|
||
_tmin = min(_tnums)
|
||
_ins = sum(1 for r in _others
|
||
if to_int(r.get("order_no"), 0) < _tmin)
|
||
_new_chs = [ch for ch in chapters if isinstance(ch, dict)]
|
||
_seq = ([(r, None) for r in _others[:_ins]]
|
||
+ [(None, ch) for ch in _new_chs]
|
||
+ [(r, None) for r in _others[_ins:]])
|
||
else:
|
||
_seq = [(None, ch) for ch in chapters if isinstance(ch, dict)]
|
||
n = 0
|
||
for idx, ent in enumerate(_seq):
|
||
row, ch = ent
|
||
if row is not None:
|
||
_no = idx + 1
|
||
if to_int(row.get("order_no"), 0) != _no:
|
||
await sor.sqlExe(
|
||
"UPDATE bid_chapters SET order_no=${o}$ WHERE id=${i}$",
|
||
{"o": _no, "i": row.get("id")})
|
||
continue
|
||
if not isinstance(ch, dict):
|
||
continue
|
||
await sor.C("bid_chapters", {
|
||
"id": new_id(), "project_id": project_id,
|
||
"chapter_no": str(ch.get("chapter_no") or (n + 1))[:32],
|
||
"title": str(ch.get("title") or "")[:255],
|
||
"section": "technical",
|
||
"outline": str(ch.get("outline") or "")[:60000],
|
||
"status": CH_PENDING, "version": 1, "revise_count": 0,
|
||
"order_no": idx + 1,
|
||
})
|
||
n += 1
|
||
await sor.sqlExe("COMMIT", {})
|
||
# 裁剪摘要落库（kind='summary'，list_tech_items 默认 kind='item' 不污染
|
||
# 评估项清单）：写者/QC 可追溯「技术章骨架按哪六类评估项裁剪、模版来源」，
|
||
# 也是 A3 段链激活过的持久证据（audit_log 之外的产出物侧记录）。
|
||
try:
|
||
await record_tech_summary(
|
||
sor, project_id, n,
|
||
template_source=str(schema.get("template_source") or ""),
|
||
template_ref=str(schema.get("template_ref") or ""))
|
||
except Exception as _e: # 摘要失败不阻塞确认主流程
|
||
logger.warning("record_tech_summary failed project=%s: %s",
|
||
project_id, str(_e)[:160])
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_human_tasks SET status='done', qc_status='passed', "
|
||
"qc_comment=${cm}$, result_data=${rd}$, "
|
||
"submitted_by=${u}$, submitted_at=NOW() WHERE id=${h}$",
|
||
{"cm": "用户已确认技术方案骨架（标准流程：只替换技术章）" if std_mode
|
||
else "用户已确认技术方案骨架",
|
||
"rd": json.dumps({"decision": "confirm", "chapters": n,
|
||
"std_mode": std_mode}, ensure_ascii=False),
|
||
"u": operator_id, "h": human_task_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
await record(sor, project_id, "bid_chapters", project_id, "template_confirmed",
|
||
to_state="pending", who=operator_id,
|
||
detail="骨架落库 %d 章%s" % (n, "（标准流程，替换技术章）" if std_mode else ""))
|
||
await sor.sqlExe("COMMIT", {})
|
||
logger.info("tech_template confirmed: project=%s chapters=%d std_mode=%s by=%s",
|
||
project_id, n, std_mode, operator_id)
|
||
return True, ("骨架已确认并落库（%d 章技术章%s），系统将自动开始分章编写。"
|
||
% (n, "，商务/报价章未动" if std_mode else ""))
|
||
|
||
# reject
|
||
await sor.sqlExe(
|
||
"UPDATE pipeline_human_tasks SET status='rejected', result_data=${rd}$, "
|
||
"submitted_by=${u}$, submitted_at=NOW() WHERE id=${h}$",
|
||
{"rd": json.dumps({"decision": "reject", "comment": (comment or "")[:2000]},
|
||
ensure_ascii=False),
|
||
"u": operator_id, "h": human_task_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
await record(sor, project_id, "pipeline_human_tasks", human_task_id,
|
||
"template_rejected", to_state="rejected", who=operator_id,
|
||
detail=(comment or "")[:500])
|
||
await sor.sqlExe("COMMIT", {})
|
||
return True, "已驳回，助手将按你的意见修订骨架后重新提案。"
|
||
|
||
|
||
async def get_template_reject_comment(sor, project_id):
|
||
"""最新一次模版驳回意见（对账器重派模版任务时注入 params，无则 ''）。"""
|
||
ht = await find_human_task(sor, project_id, HT_TEMPLATE_CONFIRM, status="rejected")
|
||
if not ht:
|
||
return ""
|
||
rd = json_loads(ht.get("result_data"), {}) or {}
|
||
if not isinstance(rd, dict):
|
||
return ""
|
||
return str(rd.get("comment") or "")
|
||
|
||
|
||
async def record_tech_summary(sor, project_id, n_chapters, template_source="",
|
||
template_ref=""):
|
||
"""技术章骨架裁剪摘要落库（bid_tech_items kind='summary'，幂等覆盖）。
|
||
|
||
2026-09-17 裁剪链挂标准流程：用户确认骨架后把「按哪六类评估项裁剪、模版来源、
|
||
骨架章数」写成摘要行——写者/QC/审计可追溯技术方案部分的裁剪依据（list_tech_items
|
||
默认 kind='item'，摘要不污染评估项清单）。失败由调用方捕获，不阻塞确认主流程。
|
||
"""
|
||
crecs = await sor.sqlExe(
|
||
"SELECT category, COUNT(*) AS c FROM " + TABLE +
|
||
" WHERE project_id=${p}$ AND kind='item' GROUP BY category", {"p": project_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
cats = {str(getattr(r, "category", "") or ""): to_int(getattr(r, "c", 0), 0)
|
||
for r in (crecs or [])}
|
||
lines = ["六类评估项覆盖：" + ("、".join(
|
||
"%s%d项" % (TECH_CATEGORY_LABELS.get(k, k), v)
|
||
for k, v in sorted(cats.items()) if v) or "（无）")]
|
||
if template_source:
|
||
lines.append("模版来源：%s%s" % (
|
||
{"kb": "公司知识库预定模版", "rag": "RAG 知识库检索",
|
||
"web": "网络检索", "manual": "按行业惯例自拟"}.get(
|
||
template_source, template_source),
|
||
"（%s）" % template_ref[:200] if template_ref else ""))
|
||
lines.append("用户确认骨架：%d 章技术章（section=technical），裁剪链完成。" % n_chapters)
|
||
await sor.sqlExe(
|
||
"DELETE FROM " + TABLE + " WHERE project_id=${p}$ AND kind='summary'",
|
||
{"p": project_id})
|
||
sid = new_id()
|
||
await sor.C(TABLE, {
|
||
"id": sid, "project_id": project_id, "kind": "summary",
|
||
"category": "", "item_name": "技术章骨架裁剪摘要",
|
||
"requirement": "\n".join(lines)[:20000],
|
||
"confidence": "high", "source_ref": "template_confirm", "order_no": 0,
|
||
})
|
||
await sor.sqlExe("COMMIT", {})
|
||
await record(sor, project_id, TABLE, sid, "record_tech_summary",
|
||
who="system", detail="%d章 source=%s" % (n_chapters, template_source or "-"))
|
||
await sor.sqlExe("COMMIT", {})
|
||
return sid
|