"""荷兰(NL)企业文档生成编排：多模板 → 合并 → 主文件 + 执照。""" from __future__ import annotations import traceback import uuid from pathlib import Path from app.core.config import settings from app.core.countries import find_templates from app.core.database import SourceSessionLocal from app.core.logger import get_logger from app.core.paths import safe_filename from app.core.response import AppError from app.repositories.nl_enterprise import NL_FILE_CATEGORIES, get_enterprise_detail, update_vat_push_status from app.repositories.vat_generate_log import insert_vat_generate_log from app.services.cos_storage import upload_file from app.services.enterprise_document import ( PHOTO_ROTATION, _build_values, _ocr_idcard_fields, _prepare_images, _write_back_annex, _write_back_status, render_templates, ) from app.services.pdf_merge import merge_pdfs logger = get_logger(__name__) # 日志级别权重：合并多条步骤日志时取最高级别 _LEVEL_ORDER = {"info": 0, "warning": 1, "error": 2} def generate_nl_documents(record: dict, files: list[dict], gen_file_prefix="", gen_file_suffix="") -> dict: """NL 文档生成：多模板 → 合并 → 主文件 + 执照，回写并返回步骤与结果。""" steps: list[dict] = [] def _record(name: str, status: str, **kw) -> None: steps.append({"step": name, "status": status, **kw}) vat_reg_info_id = record.get("VATRegInfoID") country_code = record.get("Country") code = record.get("Code") enterprise_name = record.get("NameEng") batch_id = str(uuid.uuid4()) message_lines: list[str] = [] detail_lines: list[str] = [] final_level = "info" def _log(step: str, level: str, message: str | None = None, detail: str | None = None) -> None: nonlocal final_level if _LEVEL_ORDER.get(level, 0) > _LEVEL_ORDER.get(final_level, 0): final_level = level if message: message_lines.append(f"[{level}] {step}: {message}") else: message_lines.append(f"[{level}] {step}") if detail: detail_lines.append(detail) def _flush() -> None: insert_vat_generate_log( batch_id=batch_id, code=code, vat_reg_info_id=vat_reg_info_id, enterprise_name=enterprise_name, country=country_code, level=final_level, message="\n".join(message_lines) or None, detail="\n\n".join(detail_lines) or None, ) _log("入口", "info", f"code={code} name={enterprise_name} country={country_code}") if not country_code: logger.warning("无法识别国家: %s", record.get("Country")) _log("确定国家", "error", f"无法识别国家: {record.get('Country')}") _record("确定国家", "failed", error=f"无法识别国家: {record.get('Country')}") _flush() return {"status": "failed", "steps": steps} _record("确定国家", "success", country=country_code) templates = find_templates(country_code) if not templates: logger.warning("国家 %s 无模板", country_code) _log("查找模板", "error", f"国家 {country_code} 无模板") _record("查找模板", "failed", error=f"国家 {country_code} 无模板") _flush() return {"status": "failed", "steps": steps} logger.info("找到 %d 个模板", len(templates)) _log("查找模板", "info", f"国家 {country_code} 找到 {len(templates)} 个模板") _record("查找模板", "success", count=len(templates)) try: values = _build_values(record) enterprise_name_safe = safe_filename(enterprise_name or "out") out_dir = Path(settings.output_dir) / "generate" / enterprise_name_safe out_dir.mkdir(parents=True, exist_ok=True) images, image_errors = _prepare_images(files, out_dir) if image_errors: _log("下载附件", "error", "; ".join(m for m, _ in image_errors), "\n\n".join(d for _, d in image_errors if d)) logger.info("附件图片：营业执照 %d 张 / 身份证 %d 张", len(images.get("营业执照", [])), len(images.get("身份证", []))) ocr, ocr_errors = _ocr_idcard_fields(images.get("身份证"), PHOTO_ROTATION.get("身份证", 0)) if ocr_errors: _log("OCR", "warning", "; ".join(m for m, _ in ocr_errors), "\n\n".join(d for _, d in ocr_errors)) if ocr.get("法人性别"): values["法人性别"] = ocr["法人性别"] if not values.get("签发城市") and ocr.get("签发城市"): values["签发城市"] = ocr["签发城市"] pdf_paths, license_pdf, render_errors = render_templates(templates, values, images, out_dir) if render_errors: _log("渲染模板", "error", "; ".join(m for m, _ in render_errors), "\n\n".join(d for _, d in render_errors if d)) _record("渲染模板", "success", count=len(templates)) if not pdf_paths: raise AppError("所有模板渲染失败，无可用 PDF") # 合并 main_pdf = out_dir / f"{gen_file_prefix}{enterprise_name_safe}{gen_file_suffix}.pdf" main_pdf = merge_pdfs(pdf_paths, main_pdf) logger.info("合并PDF -> %s", main_pdf) _log("合并PDF", "info", str(main_pdf)) _record("合并PDF", "success", path=str(main_pdf)) # 上传主文件 oss_url = None if settings.cos_bucket: oss_url = upload_file(main_pdf) logger.info("上传OSS -> %s", oss_url) _log("上传OSS", "info", oss_url) _record("上传OSS", "success", url=oss_url) else: logger.warning("未配置 COS，跳过上传") _log("上传OSS", "warning", "未配置 COS，跳过上传") _record("上传OSS", "skipped", error="未配置 COS") # 执照 PDF 单独上传 OSS license_oss_url = None if settings.cos_bucket and license_pdf and "license" in NL_FILE_CATEGORIES: license_oss_url = upload_file(license_pdf) logger.info("上传执照PDF -> %s", license_oss_url) _log("上传执照PDF", "info", license_oss_url) _record("上传执照PDF", "success", url=license_oss_url) # 回写 _write_back_status(vat_reg_info_id, 6, update_vat_push_status, log=_log) if oss_url: _write_back_annex(vat_reg_info_id, main_pdf, oss_url, log=_log, file_category_id=NL_FILE_CATEGORIES["main"]) if license_oss_url: _write_back_annex( vat_reg_info_id, license_pdf, license_oss_url, log=_log, file_category_id=NL_FILE_CATEGORIES["license"], ) _log("生成文档", "info", "success") _flush() return {"status": "success", "oss_url": oss_url, "merged": str(main_pdf), "steps": steps} except Exception as exc: # noqa: BLE001 logger.exception("生成文档失败: %s", exc) _log("生成文档", "error", str(exc), traceback.format_exc()) _record("生成文档", "failed", error=str(exc)) _write_back_status(vat_reg_info_id, 7, update_vat_push_status, str(exc), log=_log) _flush() return {"status": "failed", "error": str(exc), "steps": steps} def generate_enterprise_documents(code: str) -> dict: """生成企业文档（NL API 入口）：按国家找模板 → 替换 → 合并 PDF → 上传 OSS，返回步骤与结果。""" logger.info("开始生成企业文档 code=%s", code) with SourceSessionLocal() as session: rows = get_enterprise_detail(session, code) if not rows: logger.warning("企业不存在或未关联附件: %s", code) return {"status": "failed", "steps": [{"step": "获取企业详情", "status": "failed", "error": "企业不存在或未关联附件"}]} return generate_nl_documents(rows[0], rows)