alien_py_cloud/alien_contract/infrastructure/esign/contract_builder.py

import json
import logging
import re
import urllib.parse
from typing import Any

from alien_contract.infrastructure.esign.main import SIGN_POSITIONS, apply_platform_seal, 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 _is_esign_success(resp: dict[str, Any]) -> bool:
    code = resp.get("code")
    return code in (0, "0") or resp.get("success") is True


def _apply_platform_seal_if_required(contract_type: str, contract_name: str, sign_flow_id: str) -> None:
    positions = SIGN_POSITIONS.get(contract_type, {})
    if not positions.get("alien"):
        return

    seal_resp = apply_platform_seal(sign_flow_id)
    try:
        seal_json = json.loads(seal_resp)
    except json.JSONDecodeError:
        logger.error("apply_platform_seal non-json resp contract_type=%s: %s", contract_type, seal_resp)
        raise ContractBuildError(
            message=f"{contract_name}平台自动盖章失败：e签宝返回非 JSON",
            raw={"contract_type": contract_type, "sign_flow_id": sign_flow_id, "resp": seal_resp},
        )

    if not isinstance(seal_json, dict) or not _is_esign_success(seal_json):
        logger.error("apply_platform_seal failed contract_type=%s: %s", contract_type, seal_json)
        raise ContractBuildError(
            message=f"{contract_name}平台自动盖章失败",
            raw={"contract_type": contract_type, "sign_flow_id": sign_flow_id, "resp": seal_json},
        )


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},
            )

        _apply_platform_seal_if_required(contract_type, contract_name, sign_id)

        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