import json import logging import re import urllib.parse from typing import Any from alien_contract.infrastructure.esign.main import fill_in_template, create_by_file from alien_contract.infrastructure.esign.esign_config import Config logger = logging.getLogger(__name__) cfg = Config() class ContractBuildError(Exception): def __init__(self, message: str, raw: Any): super().__init__(message) self.message = message self.raw = raw def build_contract_items( configs: list[tuple[str, str, int]], template_name: str, signer_name: str, signer_id_num: str, psn_account: str, psn_name: str, ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] for contract_type, contract_name, is_master in configs: res_text = fill_in_template(template_name, contract_type=contract_type) try: res_data = json.loads(res_text) except json.JSONDecodeError: logger.error("fill_in_template non-json resp contract_type=%s: %s", contract_type, res_text) raise ContractBuildError( message=f"{contract_name}生成失败:e签宝返回非 JSON", raw={"contract_type": contract_type, "resp": res_text}, ) try: contract_url = res_data["data"]["fileDownloadUrl"] file_id = res_data["data"]["fileId"] m = re.search(r"/([^/]+)\.pdf", contract_url) if m: encoded_name = m.group(1) file_name = urllib.parse.unquote(encoded_name) else: file_name = f"{contract_type}.pdf" except Exception: logger.error("fill_in_template missing fileDownloadUrl contract_type=%s: %s", contract_type, res_data) template_id = cfg.templates_map.get(contract_type) code = res_data.get("code") if isinstance(res_data, dict) else None if code == 404: message = f"{contract_name}生成失败:模板不存在或无权限(template_id={template_id})" elif code == 401: message = f"{contract_name}生成失败:签名校验失败(template_id={template_id})" else: message = f"{contract_name}生成失败:e签宝返回缺少 fileDownloadUrl" raise ContractBuildError( message=message, raw={"contract_type": contract_type, "template_id": template_id, "resp": res_data}, ) sign_data = create_by_file( file_id, file_name, signer_name, signer_id_num, psn_account, psn_name, contract_type=contract_type, ) try: sign_json = json.loads(sign_data) except json.JSONDecodeError: logger.error("create_by_file non-json resp contract_type=%s: %s", contract_type, sign_data) raise ContractBuildError( message=f"{contract_name}发起签署失败:e签宝返回非 JSON", raw={"contract_type": contract_type, "resp": sign_data}, ) if not sign_json.get("data"): logger.error("create_by_file failed or missing data contract_type=%s: %s", contract_type, sign_json) resp_code = sign_json.get("code") if isinstance(sign_json, dict) else None resp_msg = sign_json.get("message") if isinstance(sign_json, dict) else None if resp_code == 1437328: message = f"{contract_name}发起签署失败:手机号与真实姓名不匹配,请确认联系人实名信息" elif resp_code == 1435002: message = f"{contract_name}发起签署失败:签署参数不合法({resp_msg})" elif resp_code == 1437306: message = f"{contract_name}发起签署失败:签章定位页超出文档页数,请检查模板页码配置" elif resp_code == 1435011: message = f"{contract_name}发起签署失败:签署文件或签署人为空,请检查签章坐标与签署方配置" elif resp_msg == "EMPTY_RESPONSE": message = f"{contract_name}发起签署失败:e签宝返回空响应,请检查签署参数与账号配置" else: message = f"{contract_name}发起签署失败" raise ContractBuildError( message=message, raw={"contract_type": contract_type, "resp": sign_json}, ) sign_id = sign_json["data"].get("signFlowId") if not sign_id: logger.error("create_by_file missing signFlowId contract_type=%s: %s", contract_type, sign_json) raise ContractBuildError( message=f"{contract_name}发起签署失败:e签宝返回缺少 signFlowId", raw={"contract_type": contract_type, "resp": sign_json}, ) items.append( { "contract_type": contract_type, "contract_name": contract_name, "contract_url": contract_url, "file_name": file_name, "file_id": file_id, "status": 0, "sign_flow_id": sign_id, "sign_url": "", "signing_time": "", "effective_time": "", "expiry_time": "", "contract_download_url": "", "is_master": is_master, } ) return items