tuoheng_alg_move_20240715/test/demo1.py

# -*- coding: utf-8 -*-
import datetime
import time

import json

from aliyunsdkcore.client import AcsClient
from aliyunsdkvod.request.v20170321 import GetPlayInfoRequest
from loguru import logger

from vodsdk.AliyunVodUploader import AliyunVodUploader
from vodsdk.UploadVideoRequest import UploadVideoRequest
'''
视频上传使用vod
1. 阿里云VOD文档地址：https://help.aliyun.com/product/29932.html?spm=5176.8413026.J_3895079540.5.1b4a1029mXvncc
2. 阿里云对象存储OSS SDK示例地址：https://help.aliyun.com/document_detail/64148.html?spm=a2c4g.64148.0.0.5ae54150jUecEU
4. 安装SDK：
python -m pip install aliyun-python-sdk-core -i https://pypi.tuna.tsinghua.edu.cn/simple
python -m pip install aliyun-python-sdk-live -i https://pypi.tuna.tsinghua.edu.cn/simple
python -m pip install aliyun-python-sdk-core-v3  -i https://pypi.tuna.tsinghua.edu.cn/simple
python -m pip install aliyun-python-sdk-vod  -i https://pypi.tuna.tsinghua.edu.cn/simple
python -m pip install alibabacloud_vod20170321 -i https://pypi.tuna.tsinghua.edu.cn/simple
python -m pip install oss2 -i https://pypi.tuna.tsinghua.edu.cn/simple
5. 视频域名地址：https://vod.play.t-aaron.com/
'''


class AliyunVodSdk:

    def __init__(self, requestId):
        self.__client = None
        self.__access_key = 'LTAI5tMiefafZ6br4zmrQWv9'
        self.__access_secret = 'JgzQjSCkwZ7lefZO6egOArw38YH1Tk'
        self.__regionId = "cn-shanghai"
        self.__cateId = '1000468340'
        self.__requestId = requestId

    def init_vod_client(self):
        return AcsClient(self.__access_key, self.__access_secret, self.__regionId, auto_retry=True, max_retry_time=3, timeout=30)

    '''
    根据videoId获取视频地址
    '''

    def get_play_info(self, videoId):
        logger.info("开始获取视频地址，videoId:{}", videoId)
        start = time.time()
        while True:
            try:
                clt = self.init_vod_client()
                request = GetPlayInfoRequest.GetPlayInfoRequest()
                request.set_accept_format('JSON')
                request.set_VideoId(videoId)
                request.set_AuthTimeout(3600 * 5)
                response = json.loads(clt.do_action_with_exception(request))
                play_url = response["PlayInfoList"]["PlayInfo"][0]["PlayURL"]
                logger.info("获取视频地址成功，视频地址: {}", play_url)
                return play_url
            except Exception as e:
                logger.error("获取视频地址失败，5秒后重试, requestId: {}")
                time.sleep(5)
                current_time = time.time()
                if "HTTP Status: 403" not in str(e):
                    logger.exception("获取视频地址失败: {}", e)
                    raise e
                if "HTTP Status: 403" in str(e) and ("UploadFail" in str(e) or "TranscodeFail" in str(e)):
                    logger.exception("获取视频地址失败: {}", e)
                    raise e
                diff_time = current_time - start
                if diff_time > 60 * 60 * 5:
                    logger.exception("获取视频地址失败超时异常: {}，超时时间：{}", str(e), diff_time)
                    raise e

    def upload_local_video(self, filePath, file_title, storageLocation=None):
        logger.info("开始执行vod视频上传, filePath: {}", filePath)
        uploader = AliyunVodUploader(self.__access_key, self.__access_secret, self.__requestId)
        uploadVideoRequest = UploadVideoRequest(filePath, file_title)
        uploadVideoRequest.setCateId(self.__cateId)
        if storageLocation:
            uploadVideoRequest.setStorageLocation(storageLocation)
        MAX_RETRIES = 3
        retry_count = 0
        while True:
            try:
                result = uploader.uploadLocalVideo(uploadVideoRequest)
                logger.info("vod视频上传成功, videoId:{}", result.get("VideoId"))
                return result.get("VideoId")
            except Exception as e:
                retry_count += 1
                time.sleep(1)
                logger.error("vod视频上传失败，重试次数：{}", retry_count)
                if retry_count >= MAX_RETRIES:
                    logger.exception("vod视频上传重试失败: {}", str(e))
                    raise e


YY_MM_DD_HH_MM_SS = "%Y-%m-%d %H:%M:%S"
YMDHMSF = "%Y%m%d%H%M%S%f"


def generate_timestamp():
    """根据当前时间获取时间戳，返回整数"""
    return int(time.time())


def now_date_to_str(fmt=None):
    if fmt is None:
        fmt = YY_MM_DD_HH_MM_SS
        return datetime.datetime.now().strftime(fmt)


if __name__ == "__main__":
    # 本地原视频命名
    random_time = now_date_to_str(YMDHMSF)
    # # 如果是离线视频，将 _on_or_ 替换为 _off_or_
    # orFilePath = "%s%s%s%s%s" % ('本地路径', random_time, "_on_or_", 'requestId', ".mp4")
    # # 本地AI识别后的视频命名
    # # 如果是离线视频，将 _on_ai_ 替换为 _off_ai_
    # aiFilePath = "%s%s%s%s%s" % ('本地路径', random_time, "_on_ai_", 'requestId', ".mp4")
    # filePath = "%s%s%s%s%s" % ('D:\\shipin\\', random_time, "_on_ai_", '11111111', ".mp4")
    filePath = 'D:\\shipin\\111.mp4'
    codClinet = AliyunVodSdk("1111111")
    result = codClinet.upload_local_video(filePath, 'aiOnLineVideo')
    print(result)
    url = codClinet.get_play_info(result)
    print(url)