FrameTour-RenderWorker/entity/ffmpeg.py

# 保留用于向后兼容的常量定义
import os

DEFAULT_ARGS = ("-shortest",)
ENCODER_ARGS = ("-c:v", "h264", ) if not os.getenv("ENCODER_ARGS", False) else os.getenv("ENCODER_ARGS", "").split(" ")
VIDEO_ARGS = ("-profile:v", "high", "-level:v", "4", ) if not os.getenv("VIDEO_ARGS", False) else os.getenv("VIDEO_ARGS", "").split(" ")
AUDIO_ARGS = ("-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", )
MUTE_AUDIO_INPUT = ("-f", "lavfi", "-i", "anullsrc=cl=stereo:r=48000", )


def get_mp4toannexb_filter():
    """
    Determine which mp4toannexb filter to use based on ENCODER_ARGS.
    Returns 'hevc_mp4toannexb' if ENCODER_ARGS contains 'hevc', otherwise 'h264_mp4toannexb'.
    """
    encoder_args_str = os.getenv("ENCODER_ARGS", "").lower()
    if "hevc" in encoder_args_str:
        return "hevc_mp4toannexb"
    return "h264_mp4toannexb"


class FfmpegTask(object):
    """
    兼容类：保留原有FfmpegTask接口用于向后兼容
    实际处理逻辑已迁移到新架构，该类主要用作数据载体
    """
    
    def __init__(self, input_file, task_type='copy', output_file=''):
        """保持原有构造函数签名"""
        self.annexb = False
        if type(input_file) is str:
            if input_file.endswith(".ts"):
                self.annexb = True
            self.input_file = [input_file]
        elif type(input_file) is list:
            self.input_file = input_file
        else:
            self.input_file = []
        self.zoom_cut = None
        self.center_cut = None
        self.ext_data = {}
        self.task_type = task_type
        self.output_file = output_file or ""
        self.mute = True
        self.speed = 1
        self.frame_rate = 25
        self.resolution = None
        self.subtitles = []
        self.luts = []
        self.audios = []
        self.overlays = []
        self.effects = []

    def __repr__(self):
        return f'FfmpegTask(input_file={self.input_file}, task_type={self.task_type})'

    def analyze_input_render_tasks(self):
        """分析输入中的子任务"""
        for i in self.input_file:
            if isinstance(i, FfmpegTask) and i.need_run():
                yield i

    def need_run(self):
        """判断是否需要运行"""
        if self.annexb:
            return True
        return not self.check_can_copy()

    def add_inputs(self, *inputs):
        """添加输入文件"""
        self.input_file.extend(inputs)

    def add_overlay(self, *overlays):
        """添加覆盖层"""
        for overlay in overlays:
            if str(overlay).endswith('.ass'):
                self.subtitles.append(overlay)
            else:
                self.overlays.append(overlay)
        self.correct_task_type()

    def add_audios(self, *audios):
        """添加音频"""
        self.audios.extend(audios)
        self.correct_task_type()

    def add_lut(self, *luts):
        """添加LUT"""
        self.luts.extend(luts)
        self.correct_task_type()

    def add_effect(self, *effects):
        """添加效果"""
        self.effects.extend(effects)
        self.correct_task_type()

    def get_output_file(self):
        """获取输出文件"""
        if self.task_type == 'copy':
            return self.input_file[0] if self.input_file else ""
        if not self.output_file:
            self.set_output_file()
        return self.output_file

    def correct_task_type(self):
        """校正任务类型"""
        if self.check_can_copy():
            self.task_type = 'copy'
        elif self.check_can_concat():
            self.task_type = 'concat'
        else:
            self.task_type = 'encode'

    def check_can_concat(self):
        """检查是否可以连接"""
        return (len(self.luts) == 0 and len(self.overlays) == 0 and 
                len(self.subtitles) == 0 and len(self.effects) == 0 and 
                self.speed == 1 and self.zoom_cut is None and self.center_cut is None)

    def check_can_copy(self):
        """检查是否可以复制"""
        return (len(self.luts) == 0 and len(self.overlays) == 0 and 
                len(self.subtitles) == 0 and len(self.effects) == 0 and 
                self.speed == 1 and len(self.audios) == 0 and len(self.input_file) <= 1 and
                self.zoom_cut is None and self.center_cut is None)

    def set_output_file(self, file=None):
        """设置输出文件"""
        if file is None:
            import uuid
            if self.annexb:
                self.output_file = f"rand_{uuid.uuid4()}.ts"
            else:
                self.output_file = f"rand_{uuid.uuid4()}.mp4"
        else:
            if isinstance(file, FfmpegTask):
                if file != self:
                    self.output_file = file.get_output_file()
            elif isinstance(file, str):
                self.output_file = file

    def check_annexb(self):
        """检查annexb格式"""
        return self.annexb

    def get_ffmpeg_args(self):
        """
        保留用于向后兼容，但实际逻辑已迁移到新架构
        建议使用新的 FFmpegCommandBuilder 来生成命令
        """
        # 简化版本，主要用于向后兼容
        if self.task_type == 'copy' and len(self.input_file) == 1:
            if isinstance(self.input_file[0], str):
                if self.input_file[0] == self.get_output_file():
                    return []
                return ['-y', '-hide_banner', '-i', self.input_file[0], '-c', 'copy', self.get_output_file()]
        
        # 对于复杂情况，返回基础命令结构
        # 实际处理会在新的服务架构中完成
        return ['-y', '-hide_banner', '-i'] + self.input_file + ['-c', 'copy', self.get_output_file()]