226 lines
17 KiB
Python
226 lines
17 KiB
Python
"""Snapshot-grounded report drafts, protected human text and immutable final copies."""
|
||
from copy import deepcopy
|
||
import io
|
||
from services.assistant_store import get, put, listing, event, uid, now, digest, Conflict, time_event, finish_time_event
|
||
from services.component_registry import resolve
|
||
|
||
|
||
def section_content(section_id, snapshot, run):
|
||
values = snapshot["values"]
|
||
unresolved = [f for f in run.get("findings", []) if f["status"] not in ("已解决", "不适用")]
|
||
if section_id == "overview":
|
||
return f"{snapshot['customer']}，拟融资 {values.get('facility.amount', '未提供')} 万元；租赁物价值 {values.get('lease.total_value', '未提供')} 万元。\n期间：{snapshot.get('period', '待确认')}；口径：{snapshot.get('scope', '待确认')}。\n价值覆盖率（比值）：{run.get('summary', {}).get('coverage') or '未计算'}。"
|
||
if section_id == "findings":
|
||
return "\n".join(f"{i+1}. {f['title']}（{f['result']}）：{f['suggested_action']}\n依据：{'、'.join(f.get('evidence_ids', [])) or '缺少可验证证据'}" for i, f in enumerate(unresolved)) or "本次已执行的检查未发现待处理事项；不表示未执行的检查已通过。"
|
||
if section_id in ("completeness", "verification"):
|
||
missing = [f["title"] for f in unresolved if f.get("result") in ("unknown", "error")]
|
||
return ("待补充或无法完成核验：" + "、".join(missing)) if missing else "本次必做资料检查已执行，未发现明确缺件；仍需人工核对材料真实性和时效。"
|
||
if section_id == "financial":
|
||
assets = values.get("financial.total_assets", "未提供")
|
||
liabilities = values.get("financial.total_liabilities", "未提供")
|
||
cashflow = values.get("financial.operating_cashflow", "未提供")
|
||
return f"资产总额 {assets} 万元；负债总额 {liabilities} 万元；经营现金流 {cashflow} 万元。\n计算结果及口径以本次冻结数据卡和Skill执行记录为准。"
|
||
if section_id == "scenario":
|
||
return f"当前融资金额 {values.get('facility.amount', '未提供')} 万元，租赁物价值 {values.get('lease.total_value', '未提供')} 万元，价值覆盖率 {run.get('summary', {}).get('coverage') or '未计算'}。\n额度、期限、价格和增信条件由授权人员结合情景测算确认。"
|
||
if section_id == "manager_response":
|
||
notes = [f.get("human_note") for f in run.get("findings", []) if f.get("human_note")]
|
||
return "\n".join(f"{i+1}. {note}" for i, note in enumerate(notes)) or "尚无客户经理确认说明；风险回应及补件记录保留在项目问题清单。"
|
||
responses = [s.get("outputs", {}).get("requirement_response", "") for s in run["steps"] if s.get("outputs", {}).get("requirement_response")]
|
||
failed = [s["name"] for s in run["steps"] if s["status"] in ("failed", "partial", "skipped")]
|
||
return ("\n".join(responses) or "尚无经验证的模型核查意见，请人工补充判断。") + ("\n未完整执行：" + "、".join(failed) if failed else "") + "\n本底稿仅供核查，不构成审批结论。"
|
||
|
||
|
||
class DocumentsMixin:
|
||
@staticmethod
|
||
def mark_stale(db, project_id, changed):
|
||
for finding in listing(db, "finding", project_id):
|
||
if "*" in changed or set(finding.get("fields", ["*"])) & set(changed) or not finding.get("fields"):
|
||
finding["stale"] = True
|
||
put(db, "finding", finding)
|
||
for report in listing(db, "report", project_id):
|
||
if report["status"] == "finalized":
|
||
continue
|
||
affected = False
|
||
for section in report["sections"]:
|
||
deps = section.get("dependencies", ["*"])
|
||
if "*" in deps or "*" in changed or set(deps) & set(changed):
|
||
section["stale"] = True
|
||
affected = True
|
||
if affected:
|
||
report["status"] = "stale"
|
||
put(db, "report", report)
|
||
|
||
def create_report(self, project_id, run_id, actor, template_id="template.due-diligence", idempotency_key=""):
|
||
with self.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
request_hash = digest({"project_id": project_id, "run_id": run_id, "template_id": template_id})
|
||
if idempotency_key:
|
||
prior = db.execute("SELECT request_hash,object_id FROM assistant_idempotency WHERE actor=? AND operation='report' AND key=?", (actor, idempotency_key)).fetchone()
|
||
if prior:
|
||
if prior[0] != request_hash:
|
||
raise Conflict("幂等键已用于不同底稿请求")
|
||
return get(db, "report", prior[1])
|
||
project = get(db, "project", project_id)
|
||
run = get(db, "run", run_id)
|
||
if run["project_id"] != project_id:
|
||
raise KeyError("运行记录不存在")
|
||
if run["status"] not in ("completed", "partial") or not run.get("snapshot_id"):
|
||
raise Conflict("仅可从已完成或部分完成的冻结运行生成底稿")
|
||
snapshot = get(db, "snapshot", run["snapshot_id"])
|
||
template = resolve(db, template_id)
|
||
if template["kind"] != "template" or template["scope"] != project["data_domain"]:
|
||
raise ValueError("模板类型或数据域不匹配")
|
||
stale = snapshot["data_revision"] != project["data_revision"]
|
||
sections = [{**deepcopy(s), "content": section_content(s["id"], snapshot, run), "locked": False, "confirmed": False,
|
||
"stale": stale, "source_snapshot_id": snapshot["id"], "candidate": None, "history": []} for s in template["sections"]]
|
||
if template_id in ("template.independent-review", "template.committee-brief", "template.review", "template.committee"):
|
||
decisions = listing(db, "review_decision", project_id)
|
||
judgment = next((section for section in sections if section["id"] == "judgment"), None)
|
||
if judgment and decisions:
|
||
current, seen = [], set()
|
||
for item in decisions:
|
||
key = item.get("node_id") or item.get("finding_id") or "overall"
|
||
if key in seen or (item.get("run_id") and item.get("run_id") != run_id):
|
||
continue
|
||
seen.add(key)
|
||
current.append(item)
|
||
judgment["content"] = "\n".join(
|
||
f"{item.get('node_title') or '项目整体'}｜{item.get('decision')}：{item.get('opinion') or '评审员已确认'}\n"
|
||
f"数据来源：{'、'.join(item.get('evidence_ids', [])) or '该节点未取得可验证材料'}；检查依据：{'、'.join(item.get('basis_ids', [])) or '人工专业判断'}"
|
||
for item in current
|
||
) + "\n\n以上意见由评审员确认；每项意见均关联冻结快照与证据索引，AI仅提供核查建议。"
|
||
report = {"id": uid("report"), "project_id": project_id, "run_id": run_id, "snapshot_id": snapshot["id"],
|
||
"template": deepcopy(template), "status": "stale" if stale else "draft", "sections": sections, "actor_id": actor,
|
||
"created_at": now(), "data_domain": project["data_domain"], "data_revision": snapshot["data_revision"],
|
||
"disclaimer": ("演示数据｜" if project["data_domain"] == "demo" else "") + "核查底稿｜不构成审批结论",
|
||
"evidence": deepcopy(run["evidence"])}
|
||
result = put(db, "report", report)
|
||
time_event(db, project_id, "report_drafting", "报告整理" if template_id == "template.ai-pre-review" else "评审报告编制",
|
||
"manager" if template_id == "template.ai-pre-review" else "reviewer", "active", actor, "report", report["id"],
|
||
detail={"template_id": template_id, "run_id": run_id})
|
||
if idempotency_key:
|
||
db.execute("INSERT INTO assistant_idempotency VALUES (?,?,?,?,?)", (actor, "report", idempotency_key, request_hash, result["id"]))
|
||
event(db, project_id, "report.created", actor, {"report_id": report["id"], "snapshot_id": snapshot["id"]})
|
||
return result
|
||
|
||
def report_action(self, project_id, report_id, actor, action, payload, expected):
|
||
with self.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
report = get(db, "report", report_id)
|
||
if report["project_id"] != project_id:
|
||
raise KeyError("底稿不存在")
|
||
if report["revision"] != expected or report["status"] == "finalized":
|
||
raise Conflict("底稿已更新或已定稿，不可覆盖")
|
||
if action == "regenerate":
|
||
run = get(db, "run", payload["run_id"])
|
||
if run["project_id"] != project_id or run["status"] not in ("completed", "partial"):
|
||
raise ValueError("需选择本项目已结束的运行")
|
||
snapshot = get(db, "snapshot", run["snapshot_id"])
|
||
if snapshot["data_revision"] != get(db, "project", project_id)["data_revision"]:
|
||
raise Conflict("运行快照已过期，请先重新检查")
|
||
for section in report["sections"]:
|
||
candidate = {"content": section_content(section["id"], snapshot, run), "snapshot_id": snapshot["id"], "run_id": run["id"], "data_revision": snapshot["data_revision"]}
|
||
if section["locked"] or section["confirmed"]:
|
||
section.update(candidate=candidate, stale=True)
|
||
else:
|
||
section["history"].append({"content": section["content"], "snapshot_id": section["source_snapshot_id"], "actor": actor, "time": now()})
|
||
section.update(content=candidate["content"], source_snapshot_id=snapshot["id"], stale=False, candidate=None)
|
||
report.update(run_id=run["id"], snapshot_id=snapshot["id"], data_revision=snapshot["data_revision"], evidence=run["evidence"])
|
||
elif action == "section":
|
||
section = next((s for s in report["sections"] if s["id"] == payload.get("section_id")), None)
|
||
if not section:
|
||
raise KeyError("章节不存在")
|
||
section["history"].append({"content": section["content"], "snapshot_id": section["source_snapshot_id"], "actor": actor, "time": now()})
|
||
if payload.get("accept_candidate"):
|
||
candidate = section.get("candidate")
|
||
if not candidate:
|
||
raise Conflict("没有待采纳候选")
|
||
if candidate["data_revision"] != get(db, "project", project_id)["data_revision"]:
|
||
raise Conflict("候选内容已过期")
|
||
section.update(content=candidate["content"], source_snapshot_id=candidate["snapshot_id"], candidate=None, stale=False, confirmed=False, locked=False)
|
||
else:
|
||
if section["locked"] and payload.get("content", section["content"]) != section["content"]:
|
||
raise Conflict("请先解除锁定再编辑")
|
||
if "content" in payload:
|
||
if not isinstance(payload["content"], str) or len(payload["content"]) > 20000:
|
||
raise ValueError("章节内容过长")
|
||
if section["id"] in ("overview", "findings") and payload["content"] != section["content"]:
|
||
raise ValueError("事实和机器检查清单请通过补件与复核修正；核查意见章节可人工编辑")
|
||
section["content"] = payload["content"]
|
||
section["edited_by"] = actor
|
||
if payload.get("confirmed") and section["stale"]:
|
||
raise Conflict("过期章节需先重生成或采纳新候选")
|
||
section.update(locked=bool(payload.get("locked", section["locked"])), confirmed=bool(payload.get("confirmed", False)))
|
||
elif action == "finalize":
|
||
project = get(db, "project", project_id)
|
||
if report["data_revision"] != project["data_revision"] or any(s["stale"] or not s["confirmed"] or s.get("candidate") for s in report["sections"]):
|
||
raise Conflict("所有章节须确认且来源未过期，才能定稿")
|
||
final = dict(deepcopy(report), id=uid("final"), report_id=report_id, finalized_by=actor, finalized_at=now(), status="finalized")
|
||
final["content_hash"] = digest(final)
|
||
put(db, "report_final", final, immutable=True)
|
||
report.update(status="finalized", final_id=final["id"], finalized_by=actor)
|
||
finish_time_event(db, "report", report_id)
|
||
elif action == "submit":
|
||
project = get(db, "project", project_id)
|
||
if report["data_revision"] != project["data_revision"] or any(s["stale"] for s in report["sections"]):
|
||
raise Conflict("报告来源已变化，请先重新预审并更新报告")
|
||
submissions = listing(db, "submission", project_id)
|
||
if any(item.get("status") == "submitted" for item in submissions):
|
||
raise Conflict("项目已在评审池中；如需更新，请先从评审池撤回再重新推送")
|
||
submission = {"id": uid("submission"), "project_id": project_id, "report_id": report_id, "run_id": report["run_id"],
|
||
"submitted_by": actor, "submitted_at": now(), "status": "submitted",
|
||
"sequence": 1 + max((item.get("sequence", 1) for item in submissions), default=0),
|
||
"content_hash": digest(report)}
|
||
put(db, "submission", submission, immutable=True)
|
||
report.update(status="submitted", submission_id=submission["id"], submitted_by=actor)
|
||
project.update(workflow_state="submitted", stage="待评审", submitted_at=submission["submitted_at"],
|
||
reviewer_id=None, review_claimed_at=None, review_started_at=None, review_decided_at=None)
|
||
put(db, "project", project)
|
||
finish_time_event(db, "report", report_id)
|
||
time_event(db, project_id, "review_queue", "评审排队", "reviewer", "queue", actor, "submission", submission["id"])
|
||
else:
|
||
raise ValueError("不支持的底稿动作")
|
||
if action not in ("finalize", "submit"):
|
||
report["status"] = "stale" if any(s["stale"] for s in report["sections"]) else "draft"
|
||
result = put(db, "report", report, expected=expected)
|
||
event(db, project_id, "report." + action, actor, {"report_id": report_id, "section_id": payload.get("section_id")})
|
||
return result
|
||
|
||
def export_report(self, project_id, report_id):
|
||
report = self.resource("report", report_id, project_id)
|
||
if report.get("final_id"):
|
||
report = self.resource("report_final", report["final_id"], project_id)
|
||
from docx import Document
|
||
from docx.shared import Pt, Cm, RGBColor
|
||
from docx.oxml.ns import qn
|
||
document = Document()
|
||
section = document.sections[0]
|
||
section.page_width, section.page_height = Cm(21), Cm(29.7)
|
||
section.top_margin = section.bottom_margin = Cm(2.2)
|
||
section.left_margin = section.right_margin = Cm(2.4)
|
||
for name in ("Normal", "Title", "Heading 1", "Heading 2"):
|
||
style = document.styles[name]
|
||
style.font.name = "Microsoft YaHei"
|
||
style._element.get_or_add_rPr().rFonts.set(qn("w:eastAsia"), "Microsoft YaHei")
|
||
style.font.color.rgb = RGBColor(0, 0, 0)
|
||
style.font.size = Pt(11 if name == "Normal" else 18 if name == "Title" else 13)
|
||
document.styles["Normal"].paragraph_format.line_spacing = 1.3
|
||
document.add_paragraph(report["template"]["name"], "Title")
|
||
document.add_paragraph(report["disclaimer"] + ("｜已定稿" if report["status"] == "finalized" else "｜草稿"))
|
||
document.add_paragraph(f"项目 {project_id}\n运行 {report['run_id']}\n快照 {report['snapshot_id']}")
|
||
for item in report["sections"]:
|
||
document.add_heading(item["title"], level=1)
|
||
if item["stale"]:
|
||
document.add_paragraph("本节来源已过期，需重新核查。")
|
||
for line in item["content"].splitlines():
|
||
document.add_paragraph(line)
|
||
document.add_heading("证据索引", level=1)
|
||
used = set()
|
||
for evidence in report["evidence"]:
|
||
if evidence["id"] not in used:
|
||
document.add_paragraph(f"{evidence['id']}\n{evidence['source']}，{evidence['locator']}，字段 {evidence['field']}；值 {evidence.get('value')}。")
|
||
used.add(evidence["id"])
|
||
buffer = io.BytesIO()
|
||
document.save(buffer)
|
||
return buffer.getvalue()
|