alien_py_cloud/alien_contract/infrastructure/esign/main.py

import requests
import json
from datetime import datetime
from alien_contract.infrastructure.esign.esign_config import Config
from alien_contract.infrastructure.esign.esign_algorithm import buildSignJsonHeader

cfg = Config()

SIGN_FLOW_TITLES = {
    "store_agreement": "商家入驻U店平台协议签署",
    "lawyer_agreement": "律所入驻U店平台协议签署",
    "alipay_auth": "支付宝授权函签署",
    "wechat_pay_commitment": "微信支付承诺函签署",
}

SIGN_POSITIONS = {
    "store_agreement": {"alien":{"page": 7, "x": 294, "y": 668}, "org": {"page": 7, "x": 114, "y": 666}},
    "lawyer_agreement": {"alien": {"page": 6, "x": 336, "y": 418}, "org": {"page": 6, "x": 145, "y": 418}},
    "alipay_auth": {"alien": {"page": 1, "x": 535, "y": 555}, "org": {"page": 1, "x": 535, "y": 431}},
    "wechat_pay_commitment": {"org": {"page": 1, "x": 535, "y": 515}},
}

TEMPLATE_COMPONENT_VALUES = {
    "store_agreement": lambda store_name: [
        {"componentKey": "store_name", "componentValue": store_name},
        {"componentKey": "one_name", "componentValue": store_name},
        {"componentKey": "date", "componentValue": datetime.now().strftime("%Y年%m月%d日")},
        {"componentKey": "alien_name", "componentValue": "爱丽恩严（大连）商务科技有限公司"},
    ],
    "lawyer_agreement": lambda store_name: [
        {"componentKey": "alien_name", "componentValue": "爱丽恩严（大连）商务科技有限公司"},
        {"componentKey": "alien_name_2", "componentValue": "爱丽恩严（大连）商务科技有限公司"},
        {"componentKey": "law_firm_name", "componentValue": store_name},
        {"componentKey": "law_firm_name_2", "componentValue": store_name},
        {"componentKey": "signing_date", "componentValue": datetime.now().strftime("%Y年%m月%d日")},
    ],
    "alipay_auth": lambda store_name: [],
    "wechat_pay_commitment": lambda store_name: [],
}


def _request(method: str, url: str, body: dict | None = None) -> str:
    if not url.startswith("http"):
        url = f"{cfg.host}{url}"
    headers = buildSignJsonHeader(cfg.appId, cfg.scert, method, url, body)
    if body is None:
        response = requests.request(method, url, headers=headers)
    else:
        payload = json.dumps(body, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
        response = requests.request(method, url, data=payload, headers=headers)
    if response.text:
        return response.text
    fallback = {"success": False, "http_status": response.status_code, "message": "EMPTY_RESPONSE", "request_id": response.headers.get("x-ts-request-id")}
    return json.dumps(fallback, ensure_ascii=False)


def apply_platform_seal(sign_flow_id):
    sign_url = "/v3/sign-flow/{}/sign-fields/auto-sign".format(sign_flow_id)
    return _request("POST", sign_url, {})


def _build_alien_signer(file_id: str, position: dict) -> dict:
    """构建平台方（爱丽恩严）签署人配置"""
    return {
        "signConfig": {"signOrder": 1},
        "signerType": 1,
        "signFields": [{
            "customBizNum": "9527",
            "fileId": file_id,
            "normalSignFieldConfig": {
                "autoSign": True,
                "signFieldStyle": 1,
                "signFieldPosition": {
                    "positionPage": str(position["page"]),
                    "positionX": position["x"],
                    "positionY": position["y"]
                }
            }
        }]
    }


def _build_org_signer(file_id: str, position: dict, signer_name: str, signer_id_num: str, psn_account: str, psn_name: str) -> dict:
    return {
        "signConfig": {"signOrder": 2, "forcedReadingTime": 10},
        "signerType": 1,
        "orgSignerInfo": {
            "orgName": signer_name,
            "orgInfo": {
                "orgIDCardNum": signer_id_num,
                "orgIDCardType": "CRED_ORG_USCC"
            },
            "transactorInfo": {
                "psnAccount": psn_account,
                "psnInfo": {"psnName": psn_name}
            }
        },
        "signFields": [
            {
                "customBizNum": "自定义编码001",
                "fileId": file_id,
                "normalSignFieldConfig": {
                    "signFieldStyle": 1,
                    "signFieldPosition": {
                        "positionPage": str(position["page"]),
                        "positionX": position["x"],
                        "positionY": position["y"],
                    },
                },
            }
        ],
    }


def get_auth_flow_id(org_name: str, org_id_card_num: str, legal_rep_name: str, legal_rep_id_card_num: str):
    body = {
        "orgAuthConfig": {
            "orgName": org_name,
            "orgInfo": {
                "orgIDCardNum": org_id_card_num,
                "legalRepName": legal_rep_name,
                "legalRepIDCardNum": legal_rep_id_card_num,
            },
        },
        "transactorUseSeal": True,
    }
    return _request("POST", "/v3/org-auth-url", body)


def get_template_detail():
    return _request("GET", f"/v3/doc-templates/{cfg.templates_id}")


def create_by_file(
    file_id,
    file_name,
    signer_name,
    signer_id_num,
    psn_account,
    psn_name,
    contract_type: str = "store_agreement",
):
    sign_flow_url = "/v3/sign-flow/create-by-file"
    title = SIGN_FLOW_TITLES.get(contract_type, SIGN_FLOW_TITLES["store_agreement"])
    positions = SIGN_POSITIONS.get(contract_type, SIGN_POSITIONS["store_agreement"])

    signers = []
    if positions.get("alien"):
        signers.append(_build_alien_signer(file_id, positions["alien"]))
    if positions.get("org"):
        signers.append(_build_org_signer(file_id, positions["org"], signer_name, signer_id_num, psn_account, psn_name))


    body = {
        "docs": [{"fileId": file_id, "fileName": f"{file_name}.pdf"}],
        "signFlowConfig": {
            "signFlowTitle": title,
            "autoFinish": True,
            "noticeConfig": {"noticeTypes": "1,2"},
            "notifyUrl": cfg.callback_url,
            "noticeDeveloperUrl": cfg.developer_callback_url,
            "redirectConfig": {"redirectUrl": cfg.redirect_url},
        },
        "signers": signers,
    }
    return _request("POST", sign_flow_url, body)


def sign_url(sign_flow_id, account):
    sign_url_info = "/v3/sign-flow/{}/sign-url".format(sign_flow_id)
    body = {
        "signFlowId": sign_flow_id,
        "operator": {"psnAccount": account},
        "needLogin": False,
        "urlType": 2,
        "clientType": "ALL",
    }
    return _request("POST", sign_url_info, body)


def file_download_url(sign_flow_id):
    sign_download = "/v3/sign-flow/{}/file-download-url".format(sign_flow_id)
    return _request("POST", sign_download, {"urlAvailableDate": "3600"})


def fill_in_template(store_name: str, contract_type: str = "store_agreement"):
    template_id = cfg.templates_map.get(contract_type, cfg.templates_map["store_agreement"])
    template_file_name = cfg.template_file_names.get(contract_type, cfg.template_file_names["store_agreement"])
    fill_in_data = "/v3/files/create-by-doc-template"
    component_builder = TEMPLATE_COMPONENT_VALUES.get(contract_type, TEMPLATE_COMPONENT_VALUES["store_agreement"])
    body = {
        "docTemplateId": template_id,
        "fileName": template_file_name,
        "components": component_builder(store_name),
    }
    return _request("POST", fill_in_data, body)


def get_contract_detail(file_id: str):
    detail_url = f"/v3/files/{file_id}"
    return _request("GET", detail_url)