zwf
178 lines
7.6 KiB
Python
178 lines
7.6 KiB
Python
"""
|
|
待签收工单推送。
|
|
|
|
对应文档接口：2、推送待签收工单列表
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import pandas as pd
|
|
from sqlalchemy import select
|
|
from tornado.httpclient import HTTPResponse, HTTPRequest
|
|
|
|
import models
|
|
import apps
|
|
import dock
|
|
from dock.oa import oa_api_request
|
|
from dock.govs import govs_save_sign
|
|
from dock.oa_govs import oa_sign_task
|
|
from models.govs_order_master import GovsOrderMaster
|
|
from models.govs_push_status import GovsPushStatus
|
|
from typing import Optional, Union
|
|
from paste.core.logging import echo_log
|
|
from paste.web import requests
|
|
from paste.util import udict
|
|
|
|
GOVS_MASTER_MAPPING = {
|
|
GovsOrderMaster.id.key: 'gdId',
|
|
GovsOrderMaster.order_id.key: 'workOrderNo',
|
|
GovsOrderMaster.case_content.key: 'appealContent',
|
|
GovsOrderMaster.case_goal.key: 'appealPurpose',
|
|
GovsOrderMaster.plan_sign_time.key: 'plannedSignOffTime',
|
|
GovsOrderMaster.plan_finish_time.key: 'plannedCompletionTime',
|
|
GovsOrderMaster.order_status.key: 'workOrderStatus',
|
|
GovsOrderMaster.claim_status.key: 'signOffStatus',
|
|
GovsOrderMaster.plan_back_time.key: 'returnDeadline',
|
|
GovsOrderMaster.handle_time.key: 'assignToSubordinateTime',
|
|
GovsOrderMaster.back_time.key: 'subordinateReturnTime',
|
|
GovsOrderMaster.complete_time.key: 'subordinateCompletionTime',
|
|
GovsOrderMaster.update_date.key: 'platformUpdateTime',
|
|
GovsOrderMaster.service_object_type.key: 'appealType'
|
|
}
|
|
"""
|
|
数据推送映射关系。
|
|
"""
|
|
|
|
|
|
async def after_push_order_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]):
|
|
"""
|
|
工单推送请求响应后的处理程序。
|
|
|
|
:param response: 响应对象
|
|
:param retry_queue: 重试队列
|
|
"""
|
|
body = response.body.decode()
|
|
echo_log(body)
|
|
body_data = json.loads(body)
|
|
code = udict.get_by_path(body_data, 'code')
|
|
message = udict.get_by_path(body_data, 'msg')
|
|
if code == 200:
|
|
gov_task_id_list = getattr(response.request, "gov_task_id_list", [])
|
|
order_push_data = [
|
|
{
|
|
GovsPushStatus.master_id.key: govs_task_id,
|
|
GovsPushStatus.push_order_status.key: 1
|
|
}
|
|
for govs_task_id in gov_task_id_list
|
|
]
|
|
govs_task_push_df = pd.DataFrame(order_push_data)
|
|
await GovsPushStatus.save_batch(govs_task_push_df)
|
|
echo_log(f"推送待签收工单成功.")
|
|
else:
|
|
echo_log(f"推送待签收工单失败：{message}")
|
|
|
|
if retry_queue:
|
|
echo_log(f"待签收工单重试队列中有：{retry_queue.qsize()} 个请求在等待.")
|
|
|
|
|
|
async def done_order_push(response_list: list[HTTPResponse]):
|
|
"""
|
|
待签收工单列表推送完成的回调
|
|
"""
|
|
unique_task_ids = set()
|
|
for response in response_list:
|
|
gov_task_id_list = getattr(response.request, "gov_task_id_list", [])
|
|
unique_task_ids.update(gov_task_id_list)
|
|
return unique_task_ids
|
|
|
|
|
|
async def push_order(fetch_size: int = 50, batch_size: int = 10,
|
|
task_id: Optional[Union[str, int, list[Union[str, int]]]] = None):
|
|
"""
|
|
推送待签收工单列表。
|
|
|
|
:param fetch_size: 本次推送数量
|
|
:param batch_size: 分批时，每批大小
|
|
:param task_id: 待办任务 ID 可选
|
|
"""
|
|
# 根据条件获取目标任务 ID 列表（支持指定 task_id 或分页获取）
|
|
task_query = select(GovsOrderMaster.id, GovsOrderMaster.order_id, GovsOrderMaster.case_content,
|
|
GovsOrderMaster.case_goal, GovsOrderMaster.plan_finish_time, GovsOrderMaster.plan_sign_time,
|
|
GovsOrderMaster.service_object_type, GovsOrderMaster.claim_status,
|
|
GovsOrderMaster.plan_back_time,
|
|
GovsOrderMaster.handle_time, GovsOrderMaster.back_time, GovsOrderMaster.complete_time,
|
|
GovsOrderMaster.update_date, GovsOrderMaster.order_status).order_by(GovsOrderMaster.id)
|
|
# 生产环境不推送已推送过的或已签收的
|
|
if apps.get_active_env() == 'prod':
|
|
task_query = task_query.join(
|
|
GovsPushStatus, GovsPushStatus.master_id == GovsOrderMaster.id
|
|
).where((GovsOrderMaster.govs_sign != 1) & (GovsPushStatus.push_order_status != 1))
|
|
if task_id:
|
|
if isinstance(task_id, list):
|
|
task_query = task_query.where(GovsOrderMaster.id.in_(task_id))
|
|
echo_log(f"本次推送待签收工单列表：{task_id} 的数据...")
|
|
else:
|
|
task_query = task_query.where(GovsOrderMaster.id == task_id)
|
|
echo_log(f"本次推送待签收工单：{task_id} 的数据...")
|
|
else:
|
|
task_query = task_query.limit(fetch_size)
|
|
echo_log(f"本次推送前 {fetch_size} 条待签收工单数据...")
|
|
govs_task = await GovsOrderMaster.query_as_df(task_query)
|
|
# 格式化为字符串
|
|
govs_task[GovsOrderMaster.id.key] = govs_task[GovsOrderMaster.id.key].astype(str)
|
|
for key in [GovsOrderMaster.plan_finish_time.key, GovsOrderMaster.plan_sign_time.key,
|
|
GovsOrderMaster.handle_time.key, GovsOrderMaster.back_time.key,
|
|
GovsOrderMaster.complete_time.key, GovsOrderMaster.update_date.key,
|
|
GovsOrderMaster.plan_back_time.key]:
|
|
govs_task[key] = govs_task[key].apply(
|
|
lambda x: x.strftime('%Y-%m-%d %H:%M:%S') if pd.notna(x) and hasattr(x, 'strftime') else x
|
|
)
|
|
|
|
# 处理数据映射，适应接口推送
|
|
mapped_df = govs_task.rename(columns=GOVS_MASTER_MAPPING)
|
|
# 这里把空数据都换成 None，以便存入数据库时是 null
|
|
mapped_df.replace(models.EmptyInDF + models.EmptyDatetimeInDF, '', inplace=True)
|
|
|
|
echo_log(f"正在准备请求队列...")
|
|
# 构建请求队列
|
|
govs_push_queue = asyncio.Queue()
|
|
|
|
# 向队列中填充请求对象
|
|
for start in range(0, len(mapped_df), batch_size):
|
|
batch_df: pd.DataFrame = mapped_df.iloc[start:start + batch_size]
|
|
push_list = batch_df.to_dict('records')
|
|
push_request = await oa_api_request.get_push_govs_order_master_request(push_list)
|
|
setattr(push_request, "gov_task_id_list", batch_df['gdId'].unique().tolist())
|
|
await govs_push_queue.put(push_request)
|
|
|
|
# 并发提交推送请求
|
|
echo_log(f"开始推送待签收工单数据...")
|
|
pushed_order_ids = await requests.async_concurrency(
|
|
govs_push_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
|
|
after_request=after_push_order_request, after_done=done_order_push
|
|
)
|
|
echo_log(f"待签收数据推送已经完成...")
|
|
return pushed_order_ids
|
|
|
|
|
|
async def push_full_order(fetch_size: int = 50, batch_size: int = 10,
|
|
task_id: Optional[Union[str, int, list[Union[str, int]]]] = None):
|
|
from dock.oa_govs import govs_push_process, govs_push_detail
|
|
# 推送待办工单列表，并获取推送成功的工单id
|
|
pushed_order_ids = await push_order(fetch_size, batch_size, task_id)
|
|
pushed_order_ids = list(pushed_order_ids)
|
|
# 只推送推送成功的工单的详情和办理过程
|
|
await govs_push_detail.push_order_detail(task_id=pushed_order_ids)
|
|
await govs_push_process.push_order_process(task_id=pushed_order_ids)
|
|
# 在省12345平台签收推送成功的工单
|
|
await govs_save_sign.sign_order_bypass_api(pushed_order_ids)
|
|
# OA平台签收工单
|
|
await oa_sign_task.sign_task(task_id=pushed_order_ids)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
from paste.core import aio_pool
|
|
|
|
_runner = aio_pool.get_aio_runner()
|
|
_runner(push_full_order(fetch_size=50))
|