tuoheng_alg_move_20240715/test/aliyun/vodTest.py

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import sys

from typing import List

from alibabacloud_vod20170321.client import Client as vod20170321Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_darabonba_env.client import Client as EnvClient
from alibabacloud_vod20170321 import models as vod_20170321_models
from alibabacloud_tea_console.client import Client as ConsoleClient
from alibabacloud_tea_util.client import Client as UtilClient
from vodsdk.AliyunVodUtils import *
from vodsdk.AliyunVodUploader import AliyunVodUploader
from vodsdk.UploadVideoRequest import UploadVideoRequest

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def initialization(
            region_id: str,
    ) -> vod20170321Client:
        config = open_api_models.Config()
        # 您的AccessKey ID
        config.access_key_id = EnvClient.get_env('LTAI5tSJ62TLMUb4SZuf285A')
        # 您的AccessKey Secret
        config.access_key_secret = EnvClient.get_env('MWYynm30filZ7x0HqSHlU3pdLVNeI7')
        # 您的可用区ID
        config.region_id = region_id
        return vod20170321Client(config)

    @staticmethod
    def get_play_info_sample(
            client: vod20170321Client,
            video_id: str,
    ) -> vod_20170321_models.GetPlayInfoResponse:
        request = vod_20170321_models.GetPlayInfoRequest()
        # 视频ID。
        request.video_id = video_id
        response = client.get_play_info(request)
        return response

    @staticmethod
    async def get_play_info_sample_async(
            client: vod20170321Client,
            video_id: str,
    ) -> vod_20170321_models.GetPlayInfoResponse:
        request = vod_20170321_models.GetPlayInfoRequest()
        # 视频ID。
        request.video_id = video_id
        response = await client.get_play_info_async(request)
        return response

    @staticmethod
    def main(
            args: List[str],
    ) -> None:
        try:
            region_id = args[0]
            video_id = args[1]
            client = Sample.initialization(region_id)
            response_get_play_info = Sample.get_play_info_sample(client, video_id)
            ConsoleClient.log(UtilClient.to_jsonstring(UtilClient.to_map(response_get_play_info)))
        except Exception as error:
            ConsoleClient.log(error.message)

    @staticmethod
    async def main_async(
            args: List[str],
    ) -> None:
        try:
            region_id = args[0]
            video_id = args[1]
            client = Sample.initialization(region_id)
            response_get_play_info = await Sample.get_play_info_sample_async(client, video_id)
            ConsoleClient.log(UtilClient.to_jsonstring(UtilClient.to_map(response_get_play_info)))
        except Exception as error:
            ConsoleClient.log(error.message)

accessKeyId='LTAI5tSJ62TLMUb4SZuf285A'
accessKeySecret='MWYynm30filZ7x0HqSHlU3pdLVNeI7'
filePath="/home/thsw/chenyukun/video/111111.mp4"
# 测试上传本地音视频
def testUploadLocalVideo(accessKeyId, accessKeySecret, filePath, storageLocation=None):
    try:
        # 可以指定上传脚本部署的ECS区域。如果ECS区域和视频点播存储区域相同，则自动使用内网上传，上传更快且更省公网流量。
        # ecsRegionId ="cn-shanghai"
        # uploader = AliyunVodUploader(accessKeyId, accessKeySecret, ecsRegionId)
        # 不指定上传脚本部署的ECS区域。
        uploader = AliyunVodUploader(accessKeyId, accessKeySecret)
        uploadVideoRequest = UploadVideoRequest(filePath, 'aiOnLineVideo')
        # 可以设置视频封面，如果是本地或网络图片可使用UploadImageRequest上传图片到视频点播，获取到ImageURL
        #ImageURL示例：https://example.com/sample-****.jpg
        #uploadVideoRequest.setCoverURL('<your Image URL>')
        # 标签
        # uploadVideoRequest.setTags('taa')
        if storageLocation:
            uploadVideoRequest.setStorageLocation(storageLocation)
        videoId = uploader.uploadLocalVideo(uploadVideoRequest)
        print("videoId: %s" % (videoId))

    except AliyunVodException as e:
        print(e)
# testUploadLocalVideo(accessKeyId, accessKeySecret, filePath)
from alibabacloud_tea_util import models as util_models
import time
def get_video_url(video_id):
    config = open_api_models.Config(access_key_id=accessKeyId, access_key_secret=accessKeySecret)
    config.endpoint = f'vod.aliyuncs.com'
    client = vod20170321Client(config)
    get_play_info_request = vod_20170321_models.GetPlayInfoRequest(video_id=video_id)
    runtime = util_models.RuntimeOptions()
    start = time.time()
    while True:
        try:
            # 复制代码运行请自行打印 API 的返回值
            vod_20170321_models.GetPlayInfoResponse = client.get_play_info_with_options(get_play_info_request, runtime)
            play_url = vod_20170321_models.GetPlayInfoResponse.body.play_info_list.play_info[0].play_url
            return play_url
        except Exception as error:
            print("bbbbbbbbbbbbbb")
            print(error)
            time.sleep(5)
            end = time.time()
            result = int(end - start)
            if result > 1200:
                print("aaaaaaaa")
                raise error
import json
import traceback
from aliyunsdkcore.client import AcsClient
from aliyunsdkvod.request.v20170321 import CreateUploadVideoRequest
from aliyunsdkvod.request.v20170321 import GetPlayInfoRequest
from vodsdk.AliyunVodUtils import *
from vodsdk.AliyunVodUploader import AliyunVodUploader
from vodsdk.UploadVideoRequest import UploadVideoRequest
# 获取播放地址
def init_vod_client(accessKeyId, accessKeySecret):
    regionId = 'cn-shanghai'   # 点播服务接入地域
    connectTimeout = 3         # 连接超时，单位为秒
    return AcsClient(accessKeyId, accessKeySecret, regionId, auto_retry=True, max_retry_time=3, timeout=connectTimeout)
def get_play_info(clt, videoId):
    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))
    return response


if __name__ == '__main__':
    # testUploadLocalVideo(accessKeyId, accessKeySecret, "/home/thsw/chenyukun/video/百水河7.mp4")
    # print(Sample.get_play_info_sample(Sample.initialization('cn-shanghai'), 'dfaf3d140f714d9889562bff10a6f69a'))
    # print(get_video_url('3bb41d547bad44a7a9202017b8025838'))
    try:
        clt = init_vod_client('LTAI5tSJ62TLMUb4SZuf285A', 'MWYynm30filZ7x0HqSHlU3pdLVNeI7')
        playInfo = get_play_info(clt, "43e00a1a9d334c30b743d1cd6138207a")
        print(playInfo["PlayInfoList"]["PlayInfo"][0]["PlayURL"])
        print(json.dumps(playInfo, ensure_ascii=False, indent=4))

    except Exception as e:
        print("HTTP Status: 403" not in str(e))