tuoheng_airport_media_discard_20240715/util/AliyunUtil.py

# -*- coding: utf-8 -*-
from json import loads
from os.path import join, exists
from traceback import format_exc

import time

from aliyunsdkcore.client import AcsClient
from loguru import logger

from enums.ExceptionEnum import ExceptionType
from exception.CustomerException import ServiceException
from osssdk import Auth, Bucket, ResumableStore
from osssdk.resumable import resumable_upload1
from util.FileUtil import create_dir
from util.RWUtils import getConfigs
from vodsdk.AliyunVodUploader import AliyunVodUploader
from vodsdk.UploadVideoRequest import UploadVideoRequest
from aliyunsdkvod.request.v20170321.GetPlayInfoRequest import GetPlayInfoRequest


class OssUtil:
    __slots__ = ('bucket', '__base_dir', '__config', 'progress', 'image_url', 'exception', "status")

    def __init__(self, base_dir):
        self.bucket = None
        self.__base_dir = base_dir
        self.__config = getConfigs(base_dir, 'config/aliyun.yml')
        self.progress = "0.0000"
        self.image_url = ""
        self.exception = None
        self.status = False

    def get_oss_bucket(self):
        if self.bucket is None:
            logger.info("初始化oss桶")
            auth = Auth(self.__config["access_key"], self.__config["access_secret"])
            self.bucket = Bucket(auth, self.__config["oss"]["endpoint"], self.__config["oss"]["bucket"])

    def percentage(self, useData, consumed_bytes, total_bytes):
        if total_bytes:
            # filePath = useData
            rate = round(float(float(consumed_bytes) / float(total_bytes)), 4)
            self.progress = str(rate)
            # logger.info("{}文件上传任务进度: {}", filePath, rate)

    """
        上传本地文件
    """

    def put_object_from_file(self, uploadPath, filePath):
        logger.info("开始上传文件, key: {}, 文件: {}, ", uploadPath, filePath)
        MAX_RETRIES = 3
        retry_count = 0
        while True:
            try:
                self.progress = "0.0000"
                if not exists(filePath):
                    break
                self.get_oss_bucket()
                useData = filePath
                resp = self.bucket.put_object_from_file_1(uploadPath, filePath, progress_callback=self.percentage,
                                                          useData=useData)
                if resp.status == 200:
                    self.image_url = join(self.__config['oss']['host_url'], uploadPath)
                    self.progress = "1.0000"
                else:
                    raise Exception("上传文件失败")
                logger.info("上传文件到oss成功! key: {}, filePath:{}, 状态码: {}", uploadPath, filePath, resp.status)
                break
            except Exception as e:
                self.bucket = None
                retry_count += 1
                time.sleep(1)
                logger.info("上传文件到oss失败, 重试次数：{}, key: {}, filePath:{}", uploadPath, filePath)
                if retry_count > MAX_RETRIES:
                    logger.error("上传文件到oss重试失败:{}, key: {}, filePath:{}", format_exc(), uploadPath, filePath)
                    raise e

    """
        断点续传上传本地文件
    """

    def resumable_upload(self, uploadPath, filePath):
        logger.info("开始上传文件, key: {}, 文件: {}, ", uploadPath, filePath)
        MAX_RETRIES = 3
        retry_count = 0
        while True:
            try:
                self.progress = "0.0000"
                if not exists(filePath):
                    break
                if self.status:
                    break
                self.get_oss_bucket()
                useData = filePath
                size = 1024 * 1024 * 10
                tmp = join(join(self.__base_dir, "tmp"), 'oss')
                create_dir(tmp)
                resp = resumable_upload1(self.bucket, uploadPath, filePath,
                                         store=ResumableStore(root=tmp),
                                         multipart_threshold=size, part_size=size,
                                         progress_callback=self.percentage, num_threads=2, useData=useData)
                if resp.status == 200:
                    self.image_url = "%s/%s" % (self.__config['oss']['host_url'], uploadPath)
                    self.progress = "1.0000"
                else:
                    raise Exception("上传文件失败")
                logger.info("上传文件到oss成功! key: {}, filePath:{}, 状态码: {}", uploadPath, filePath, resp.status)
                break
            except Exception as e:
                self.bucket = None
                retry_count += 1
                time.sleep(2)
                logger.info("上传文件到oss失败: {}, 重试次数：{}, key: {}, filePath:{}", format_exc(), retry_count,
                            uploadPath, filePath)
                if retry_count > MAX_RETRIES:
                    logger.error("上传文件到oss重试失败:{}, key: {}, filePath:{}", format_exc(), uploadPath, filePath)
                    raise e

    def get_progress(self):
        return self.progress

    def get_image_url(self):
        return self.image_url


class VodUtil:
    __slots__ = ('__config', 'progress', "__videoUrl", "exception", "uploader", "status")

    def __init__(self, base_dir):
        self.progress = "0.0000"
        self.__config = getConfigs(base_dir, 'config/aliyun.yml')
        self.__videoUrl = ""
        self.exception = None
        self.uploader = None
        self.status = False

    def init_vod_client(self, accessKeyId, accessKeySecret):
        regionId = self.__config["vod"]["ecsRegionId"]
        return AcsClient(accessKeyId, accessKeySecret, regionId, auto_retry=True, max_retry_time=3, timeout=60)

    def uploadProgressCallback(self, consumedBytes, totalBytes):
        try:
            if totalBytes:
                rate = round(float(float(consumedBytes) / float(totalBytes)), 4)
            else:
                rate = 0
            if rate != 0 and rate != 1:
                self.progress = str(rate)
            logger.info('视频上传中: {} bytes, percent:{}', consumedBytes, rate)
        except Exception as e:
            logger.exception("打印视频上传进度回调方法异常: {}", e)

    def get_play_info(self, videoId, filePath, file_title):
        logger.info("开始获取视频地址，videoId:{}, filePath: {}, file_title: {}", videoId, filePath, file_title)
        start = time.time()
        timeout = None
        retry_count = 0
        while True:
            try:
                if self.status:
                    return ""
                if float(self.progress) < 0.8:
                    self.progress = "0.8000"
                clt = self.init_vod_client(self.__config["access_key"],
                                           self.__config["access_secret"])
                request: GetPlayInfoRequest = GetPlayInfoRequest()
                request.set_accept_format('JSON')
                request.set_VideoId(videoId)
                request.set_AuthTimeout(3600 * 5)
                response = loads(clt.do_action_with_exception(request))
                play_url = response["PlayInfoList"]["PlayInfo"][0]["PlayURL"]
                logger.info("获取视频地址成功，视频地址: {}, file_title: {}", play_url, file_title)
                if play_url is None or len(play_url) == 0:
                    raise Exception("获取视频地址失败")
                self.progress = "1.0000"
                return play_url
            except Exception as e:
                logger.info("获取视频地址失败，5秒后重试, 异常: {}, filePath:{}, file_title:{}", format_exc(), filePath,
                            file_title)
                time.sleep(5)
                current_time = time.time()
                if "Uploading" in str(e):
                    if timeout is None:
                        timeout = time.time()
                    if time.time() - timeout > 1800:
                        logger.error("获取视频地址超时, filePath:{}, file_title:{}", filePath,
                                     file_title)
                        raise ServiceException(ExceptionType.GET_VIDEO_URL_EXCEPTION.value[0],
                                               ExceptionType.GET_VIDEO_URL_EXCEPTION.value[1])
                if "UploadSucc" not in str(e) and "Transcoding" not in str(e) and "Uploading" not in str(e):
                    retry_count += 1
                    if retry_count > 3:
                        logger.error("获取视频地址失败: {}, filePath:{}, file_title:{}", format_exc(), filePath, file_title)
                        raise ServiceException(ExceptionType.GET_VIDEO_URL_EXCEPTION.value[0],
                                               ExceptionType.GET_VIDEO_URL_EXCEPTION.value[1])
                diff_time = current_time - start
                if diff_time > 60 * 60 * 5:
                    logger.error("获取视频地址失败超时异常: {}，超时时间：{}, filePath:{}, file_title:{}", format_exc(),
                                 diff_time, filePath, file_title)
                    raise ServiceException(ExceptionType.GET_VIDEO_URL_TIMEOUT_EXCEPTION.value[0],
                                           ExceptionType.GET_VIDEO_URL_TIMEOUT_EXCEPTION.value[1])

    def upload_local_video(self, filePath, file_title):
        logger.info("开始执行vod视频上传, filePath: {}, fileTitle: {}", filePath, file_title)
        self.uploader = AliyunVodUploader(self.__config["access_key"], self.__config["access_secret"],
                                     progress_callback=self.uploadProgressCallback)
        uploadVideoRequest: UploadVideoRequest = UploadVideoRequest(filePath, file_title)
        logger.info("视频分类:{}", self.__config["vod"]["cateId"])
        uploadVideoRequest.setCateId(self.__config["vod"]["cateId"])
        MAX_RETRIES = 3
        retry_count = 0
        while True:
            try:
                self.progress = "0.0000"
                if not exists(filePath):
                    return None
                if self.status:
                    return None
                result = self.uploader.uploadLocalVideo(uploadVideoRequest)
                if self.status:
                    return None
                logger.info("vod视频上传成功, videoId:{}", result.get("VideoId"))
                return result.get("VideoId")
            except Exception:
                retry_count += 1
                time.sleep(1)
                logger.error("vod视频上传失败:{}，重试次数：{}", format_exc(), retry_count)
                if retry_count >= MAX_RETRIES:
                    raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                           ExceptionType.SERVICE_INNER_EXCEPTION.value[1])

    def get_play_url(self, filePath, file_title):
        videoId = self.upload_local_video(filePath, file_title)
        if self.status:
            return
        if videoId is None or len(videoId) == 0:
            logger.error("上传视频失败！filePath: {}, file_title: {}", filePath, file_title)
            raise ServiceException(ExceptionType.UPLOAD_VIDEO_URL_EXCEPTION.value[0],
                                   ExceptionType.UPLOAD_VIDEO_URL_EXCEPTION.value[1])
        self.__videoUrl = self.get_play_info(videoId, filePath, file_title)

    def get_progress(self):
        return self.progress

    def get_video_url(self):
        return self.__videoUrl

    # if __name__ == "__main__":
#     vod = VodUtil(r"D:\tuoheng\codenew\tuoheng_airprt_media")
#     Thread(target=vod.get_play_url, args=(r"D:\test\video\222.mp4", 'aaa')).start()
#     while True:
#         time.sleep(1)
#         if vod.uploader:
#             print(vod.uploader.progress)