alien_py_cloud/common/esigntool/contract_builder.py

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