video_rewriting/common/ffmpeg.py

import configparser
import glob
import os
import random
import re
import subprocess
import sys
import time
import urllib.parse
import json

import requests
from datetime import datetime, timedelta
from urllib.parse import urlencode

class FFmpeg():

    """
    时间转换
    """
    @classmethod
    def seconds_to_srt_time(cls, seconds):
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        seconds = seconds % 60
        milliseconds = int((seconds - int(seconds)) * 1000)
        return f"{hours:02d}:{minutes:02d}:{int(seconds):02d},{milliseconds:03d}"

    """
    获取单个视频时长
    """
    @classmethod
    def get_video_duration(cls, video_url):
        ffprobe_cmd = [
            "ffprobe",
            "-i", video_url,
            "-show_entries", "format=duration",
            "-v", "quiet",
            "-of", "csv=p=0"
        ]
        output = subprocess.check_output(ffprobe_cmd).decode("utf-8").strip()
        return float(output)

    """
    获取视频文件的时长（秒）
    """
    @classmethod
    def get_videos_duration(cls, video_file):
        result = subprocess.run(
            ["ffprobe", "-v", "error", "-show_entries", "format=duration",
             "-of", "default=noprint_wrappers=1:nokey=1", video_file],
            capture_output=True, text=True)
        return float(result.stdout)

    """
    视频裁剪
    """
    @classmethod
    def video_tailor(cls, video_url):
        output_video_path = ''
        try:
            # 获取视频的原始宽高信息
            ffprobe_cmd = f"ffprobe -v error -select_streams v:0 -show_entries stream=width,height -of csv=p=0 {video_url}"
            ffprobe_process = subprocess.Popen(ffprobe_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
            output, _ = ffprobe_process.communicate()
            width, height = map(int, output.decode().strip().split(','))
            # 计算裁剪后的高度
            new_height = int(height * 0.7)
            # 构建 FFmpeg 命令,裁剪视频高度为原始高度的70%，并将宽度缩放为320x480
            ffmpeg_cmd = [
                "ffmpeg",
                "-i", video_url,
                "-vf", f"crop={width}:{new_height},scale=320:480",
                "-c:v", "libx264",
                "-c:a", "aac",
                "-y",
                output_video_path
            ]
            # 执行 FFmpeg 命令
            subprocess.run(ffmpeg_cmd, check=True)
            return output_video_path
        except Exception as e:
            return None

    """
     截取原视频最后一帧
    """
    @classmethod
    def video_png(cls, new_video_path, video_path_url, pw_random_id):
        """
        jpg_url 生成图片位置
        :param new_video_path: 视频地址
        :return:
        """
        jpg_url = video_path_url + str(pw_random_id) + 'png.jpg'
        # 获取视频时长
        total_duration = cls.get_video_duration(new_video_path)
        time_offset = total_duration - 1  # 提取倒数第一秒的帧
        # 获取视频最后一秒，生成.jpg
        subprocess.run(
            ['ffmpeg', '-ss', str(time_offset), '-i', new_video_path, '-t', str(total_duration), '-vf', 'fps=1', "-y", jpg_url])
        return jpg_url

    """
    获取视频音频
    """
    @classmethod
    def get_video_mp3(cls, video_file, video_path_url, pw_random_id):
        pw_mp3_path = video_path_url + str(pw_random_id) +'pw_video.mp3'
        command = [
            'ffmpeg',
            '-i', video_file,
            '-vn',  # 禁用视频流
            '-q:a', '0',  # 指定音频质量
            pw_mp3_path
        ]
        subprocess.run(command)
        time.sleep(1)
        return pw_mp3_path

    """
     生成片尾视频
    """
    @classmethod
    def pw_video(cls, jpg_url, video_path_url, pw_url, pw_srt, pw_random_id, pw_mp3_path):
        # 添加音频到图片
        """
        jpg_url 图片地址
        pw_video 提供的片尾视频
        pw_duration  提供的片尾视频时长
        new_video_path 视频位置
        subtitle_cmd 字幕
        pw_url 生成视频地址
        :return:
        """
        pw_srt_path = video_path_url + str(pw_random_id) +'pw_video.srt'
        # 创建临时字幕文件
        with open(pw_srt_path, 'w') as f:
            f.write(pw_srt)
        # 片尾位置
        pw_url_path = video_path_url + str(pw_random_id) + 'pw_video.mp4'
        # 获取视频时长
        pw_duration = cls.get_video_duration(pw_url)

        time.sleep(2)
        # 添加字幕 wqy-zenhei  Hiragino Sans GB
        subtitle_cmd = f"subtitles={pw_srt_path}:force_style='Fontsize=14,Fontname=wqy-zenhei,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000,Bold=1,MarginV=155'"
        background_cmd = "drawbox=y=366/2:color=yellow@1.0:width=iw:height=50:t=fill"
        ffmpeg_cmd = [
            'ffmpeg',
            '-loop', '1',
            '-i', jpg_url,  # 输入的图片文件
            '-i', pw_mp3_path,  # 输入的音频文件
            '-c:v', 'libx264',  # 视频编码格式
            '-t', str(pw_duration),  # 输出视频的持续时间，与音频持续时间相同
            '-pix_fmt', 'yuv420p',  # 像素格式
            '-c:a', 'aac',  # 音频编码格式
            '-strict', 'experimental',  # 使用实验性编码器
            '-shortest',  # 确保输出视频的长度与音频一致
            '-vf', f"scale=320x480,{background_cmd},{subtitle_cmd}",  # 视频过滤器，设置分辨率和其他过滤器
            pw_url_path  # 输出的视频文件路径
        ]
        subprocess.run(ffmpeg_cmd)
        return pw_url_path


    """
    设置统一格式拼接视频
    """
    @classmethod
    def concatenate_videos(cls, video_list, video_path_url):
        concatenate_videos_url = video_path_url + 'concatenate_videos.mp4'
        # 拼接视频
        VIDEO_COUNTER = 0
        FF_INPUT = ""
        FF_SCALE = ""
        FF_FILTER = ""
        ffmpeg_cmd = ["ffmpeg"]
        for videos in video_list:
            # 添加输入文件
            FF_INPUT += f" -i {videos}"
            # 为每个视频文件统一长宽，并设置SAR(采样宽高比)
            FF_SCALE += f"[{VIDEO_COUNTER}:v]scale=320x480,setsar=1[v{VIDEO_COUNTER}];"
            # 为每个视频文件创建一个输入流，并添加到-filter_complex参数中
            FF_FILTER += f"[v{VIDEO_COUNTER}][{VIDEO_COUNTER}:a]"
            # 增加视频计数器
            VIDEO_COUNTER += 1
        # 构建最终的FFmpeg命令
        ffmpeg_cmd.extend(FF_INPUT.split())
        ffmpeg_cmd.extend(["-filter_complex", f"{FF_SCALE}{FF_FILTER}concat=n={VIDEO_COUNTER}:v=1:a=1[v][a]",
                           "-map", "[v]", "-map", "[a]", "-y", concatenate_videos_url])
        subprocess.run(ffmpeg_cmd)
        return concatenate_videos_url


    """
    单个视频拼接
    """
    @classmethod
    def single_video(cls, new_video_path, video_share, video_path_url, zm):
        single_video_url = video_path_url + 'single_video.mp4'
        single_video_srt = video_path_url + 'single_video.srt'
        # 获取时长
        duration = cls.get_video_duration(new_video_path)
        start_time = cls.seconds_to_srt_time(0)
        end_time = cls.seconds_to_srt_time(duration)
        single_video_txt = video_path_url + 'single_video.txt'
        with open(single_video_txt, 'w') as f:
            f.write(f"file '{new_video_path}'\n")
        with open(single_video_srt, 'w') as f:
            f.write(f"1\n{start_time} --> {end_time}\n\u2764\uFE0F{zm}\n\n")
        background_cmd = "drawbox=y=370/1:color=yellow@1.0:width=iw:height=70:t=fill"
        if video_share == '有':
            # 添加字幕 wqy-zenhei  Hiragino Sans GB
            subtitle_cmd = f"subtitles={single_video_srt}:force_style='Fontsize=14,Fontname=wqy-zenhei,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000,Bold=1,MarginV=30'"
            draw = f"{background_cmd},{subtitle_cmd}"
        else:
            subtitle_cmd = f"subtitles={single_video_srt}:force_style='Fontsize=14,Fontname=wqy-zenhei,Outline=2,PrimaryColour=&H00FFFF,SecondaryColour=&H000000,Bold=1,MarginV=30'"
            draw = f"{subtitle_cmd}"
        # 多线程数
        num_threads = 4
        # 构建 FFmpeg 命令,生成视频
        ffmpeg_cmd_oss = [
                "ffmpeg",
                "-f", "concat",
                "-safe", "0",
                "-i",  f"{single_video_txt}",
                "-c:v", "libx264",
                "-c:a", "aac",
                "-threads", str(num_threads),
                "-vf", f"scale=320x480,{draw}",
                 "-y",
                 single_video_url

        ]
        subprocess.run(ffmpeg_cmd_oss)
        return single_video_url