import asyncio import datetime import io import json import logging from tornado.httpclient import HTTPResponse, HTTPRequest import apps import dock from dock.dcm import dcm_api, dcm_push_conv_rollback, dcm_send_sms, dcm_push_upload from dock.oa import oa_api_request, oa_result_notify, PushException from models.dcm_rollback import DcmRollback from models.dcm_task import DcmTask from paste.core.logging import echo_log from paste.util import ufile, udict from paste.web import requests async def get_rollback_request(dcm_rollback: DcmRollback, dcm_task: DcmTask, media_ids: str, media_types: str, add_num: int, extra_opinion: str): """ 创建回退请求对象。方法仅创建请求对象，并未实际提交请求，具体由调度方法处理。 :param dcm_rollback: 回退对象 :param dcm_task: 工单对象 :param media_ids: 上传的附件 MediaID 字符串，多个用英文逗号分隔 :param media_types: 上传的附件的MediaType字符串，多个用英文逗号分隔 :param add_num: 上传的附件数 :param extra_opinion: 便民表单的extendInfo文本 :return: HTTPRequest 对象 """ api_url = '/home/workflow/rollback' body_dict = { 'actID': int(dcm_rollback.act_id), 'opinion': dcm_rollback.opinion if not extra_opinion else dcm_rollback.opinion + ' ' + extra_opinion, 'transInfo': dcm_rollback.trans_info, 'saveOldActFlag': dcm_rollback.save_old_act_flag, 'rollbackReasonID': dcm_rollback.rollback_reason_id, 'transListNum': 0, 'mediaIDs': media_ids, 'mediaUsage': '回退', 'relationTypeID': 10, 'relationID': dcm_task.rec_id, 'relationMainID': 45, 'relationSubID': dcm_task.act_id, 'tempUsage': 'rollback_tmp', 'mediaTypes': media_types, 'addNum': add_num } rollback_request = await dcm_api.new_api_request(api_url, body_dict) return rollback_request async def on_attachment_download_error(request: HTTPRequest, exc: Exception, retry_queue: asyncio.Queue = None): """ 下载附件时的出错处理。 :param request: 请求对象 :param exc: 异常对象 :param retry_queue: :return: """ retry = getattr(request, 'retry', 0) max_retry = getattr(request, 'max_retry', 0) if retry < max_retry - 1: # 非最后一次尝试，不处理 return message = f"下载回退附件时发生错误，下载地址：{request.url}" echo_log(message) echo_log(exc, logging.ERROR, is_log_exc=True) # 保存异常 dcm_rollback: DcmRollback = getattr(request, 'dcm_rollback') exc_list = getattr(dcm_rollback, 'exc_list') exc_list.append(PushException(message, dcm_rollback.flow_token, 3)) async def done_attachment_download(response_list: list[HTTPResponse]): """ 所有附件下载完成执行的处理程序。 :param response_list: 附件下载响应列表 :return: 返回附件字典 """ # 遍历取出附件数据 date_str = datetime.datetime.now().strftime("%Y%m%d") files_body: dict[str, io.IOBase] = {} for i, response in enumerate(response_list): _type = ufile.inspect_type(response.body) if isinstance(_type, tuple): _fn = f"CT{date_str}{i + 1:04d}.{_type[0]}" elif isinstance(_type, str) and _type: _fn = f"CT{date_str}{i + 1:04d}.{_type}" else: _fn = f"CT{date_str}{i + 1:04d}" file_obj = io.BytesIO(response.body) files_body[_fn] = file_obj return files_body async def after_push_rollback_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]): """ 提交数字城管后的处理程序。 :param response: 响应对象 :param retry_queue: 重试队列 """ echo_log(response.body.decode()) echo_log('回退请求成功.') async def push_rollback(dcm_rollback: DcmRollback, dcm_task: DcmTask): """ 推送回退请求。 :param dcm_rollback: 保存在数据库的回退对象 :param dcm_task: 待办工单 """ try: # 如果来源是12345热线，提交便民回退 extra_opinion = '' if dcm_task.event_src_name == '12345热线': conv_response_list = await dcm_push_conv_rollback.push_conv_rollback_request(dcm_rollback, dcm_task) if conv_response_list: conv_response = conv_response_list[0].body.decode() conv_response = json.loads(conv_response) extra_opinion = udict.get_by_path(conv_response, 'resultInfo.data.extendInfo', '') echo_log(f"正在准备下载队列...") setattr(dcm_rollback, 'exc_list', []) files_body = {} if dcm_rollback.attachments: # 创建并填充下载请求队列 attachment_download_queue = asyncio.Queue() attachment_id_list = dcm_rollback.attachments.split(',') for attachment_id in attachment_id_list: download_request = await oa_api_request.get_download_request(attachment_id) setattr(download_request, 'dcm_rollback', dcm_rollback) await attachment_download_queue.put(download_request) # 启动下载，得到所有的附件 files_body: dict[str, io.IOBase] = await requests.async_concurrency( attachment_download_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT, after_done=done_attachment_download, on_error=on_attachment_download_error ) # 读取异常有则报错 exc_list: list[PushException] = getattr(dcm_rollback, 'exc_list', []) if exc_list: message = ";".join([f"{exc}" for exc in exc_list]) raise PushException(message, dcm_rollback.flow_token, 3) echo_log(f"附件下载完成，正在准备提交数字城管...") # 仅生产环境真实提交，其他环境不实际提交 if apps.get_active_env() in ('dev', '', None): echo_log(f"非生产环境，不实际提交.") return # 上传附件，同步得到 MediaIds media_ids, media_types, add_num = await dcm_push_upload.push_upload( dcm_task, dcm_push_upload.RollbackTmp, files_body ) # 创建并填充数字城管请求队列 rollback_request = await get_rollback_request( dcm_rollback, dcm_task, media_ids, media_types, add_num, extra_opinion ) setattr(rollback_request, 'dcm_rollback', dcm_rollback) rollback_push_queue = asyncio.Queue() await rollback_push_queue.put(rollback_request) # 提交数字城管-回退请求 rollback_push_response_list = await requests.async_concurrency( rollback_push_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT, after_request=after_push_rollback_request ) # 检查是否推送成功，失败直接报错 if len(rollback_push_response_list) != 1: raise PushException("回退请求发生错误.", dcm_rollback.flow_token, 3) # 检查响应的内容，验证是否推送成功 response_data = rollback_push_response_list[0].body.decode() response_data = json.loads(response_data) if udict.get_by_path(response_data, 'resultInfo.message', '') != '回退成功！' or udict.get_by_path( response_data, 'resultInfo.success') is not True: raise PushException("回退请求发生错误.", dcm_rollback.flow_token, 3) # 保存成功状态 dcm_rollback.status = 1 await dcm_rollback.async_save() # 回退请求提交后，通知回退成功 await oa_result_notify.push_result_notify( dcm_rollback.flow_token, '回退成功', 1 ) # 如果需要发送短信，后台发送 if dcm_rollback.send_message in (1, '1'): await dcm_send_sms.send_sms({ 'actDefID': '45', 'partStr': '254,role,市受理员', 'itemType': 'rollback', 'recID': dcm_task.rec_id, 'actID': dcm_task.act_id, 'actPropertyID': '7' }) except PushException as e: # 任何异常都意味着失败，通知 OA echo_log(f'回退发生错误.', logging.ERROR) echo_log(e, logging.ERROR, is_log_exc=True) # 保存失败状态 dcm_rollback.status = 0 await dcm_rollback.async_save() # 回退发生异常，通知回退失败 await oa_result_notify.push_result_notify( e.flow_token, f"{e}", e.return_code ) except Exception as e: # 其他异常都意味着失败 echo_log(f'回退发生错误.', logging.ERROR) echo_log(e, logging.ERROR, is_log_exc=True) if __name__ == "__main__": from paste.core import aio_pool async def push(): dcm_task = await DcmTask(id=2054174091237265413).async_find_first() task_rollback = await DcmRollback(id=2058802518821048320).async_find_first() await push_rollback(task_rollback, dcm_task) _runner = aio_pool.get_aio_runner() _runner(push())