DataCenterGame/data_processing/video_processing.py

import logging
import time
import requests
import hashlib
import ffmpeg
import os
import oss2
from model.sql_models import DB
from config import using_config as db_config

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)

    # 获取视频---matedata-title
    video_info_input = ffmpeg.probe(input_file)['format']
    video_title = ''
    try:
        if 'zx' in video_info_input['tags']['title']:
            video_title = video_info_input['tags']['title']
    except:
        pass

    if video_title[:2] == 'zx':
        # 如果有这一类数据,数据库比对matedata,获取download_path,等其他信息
        db_qc = DB(config=db_config.quchen_text)
        sql = '''
        select download_path ,video_meta_data ,size,video_length ,byte_rate ,
        width ,height ,type
        from video_info
        where video_meta_data='{}'
        '''.format(video_title)
        cursor = db_qc.session.execute(sql)
        lines = []
        for line in cursor.fetchall():
            lines.append(line)
        if len(lines) > 0:
            logging.info(video_url + ' 视频已经存在')
            cloud_filepath, metadata_title, video_size, duration, bit_rate, width, height, format = lines[0]
            return cloud_filepath, metadata_title, video_size, duration, bit_rate, width, height, format

    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):
    logging.info('准备上传视频')
    oss_config = db_config.oss_config
    access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', oss_config['OSS_TEST_ACCESS_KEY_ID'])
    access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', oss_config['OSS_TEST_ACCESS_KEY_SECRET'])
    bucket_name = os.getenv('OSS_TEST_BUCKET', oss_config['OSS_TEST_BUCKET'])
    endpoint = os.getenv('OSS_TEST_ENDPOINT', oss_config['OSS_TEST_ENDPOINT'])

    # 创建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()

    import logging
    from logging import handlers

    logging.basicConfig(
        handlers=[
            logging.handlers.RotatingFileHandler('video_process.log',
                                                 maxBytes=10 * 1024 * 1024,
                                                 backupCount=5,
                                                 encoding='utf-8')
            , logging.StreamHandler()  # 供输出使用
        ],
        level=logging.INFO,
        format="%(asctime)s - %(levelname)s %(filename)s %(funcName)s %(lineno)s - %(message)s"
    )
    # change_format(
    #     'https://zx-media-database.oss-cn-hangzhou.aliyuncs.com/video/zx_beb8a2fd5980bb6fb4f9cef486546338.mp4',
    #     file_md5='beb8a2fd5980bb6fb4f9cef486546338')
    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')