import json import time import pymssql import urllib.request import urllib.error from datetime import datetime, timedelta, date # API 任务签名文件 fileUrl 为相对路径时拼接的 OSS 域名（与 es_haiya_epr config source_oss_base_url 默认值一致） SOURCE_OSS_BASE_URL = 'https://usaeu-1259285998.cos.ap-guangzhou.myqcloud.com' # API流程结果通知接口（字段检查失败时把错误信息发给调用方；URL 由 RPA 程序固定配置，可通过参数覆盖； # 默认值与 PHP .env ES_HAGUE_RESULT_CALLBACK_URL / es_haiya 同为 test-cloud 回调地址） DEFAULT_CALLBACK_URL = 'https://test-cloud.usaeu.com/prod-api/delivery/rpa/callback' # HTTP 超时秒数，与 PHP AsyncResultNotifier 默认一致 CALLBACK_TIMEOUT = 10 # 发送失败重试次数（总尝试 = 1+重试次数，0 只发一次），与 PHP AsyncResultNotifier / rpa_030_save 默认值一致 CALLBACK_MAX_RETRIES = 3 # 每次重试间隔秒数 CALLBACK_RETRY_DELAY = 5 def post_json(url, payload, timeout=CALLBACK_TIMEOUT): """ POST JSON 请求 参数: url: 请求地址 payload: 请求体（dict，将被 JSON 编码） timeout: 超时秒数 返回: dict: {success, http_code, response, error} """ data = json.dumps(payload, ensure_ascii=False).encode('utf-8') req = urllib.request.Request( url, data=data, headers={'Content-Type': 'application/json; charset=utf-8'}, method='POST' ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode('utf-8', errors='replace') success = 200 <= resp.status < 300 return { 'success': success, 'http_code': resp.status, 'response': body, 'error': None if success else f'HTTP {resp.status}', } except urllib.error.HTTPError as e: body = e.read().decode('utf-8', errors='replace') return { 'success': False, 'http_code': e.code, 'response': body, 'error': f'HTTP {e.code}', } except Exception as e: return { 'success': False, 'http_code': None, 'response': None, 'error': str(e), } def parse_biz_param_from_row(result): """ 从查询行 TaskData JSON 中取 bizParam（受理时原样回传的业务参数） 参数: result: 查询结果字典（含 TaskData 字段） 返回: dict 或 None """ try: task_data = json.loads(result.get('TaskData')) if result and result.get('TaskData') else {} if isinstance(task_data, dict): return task_data.get('bizParam') except (ValueError, TypeError): pass return None def notify_api_result(task_id, callback_url, success, error='', biz_param=None): """ 调用异步结果通知接口（API流程专用），把字段检查失败等错误信息发给调用方（发送失败自动重试） 请求体信封与 PHP AsyncResultNotifier 一致：{code, msg, ProcessMode, data, bizParam} - 失败: code=500, msg=错误原因, data=null（非 200 时 data 统一为 null，错误原因放 msg， 对齐 UNIFIED_API_DESIGN §13.2 失败回调格式） 重试: CALLBACK_MAX_RETRIES（总尝试=1+重试次数）+ CALLBACK_RETRY_DELAY 间隔，与 PHP 侧/ rpa_030_save 一致；2xx 视为成功，不捕获对方响应体中的业务失败。 本脚本无 Notify* 落库（字段检查失败已置 pdf_result=3 在任务行可见；落库仅在 rpa_030_save 侧实现， es_haiya 取数脚本本无通知，此为 epr 增强） 参数: task_id: 任务ID callback_url: 结果通知接口地址（为空时用 DEFAULT_CALLBACK_URL） success: 是否成功（get 脚本仅失败场景调用） error: 失败原因 biz_param: 受理时原样回传的业务参数 返回: dict: {success, status, attempts, http_code, response, error} """ url = callback_url or DEFAULT_CALLBACK_URL if success: data = { 'task_id': task_id, 'status': 'success', } msg = 'success' else: # 非 200：data 统一为 null，错误原因放 msg data = None msg = error or '未知错误' payload = { 'code': 200 if success else 500, 'msg': msg, 'ProcessMode': 'async', 'data': data, 'bizParam': biz_param if biz_param else None, } result = { 'success': False, 'status': 'failed', 'attempts': 0, 'http_code': None, 'response': None, 'error': None, } total_attempts = CALLBACK_MAX_RETRIES + 1 for attempt in range(1, total_attempts + 1): send_result = post_json(url, payload) result['attempts'] = attempt result['http_code'] = send_result['http_code'] result['response'] = send_result['response'] result['error'] = send_result['error'] if send_result['success']: result['success'] = True result['status'] = 'success' print(f"结果通知接口调用成功: task_id={task_id}, " f"attempt={attempt}, http_code={send_result['http_code']}") return result print(f"结果通知接口调用失败: task_id={task_id}, " f"attempt={attempt}/{total_attempts}, " f"http_code={send_result['http_code']}, error={send_result['error']}") if attempt < total_attempts and CALLBACK_RETRY_DELAY > 0: time.sleep(CALLBACK_RETRY_DELAY) return result def process_datetime_fields(result): """ 处理结果中的datetime字段，将datetime对象转换为字符串 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ for key, value in result.items(): if isinstance(value, datetime): result[key] = value.strftime('%Y-%m-%d %H:%M:%S') elif isinstance(value, date): # 新增：处理 date 类型 result[key] = value.strftime('%Y-%m-%d') def process_legal_person_birth_date(result): val = result.get('LegalPersonBirthDate') if val: # 兼容 datetime.date、datetime.datetime、字符串三种情况 if isinstance(val, (datetime, date)): result['LegalPersonBirthDate_Y'] = val.year result['LegalPersonBirthDate_M'] = val.month result['LegalPersonBirthDate_D'] = val.day elif isinstance(val, str): try: birth_date = datetime.strptime(val[:19], '%Y-%m-%d %H:%M:%S') except ValueError: try: birth_date = datetime.strptime(val[:10], '%Y-%m-%d') except ValueError: birth_date = None if birth_date: result['LegalPersonBirthDate_Y'] = birth_date.year result['LegalPersonBirthDate_M'] = birth_date.month result['LegalPersonBirthDate_D'] = birth_date.day else: result['LegalPersonBirthDate_Y'] = '' result['LegalPersonBirthDate_M'] = '' result['LegalPersonBirthDate_D'] = '' else: result['LegalPersonBirthDate_Y'] = '' result['LegalPersonBirthDate_M'] = '' result['LegalPersonBirthDate_D'] = '' else: result['LegalPersonBirthDate_Y'] = '' result['LegalPersonBirthDate_M'] = '' result['LegalPersonBirthDate_D'] = '' del result['LegalPersonBirthDate'] def process_legal_person_pinyin_name(result): """ 处理LegalPersonFullNamePinYin字段，分解为姓和名，删除原字段 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonFullNamePinYin'): pinyin_name = result['LegalPersonFullNamePinYin'].strip().upper() parts = pinyin_name.split() if len(parts) >= 2: # 最后一个部分为姓，前面的为名 result['LegalPersonFullNamePinYin_X'] = parts[-1] result['LegalPersonFullNamePinYin_M'] = ''.join(parts[:-1]) elif len(parts) == 1: # 只有一个部分时，作为姓处理 result['LegalPersonFullNamePinYin_X'] = parts[0] result['LegalPersonFullNamePinYin_M'] = '' else: # 空字符串处理 result['LegalPersonFullNamePinYin_X'] = '' result['LegalPersonFullNamePinYin_M'] = '' # 删除原字段 del result['LegalPersonFullNamePinYin'] def process_legal_person_city_english_name(result): """ 处理LegalPersonCityEngName字段，将字符全部变为大写，去除中间的空格，去掉最后的shi拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonCityEngName'): city_name = result['LegalPersonCityEngName'].strip() # 转换为大写 city_name = city_name.upper() # 去除中间的空格 city_name = city_name.replace(' ', '') # 去掉最后的shi拼音 if city_name.endswith('SHI'): city_name = city_name[:-3] # 直接在原字段上重新赋值 result['LegalPersonCityEngName'] = city_name def process_legal_person_address_province_english_name(result): """ 处理LegalPersonAddressProvinceEn字段，将字符全部变为大写，去除中间的空格，去掉最后的sheng拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonAddressProvinceEn'): province_name = result['LegalPersonAddressProvinceEn'].strip() # 转换为大写 province_name = province_name.upper() # 去除中间的空格 province_name = province_name.replace(' ', '') # 去掉最后的sheng拼音 if province_name.endswith('SHENG'): province_name = province_name[:-5] # 直接在原字段上重新赋值 result['LegalPersonAddressProvinceEn'] = province_name def process_legal_person_country(result): """ 处理LegalPersonCountry字段，将字符全部变为大写 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonCountry'): # 直接在原字段上重新赋值 result['LegalPersonCountry'] = result['LegalPersonCountry'].strip().upper() if result.get('CompanyCountry'): # 直接在原字段上重新赋值 result['CompanyCountry'] = result['CompanyCountry'].strip().upper() def process_city_english_name(result): """ 处理CityEngName字段，将字符全部变为大写，去除中间的空格，去掉最后的shi拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('CityEngName'): city_name = result['CityEngName'].strip() # 转换为大写 city_name = city_name.upper() # 去除中间的空格 city_name = city_name.replace(' ', '') # 去掉最后的shi拼音 if city_name.endswith('SHI'): city_name = city_name[:-3] # 直接在原字段上重新赋值 result['CityEngName'] = city_name def process_company_address_province_english_name(result): """ 处理CompanyAddressProvinceEn字段，将字符全部变为大写，去除中间的空格，去掉最后的sheng拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('CompanyAddressProvinceEn'): province_name = result['CompanyAddressProvinceEn'].strip() # 转换为大写 province_name = province_name.upper() # 去除中间的空格 province_name = province_name.replace(' ', '') # 去掉最后的sheng拼音 if province_name.endswith('SHENG'): province_name = province_name[:-5] # 直接在原字段上重新赋值 result['CompanyAddressProvinceEn'] = province_name def process_ar(result): """ 处理AR字段（授权机构模板选择），归一化为大写 M/O 030 文件按 AR 选择模板：M=原 mokj 模板，O=Onesea 模板（RPA 030 程序据此选模板） 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ ar_value = result.get('AR') if ar_value is not None: result['AR'] = str(ar_value).strip().upper() def process_id_number_and_passport(result): """ 处理身份证号和护照号码字段 LegalPersonIDCardType为IDCard时，LegalPersonIdNumber值不变，LegalPersonPassportNumber为空 LegalPersonIDCardType为其他值时，LegalPersonPassportNumber为LegalPersonIdNumber的值，LegalPersonIdNumber的值为空 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ id_card_type = result.get('LegalPersonIDCardType', '').strip() id_number = result.get('LegalPersonIdNumber', '') if id_card_type == 'IDCard': # 身份证类型，保持身份证号不变，护照号为空 result['LegalPersonIdNumber'] = id_number result['LegalPersonPassportNumber'] = '' else: # 其他类型，将身份证号的值移到护照号，身份证号设为空 result['LegalPersonPassportNumber'] = id_number result['LegalPersonIdNumber'] = '' def process_current_date(result): """ 处理当前日期，计算当天的年月日并保存到各个字段 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ current_date = datetime.now() result['current_y'] = current_date.year result['current_m'] = current_date.month result['current_d'] = current_date.day def process_legal_signed_file(result, ip, port, username, password, vat_database): """ 处理LegalSignedFile字段，替换为可下载的完整URL API 任务（DataSource='api'）：LegalSignedFile 为 [{"fileUrl":...,"fileName":...}] JSON 数组， 取 fileUrl（相对路径拼 SOURCE_OSS_BASE_URL），不查 vat_db； source 任务：LegalSignedFile 为 vat_db 附件表 F_Id，查 Base_AnnexesFile 取文件路径并替换域名前缀。 参数: result: 查询结果字典 ip: 数据库IP地址 port: 数据库端口 username: 数据库用户名 password: 数据库密码 vat_database: vat_db 数据库名（source 任务查签名附件用） 返回: None: 直接修改原字典 """ if not result.get('LegalSignedFile'): result['LegalSignedFile'] = '' return is_api_task = (result.get('DataSource') or '').strip().lower() == 'api' if is_api_task: # API 任务：解析 JSON 数组取 fileUrl；相对路径拼 OSS 域名，完整 URL 原样使用 try: items = json.loads(result['LegalSignedFile']) file_url = items[0].get('fileUrl', '') if isinstance(items, list) and items else '' except (ValueError, TypeError): file_url = '' if file_url: if file_url.startswith('http://') or file_url.startswith('https://'): result['LegalSignedFile'] = file_url else: result['LegalSignedFile'] = SOURCE_OSS_BASE_URL.rstrip('/') + '/' + file_url.lstrip('/') else: result['LegalSignedFile'] = '' return # source 任务：查 vat_db 附件表 conn = None cursor = None try: conn = pymssql.connect( server=ip, port=port, user=username, password=password, database=vat_database, charset='utf8' ) cursor = conn.cursor() # 查询文件路径 query_sql = """ SELECT F_FilePath FROM Base_AnnexesFile WHERE F_Id = %s """ cursor.execute(query_sql, (result['LegalSignedFile'],)) file_result = cursor.fetchone() if file_result and file_result[0]: file_path = file_result[0] # 替换字符串路径 processed_path = file_path.replace('G:/fileAnnexes', 'https://file.usaeu.com') # 直接在原字段上重新赋值，删除原字段值 result['LegalSignedFile'] = processed_path else: # 附件记录不存在，置空（避免把 F_Id 原样交给 RPA） print(f"LegalSignedFile 附件记录未找到(F_Id={result['LegalSignedFile']})，置空") result['LegalSignedFile'] = '' except Exception as e: print(f"处理LegalSignedFile时发生错误: {e}") # 出错时设置为空字符串 result['LegalSignedFile'] = '' finally: # 关闭游标和连接 if cursor: cursor.close() if conn: conn.close() def process_result_fields(result, ip=None, port=None, username=None, password=None, vat_database='vat_db'): """ 处理查询结果中的各种字段 参数: result: 查询结果字典 ip: 数据库IP地址 (用于LegalSignedFile处理) port: 数据库端口 (用于LegalSignedFile处理) username: 数据库用户名 (用于LegalSignedFile处理) password: 数据库密码 (用于LegalSignedFile处理) vat_database: vat_db 数据库名 (用于LegalSignedFile处理；默认 'vat_db' 与原脚本行为一致) 返回: None: 直接修改原字典 """ process_datetime_fields(result) process_legal_person_birth_date(result) # 分解为年月日，删除原字段 process_legal_person_pinyin_name(result) # 分解为姓和名，删除原字段 process_legal_person_city_english_name(result) # 直接修改原字段 process_legal_person_address_province_english_name(result) # 直接修改原字段 process_legal_person_country(result) # 直接修改原字段 process_city_english_name(result) # 直接修改原字段 process_company_address_province_english_name(result) # 直接修改原字段 process_ar(result) # 授权机构模板选择字段归一化为大写 M/O process_id_number_and_passport(result) # 处理身份证号和护照号 process_current_date(result) # 新增当前日期字段 # 处理LegalSignedFile，需要数据库连接参数 if all([ip, port, username, password, vat_database]): process_legal_signed_file(result, ip, port, username, password, vat_database) # 直接修改原字段 else: result['LegalSignedFile'] = '' # 可以在这里添加更多字段处理函数 def query_and_update_vat_task(ip, port, username, password, database, vat_database='vat_db', callback_url=None): """ 查询EPRHaiyaProcessingTasks表中符合条件的一条记录，并更新其状态 source/API 任务都支持：source 任务要求 EPRRegInfoId 非空， API 任务（DataSource='api'）EPRRegInfoId 为 NULL，单独放行。 API 任务字段检查失败或处理异常时，调用异步结果通知接口把错误信息发给调用方。 参数: ip: 数据库IP地址 port: 数据库端口 username: 数据库用户名 password: 数据库密码 database: 任务表所在数据库名（EPRHaiyaProcessingTasks） vat_database: vat_db 数据库名（source 任务查签名附件；默认 'vat_db' 与原脚本行为一致） callback_url: 结果通知接口地址（API任务错误通知用；为空时用 DEFAULT_CALLBACK_URL） 返回: dict: 查询到的数据字典，如果没有查到则返回None """ conn = None cursor = None result = None try: # 连接数据库 conn = pymssql.connect( server=ip, port=port, user=username, password=password, database=database, charset='utf8' ) cursor = conn.cursor(as_dict=True) # 查询一条符合条件的记录 query_sql = """ SELECT TOP 1 * FROM EPRHaiyaProcessingTasks WHERE TaskStatus = 2 AND pdf_result IN (0, 1) AND (EPRRegInfoId IS NOT NULL OR DataSource = 'api') ORDER BY Id DESC """ cursor.execute(query_sql) result = cursor.fetchone() # 如果查询到数据，则验证字段完整性 if result: # 定义必须非空的字段列表（排除LegalSignedFile） required_fields = [ 'EPRBusinessRecordId', 'LegalPersonFullNamePinYin', 'LegalPersonIdNumber', 'LegalPersonIDCardType', 'LegalPersonCountry', 'LegalPersonGender', 'LegalPersonBirthDate', 'LegalPersonCityEngName', 'LegalPersonAddressProvinceEn', 'CompanyAddressLine1En', 'CityEngName', 'CompanyAddressPostcode', 'CompanyAddressProvinceEn', 'CompanyCountry', 'CompanyCountryCode', 'AR', 'CreatedAt' ] # 检查所有必填字段是否为空 empty_field = None for field in required_fields: if result.get(field) is None or result.get(field) == '': empty_field = field break # 如果有必填字段为空，设置pdf_result为3并提交更新 if empty_field: update_sql = """ UPDATE EPRHaiyaProcessingTasks SET pdf_result = 3 WHERE Id = %s """ cursor.execute(update_sql, (result.get('Id'),)) conn.commit() # API 任务：字段检查失败，调用异步结果通知接口把错误信息发给调用方（通知失败不阻断主流程） error_msg = f'必填字段为空: {empty_field}' if (result.get('DataSource') or '').strip().lower() == 'api': try: notify_api_result(result.get('Id'), callback_url, False, error_msg, biz_param=parse_biz_param_from_row(result)) except Exception as notify_err: print(f'结果通知调用失败(不阻断主流程): {notify_err}') print(f'任务 Id={result.get("Id")} 字段检查失败: {error_msg}') return None # 处理结果中的各种字段，传入数据库连接参数用于LegalSignedFile处理 process_result_fields(result, ip, port, username, password, vat_database) # 更新pdf_result为1（处理中） update_sql = """ UPDATE EPRHaiyaProcessingTasks SET pdf_result = 1 WHERE Id = %s """ cursor.execute(update_sql, (result.get('Id'),)) # 提交更新 conn.commit() return result else: return None except Exception as e: print(f"数据库操作错误: {e}") # API 任务：处理异常时把错误信息发给调用方（尽力而为，通知失败不阻断） if result is not None and (result.get('DataSource') or '').strip().lower() == 'api': try: notify_api_result(result.get('Id'), callback_url, False, str(e), biz_param=parse_biz_param_from_row(result)) except Exception as notify_err: print(f'结果通知调用失败(不阻断主流程): {notify_err}') raise finally: # 关闭游标和连接 if cursor: cursor.close() if conn: conn.close()