"""企业文档生成（共享底层）：字段映射 / 图片与 OCR / 模板渲染 / 回写 helper。""" from __future__ import annotations import json import re import tempfile import traceback from datetime import datetime from pathlib import Path from docx import Document from app.core.database import SourceSessionLocal from app.core.download import download_file from app.core.image import normalize_image_bytes, pdf_to_images from app.core.logger import get_logger from app.repositories.common_enterprise import insert_generated_file from app.repositories.fr_enterprise import get_epr_category_names from app.services.converter import convert_to_pdf from app.services.document import render_word from app.services.fr_annuaire import get_naf_code from app.services.renderers.pdf import render_pdf from app.services.renderers.xlsx import render_xlsx logger = get_logger(__name__) # 照片转正策略：仅按 EXIF 方向自动转正；无 EXIF 方向信息的照片（无法可靠自动判向）维持原样。 # 保留按文档类型配置固定顺时针角度的能力，默认均为 0（不额外旋转）。 PHOTO_ROTATION: dict[str, int] = {"营业执照": 0, "身份证": 0} # 电池分类占位符（FR 电池法模板用） _BATTERY_PLACEHOLDERS: tuple[str, ...] = ( "便携式电池", "轻型交通工具用电池", "工业电池", "启动、照明和点火电池", "电动汽车电池", ) # 电池分类映射：eprCategoryName → 占位符 key（恒等映射；如日后名称不一致仅改这里） _BATTERY_TYPE_MAP: dict[str, str] = { "便携式电池": "便携式电池", "轻型交通工具用电池": "轻型交通工具用电池", "工业电池": "工业电池", "启动、照明和点火电池": "启动、照明和点火电池", "电动汽车电池": "电动汽车电池", } # 性别编码/中文 → 英文（OCR 返回「男/女」，DB 列 LegalPersonGender 存「1/2」） _GENDER_EN: dict[str, str] = {"1": "Male", "2": "Female", "男": "Male", "女": "Female"} # 性别 → 法国称呼 MR/MS（法国电池法模板 Civility 用） _GENDER_FR: dict[str, str] = {"1": "MR", "2": "MS", "男": "MR", "女": "MS", "Male": "MR", "Female": "MS"} # 主要城市中英对照（OCR 识别出中文签发城市后转英文；可按需扩展） _CITY_CN2EN: dict[str, str] = { "杭州": "Hangzhou", "深圳": "Shenzhen", "广州": "Guangzhou", "北京": "Beijing", "上海": "Shanghai", "南京": "Nanjing", "苏州": "Suzhou", "成都": "Chengdu", "武汉": "Wuhan", "天津": "Tianjin", "重庆": "Chongqing", "宁波": "Ningbo", "厦门": "Xiamen", "青岛": "Qingdao", "东莞": "Dongguan", "佛山": "Foshan", "西安": "Xi'an", "长沙": "Changsha", "郑州": "Zhengzhou", "福州": "Fuzhou", "合肥": "Hefei", "济南": "Jinan", "大连": "Dalian", "珠海": "Zhuhai", "中山": "Zhongshan", "无锡": "Wuxi", "常州": "Changzhou", "温州": "Wenzhou", "嘉兴": "Jiaxing", "金华": "Jinhua", "绍兴": "Shaoxing", "台州": "Taizhou", "湖州": "Huzhou", "义乌": "Yiwu", } def _date_str(value) -> str | None: """DATE/DATETIME 转字符串（只取日期 YYYY-MM-DD）；None 或字符串原样返回。""" if value is None: return None if hasattr(value, "strftime"): return value.strftime("%Y-%m-%d") return str(value) def _gender_from_id(id_number: str) -> str | None: """18 位中国居民身份证：第 17 位奇数为男、偶数为女，返回英文 Male/Female。""" if id_number and len(id_number) == 18: try: digit = int(id_number[16]) except ValueError: return None return "Male" if digit % 2 else "Female" return None def _gender_en(value) -> str | None: """性别编码/中文 → 英文 Male/Female；无法识别返回 None。""" if value is None: return None return _GENDER_EN.get(str(value).strip()) def _gender_fr(value) -> str | None: """性别编码/中文/英文 → 法国称呼 MR/MS；无法识别返回 None。""" if value is None: return None return _GENDER_FR.get(str(value).strip()) _PHONE_SEPARATOR_RE = re.compile(r"[^0-9A-Za-z+]+") def _format_phone(value) -> str: """联系电话格式化：中间特殊字符（-、.、/、()、空格等）折叠为单个空格；保留数字、字母与开头的 +。""" phone = str(value).strip() if value else "" return _PHONE_SEPARATOR_RE.sub(" ", phone).strip() def _city_from_authority(authority: str) -> str: """从身份证「签发机关」提取城市级行政区划（如「杭州市公安局西湖分局」→「杭州市」）。""" if not authority: return "" # 城市紧邻「公安局」之前，其前可能带省级前缀（XX省 / XX自治区）： # 先用贪婪匹配吃掉省级前缀，再取「城市+公安局」（如「广西壮族自治区南宁市」→「南宁市」）。 m = re.search( r"(?:[一-龥]{2,}(?:省|自治区))?" r"([一-龥]{2,6}?(?:市|自治州|地区|盟))公安局", authority, ) if m: return m.group(1) # 兜底：无「公安局」锚点时取首个城市级行政区划 m = re.search(r"[一-龥]{2,6}?(?:市|自治州|地区|盟)", authority) return m.group(0) if m else "" def _city_en(cn_city: str) -> str: """中文城市 → 英文城市名（映射表）；未命中返回空串。""" if not cn_city: return "" for suffix in ("市", "自治州", "地区", "盟"): if cn_city.endswith(suffix): cn_city = cn_city[: -len(suffix)] break return _CITY_CN2EN.get(cn_city, "") def _split_foreign_name(full_name) -> tuple[str, str]: """把国外姓名拼音拆成 (姓, 名)，仅按西式规则（姓在名后）。 - "John Smith" -> ("Smith", "John") - "John Michael Smith" -> ("Smith", "John Michael") - "Smith, John" -> ("Smith", "John") # 逗号前是姓 - 空 / 单名 -> (原词, "") """ if not full_name: return "", "" name = str(full_name).strip() if not name: return "", "" if "," in name: parts = [p.strip() for p in name.split(",") if p.strip()] surname = parts[0] if parts else "" given = " ".join(parts[1:]) return surname, given words = name.split() if len(words) == 1: return words[0], "" return words[-1], " ".join(words[:-1]) # 方案二是否启用：True=按 eprCategory 查 EPRCategory 表；False=直接用 JSON 里的 eprCategoryName _USE_EPR_CATEGORY_DB = True def _parse_category_brand_json(value) -> list[dict]: """把 CategoryBrandJson 字段解析为 list（JSON 字符串或已是 list）。""" if value is None: return [] if isinstance(value, list): return [i for i in value if isinstance(i, dict)] if isinstance(value, str): value = value.strip() if not value: return [] try: parsed = json.loads(value) except (json.JSONDecodeError, TypeError): logger.warning("CategoryBrandJson 解析失败: %s", value) return [] return [i for i in parsed if isinstance(i, dict)] if isinstance(parsed, list) else [] return [] def _battery_type_values_from_names(items: list[dict]) -> dict: """方案一（现启用）：取每项 eprCategoryName 与 _BATTERY_TYPE_MAP 匹配，命中填 YES。""" matched: set[str] = set() for item in items: name = item.get("eprCategoryName") if name is None: continue value = _BATTERY_TYPE_MAP.get(str(name).strip()) if value: matched.add(value) return {k: ("YES" if k in matched else "") for k in _BATTERY_PLACEHOLDERS} def _battery_type_values_by_db(items: list[dict]) -> dict: """方案二（备用）：按 eprCategory 查 EPRCategory 表取名称，再匹配。""" ids = [str(i["eprCategory"]) for i in items if i.get("eprCategory")] matched: set[str] = set() if ids: with SourceSessionLocal() as session: rows = get_epr_category_names(session, ids) for row in rows: name = row.get("CategoryName") if name is None: continue value = _BATTERY_TYPE_MAP.get(str(name).strip()) if value: matched.add(value) return {k: ("YES" if k in matched else "") for k in _BATTERY_PLACEHOLDERS} def _battery_type_values(category_brand_json) -> dict: """根据 CategoryBrandJson 返回电池分类占位符值（命中填 YES，其余空白）。""" items = _parse_category_brand_json(category_brand_json) if _USE_EPR_CATEGORY_DB: return _battery_type_values_by_db(items) return _battery_type_values_from_names(items) # 法国国家名归一化集合（CompanyCountry 可能是 France / FR / 法国 等） _FRANCE_COUNTRY_VALUES = {"france", "fr", "fra", "法国", "法兰西"} def _is_france(country) -> bool: return str(country or "").strip().lower() in _FRANCE_COUNTRY_VALUES def _build_values(first: dict) -> dict: """把 DB 字段映射到模板占位符 key（【key】）。""" id_number = first.get("LegalPersonIdNumber") or "" id_start = _date_str(first.get("LegalPersonIdStartDate")) id_end = _date_str(first.get("LegalPersonIdEndDate")) validity = " - ".join(x for x in (id_start, id_end) if x) city_eng = first.get("CityEngName") or "" province_eng = first.get("CompanyAddressProvinceEn") or "" city_province = ", ".join(x for x in (city_eng, province_eng) if x) surname, given = _split_foreign_name(first.get("LegalPersonFullNamePinYin")) naf_code = "" if _is_france(first.get("CompanyCountry")): naf_code = get_naf_code(first.get("RegNumber")) or "" return { "营业执照号码": first.get("RegNumber"), "公司名字拼音": (first.get("NameEng") or "").upper(), "公司名字英文或拼音": first.get("NameEng"), "公司地址拼音或英文": (first.get("RegAddressEng") or "").upper(), "地址拼音-号码，街道": first.get("RegAddressEng"), "邮编": first.get("CompanyAddressPostcode"), "VAT税号": first.get("VATNumber"), "NAF Code": naf_code, "公司国家": first.get("CompanyCountry"), "公司国家en": first.get("CountryNameEn"), "城市，省份": (city_province or "").upper(), "法人名字拼音": first.get("LegalPersonFullNamePinYin") or first.get("LegalPersonName"), "法人签名": first.get("LegalPersonFullNamePinYin") or first.get("LegalPersonName"), "法人姓氏": (surname or "").upper(), "法人名称": (given or "").upper(), "法人身份证号": id_number, "身份证号码": id_number, "法人出生年月日": _date_str(first.get("LegalPersonBirthDate")), "法人出生年月日或者日月年": _date_str(first.get("LegalPersonBirthDate")), "法人邮箱": first.get("LegalPersonEmail"), # "身份证地址": first.get("LegalPersonIdCardAddress"), #使用拼音地址 "身份证地址": first.get("LegalPersonIdCardAddressEng"), "身份证地址拼音或英文": first.get("LegalPersonIdCardAddressEng") or first.get("LegalPersonIdCardAddress"), "法人性别": _gender_en(first.get("LegalPersonGender")) or _gender_from_id(id_number), "法人性别（法国）": _gender_fr(first.get("LegalPersonGender")) or _gender_fr(_gender_from_id(id_number)), "签发城市": first.get("LegalPersonCityEngName"), "有效期": validity, "联系电话": _format_phone(first.get("LegalPersonPhone")), "区号-电话号码": first.get("LegalPersonPhone"), "成立日月年或者年月日": _date_str(first.get("EstablishmentDate")), "签发日月年": _date_str(first.get("SignatureDate")), "城市拼音": first.get("CityEngName"), "签署日月年（可以是今日）": datetime.now().strftime("%Y-%m-%d"), "卖家ID/TOKEN ID": first.get("SellerID") or "", "经营的产品名字英文": first.get("ProductRange") or "", "日期": datetime.now().strftime("%Y-%m-%d"), **_battery_type_values(first.get("CategoryBrandJson")), } def _classify_file_role(role: str | None, name: str) -> str | None: """按 DB 角色优先、文件名兜底，返回 license/id_front/id_back/None。""" if role in ("license", "id_front", "id_back"): return role name = name or "" if "营业执照" in name or "business" in name.lower() or "license" in name.lower(): return "license" if "身份证" in name: return "id_front" if "正" in name else "id_back" return None def _prepare_images(rows: list[dict], out_dir: Path) -> tuple[dict, list[tuple[str, str]]]: """下载证件照/营业执照并转成图片。 返回 (images, errors)：images 为 {'营业执照': [图...], '身份证': [正, 反]}； errors 为单个附件失败的 (描述, 堆栈) 列表（失败跳过继续，不中断）。 """ images: dict[str, list[Path]] = {"营业执照": [], "身份证": []} errors: list[tuple[str, str]] = [] for r in rows: name = r["F_FileName"] or "" url = r["F_FilePath"] ftype = (r["F_FileType"] or "").lower() if not url: continue try: local = out_dir / name download_file(url, local) imgs = pdf_to_images(local, out_dir) if ftype == "pdf" else [local] img = imgs[0] if imgs else None if not img: errors.append((f"{name}: 转图片失败（无图片）", "")) continue except Exception as exc: # noqa: BLE001 errors.append((f"{name}: {exc}", traceback.format_exc())) continue role = _classify_file_role(r.get("file_role"), name) logger.debug("下载并分类附件 %s -> %s", name, role or "忽略") if role == "license": images["营业执照"].append(img) elif role == "id_front": images["身份证"].insert(0, img) elif role == "id_back": images["身份证"].append(img) return images, errors def _replace_images(docx_path: Path, image_paths: list[Path], rotation: int = 0) -> None: """把 docx 前 N 个内嵌图按顺序替换为给定图片（保持原位置与尺寸）。 图片先经 normalize_image_bytes 归一化（EXIF 转正 + 可选旋转 + 转 PNG）。 """ from docx.oxml.ns import qn doc = Document(str(docx_path)) for i, img in enumerate(image_paths): if i >= len(doc.inline_shapes): break blip = doc.inline_shapes[i]._inline.graphic.graphicData.pic.blipFill.blip r_id = blip.get(qn("r:embed")) doc.part.related_parts[r_id]._blob = normalize_image_bytes(img, rotation) doc.save(str(docx_path)) def _ocr_idcard_fields(id_images: list[Path], rotation: int) -> tuple[dict, list[tuple[str, str]]]: """对身份证正/反面做 OCR，返回 ({签发城市, 法人性别}, errors)；失败或未配置时不阻断生成。""" if not id_images: return {}, [] from app.services.tencent_ocr import recognize_idcard front = id_images[0] if len(id_images) > 0 else None back = id_images[1] if len(id_images) > 1 else None def _recognize(img: Path, side: str) -> dict: # 身份证原图可能横向，先转正再识别 with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as fh: fh.write(normalize_image_bytes(img, rotation)) tmp = fh.name try: return recognize_idcard(tmp, side) finally: try: Path(tmp).unlink() except OSError: pass result: dict = {} errors: list[tuple[str, str]] = [] try: if front: sex = (_recognize(front, "FRONT").get("sex") or "") if sex: result["法人性别"] = _gender_en(sex) if back: authority = (_recognize(back, "BACK").get("authority") or "") city = _city_en(_city_from_authority(authority)) if city: result["签发城市"] = city except Exception as exc: # noqa: BLE001 errors.append((f"身份证 OCR: {exc}", traceback.format_exc())) logger.warning("身份证 OCR 识别失败，签发城市/性别回退: %s", exc) logger.debug("身份证 OCR 完成: %s", result) return result, errors def render_templates(templates, values: dict, images: dict, out_dir: Path) -> tuple[list[Path], Path | None, list[tuple[str, str]]]: """渲染每个模板：PDF 表单填字段；DOCX 渲染 + 图片替换后转 PDF；单模板失败跳过继续。 返回 (pdf_paths, license_pdf, render_errors)。 """ pdf_paths: list[Path] = [] license_pdf: Path | None = None render_errors: list[tuple[str, str]] = [] for tpl in templates: logger.info("渲染模板 %s", tpl.name) try: ext = tpl.suffix.lower() if ext == ".pdf": out_pdf = out_dir / f"{tpl.stem}.pdf" render_pdf(tpl, values, out_pdf) pdf_paths.append(out_pdf) elif ext in (".xlsx", ".xlsm"): out_xlsx = out_dir / f"{tpl.stem}{ext}" render_xlsx(tpl, values, out_xlsx) out_pdf = convert_to_pdf(out_xlsx, out_dir / f"{tpl.stem}.pdf") # xlsx 保留原格式（.xlsx）上传，不转 PDF；下游按源文件后缀区分格式 # pdf_paths.append(out_pdf) pdf_paths.append(out_xlsx) else: out_docx = out_dir / f"{tpl.stem}.docx" render_word(tpl, values, out_docx, to_pdf=False) if "执照" in tpl.name and images.get("营业执照"): _replace_images(out_docx, images["营业执照"], PHOTO_ROTATION.get("营业执照", 0)) logger.debug("替换营业执照照片 x%d", len(images["营业执照"])) if "身份证" in tpl.name and images.get("身份证"): _replace_images(out_docx, images["身份证"], PHOTO_ROTATION.get("身份证", 0)) logger.debug("替换身份证照片 x%d", len(images["身份证"])) out_pdf = convert_to_pdf(out_docx, out_dir / f"{tpl.stem}.pdf") pdf_paths.append(out_pdf) if "执照" in tpl.name: license_pdf = out_pdf except Exception as exc: # noqa: BLE001 logger.exception("渲染模板失败 %s: %s", tpl.name, exc) render_errors.append((f"{tpl.name}: {exc}", traceback.format_exc())) return pdf_paths, license_pdf, render_errors def _write_back_status(vat_reg_info_id: str | None, push_status: int, update_status, error_msg: str | None = None, log=None) -> None: """回写生成状态（NL=VATRegInfo，FR=EPRRegInfo）；回写失败仅记日志（可传 log 回调落表），不阻断主流程。""" if not vat_reg_info_id: logger.warning("无 VATRegInfoID，跳过状态回写") return try: with SourceSessionLocal() as session: update_status(session, vat_reg_info_id, push_status, error_msg) logger.info("回写推送状态 VATRegInfoID=%s push_status=%d", vat_reg_info_id, push_status) except Exception as exc: # noqa: BLE001 logger.error("回写推送状态失败: %s", exc) if log: log("回写状态", "error", f"回写推送状态失败: {exc}") def _write_back_annex( vat_reg_info_id: str | None, pdf_file: Path, oss_url: str, file_category_id: str, log=None, ) -> None: """把生成 PDF 写回 Base_AnnexesFile 附件表；失败仅记日志（可传 log 回调落表），不阻断主流程。""" if not vat_reg_info_id: logger.warning("无 VATRegInfoID，跳过附件写回") return try: file_size = pdf_file.stat().st_size with SourceSessionLocal() as session: insert_generated_file( session, vat_reg_info_id=vat_reg_info_id, file_name=pdf_file.stem, file_size=file_size, file_url=oss_url, file_ext=pdf_file.suffix, file_type=pdf_file.suffix.lstrip("."), file_category_id=file_category_id, ) logger.info("写回生成文件附件 VATRegInfoID=%s -> %s", vat_reg_info_id, oss_url) except Exception as exc: # noqa: BLE001 logger.error("写回生成文件附件失败: %s", exc) if log: log("回写附件", "error", f"写回生成文件附件失败: {exc}")