tuoheng_airport_media_discard_20240715/concurrency/mqtt/MqttUploadFileProcess.py

# -*- coding: utf-8 -*-
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Process, Queue
from os.path import join, exists
from threading import Thread
from time import sleep, time
from traceback import format_exc

from loguru import logger

from bean.Feedback import upload_result
from enums.ExceptionEnum import UploadStrExceptionType
from enums.StatusEnum import UploadTaskStatusType, UploadStatusType
from exception.CustomerException import ServiceException
from util.AliyunUtil import OssUtil, VodUtil
from util.FileUtil import get_all_images, get_all_videos, get_file_name, delete_file, create_dir, \
    move_file, remove_after_create
from util.LogUtils import init_log
from util.QueUtil import put_queue, get_no_block_queue, clear_queue
from util.ThreadUtil import ThreadPoolExecutorNew
from util.TimeUtils import now_date_to_str, YMDHMS

'''
    视频上传
'''


class MqttUploadFileProcess(Process):
    __slots__ = ('__fb_queue', '__event_queue', '__service_config', '__requestId', '__current_time', 'videoPath',
                 'imagePath', 'image_backup', 'video_backup')

    def __init__(self, fb_queue, service_config, request_id):
        super().__init__()
        self.__fb_queue = fb_queue
        self.__requestId = request_id
        self.__event_queue = Queue()
        self.__service_config = service_config
        self.__current_time = now_date_to_str(YMDHMS)
        backup = join(service_config["file"]["backup"], self.__current_time)
        self.videoPath, self.imagePath = service_config["file"]['videoPath'], service_config["file"]['imagePath']
        self.image_backup, self.video_backup = join(backup, 'images'), join(backup, 'videos')
        upload_result(fb_queue, request_id, status=UploadTaskStatusType.WAITING.value[0])

    def sendEvent(self, result):
        put_queue(self.__event_queue, result, timeout=2, is_throw_ex=True)

    @staticmethod
    def uploadImage(oss, uploadPath, filePath):
        try:
            oss.resumable_upload(uploadPath, filePath)
        except Exception as e:
            oss.exception = e
            logger.error("oss上传文件异常: {}, uploadPath:{}, filePath:{}", format_exc(), uploadPath, filePath)
            raise e

    @staticmethod
    def uploadVideo(vod, file_title, filePath):
        try:
            vod.get_play_url(filePath, file_title)
        except Exception as e:
            vod.exception = e
            logger.error("oss上传文件异常: {}, file_title:{}, filePath:{}", format_exc(), file_title, filePath)
            raise e

    @staticmethod
    def move_method(file, backup):
        move_file(file, backup)

    @staticmethod
    def hdThread(imageList, videoList, fb_queue, hb_queue, requestId):
        try:
            logger.info("启动文件上传心跳线程, requestId:{}", requestId)
            start_time = time()
            count = 0
            while True:
                if time() - start_time > 43200:
                    logger.error("心跳线程运行超时！！！！requestId:{}", requestId)
                    break
                command = get_no_block_queue(hb_queue)
                if command is not None:
                    if "stop" == command.get("command"):
                        logger.info("开始停止心跳线程！！！！requestId:{}", requestId)
                        break
                if count % 5 == 0:
                    upload_result(fb_queue, requestId, status=UploadTaskStatusType.RUNNING.value[0],
                                  imageList=imageList, videoList=videoList)
                    count = 0
                count += 1
                sleep(1)
        except Exception:
            logger.error("心跳线程异常:{}, requestId:{}", format_exc(), requestId)
        logger.info("心跳线程停止完成！requestId:{}", requestId)

    def upload_method(self, service_config, video_path_array, videoTask, videoList, vt, image_path_array, imageTask,
                      imageList, it):
        current_time = self.__current_time
        for i in video_path_array:
            vod = VodUtil(service_config["base_dir"])
            filename = get_file_name(i)
            video_result = vt.submit(self.uploadVideo, vod, filename, i)
            videoTask[filename] = [video_result, vod, i]
            videoList.append({"fileName": filename, "videoUrl": "", "status": UploadStatusType.WAITING.value[0],
                              "progress": vod.get_progress()})
        for i in image_path_array:
            oss = OssUtil(service_config["base_dir"])
            fileName = get_file_name(i)
            uploadPath = "%s/%s" % (current_time, fileName)
            image_result = it.submit(self.uploadImage, oss, uploadPath, i)
            imageTask[fileName] = [image_result, oss, i]
            imageList.append({"fileName": fileName, "imageUrl": "", "status": UploadStatusType.WAITING.value[0],
                              "progress": oss.progress})

    def start_hd_thread(self, imageList, videoList, fb_queue, hb_queue, requestId):
        hd = Thread(target=self.hdThread, args=(imageList, videoList, fb_queue, hb_queue, requestId))
        hd.setDaemon(True)
        hd.start()
        return hd

    @staticmethod
    def stop_hd_thread(hb, hb_queue, requestId):
        if hb is not None and hb.is_alive():
            put_queue(hb_queue, {"command": "stop"}, timeout=2, is_throw_ex=False)
            start = time()
            hb.join(60)
            if time() - start > 60:
                logger.error("心跳线程停止超时, requestId:{}", requestId)
            clear_queue(hb_queue)

    @staticmethod
    def change_status(imageList, num, imageTask, videoList, videoTask, requestId, deleteImageList, deleteVideoList):
        for image in imageList:
            if image["status"] < UploadStatusType.SUCCESS.value[0]:
                num += 1
                image_task = imageTask.get(image.get("fileName"))
                # (image_result, oss, i)
                if image_task:
                    if not image_task[0].done():
                        image["status"] = UploadStatusType.RUNNING.value[0]
                        image["progress"] = image_task[1].get_progress()
                        continue
                    if image_task[0].done():
                        try:
                            image_task[0].result()
                            image["imageUrl"] = image_task[1].get_image_url()
                            image["status"] = UploadStatusType.SUCCESS.value[0]
                            image["progress"] = "1.0000"
                            deleteImageList.append(image_task[2])
                            # delete_file(image_task[2])
                        except Exception:
                            logger.error("文件{}上传失败, 异常: {}, requestId: {}", image_task[2], format_exc(),
                                         requestId)
                            image["status"] = UploadStatusType.FAILED.value[0]
                            image["progress"] = image_task[1].get_progress()
        for video in videoList:
            if video["status"] < UploadStatusType.SUCCESS.value[0]:
                num += 1
                video_task = videoTask.get(video.get("fileName"))
                if video_task:
                    # 如果任务已经完成
                    # (video_result, vod, i)
                    if not video_task[0].done():
                        video["status"] = UploadStatusType.RUNNING.value[0]
                        video["progress"] = video_task[1].get_progress()
                        continue
                    if video_task[0].done():
                        try:
                            video_task[0].result()
                            video["videoUrl"] = video_task[1].get_video_url()
                            video["status"] = UploadStatusType.SUCCESS.value[0]
                            video["progress"] = "1.0000"
                            deleteVideoList.append(video_task[2])
                            # delete_file(video_task[2])
                        except Exception:
                            logger.error("文件{}上传失败, 异常：{}, requestId: {}", video_task[2], format_exc(),
                                         requestId)
                            video["status"] = UploadStatusType.FAILED.value[0]
                            video["progress"] = video_task[1].get_progress()
        return num

    @staticmethod
    def stop_all_task(imageList, imageTask, videoList, videoTask):
        for image in imageList:
            if image["status"] < UploadStatusType.SUCCESS.value[0]:
                image_task = imageTask.get(image.get("fileName"))
                # (image_result, oss, i)
                if image_task:
                    if not image_task[0].done():
                        # 如果任务没有结束, 停止设置取消任务
                        image_task[1].status = True
        for video in videoList:
            if video["status"] < UploadStatusType.SUCCESS.value[0]:
                video_task = videoTask.get(video.get("fileName"))
                if video_task:
                    # 如果任务已经完成
                    # (video_result, vod, i)
                    if not video_task[0].done():
                        video_task[1].status = True
                        if video_task[1].uploader is not None:
                            if video_task[1].uploader.uploader is not None:
                                video_task[1].uploader.uploader.status = True

    @staticmethod
    def stop_change_status(imageList, imageTask, videoList, videoTask, deleteImageList, deleteVideoList):
        for image in imageList:
            if image["status"] < UploadStatusType.SUCCESS.value[0]:
                image_task = imageTask.get(image.get("fileName"))
                # (image_result, oss, i)
                if image_task:
                    oss = image_task[1]
                    status = oss.status
                    image_url = oss.get_image_url()
                    if image_url is not None and len(image_url) > 0:
                        image["imageUrl"] = image_url
                        image["status"] = UploadStatusType.SUCCESS.value[0]
                        image["progress"] = "1.0000"
                        deleteImageList.append(image_task[2])
                        # delete_file(image_task[2])
                    else:
                        if status:
                            image["status"] = UploadStatusType.CANCEL.value[0]
                            image["progress"] = oss.get_progress()
                        else:
                            image["status"] = UploadStatusType.FAILED.value[0]
                            image["progress"] = oss.get_progress()
        for video in videoList:
            if video["status"] < UploadStatusType.SUCCESS.value[0]:
                video_task = videoTask.get(video.get("fileName"))
                if video_task:
                    # 如果任务已经完成
                    # (video_result, vod, i)
                    vod = video_task[1]
                    status = vod.status
                    video_url = vod.get_video_url()
                    if video_url is not None and len(video_url) > 0:
                        video["videoUrl"] = video_url
                        video["status"] = UploadStatusType.SUCCESS.value[0]
                        video["progress"] = "1.0000"
                        deleteVideoList.append(video_task[2])
                        # delete_file(video_task[2])
                    else:
                        if status:
                            video["status"] = UploadStatusType.CANCEL.value[0]
                            video["progress"] = vod.get_progress()
                        else:
                            video["status"] = UploadStatusType.FAILED.value[0]
                            video["progress"] = vod.get_progress()

    @staticmethod
    def wait_thread(it, vt, requestId):
        if it:
            it.shutdown()
            logger.info("it线程池关闭完成m, requestId:{}", requestId)
        if vt:
            vt.shutdown()
            logger.info("vt线程池关闭完成m, requestId:{}", requestId)

    def run(self):
        imageList, videoList = [], []
        deleteImageList, deleteVideoList = [], []
        # 线程任务
        imageTask, videoTask = {}, {}
        # 备份路径
        image_backup, video_backup = self.image_backup, self.video_backup
        imagePath, videoPath = self.imagePath, self.videoPath
        # 队列
        fb_queue, hb_queue, event_queue = self.__fb_queue, Queue(), self.__event_queue
        requestId, service_config = self.__requestId, self.__service_config
        it, vt, ex, hd = None, None, None, None
        try:
            init_log(service_config["base_dir"])
            logger.info("启动文件上传进程！requestId:{}", requestId)
            # 检查文件路径是否存在，不存在则路径有问题
            if not exists(imagePath) or not exists(videoPath):
                logger.error("图片、视频路径检测失败, 图片或视频路径不存在！requestId:{}", requestId)
                logger.error("图片路径: {}, 视频路径: {}, requestId:{}", imagePath, videoPath, requestId)
                raise ServiceException(UploadStrExceptionType.FILE_PATH_IS_NOT_AREADLY.value[0],
                                       UploadStrExceptionType.FILE_PATH_IS_NOT_AREADLY.value[1])
            # 检查是否有视频和图片，没有直接返回
            image_path_array, video_path_array = get_all_images(imagePath), get_all_videos(videoPath)
            if len(image_path_array) == 0 and len(video_path_array) == 0:
                logger.info("未查询到本地视频及图片文件！！！requestId:{}", requestId)
                # upload_result(fb_queue, requestId, status=UploadTaskStatusType.SUCCESS.value[0])
                return
            # 启动心跳线程
            hd = self.start_hd_thread(imageList, videoList, fb_queue, hb_queue, requestId)
            # 初始化图片、视频线程池
            it = ThreadPoolExecutorNew(max_workers=10, thread_name_prefix='图片线程池')
            vt = ThreadPoolExecutorNew(max_workers=2, thread_name_prefix='视频线程线程池')
            self.upload_method(service_config, video_path_array, videoTask, videoList, vt, image_path_array, imageTask,
                               imageList, it)
            start_time = time()
            while True:
                # 超时检查
                if time() - start_time >= 43200:
                    logger.error("上传文件任务执行超时！requestId: {}", requestId)
                    raise ServiceException(UploadStrExceptionType.TASK_IS_RUN_TIMEOUT.value[0],
                                           UploadStrExceptionType.TASK_IS_RUN_TIMEOUT.value[1])
                # 心跳检查
                if hd is not None and not hd.is_alive():
                    logger.error("心跳线程异常停止, requestId:{}", requestId)
                    raise Exception("心跳线程异常停止！")
                event_result = get_no_block_queue(event_queue)
                # 终止任务
                if event_result is not None and event_result.get("command") == "stop":
                    # 清空线程队列
                    it.clear_work_queue()
                    vt.clear_work_queue()
                    # 停止所有任务
                    self.stop_all_task(imageList, imageTask, videoList, videoTask)
                    # 等待线程任务完成
                    it.shutdown()
                    vt.shutdown()
                    # 更新任务状态
                    self.stop_change_status(imageList, imageTask, videoList, videoTask, deleteImageList,
                                            deleteVideoList)
                    break
                num = 0
                num = self.change_status(imageList, num, imageTask, videoList, videoTask, requestId, deleteImageList,
                                         deleteVideoList)
                if num == 0:
                    break
                sleep(1)
        except ServiceException as s:
            ex = s.code, s.msg
            if it:
                it.clear_work_queue()
            if vt:
                vt.clear_work_queue()
            # 停止所有任务
            self.stop_all_task(imageList, imageTask, videoList, videoTask)
            if it:
                it.shutdown()
            if vt:
                vt.shutdown()
            self.stop_change_status(imageList, imageTask, videoList, videoTask, deleteImageList, deleteVideoList)
            logger.error("上传文件任务异常失败: {}, requestId:{}", s.msg, requestId)
        except Exception:
            ex = UploadStrExceptionType.SERVICE_INNER_EXCEPTION.value[0], \
                UploadStrExceptionType.SERVICE_INNER_EXCEPTION.value[1]
            if it:
                it.clear_work_queue()
            if vt:
                vt.clear_work_queue()
            # 停止所有任务
            self.stop_all_task(imageList, imageTask, videoList, videoTask)
            if it:
                it.shutdown()
            if vt:
                vt.shutdown()
            self.stop_change_status(imageList, imageTask, videoList, videoTask, deleteImageList, deleteVideoList)
            logger.error("上传文件任务异常失败: {}, requestId:{}", format_exc(), requestId)
        finally:
            try:
                if len(deleteImageList) > 0 or len(deleteVideoList) > 0:
                    with ThreadPoolExecutor(max_workers=20) as de:
                        for imageFile in deleteImageList:
                            logger.error("删除图片文件, 文件名: {}, requestId: {}", imageFile, requestId)
                            de.submit(delete_file, imageFile)
                        for videoFile in deleteVideoList:
                            logger.error("删除视频文件, 文件名: {}, requestId: {}", videoFile, requestId)
                            de.submit(delete_file, videoFile)
                        de.shutdown(wait=True)
                self.wait_thread(it, vt, requestId)
                image_path_array1, video_path_array1 = get_all_images(imagePath), get_all_videos(videoPath)
                if len(image_path_array1) > 0 or len(video_path_array1) > 0:
                    with ThreadPoolExecutor(max_workers=20) as ec:
                        for i in image_path_array1:
                            create_dir(image_backup)
                            logger.error("上传文件失败文件迁移备份, 文件名: {}, requestId: {}", i, requestId)
                            ec.submit(move_file, i, image_backup)
                        for i in video_path_array1:
                            create_dir(video_backup)
                            logger.error("上传文件失败文件迁移备份, 文件名: {}, requestId: {}", i, requestId)
                            ec.submit(move_file, i, video_backup)
                        ec.shutdown(wait=True)
                self.stop_hd_thread(hd, hb_queue, requestId)
                if ex:
                    code, msg = ex
                    upload_result(fb_queue, requestId, errorCode=code, errorMsg=msg,
                                  status=UploadTaskStatusType.FAILED.value[0], imageList=imageList, videoList=videoList)
                else:
                    upload_result(fb_queue, requestId, status=UploadTaskStatusType.SUCCESS.value[0], imageList=imageList,
                              videoList=videoList)
            except Exception:
                logger.error("未知异常: {}", format_exc())
                self.stop_hd_thread(hd, hb_queue, requestId)
                upload_result(fb_queue, requestId,
                              errorCode=UploadStrExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                              errorMsg=UploadStrExceptionType.SERVICE_INNER_EXCEPTION.value[1],
                              status=UploadTaskStatusType.FAILED.value[0],
                              imageList=imageList, videoList=videoList)
            finally:
                self.stop_hd_thread(hd, hb_queue, requestId)
                clear_queue(event_queue, is_ex=False)
                clear_queue(hb_queue, is_ex=False)
                tmp = join(service_config["base_dir"], r'tmp\oss')
                remove_after_create(tmp)
                logger.info("上传文件任务完成, requestId: {}", requestId)