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)