Python을 사용하여 클라우드 인터페이스에 연결하여 비디오 트랜스코딩 및 편집을 수행하는 방법

pip install requests
pip install json

import requests

def upload_video_file(api_key, api_secret, bucket_name, local_file_path):
    url = f'https://{bucket_name}.video.upyun.com/{local_file_path}'
    authorization = api_key + ":" + api_secret
    headers = {
        'Authorization': 'Basic ' + base64.b64encode(authorization.encode()).decode()
    }
    with open(local_file_path, 'rb') as file:
        data = file.read()
    response = requests.put(url, headers=headers, data=data)
    if response.status_code == 200:
        print("视频文件上传成功！")
    else:
        print("视频文件上传失败！")

api_key = 'your_api_key'
api_secret = 'your_api_secret'
bucket_name = 'your_bucket_name'
local_file_path = 'example.mp4'

upload_video_file(api_key, api_secret, bucket_name, local_file_path)

import requests

def transcode_video(api_key, api_secret, bucket_name, local_file_name, target_file_format, target_resolution):
    url = f'https://{bucket_name}.video.upyun.com/transcoding/'
    authorization = api_key + ":" + api_secret
    headers = {
        'Authorization': 'Basic ' + base64.b64encode(authorization.encode()).decode(),
        'X-Transcode-Target': target_file_format,
        'X-Transcode-Resolution': target_resolution
    }
    data = {
        'source': local_file_name
    }
    response = requests.post(url, headers=headers, data=data)
    if response.status_code == 201:
        task_id = response.json()['task_id']
        print(f"转码任务已创建，任务ID为{task_id}")
    else:
        print("转码任务创建失败！")

api_key = 'your_api_key'
api_secret = 'your_api_secret'
bucket_name = 'your_bucket_name'
local_file_name = 'example.mp4'
target_file_format = 'mp4'
target_resolution = '720p'

transcode_video(api_key, api_secret, bucket_name, local_file_name, target_file_format, target_resolution)

import requests

def query_task_status(api_key, api_secret, bucket_name, task_id):
    url = f'https://{bucket_name}.video.upyun.com/tasks/{task_id}/'
    authorization = api_key + ":" + api_secret
    headers = {
        'Authorization': 'Basic ' + base64.b64encode(authorization.encode()).decode()
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        task_status = response.json()['status']
        print(f"任务状态为{task_status}")
    else:
        print("查询任务状态失败！")

api_key = 'your_api_key'
api_secret = 'your_api_secret'
bucket_name = 'your_bucket_name'
task_id = 'your_task_id'

query_task_status(api_key, api_secret, bucket_name, task_id)

import requests

def download_transcoded_video(api_key, api_secret, bucket_name, task_id, local_file_name):
    url = f'https://{bucket_name}.video.upyun.com/tasks/{task_id}/download'
    authorization = api_key + ":" + api_secret
    headers = {
        'Authorization': 'Basic ' + base64.b64encode(authorization.encode()).decode()
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        with open(local_file_name, 'wb') as file:
            file.write(response.content)
        print("视频文件下载成功！")
    else:
        print("视频文件下载失败！")

api_key = 'your_api_key'
api_secret = 'your_api_secret'
bucket_name = 'your_bucket_name'
task_id = 'your_task_id'
local_file_name = 'output.mp4'

download_transcoded_video(api_key, api_secret, bucket_name, task_id, local_file_name)