import asyncio import datetime import io import logging from tornado.httpclient import HTTPResponse, HTTPRequest import apps import dock from dock.dcm import dcm_api, dcm_push_upload from dock.oa import oa_api_request, oa_result_notify, PushException from dock.oa_dcm import oa_update_post_delay from models.dcm_apply_delay import DcmApplyPostpone from models.dcm_task import DcmTask from paste.core.logging import echo_log from paste.util import ufile from paste.web import requests async def get_apply_postpone_request(dcm_apply_postpone: DcmApplyPostpone, dcm_task: DcmTask, media_ids: str, media_types: str, add_num: int): """ 创建申请延期请求对象。方法仅创建请求对象，并未实际提交请求，具体由调度方法处理。 :param dcm_apply_postpone: 申请延期对象 :param dcm_task: 工单对象 :param media_ids: 上传的附件 MediaID 字符串，多个用英文逗号分隔 :param media_types: 上传的附件的MediaType字符串，多个用英文逗号分隔 :param add_num: 上传的附件数 :return: HTTPRequest 对象 """ api_url = '/home/workflow/applypostpone' postpone_desc = f'延长 {dcm_apply_postpone.delay_multiple}倍小类时限（{dcm_apply_postpone.delay_multiple * 24}小时）' body_dict = { 'applyActID': int(dcm_apply_postpone.act_id), 'replyPartID': int(dcm_apply_postpone.reply_part_id), 'ardLevel': dcm_apply_postpone.ard_level, 'ardTypeID': int(dcm_apply_postpone.ard_type_id), 'applyMemo': f'{dcm_apply_postpone.apply_memo}\n{postpone_desc}', 'timeNum': dcm_apply_postpone.delay_multiple * 24, 'timeUnit': dcm_apply_postpone.time_unit, 'postponeDate': dcm_apply_postpone.postpone_date, 'addNum': add_num, 'mediaUsage': '申请延期', 'mediaIDs': media_ids, 'relationTypeID': 10, 'relationID': dcm_task.rec_id, 'relationMainID': 21, 'relationSubID': -1, 'tempUsage': 'apply_postpone_tmp', 'mediaTypes': media_types } apply_postpone_request = await dcm_api.new_api_request(api_url, body_dict) return apply_postpone_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_apply_postpone: DcmApplyPostpone = getattr(request, 'dcm_apply_postpone') exc_list = getattr(dcm_apply_postpone, 'exc_list') exc_list.append(PushException(message, dcm_apply_postpone.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_apply_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]): """ 提交数字城管后的处理程序。 :param response: 响应对象 :param retry_queue: 重试队列 """ echo_log(response.body.decode()) echo_log('申请延期请求成功.') async def push_apply_postpone(dcm_apply_postpone: DcmApplyPostpone, dcm_task: DcmTask): """ 推送申请延期请求。 :param dcm_apply_postpone: 保存在数据库的申请延期对象 :param dcm_task: 待办工单 """ try: files_body = {} if dcm_apply_postpone.attachments: echo_log(f"正在准备下载队列...") setattr(dcm_apply_postpone, 'exc_list', []) # 创建并填充下载请求队列 attachment_download_queue = asyncio.Queue() attachment_id_list = dcm_apply_postpone.attachments.split(',') for attachment_id in attachment_id_list: download_request = await oa_api_request.get_download_request(attachment_id) setattr(download_request, 'dcm_apply_postpone', dcm_apply_postpone) 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_apply_postpone, 'exc_list', []) if exc_list: message = ";".join([f"{exc}" for exc in exc_list]) raise PushException(message, dcm_apply_postpone.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.ApplyPostponeTmp, files_body, -1, 21 ) # 创建并填充数字城管请求队列 apply_postpone_request = await get_apply_postpone_request( dcm_apply_postpone, dcm_task, media_ids, media_types, add_num ) setattr(apply_postpone_request, 'dcm_apply_postpone', dcm_apply_postpone) apply_postpone_push_queue = asyncio.Queue() await apply_postpone_push_queue.put(apply_postpone_request) # 提交数字城管-申请延期请求 apply_postpone_push_response_list = await requests.async_concurrency( apply_postpone_push_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT, after_request=after_push_apply_request ) # 检查是否推送成功，失败直接报错 if len(apply_postpone_push_response_list) != 1: raise PushException("申请延期请求发生错误.", dcm_apply_postpone.flow_token, 3) # TODO: 这里要做实际的推送成功检查 # 保存成功状态 dcm_apply_postpone.status = 1 await dcm_apply_postpone.async_save() # 申请延期请求提交后，通知申请延期成功 await oa_result_notify.push_result_notify( dcm_apply_postpone.flow_token, '申请延期成功', 1 ) # 更新延期时间 await oa_update_post_delay.update_process_delay(dcm_apply_postpone.dcm_task_id) except PushException as e: # 任何异常都意味着失败，通知 OA echo_log(f'申请延期发生错误.', logging.ERROR) echo_log(e, logging.ERROR, is_log_exc=True) # 保存失败状态 dcm_apply_postpone.status = 0 await dcm_apply_postpone.async_save() # 申请延期发生异常，通知申请延期失败 await oa_result_notify.push_result_notify( e.flow_token, f"{e}", e.return_code ) except Exception as e: # 其他异常都意味着失败，通知 OA 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 test(): dcm_task = await DcmTask(id=2054174091237265408).async_find_first() task_apply_postpone = await DcmApplyPostpone(id=2055639783446810624).async_find_first() await push_apply_postpone(task_apply_postpone, dcm_task) _runner = aio_pool.get_aio_runner() _runner(test())