ks_automation/common/ffmpeg.py

import subprocess
import time



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:
            # 获取视频的原始宽高信息
            width, height = cls.get_w_h_size(video_url)
            # 计算裁剪后的高度
            new_height = int(height * 0.8)
            # 构建 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 get_w_h_size(cls, new_video_path):
        try:
            # 获取视频的原始宽高信息
            ffprobe_cmd = f"ffprobe -v error -select_streams v:0 -show_entries stream=width,height -of csv=p=0 {new_video_path}"
            ffprobe_process = subprocess.Popen(ffprobe_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
            output, _ = ffprobe_process.communicate()
            output_decoded = output.decode().strip()
            split_output = [value for value in output_decoded.split(',') if value.strip()]
            height, width = map(int, split_output)
            return width, height
        except ValueError as e:
            return 1920, 1080

    """
    视频裁剪
    """
    @classmethod
    def video_crop(cls, new_video_path, video_path_url, pw_random_id):
        crop_url = video_path_url + str(pw_random_id) + 'crop.mp4'
        # 获取视频的原始宽高信息
        ffprobe_cmd = f"ffprobe -v error -select_streams v:0 -show_entries stream=width,height -of csv=p=0 {new_video_path}"
        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.8)

        # 构建 FFmpeg 命令,裁剪视频高度为原始高度的80%
        ffmpeg_cmd = [
            "ffmpeg",
            "-i", new_video_path,
            "-vf", f"crop={width}:{new_height}",
            "-c:v", "libx264",
            "-c:a", "aac",
            "-y",
            crop_url
        ]
        subprocess.run(ffmpeg_cmd)
        return crop_url

    """
    视频裁剪
    """
    @classmethod
    def video_ggduration(cls, new_video_path, video_path_url, pw_random_id, gg_duration_total):
        gg_duration_url = video_path_url + str(pw_random_id) + 'gg_duration.mp4'
        # 获取视频时长
        total_duration = cls.get_video_duration(new_video_path)
        duration = int(total_duration) - int(gg_duration_total)
        if int(total_duration) < int(gg_duration_total):
            return new_video_path
        ffmpeg_cmd = [
            "ffmpeg",
            "-i", new_video_path,
            "-c:v", "libx264",
            "-c:a", "aac",
            "-t", str(duration),
            "-y",
            gg_duration_url
        ]
        subprocess.run(ffmpeg_cmd)
        return gg_duration_url

    """
     截取原视频最后一帧
    """
    @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,
            '-q:a', '0',
            '-map', 'a',
            # '-codec:a', 'libmp3lame',  # 指定 MP3 编码器
            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
        height = 1080
        margin_v = int(height) // 8  # 可根据需要调整字幕和背景之间的距离
        subtitle_cmd = f"subtitles={pw_srt_path}:force_style='Fontsize=14,Fontname=wqy-zenhei,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000,Bold=1,MarginV={margin_v}'"
        bg_position_offset = (int(height) - margin_v) / 1.75
        background_cmd = f"drawbox=y=(ih-{int(height)}/2-{bg_position_offset}):color=yellow@1.0:width=iw:height={int(height)}/4: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=1080x1920,{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'
        # 获取视频的原始宽高信息
        width, height = cls.get_w_h_size(video_list[0])
        # 拼接视频
        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={int(height)}x{int(width)},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")
        width, height = cls.get_w_h_size(new_video_path)
        box_height = int(int(height) / 4)  # 框的高度为视频高度的四分之一

        background_cmd = f"drawbox=y=ih-{70 + box_height}-{int(box_height / 20)}:color=yellow@1.0:width=iw:height={box_height}: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=20'"
            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=20'"
            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"{draw}",
                 "-y",
                 single_video_url

        ]
        subprocess.run(ffmpeg_cmd_oss)
        return single_video_url