"""整书评分能力（agent.bid_scorer）：按评分项和得分规则对整个标书打分。 对应业务：对整个标书评审，并按照评分项和得分规则对标书进行评分；如果不合适，提出改进意见， 按照意见对问题所在章节修改、评审……直到得分通过。 机制：add_score_item_result 逐项落分（含 improvement + target_chapter_id）→ finalize_bid_score 汇总判定；不通过则把命中章节置 rejected（reconciler 自动派重写任务），通过则发人类交付确认任务。 """ import json import logging from .bid_common import ( get_db, new_id, rec_to_dict, rows_to_dicts, to_float, to_int, get_thresholds, record, create_human_task, find_human_task, resolve_project_id, create_role_task, CH_APPROVED, CH_REJECTED, DOC_DRAFT, DOC_REVIEWING, DOC_PASSED, DOC_REJECTED, HT_DELIVERY_CONFIRM, ) logger = logging.getLogger("pipeline.bidding.score") async def _latest_doc(sor, project_id, document_id=""): if document_id: recs = await sor.sqlExe("SELECT * FROM bid_documents WHERE id=${d}$", {"d": document_id}) else: recs = await sor.sqlExe( "SELECT * FROM bid_documents WHERE project_id=${p}$ ORDER BY version DESC LIMIT 1", {"p": project_id}) await sor.sqlExe("COMMIT", {}) return rec_to_dict(recs[0]) if recs else {} async def start_bid_score(project_id, document_id="", who=None, agent_id=None, task_id=""): """开启一轮整书评分：创建 bid_reviews(scope=whole) 并返回 review_id + 评分项清单。""" 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: doc = await _latest_doc(sor, project_id, document_id) if not doc: return False, "项目下无合成标书，请先 compose_bid" th = await get_thresholds(sor) rn = await sor.sqlExe( "SELECT COUNT(*) AS c FROM bid_reviews WHERE project_id=${p}$ AND scope='whole'", {"p": project_id}) await sor.sqlExe("COMMIT", {}) rnd = to_int(getattr(rn[0], "c", 0) if rn else 0) + 1 # 同一标书版本已有未完成评审 → 复用，避免重复开轮 open_r = await sor.sqlExe( "SELECT id FROM bid_reviews WHERE project_id=${p}$ AND scope='whole' " "AND document_id=${d}$ AND score IS NULL ORDER BY created_at DESC LIMIT 1", {"p": project_id, "d": doc.get("id")}) await sor.sqlExe("COMMIT", {}) if open_r: rid = getattr(open_r[0], "id", "") else: rid = new_id() await sor.C("bid_reviews", { "id": rid, "project_id": project_id, "scope": "whole", "document_id": doc.get("id"), "round": rnd, "reviewer": (who or "agent.bid_scorer")[:60], "review_type": "agent", "pass_ratio": th["bid_pass_ratio"], "passed": "0", "max_score": to_float(doc.get("max_score"), 0.0) or None, "task_id": task_id or "", }) await sor.sqlExe( "UPDATE bid_documents SET status=${st}$, review_id=${r}$, updated_at=NOW() " "WHERE id=${d}$", {"st": DOC_REVIEWING, "r": rid, "d": doc.get("id")}) await sor.sqlExe("COMMIT", {}) items = await sor.sqlExe( "SELECT s.id, s.item_no, s.item_name, s.max_score, s.is_veto, s.section, " "s.scoring_rule, s.target_chapter_no FROM bid_scoring_items s " "WHERE s.project_id=${p}$ ORDER BY s.section, s.order_no", {"p": project_id}) await sor.sqlExe("COMMIT", {}) rows = rows_to_dicts(items, limit=300) for r in rows: r["scoring_rule"] = (r.get("scoring_rule") or "")[:800] return True, json.dumps({ "review_id": rid, "round": rnd, "document_id": doc.get("id"), "document_version": doc.get("version"), "file_path": doc.get("file_path"), "pass_ratio": th["bid_pass_ratio"], "max_round": th["max_round"], "scoring_items": rows, "note": ("请逐个评分项调用 add_score_item_result 落分（不合格项必须给 improvement 与 " "target_chapter_no/target_chapter_id），全部落完后调用 finalize_bid_score 汇总。"), }, ensure_ascii=False, default=str) async def add_score_item_result(review_id, scoring_item_id="", score=0, max_score="", reason="", improvement="", target_chapter_no="", target_chapter_id="", item_name="", who=None, agent_id=None): """逐个评分项落分。不合格项（得分率 < 门限）必须给 improvement + 目标章节。""" if not review_id: return False, "缺少 review_id" db, dbname = get_db() async with db.sqlorContext(dbname) as sor: rv = await sor.sqlExe("SELECT * FROM bid_reviews WHERE id=${r}$", {"r": review_id}) await sor.sqlExe("COMMIT", {}) if not rv: return False, "评审记录不存在: %s" % review_id review = rec_to_dict(rv[0]) pid = review.get("project_id") th = await get_thresholds(sor) it = {} if scoring_item_id: irec = await sor.sqlExe( "SELECT * FROM bid_scoring_items WHERE id=${i}$", {"i": scoring_item_id}) await sor.sqlExe("COMMIT", {}) it = rec_to_dict(irec[0]) if irec else {} ms = to_float(max_score, 0.0) or to_float(it.get("max_score"), 0.0) if ms <= 0: return False, "缺少 max_score（评分项满分）" sc = to_float(score, -1.0) if sc < 0: return False, "score 必须为数字" ratio = sc / ms weak = ratio < th["bid_pass_ratio"] if weak and not (improvement or "").strip(): return False, ("该评分项得分率 %.0f%% 低于门限，必须给出 improvement（改进意见）" "与目标章节 target_chapter_no/target_chapter_id" % (ratio * 100)) cid = target_chapter_id or "" cno = (target_chapter_no or it.get("target_chapter_no") or "").strip() if not cid and cno: crec = await sor.sqlExe( "SELECT id FROM bid_chapters WHERE project_id=${p}$ AND chapter_no=${c}$ LIMIT 1", {"p": pid, "c": cno}) await sor.sqlExe("COMMIT", {}) cid = getattr(crec[0], "id", "") if crec else "" if not cid and scoring_item_id: crec = await sor.sqlExe( "SELECT id FROM bid_chapters WHERE project_id=${p}$ " "AND scoring_item_ids LIKE ${kw}$ LIMIT 1", {"p": pid, "kw": "%" + scoring_item_id + "%"}) await sor.sqlExe("COMMIT", {}) cid = getattr(crec[0], "id", "") if crec else "" exist = await sor.sqlExe( "SELECT id FROM bid_scores WHERE review_id=${r}$ AND scoring_item_id=${i}$ LIMIT 1", {"r": review_id, "i": scoring_item_id or ""}) await sor.sqlExe("COMMIT", {}) data = { "project_id": pid, "review_id": review_id, "scoring_item_id": scoring_item_id or "", "item_name": (item_name or it.get("item_name") or "")[:250], "score": sc, "max_score": ms, "reason": reason or "", "improvement": improvement or "", "target_chapter_id": cid, } if exist: data["id"] = getattr(exist[0], "id", "") await sor.U("bid_scores", data) else: data["id"] = new_id() await sor.C("bid_scores", data) await sor.sqlExe("COMMIT", {}) return True, ("评分项「%s」已落分 %.2f/%.2f%s" % (data["item_name"], sc, ms, "（需改进 → 章节 %s）" % (cid or cno or "未映射") if weak else "")) async def finalize_bid_score(review_id, comments="", who=None, agent_id=None): """汇总评分：判定通过/不通过。 不通过 → 命中章节置 rejected（带改进意见）→ reconciler 自动派重写任务； 通过 → 标书 passed + 发人类交付确认任务； 轮次超上限 → 抛人工介入。 """ if not review_id: return False, "缺少 review_id" db, dbname = get_db() async with db.sqlorContext(dbname) as sor: rv = await sor.sqlExe("SELECT * FROM bid_reviews WHERE id=${r}$", {"r": review_id}) await sor.sqlExe("COMMIT", {}) if not rv: return False, "评审记录不存在: %s" % review_id review = rec_to_dict(rv[0]) pid = review.get("project_id") th = await get_thresholds(sor) srecs = await sor.sqlExe( "SELECT * FROM bid_scores WHERE review_id=${r}$", {"r": review_id}) await sor.sqlExe("COMMIT", {}) scores = rows_to_dicts(srecs, limit=500) if not scores: return False, "本轮尚无评分明细，请先用 add_score_item_result 逐项落分" total = sum(to_float(s.get("score"), 0.0) for s in scores) maxsum = sum(to_float(s.get("max_score"), 0.0) for s in scores) # 否决项检查：满分项得 0 分且 is_veto=1 视为致命 veto_ids = await sor.sqlExe( "SELECT id, item_name FROM bid_scoring_items WHERE project_id=${p}$ AND is_veto='1'", {"p": pid}) await sor.sqlExe("COMMIT", {}) veto_map = {getattr(r, "id", ""): getattr(r, "item_name", "") for r in (veto_ids or [])} veto_fail = [veto_map[s["scoring_item_id"]] for s in scores if s.get("scoring_item_id") in veto_map and to_float(s.get("score"), 0.0) <= 0] ratio = (total / maxsum) if maxsum > 0 else 0.0 passed = (ratio >= th["bid_pass_ratio"]) and not veto_fail weak = [s for s in scores if to_float(s.get("max_score"), 0.0) > 0 and to_float(s.get("score"), 0.0) / to_float(s.get("max_score"), 1.0) < th["bid_pass_ratio"]] summary_lines = ["总分 %.2f/%.2f（%.1f%%，门限 %.0f%%）" % (total, maxsum, ratio * 100, th["bid_pass_ratio"] * 100)] if veto_fail: summary_lines.append("⛔ 否决项未响应：" + "、".join(veto_fail[:5])) for s in weak[:20]: summary_lines.append("- %s %.1f/%.1f → %s" % (s.get("item_name"), to_float(s.get("score")), to_float(s.get("max_score")), (s.get("improvement") or "")[:200])) summary = "\n".join(summary_lines) if comments: summary += "\n\n评审说明：" + comments await sor.sqlExe( "UPDATE bid_reviews SET score=${sc}$, max_score=${ms}$, passed=${ps}$, " "comments=${cm}$, updated_at=NOW() WHERE id=${r}$", {"sc": total, "ms": maxsum, "ps": "1" if passed else "0", "cm": summary[:60000], "r": review_id}) doc_status = DOC_PASSED if passed else DOC_REJECTED await sor.sqlExe( "UPDATE bid_documents SET status=${st}$, total_score=${sc}$, max_score=${ms}$, " "pass_ratio=${pr}$, updated_at=NOW() WHERE id=${d}$", {"st": doc_status, "sc": total, "ms": maxsum, "pr": th["bid_pass_ratio"], "d": review.get("document_id") or ""}) await sor.sqlExe("COMMIT", {}) await record(sor, pid, "bid_documents", review.get("document_id") or "", "score", to_state=doc_status, who=who, agent_id=agent_id, detail="%.2f/%.2f round=%s" % (total, maxsum, review.get("round"))) if passed: exist = await find_human_task(sor, pid, HT_DELIVERY_CONFIRM, status="pending") if not exist: await create_human_task( sor, pid, HT_DELIVERY_CONFIRM, "标书评分通过，请确认交付", ("标书评分已达通过门限。\n%s\n\n" "请人工复核并确认交付（用印/密封/份数/递交方式按招标文件要求执行）。" % summary[:3000]), ) await sor.sqlExe("COMMIT", {}) return True, "整书评分通过：%s\n已发人工交付确认任务。" % summary # 不通过：轮次门禁 if to_int(review.get("round"), 1) >= th["max_round"]: await create_human_task( sor, pid, "general", "标书评分连续不达标需人工介入（第 %s 轮）" % review.get("round"), ("整书评分已进行 %s 轮仍未达门限：\n%s\n\n" "请人工决策：调整评分预期/补充关键素材/重构章节，处理完成后完成本任务。" % (review.get("round"), summary[:3000])), ) await sor.sqlExe("COMMIT", {}) return True, ("整书评分不通过且已达轮次上限 %d → 已抛人工介入。\n%s" % (th["max_round"], summary)) # 把改进意见落到问题章节：approved → rejected（reconciler 会派重写任务） touched = 0 for s in weak: cid = s.get("target_chapter_id") or "" if not cid: continue crec = await sor.sqlExe( "SELECT id, status, revise_count, title FROM bid_chapters WHERE id=${c}$", {"c": cid}) await sor.sqlExe("COMMIT", {}) if not crec: continue c = rec_to_dict(crec[0]) imp = ("【整书评分改进意见 · 第%s轮】评分项「%s」得 %.1f/%.1f：\n%s" % (review.get("round"), s.get("item_name"), to_float(s.get("score")), to_float(s.get("max_score")), s.get("improvement") or "")) await sor.sqlExe( "UPDATE bid_chapters SET status=${st}$, review_comment=${cm}$, " "revise_count=${rc}$, updated_at=NOW() WHERE id=${c}$", {"st": CH_REJECTED, "cm": imp[:60000], "rc": to_int(c.get("revise_count"), 0) + 1, "c": cid}) await sor.sqlExe("COMMIT", {}) await record(sor, pid, "bid_chapters", cid, "score_reject", from_state=c.get("status"), to_state=CH_REJECTED, who=who, agent_id=agent_id, detail=(s.get("improvement") or "")[:400]) # 退回也同步镜像（文件头状态/评审分/改进意见要与库一致，2026-09-16） try: from .bid_chapter_mirror import sync_chapter_mirror await sync_chapter_mirror(sor, pid, cid) except Exception as e: # noqa: BLE001 logger.warning("章节镜像同步失败（不阻断）: chapter=%s err=%s", cid, e) touched += 1 if touched == 0 and weak: await create_human_task( sor, pid, "general", "整书评分不通过但改进意见未映射到章节", ("以下评分项不达标，但 target_chapter_id 为空，系统无法自动定位待改章节：\n%s\n\n" "请人工指定应修改的章节（或补 target_chapter_no 后重新评分）。" % summary[:3000]), ) await sor.sqlExe("COMMIT", {}) return True, ("整书评分不通过：%s\n已退回 %d 个问题章节重写，系统将自动派发重写任务。" % (summary, touched)) async def request_chapter_revise(chapter_id, improvement, who=None, agent_id=None): """按改进意见直接退回单个章节（评分员定点回退用）。""" if not (improvement or "").strip(): return False, "必须给出 improvement（改进意见）" from .bid_review_capability import reject_chapter return await reject_chapter(chapter_id, comment=improvement, who=who, agent_id=agent_id) async def bid_score_report(project_id, limit=50): """评分报告：最近一轮整书评分的总分 + 逐项明细。""" try: project_id = await resolve_project_id(project_id) except ValueError as e: return "缺少 project_id" db, dbname = get_db() async with db.sqlorContext(dbname) as sor: rv = await sor.sqlExe( "SELECT * FROM bid_reviews WHERE project_id=${p}$ AND scope='whole' " "ORDER BY created_at DESC LIMIT 1", {"p": project_id}) await sor.sqlExe("COMMIT", {}) if not rv: return "尚无整书评分记录" r = rec_to_dict(rv[0]) srecs = await sor.sqlExe( "SELECT item_name, score, max_score, LEFT(reason,300) AS reason, " "LEFT(improvement,300) AS improvement, target_chapter_id " "FROM bid_scores WHERE review_id=${r}$ ORDER BY created_at LIMIT " + str(to_int(limit, 50)), {"r": r.get("id")}) await sor.sqlExe("COMMIT", {}) return json.dumps({ "review_id": r.get("id"), "round": r.get("round"), "score": r.get("score"), "max_score": r.get("max_score"), "passed": r.get("passed"), "pass_ratio": r.get("pass_ratio"), "comments": (r.get("comments") or "")[:3000], "items": rows_to_dicts(srecs, limit=to_int(limit, 50)), }, ensure_ascii=False, default=str) R_SCORER = "agent.bid_scorer" async def reject_delivery(human_task_id, comment="", operator_id=None): """用户退回交付（专用端点调用，agent 无此工具——退回是用户裁决）。 2026-09-17 用户报障：交付确认待办只有「确认交付」单按钮，没有退回通道—— 复核发现问题（内容/用印/密封/口径）时无法把标书打回，只能确认或干等。 对齐 tech_template_confirm 的 confirm/reject 双通道模式。 机制（复用既有两级回退循环，不另建链路）： ① 待办置 rejected，退回意见存 result_data（tech_template 驳回同款约定）； ② 最新 passed 标书置 rejected（否则 G 段按 passed 立刻重发确认待办）； ③ 派 scorer「交付驳回处理」任务（task_kind=bid_delivery_reject，params 带 reject_comment）——scorer 按用户意见 request_chapter_revise 退回问题章节， 之后 C→D→E→F→G 既有链路自动：重写→重审→重合成→重评分→重发交付确认； ④ 全部 approved 章节 revise_count 复位（用户级退回=新一轮修订预算；不复位则 上轮烧满 max_revise 的章节被 C 段门禁拦死，退回即死锁——对齐 C1 段人工 完成待办后复位 revise_count 的既有语义）； ⑤ E 段配套门禁（bid_flow）：bid_delivery_reject 任务在办期间不派合成—— 否则章节还没被退回就抢先重合成同内容新版，用户意见被静默绕过。 """ if not human_task_id: return False, "缺少 human_task_id" if not operator_id: return False, "未登录" if 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_DELIVERY_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 "" if not project_id: return False, "待办未挂项目" # 权限：复用引擎 _check_confirm_operator（同机构 + owner.superuser/项目创建者） # ——与 flow_plan/tech_template 确认门语义一致，不自写校验（2026-09-15 实测 # 自写「仅同机构」会误拒超管）。 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, "仅同机构用户可退回该项目交付" # ① 待办置 rejected + 意见落 result_data（与 tech_template 驳回分支同款约定； # 注：pipeline_human_tasks 无 updated_at 列） await sor.sqlExe( "UPDATE pipeline_human_tasks SET status='rejected', result_data=${rd}$, " "submitted_by=${u}$, submitted_at=NOW() WHERE id=${h}$ AND status='pending'", {"rd": json.dumps({"decision": "reject", "comment": comment[: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, "delivery_rejected", to_state="rejected", who=operator_id, detail=comment[:500]) # ② 最新标书 passed → rejected（draft/reviewing 本就未过评分，不动） doc = await _latest_doc(sor, project_id) doc_id = doc.get("id") or "" if doc and doc.get("status") == DOC_PASSED: await sor.sqlExe( "UPDATE bid_documents SET status=${st}$, updated_at=NOW() " "WHERE id=${d}$ AND status=${os}$", {"st": DOC_REJECTED, "d": doc_id, "os": DOC_PASSED}) await sor.sqlExe("COMMIT", {}) # ④ approved 章节重做预算复位（幂等：只动 approved，正在重写的 rejected 章不动） await sor.sqlExe( "UPDATE bid_chapters SET revise_count=0 WHERE project_id=${p}$ " "AND status=${st}$ AND revise_count>0", {"p": project_id, "st": CH_APPROVED}) await sor.sqlExe("COMMIT", {}) # ③ 派 scorer 交付驳回处理任务（在办任务幂等：已有同类在办不重派） from .bid_flow import _open_tech_tasks if await _open_tech_tasks(sor, project_id, "bid_delivery_reject") > 0: return True, "交付已退回（驳回处理任务已在办，勿重复退回）" prec = await sor.sqlExe("SELECT name FROM sd_projects WHERE id=${p}$", {"p": project_id}) await sor.sqlExe("COMMIT", {}) pname = getattr(prec[0], "name", "") if prec else "" tid = await create_role_task( sor, project_id, R_SCORER, "%s 交付驳回处理（按用户退回意见打回章节）" % pname, {"stage": "score", "task_kind": "bid_delivery_reject", "document_id": doc_id, "reject_comment": comment[:60000]}) await sor.sqlExe("COMMIT", {}) await record(sor, project_id, "bid_documents", doc_id, "delivery_reject", from_state=DOC_PASSED, to_state=DOC_REJECTED, who=operator_id, detail=comment[:480]) return True, ("交付已退回：标书 v%s 置 rejected，评分员将按你的意见退回问题章节" "重写（重写→重审→重合成→重评分后自动重发交付确认）。" % (doc.get("version") or "?"))