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 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 "已生成草稿导出文件，正式版仍被风险项阻断。" ) return { "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, }, } 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"{escape(line)}" for line in lines ) return ( "" "" "" f"{paragraphs}" "" "" "" ) def _content_types_xml() -> str: return ( "" "" "" "" "" "" ) def _root_rels_xml() -> str: return ( "" "" "" "" ) def _document_rels_xml() -> str: return ( "" "" )