Haokai/d3i-szct
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

恢复项目

Browse Source
This commit is contained in:
zwf
2026-06-03 09:58:49 +08:00
parent dd72f9008f
commit 85ec8bf063
20 changed files with 305 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dock/govc/govc_scrape.py
+96 -10
View File
@@ -7,7 +7,9 @@ from typing import Optional, Union
from sqlalchemy import select, desc
import dock
from dock.govc import govc_scrape_dept_feedback, govc_scrape_return_visit, govc_scrape_finish_info, govc_scrape_order
from dock.govc import govc_scrape_dept_feedback, govc_scrape_return_visit, govc_scrape_finish_info, govc_scrape_detail, \
govc_scrape_process, govc_scrape_requester, govc_scrape_delay_info, govc_scrape_contact_info, \
govc_scrape_order_status, govc_scrape_history_order, govc_scrape_supervision, govc_scrape_order
from models.govc_task import GovcTask
from paste.core.logging import echo_log
from paste.web import requests
@@ -52,8 +54,16 @@ async def fetch_govc_task(fetch_size: int = 60, task_id: Optional[Union[str, int
# 构建请求队列
feedback_queue = asyncio.Queue()
result_info_queue = asyncio.Queue()
finish_info_queue = asyncio.Queue()
contact_info_queue = asyncio.Queue()
delay_info_queue = asyncio.Queue()
detail_queue = asyncio.Queue()
status_queue = asyncio.Queue()
process_queue = asyncio.Queue()
requester_queue = asyncio.Queue()
return_visit_queue = asyncio.Queue()
history_order_queue = asyncio.Queue()
supervision_queue = asyncio.Queue()
# 向队列中填充请求对象
echo_log(f"正在准备请求队列...")
for _h, _row in task_df.iterrows():
@@ -61,15 +71,59 @@ async def fetch_govc_task(fetch_size: int = 60, task_id: Optional[Union[str, int
_row.get(GovcTask.c_guid.key))
setattr(_feedback_request, 'task_id', _row.get(GovcTask.id.key))
await feedback_queue.put(_feedback_request)
_result_info_request = await govc_scrape_return_visit.get_return_visit_request(_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key))
setattr(_result_info_request, 'task_id', _row.get(GovcTask.id.key))
await result_info_queue.put(_result_info_request)
_finish_info_request = await govc_scrape_finish_info.get_finish_info_request(_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key))
setattr(_finish_info_request, 'task_id', _row.get(GovcTask.id.key))
await finish_info_queue.put(_finish_info_request)
_contact_info_request = await govc_scrape_contact_info.get_contact_request(_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key))
setattr(_contact_info_request, 'task_id', _row.get(GovcTask.id.key))
await contact_info_queue.put(_contact_info_request)
_delay_info_request = await govc_scrape_delay_info.get_delay_request(_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key))
setattr(_delay_info_request, 'task_id', _row.get(GovcTask.id.key))
await delay_info_queue.put(_delay_info_request)
_detail_request = await govc_scrape_detail.get_detail_request(_row.get(GovcTask.c_guid.key))
setattr(_detail_request, 'task_id', _row.get(GovcTask.id.key))
await detail_queue.put(_detail_request)
_status_request = await govc_scrape_order_status.get_fetch_status_request(_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key))
setattr(_status_request, 'task_id', _row.get(GovcTask.id.key))
await status_queue.put(_status_request)
_process_request = await govc_scrape_process.get_process_request(_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key))
setattr(_process_request, 'task_id', _row.get(GovcTask.id.key))
await process_queue.put(_process_request)
_requester_request = await govc_scrape_requester.get_requster_request(_row.get(GovcTask.c_guid.key))
setattr(_requester_request, 'task_id', _row.get(GovcTask.id.key))
await requester_queue.put(_requester_request)
_return_visit_request = await govc_scrape_return_visit.get_return_visit_request(_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key))
setattr(_return_visit_request, 'task_id', _row.get(GovcTask.id.key))
await return_visit_queue.put(_return_visit_request)
_history_order_request = await govc_scrape_history_order.get_history_order_request(
_row.get(GovcTask.c_guid.key))
setattr(_history_order_request, 'task_id', _row.get(GovcTask.id.key))
await history_order_queue.put(_history_order_request)
for supervise_type in ('zx', 'bm', 'bmhj'):
_supervision_request = await govc_scrape_supervision.get_supervision_request(
_row.get(GovcTask.pvi_guid.key),
_row.get(GovcTask.c_guid.key),
supervise_type)
setattr(_supervision_request, 'task_id', _row.get(GovcTask.id.key))
setattr(_supervision_request, 'supervise_type', supervise_type)
await supervision_queue.put(_supervision_request)
echo_log(f"抓取待办详细数据...")
tasks = [
requests.async_concurrency(
@@ -77,12 +131,44 @@ async def fetch_govc_task(fetch_size: int = 60, task_id: Optional[Union[str, int
after_request=govc_scrape_dept_feedback.after_feedback_request
),
requests.async_concurrency(
result_info_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_result_info.after_result_info_request
finish_info_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_finish_info.after_finish_info_request),
requests.async_concurrency(
contact_info_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_contact_info.after_contact_request
),
requests.async_concurrency(
finish_info_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_finish_info.after_finish_info_request)
delay_info_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_delay_info.after_delay_request
),
requests.async_concurrency(
detail_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_detail.after_detail_request
),
requests.async_concurrency(
status_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_order_status.after_fetch_status_request
),
requests.async_concurrency(
process_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_process.after_process_request
),
requests.async_concurrency(
request_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_requester.after_requester_request
),
requests.async_concurrency(
return_visit_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_return_visit.after_return_visit_request
),
requests.async_concurrency(
history_order_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_history_order.after_history_order_request
),
requests.async_concurrency(
supervision_queue, con_count=dock.CONCURRENCY_COUNT, retry=dock.MAX_RETRY_COUNT,
after_request=govc_scrape_supervision.after_return_visit_request
)
]
await asyncio.gather(*tasks)
dock/govc/govc_scrape_dept_feedback.py
+8 -6
View File
@@ -47,12 +47,14 @@ async def after_feedback_request(response: HTTPResponse, retry_queue: asyncio.Qu
forward_mapping = {dict_f: table_f for table_f, dict_f in GovcTaskDeptFeedback.FieldMapping.items()}
mapped_df = mapped_df.rename(columns=forward_mapping)
# 把字典、列表改为字符串
mapped_df[GovcTaskDeptFeedback.zxhf_info.key] = mapped_df[GovcTaskDeptFeedback.zxhf_info.key].apply(
lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (list, dict)) else x
)
mapped_df[GovcTaskDeptFeedback.back_info.key] = mapped_df[GovcTaskDeptFeedback.back_info.key].apply(
lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (list, dict)) else x
)
if GovcTaskDeptFeedback.zxhf_info.key in mapped_df:
mapped_df[GovcTaskDeptFeedback.zxhf_info.key] = mapped_df[GovcTaskDeptFeedback.zxhf_info.key].apply(
lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (list, dict)) else x
)
if GovcTaskDeptFeedback.back_info.key in mapped_df:
mapped_df[GovcTaskDeptFeedback.back_info.key] = mapped_df[GovcTaskDeptFeedback.back_info.key].apply(
lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (list, dict)) else x
)
# 这里把空数据都换成 None，以便存入数据库时是 null
mapped_df.replace(models.EmptyInDF + models.EmptyDatetimeInDF, None, inplace=True)
task_id = getattr(response.request, GovcTaskDeptFeedback.task_id.key)
dock/govc/govc_scrape_detail.py
+4 -2
View File
@@ -1,5 +1,6 @@
import asyncio
import json
from urllib.parse import urlencode
import pandas as pd
from tornado.httpclient import HTTPResponse, HTTPRequest
@@ -19,15 +20,16 @@ async def get_detail_request(cguid: str):
:param cguid: 工单列表请求返回的cguid
"""
api_url = '/rest/sztaskworkordercommonrest/getDetail'
api_url = '/rest/sztaskworkordercommonrest/getDetail?'
headers = {
'Referer': f'{govc_api.ApiUrl}/rest/sztaskworkordercommonrest/getDetail'
}
request_body = {
'caseguid': cguid, 'secret': 1
}
api_url += urlencode(request_body)
# 构造 API 请求
return await govc_api.new_api_request(api_url, request_body, headers=headers, method='GET')
return await govc_api.new_api_request(api_url,{}, headers=headers)
async def after_detail_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]):
dock/govc/govc_scrape_history_order.py
+53
View File
@@ -0,0 +1,53 @@
import asyncio
import json
import pandas as pd
from tornado.httpclient import HTTPResponse, HTTPRequest
from dock.govc import govc_api
import models
from models.govc_task_history import GovcTaskHistory
from paste.util import udict
from paste.core.logging import echo_log
async def get_history_order_request(cguid: str):
"""
获取市12345的工单的历史工单信息，请求响应是单条工单的数据
:param cguid: 工单列表请求返回的cguid
"""
api_url = f'/rest/sztaskworkordercommonrest/getHistoryWorkOrder?caseguid={cguid}'
headers = {
'Referer': f'{govc_api.ApiUrl}/rest/sztaskworkordercommonrest/getHistoryWorkOrder'
}
# 构造 API 请求
return await govc_api.new_api_request(api_url, {}, headers=headers)
async def after_history_order_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]):
"""
任务请求响应后的处理程序。
:param response: 响应对象
:param retry_queue: 重试队列
"""
response_body = response.body.decode()
response_data = json.loads(response_body)
history_order_list = udict.get_by_path(response_data, 'params.list')
if history_order_list:
mapped_df = pd.DataFrame(history_order_list)
# 更换映射方向，用于将源数据列名改为与数据库表对应
forward_mapping = {dict_f: table_f for table_f, dict_f in GovcTaskHistory.FieldMapping.items()}
mapped_df = mapped_df.rename(columns=forward_mapping)
# 这里把空数据都换成 None，以便存入数据库时是 null
mapped_df.replace(models.EmptyInDF + models.EmptyDatetimeInDF, None, inplace=True)
task_id = getattr(response.request, GovcTaskHistory.task_id.key)
mapped_df[GovcTaskHistory.task_id.key] = task_id
# 筛选数据状态
_created, _updated = await GovcTaskHistory.save_batch(mapped_df)
echo_log(f"成功创建历史工单信息：{_created}条，更新：{_updated}条.")
else:
echo_log('未获取到历史工单信息')
if retry_queue:
echo_log(f"历史工单信息重试队列中有：{retry_queue.qsize()} 个请求在等待.")
dock/govc/govc_scrape_process.py
+4 -2
View File
@@ -1,5 +1,6 @@
import asyncio
import json
from urllib.parse import urlencode
import pandas as pd
from tornado.httpclient import HTTPResponse, HTTPRequest
@@ -18,7 +19,7 @@ async def get_process_request(pviguid: str, cguid: str):
:param pviguid: 工单列表请求返回的pviguid
:param cguid: 工单列表请求返回的cguid
"""
api_url = '/rest/sztaskworkordercommonrest/getTracing'
api_url = '/rest/sztaskworkordercommonrest/getTracing?'
headers = {
'Referer': f'{govc_api.ApiUrl}/rest/sztaskworkordercommonrest/getTracing'
}
@@ -26,8 +27,9 @@ async def get_process_request(pviguid: str, cguid: str):
'caseguid': cguid,
'pviguid': pviguid
}
api_url += urlencode(request_body)
# 构造 API 请求
return await govc_api.new_api_request(api_url, request_body, headers=headers, method='GET')
return await govc_api.new_api_request(api_url, {}, headers=headers)
async def after_process_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]):
dock/govc/govc_scrape_requester.py
+4 -2
View File
@@ -1,5 +1,6 @@
import asyncio
import json
from urllib.parse import urlencode
import pandas as pd
from tornado.httpclient import HTTPResponse, HTTPRequest
@@ -17,15 +18,16 @@ async def get_requster_request(cguid: str):
:param cguid: 工单列表请求返回的cguid
"""
api_url = '/rest/sztaskworkordercommonrest/getInformation'
api_url = '/rest/sztaskworkordercommonrest/getInformation?'
headers = {
'Referer': f'{govc_api.ApiUrl}/rest/sztaskworkordercommonrest/getInformation'
}
request_body = {
'caseguid': cguid, 'secret': 0
}
api_url += urlencode(request_body)
# 构造 API 请求
return await govc_api.new_api_request(api_url, request_body, headers=headers, method='GET')
return await govc_api.new_api_request(api_url, {}, headers=headers)
async def after_requester_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]):
dock/govc/govc_scrape_supervision.py
+63
View File
@@ -0,0 +1,63 @@
import asyncio
import json
import pandas as pd
from tornado.httpclient import HTTPResponse, HTTPRequest
from dock.govc import govc_api
import models
from models.govc_task_supervision import GovcTaskSupervision
from paste.util import udict
from paste.core.logging import echo_log
async def get_supervision_request(pviguid: str, cguid: str, supervisetype: str):
"""
获取市12345的工单的监察信息，请求响应是单条工单的数据
:param pviguid: 工单列表请求返回的pviguid
:param cguid: 工单列表请求返回的cguid
:param supervisetype: 监察类型 zx:座席区 bm:部门监察 bmhj:部门核减监察
"""
api_url = f'/rest/sztaskworkordercommonrest/getSupervision?supervisetype={supervisetype}'
headers = {
'Referer': f'{govc_api.ApiUrl}/rest/sztaskworkordercommonrest/getSupervision'
}
request_body = {
"ProcessVersionInstanceGuid": pviguid,
'caseguid': cguid,
'yearflag': 'undefined'
}
# 构造 API 请求
return await govc_api.new_api_request(api_url, request_body, headers=headers)
async def after_return_visit_request(response: HTTPResponse, retry_queue: asyncio.Queue[HTTPRequest]):
"""
任务请求响应后的处理程序。
:param response: 响应对象
:param retry_queue: 重试队列
"""
response_body = response.body.decode()
response_data = json.loads(response_body)
zx_list = udict.get_by_path(response_data, 'params.zxlist')
if zx_list:
mapped_df = pd.DataFrame(zx_list)
# 更换映射方向，用于将源数据列名改为与数据库表对应
forward_mapping = {dict_f: table_f for table_f, dict_f in GovcTaskSupervision.FieldMapping.items()}
mapped_df = mapped_df.rename(columns=forward_mapping)
supervise_type = getattr(response.request, 'supervise_type', None)
mapped_df[GovcTaskSupervision.supervise_type.key] = supervise_type
# 这里把空数据都换成 None，以便存入数据库时是 null
mapped_df.replace(models.EmptyInDF + models.EmptyDatetimeInDF, None, inplace=True)
mapped_df.fillna('', inplace=True)
task_id = getattr(response.request, GovcTaskSupervision.task_id.key)
mapped_df[GovcTaskSupervision.task_id.key] = task_id
# 筛选数据状态
_created, _updated = await GovcTaskSupervision.save_batch(mapped_df)
echo_log(f"成功创建监察信息：{_created}条，更新：{_updated}条.")
else:
echo_log('未获取到监察信息')
if retry_queue:
echo_log(f"监察信息重试队列中有：{retry_queue.qsize()} 个请求在等待.")
dock/govs/govs_scrape.py
+1 -1
View File
@@ -103,4 +103,4 @@ if __name__ == "__main__":
from paste.core import aio_pool
_runner = aio_pool.get_aio_runner()
_runner(fetch_govs_task(dept_page_tag=1, num_per_page=50))
_runner(fetch_govs_task(dept_page_tag=0, num_per_page=50))