250 lines
9.9 KiB
Python
250 lines
9.9 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from xml.sax.saxutils import escape
|
||
import zipfile
|
||
|
||
from django.conf import settings
|
||
|
||
from agent_core.governance import load_governance_config
|
||
from apps.documents.services import create_export_record
|
||
|
||
|
||
def generate_registration_export(*, batch, conversation, upstream_summary: dict | None = None) -> dict:
|
||
"""
|
||
基于当前会话上下文生成最小可下载的 Word 导出结果。
|
||
|
||
这里故意保持为 Django 服务层职责：
|
||
- 根据风险结果判断正式版/草稿版
|
||
- 选择治理台中启用的模板摘要
|
||
- 生成离线可演示的 `.docx` 文件与下载链接
|
||
"""
|
||
upstream_summary = upstream_summary or {}
|
||
template_mapping = _resolve_template_mapping()
|
||
blocked_items = _collect_blocked_items(upstream_summary)
|
||
can_export_formally = _can_export_formally(upstream_summary, blocked_items)
|
||
export_mode = "formal" if can_export_formally else "draft"
|
||
export_status = "completed" if can_export_formally else "draft_only"
|
||
relative_path = _build_relative_export_path(batch.batch_id, export_mode)
|
||
absolute_path = Path(settings.MEDIA_ROOT) / relative_path
|
||
absolute_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
fillable_items = _build_fillable_items(batch, conversation)
|
||
_write_minimal_docx(
|
||
absolute_path,
|
||
product_name=batch.product_name or conversation.product_name or "未命名资料包",
|
||
batch_id=batch.batch_id,
|
||
export_mode=export_mode,
|
||
summary=upstream_summary.get("summary", ""),
|
||
fillable_items=fillable_items,
|
||
blocked_items=blocked_items,
|
||
template_mapping=template_mapping,
|
||
)
|
||
|
||
file_name = absolute_path.name
|
||
download_url = f"/{settings.MEDIA_URL.strip('/')}/{relative_path.as_posix()}"
|
||
summary = (
|
||
"已生成正式版导出文件。"
|
||
if can_export_formally
|
||
else "已生成草稿导出文件，正式版仍被风险项阻断。"
|
||
)
|
||
report = {
|
||
"output_type": "registration_word_export_report",
|
||
"summary": summary,
|
||
"template_name": template_mapping["template_name"],
|
||
"template_version": template_mapping["version"],
|
||
"export_status": export_status,
|
||
"draft_export_status": "completed",
|
||
"formal_export_status": "completed" if can_export_formally else "blocked",
|
||
"can_export_formally": can_export_formally,
|
||
"fillable_items": fillable_items,
|
||
"fillable_field_count": len(fillable_items),
|
||
"filled_field_count": len(fillable_items),
|
||
"blocked_items": blocked_items,
|
||
"blocked_field_count": len(blocked_items),
|
||
"manual_review_field_count": len(blocked_items),
|
||
"layout_check_status": "passed",
|
||
"download_url": download_url,
|
||
"output_file": {
|
||
"file_name": file_name,
|
||
"relative_path": relative_path.as_posix(),
|
||
"absolute_path": str(absolute_path),
|
||
"export_mode": export_mode,
|
||
},
|
||
}
|
||
create_export_record(
|
||
batch=batch,
|
||
conversation_id=conversation.conversation_id,
|
||
product_name=batch.product_name or conversation.product_name,
|
||
template_name=report["template_name"],
|
||
template_version=report["template_version"],
|
||
export_mode=export_mode,
|
||
output_type=report["output_type"],
|
||
file_name=file_name,
|
||
relative_path=relative_path.as_posix(),
|
||
download_url=download_url,
|
||
)
|
||
return report
|
||
|
||
|
||
def update_conversation_with_export_report(conversation, export_report: dict) -> None:
|
||
latest_summary = dict(conversation.latest_summary or {})
|
||
previous_structured_output = latest_summary.get("structured_output") or {}
|
||
if previous_structured_output.get("output_type") != "registration_word_export_report":
|
||
latest_summary["upstream_structured_output"] = previous_structured_output
|
||
latest_summary["structured_output"] = export_report
|
||
latest_summary["answer"] = export_report.get("summary", "")
|
||
latest_summary["status"] = "success"
|
||
conversation.latest_summary = latest_summary
|
||
conversation.node_results = _update_word_export_node(conversation.node_results, export_report)
|
||
conversation.save(update_fields=["latest_summary", "node_results", "updated_at"])
|
||
|
||
|
||
def _resolve_template_mapping() -> dict:
|
||
governance_config = load_governance_config()
|
||
for item in governance_config["template_mappings"]:
|
||
if item.get("status") == "启用":
|
||
return item
|
||
return {
|
||
"template_name": "注册证导出模板",
|
||
"version": "V1.0",
|
||
"field_mapping_summary": "产品名称 / 批次号 / 风险结论",
|
||
}
|
||
|
||
|
||
def _collect_blocked_items(upstream_summary: dict) -> list[str]:
|
||
blocked_items = []
|
||
for item in upstream_summary.get("manual_review_items") or []:
|
||
if isinstance(item, str) and item.strip():
|
||
blocked_items.append(item.strip())
|
||
for item in upstream_summary.get("risk_items") or []:
|
||
if isinstance(item, dict):
|
||
title = (item.get("title") or item.get("issue") or "").strip()
|
||
if title:
|
||
blocked_items.append(title)
|
||
unique_items = []
|
||
for item in blocked_items:
|
||
if item not in unique_items:
|
||
unique_items.append(item)
|
||
return unique_items
|
||
|
||
|
||
def _can_export_formally(upstream_summary: dict, blocked_items: list[str]) -> bool:
|
||
pass_status = upstream_summary.get("pass_status")
|
||
if pass_status in {"blocked", "failed", "review_required", "manual_review"}:
|
||
return False
|
||
highest_risk_level = str(upstream_summary.get("highest_risk_level", "")).lower()
|
||
if highest_risk_level == "high":
|
||
return False
|
||
return not blocked_items
|
||
|
||
|
||
def _build_fillable_items(batch, conversation) -> list[dict]:
|
||
return [
|
||
{"placeholder": "{{ product_name }}", "field_name": "产品名称", "field_value": batch.product_name},
|
||
{"placeholder": "{{ batch_id }}", "field_name": "批次号", "field_value": batch.batch_id},
|
||
{"placeholder": "{{ conversation_id }}", "field_name": "会话编号", "field_value": conversation.conversation_id},
|
||
{"placeholder": "{{ file_count }}", "field_name": "文件数", "field_value": str(batch.file_count)},
|
||
{"placeholder": "{{ page_count }}", "field_name": "页数", "field_value": str(batch.page_count)},
|
||
]
|
||
|
||
|
||
def _build_relative_export_path(batch_id: str, export_mode: str) -> Path:
|
||
file_name = f"{batch_id}-{export_mode}.docx"
|
||
return Path("exports") / date.today().strftime("%Y%m%d") / file_name
|
||
|
||
|
||
def _update_word_export_node(node_results: list[dict], export_report: dict) -> list[dict]:
|
||
updated_nodes = []
|
||
export_status = export_report.get("export_status")
|
||
node_status = "已完成" if export_status == "completed" else "待复核"
|
||
for node in node_results or []:
|
||
current = dict(node)
|
||
if current.get("label") == "Word 回填导出":
|
||
current["status"] = node_status
|
||
current["summary"] = export_report.get("summary", "")
|
||
updated_nodes.append(current)
|
||
return updated_nodes
|
||
|
||
|
||
def _write_minimal_docx(
|
||
output_path: Path,
|
||
*,
|
||
product_name: str,
|
||
batch_id: str,
|
||
export_mode: str,
|
||
summary: str,
|
||
fillable_items: list[dict],
|
||
blocked_items: list[str],
|
||
template_mapping: dict,
|
||
) -> None:
|
||
document_lines = [
|
||
f"注册审核导出文件（{'正式版' if export_mode == 'formal' else '草稿版'}）",
|
||
f"产品名称：{product_name}",
|
||
f"批次号：{batch_id}",
|
||
f"模板：{template_mapping.get('template_name', '')} {template_mapping.get('version', '')}".strip(),
|
||
f"风险摘要：{summary or '无'}",
|
||
"回填字段：",
|
||
]
|
||
document_lines.extend(
|
||
f"- {item['field_name']}：{item['field_value']}" for item in fillable_items
|
||
)
|
||
if blocked_items:
|
||
document_lines.append("阻断项：")
|
||
document_lines.extend(f"- {item}" for item in blocked_items)
|
||
else:
|
||
document_lines.append("阻断项：无")
|
||
|
||
document_xml = _build_document_xml(document_lines)
|
||
with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
|
||
archive.writestr("[Content_Types].xml", _content_types_xml())
|
||
archive.writestr("_rels/.rels", _root_rels_xml())
|
||
archive.writestr("word/document.xml", document_xml)
|
||
archive.writestr("word/_rels/document.xml.rels", _document_rels_xml())
|
||
|
||
|
||
def _build_document_xml(lines: list[str]) -> str:
|
||
paragraphs = "".join(
|
||
f"<w:p><w:r><w:t xml:space='preserve'>{escape(line)}</w:t></w:r></w:p>" for line in lines
|
||
)
|
||
return (
|
||
"<?xml version='1.0' encoding='UTF-8' standalone='yes'?>"
|
||
"<w:document xmlns:w='http://schemas.openxmlformats.org/wordprocessingml/2006/main'>"
|
||
"<w:body>"
|
||
f"{paragraphs}"
|
||
"<w:sectPr/>"
|
||
"</w:body>"
|
||
"</w:document>"
|
||
)
|
||
|
||
|
||
def _content_types_xml() -> str:
|
||
return (
|
||
"<?xml version='1.0' encoding='UTF-8'?>"
|
||
"<Types xmlns='http://schemas.openxmlformats.org/package/2006/content-types'>"
|
||
"<Default Extension='rels' ContentType='application/vnd.openxmlformats-package.relationships+xml'/>"
|
||
"<Default Extension='xml' ContentType='application/xml'/>"
|
||
"<Override PartName='/word/document.xml' "
|
||
"ContentType='application/vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml'/>"
|
||
"</Types>"
|
||
)
|
||
|
||
|
||
def _root_rels_xml() -> str:
|
||
return (
|
||
"<?xml version='1.0' encoding='UTF-8'?>"
|
||
"<Relationships xmlns='http://schemas.openxmlformats.org/package/2006/relationships'>"
|
||
"<Relationship Id='rId1' "
|
||
"Type='http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument' "
|
||
"Target='word/document.xml'/>"
|
||
"</Relationships>"
|
||
)
|
||
|
||
|
||
def _document_rels_xml() -> str:
|
||
return (
|
||
"<?xml version='1.0' encoding='UTF-8'?>"
|
||
"<Relationships xmlns='http://schemas.openxmlformats.org/package/2006/relationships'/>"
|
||
)
|