import json import logging import re import urllib.parse from typing import Any from common.esigntool.main import fill_in_template, create_by_file logger = logging.getLogger(__name__) 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) raise ContractBuildError( message=f"{contract_name}生成失败:e签宝返回缺少 fileDownloadUrl", raw={"contract_type": contract_type, "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) raise ContractBuildError( message=f"{contract_name}发起签署失败", 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