|
- # -*- coding: utf-8 -*-
- import time
- from json import loads
- from threading import Thread
- from traceback import format_exc
-
- from loguru import logger
-
- from util.QueUtil import get_block_queue
- from util.RequestUtil import HttpRequests
-
-
- class HttpFeedbackThread(Thread):
- __slots__ = '__fb_queue'
-
- def __init__(self, fb_queue):
- super().__init__()
- self.__fb_queue = fb_queue
-
- def run(self):
- logger.info("启动反馈线程")
- fb_queue = self.__fb_queue
- request = HttpRequests()
- try:
- headers = {'content-type': "application/json"}
- while True:
- try:
- fb = get_block_queue(fb_queue)
- if fb is not None and len(fb) > 0:
- if fb[0] == "upload":
- upload_req(request, fb[1], headers)
- del fb
- else:
- time.sleep(1)
- except Exception:
- logger.error("反馈异常：{}", format_exc())
- if request:
- request.close_session()
- request = HttpRequests()
- finally:
- request.close_session()
- logger.info("反馈线程执行完成")
-
-
- def upload_req(request, upload, headers):
- logger.info("开始发送文件上传回调请求, 回调地址: {}", upload['callback_url'])
- logger.info("开始发送文件上传回调请求, 回调请求体: {}", upload['data'])
- try:
- response = request.send_request('POST', upload['callback_url'], data=upload['data'], headers=headers, timeout=3)
- if response.status_code != 200:
- logger.error("上传回调请求失败! 状态码: {}, {}", response.status_code, response.__dict__)
- else:
- content = response.content.decode('utf-8')
- if content is not None and len(content) > 0:
- content = loads(content)
- code = content.get('code')
- if code is not None and code != 0:
- logger.error("上传回调请求失败! 失败描述: {}", content.get('msg'))
- except Exception:
- logger.error("回调请求失败, 请求体: {}, 回调地址: {}, 异常信息: {}", upload['data'], upload['callback_url'],
- format_exc())
|
