DataCenterGame/data_processing/video_processing.py

import logging
import time
import requests
import hashlib
import ffmpeg
import os
import oss2


def change_format(video_url, file_md5=None):
    now_time = int(time.time() * 1000)
    input_file = '/tmp/{}.mp4'.format(now_time)
    output_file = '/tmp/{}.mp4'.format(now_time + 1)
    rsp = requests.get(video_url)
    with open(input_file, 'wb') as f:
        f.write(rsp.content)
    if not file_md5:
        file_md5 = hashlib.md5(rsp.content).hexdigest()
        logging.info(file_md5)  # ac3ee699961c58ef80a78c2434efe0d0

    metadata_title = 'zx_' + file_md5
    ffmpeg.input(input_file).output(output_file, preset='slower', vcodec='h264',
                                    metadata='title=' + metadata_title).overwrite_output().run()
    video_info = ffmpeg.probe(output_file)['format']
    duration = video_info['duration']
    video_size = int(video_info['size'])
    bit_rate = (video_size / (float(duration) / 8))
    probe = ffmpeg.probe(input_file)
    format = probe['format']
    duration = format['duration']
    video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None)
    width = int(video_stream['width'])
    height = int(video_stream['height'])

    logging.info('视频时长：{}  视频大小：{}  视频比特率：{} 视频宽:{} 视频长:{}'.format(duration, video_size, bit_rate, width, height))

    cloud_filepath = 'video/{}.mp4'.format(metadata_title)

    video_format=video_url.split('.')[-1]
    format='mp4' if len(video_format)>4 else video_format

    # 上传视频
    update_to_aliyun(local_filepath=output_file, cloud_filepath=cloud_filepath)

    # 删除视频,上传视频
    os.remove(input_file)
    os.remove(output_file)
    return cloud_filepath, metadata_title, video_size, duration, bit_rate, width, height, format


def update_to_aliyun(local_filepath, cloud_filepath):
    # TODO:之后oss相关密码信息写入config
    access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', 'LTAI5tGAUywhGk8xDuXosquq')
    access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', 'VMVfRv6DdSYuBipIMU5UdtHBYB0Jjb')
    bucket_name = os.getenv('OSS_TEST_BUCKET', 'zx-media-database')
    endpoint = os.getenv('OSS_TEST_ENDPOINT', 'http://oss-cn-hangzhou.aliyuncs.com')

    # 创建Bucket对象，所有Object相关的接口都可以通过Bucket对象来进行
    bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

    # 查询文件是否已经在云上
    name_set = set()
    for i, object_info in enumerate(oss2.ObjectIterator(bucket)):
        name_set.add(object_info.key)
    if cloud_filepath not in name_set:
        logging.info('上传视频' + cloud_filepath)
        # 数据不存在于云上,则进行数据存储

        # 进行数据存储
        bucket.put_object_from_file(cloud_filepath, local_filepath)


if __name__ == '__main__':
    import threading
    for i in range(100):
        x_thread=threading.Thread(target=change_format,args=('http://wxsnsdy.wxs.qq.com/131/20210/snssvpdownload/SH/reserved/ads_svp_video__0b53qybncaaciaadpfl53jqbrbqe2gdafuka.f110002.mp4?dis_k=2982f4fc3f191cb6199045ca072ed033&dis_t=1618995038&m=beb8a2fd5980bb6fb4f9cef486546338',))
        x_thread.start()

    change_format('http://wxsnsdy.wxs.qq.com/131/20210/snssvpdownload/SH/reserved/ads_svp_video__0b53qybncaaciaadpfl53jqbrbqe2gdafuka.f110002.mp4?dis_k=2982f4fc3f191cb6199045ca072ed033&dis_t=1618995038&m=beb8a2fd5980bb6fb4f9cef486546338')