ffmpeg_concat_videos.py

import subprocess
from pathlib import Path
import os
import json

class ConcatVideos:
    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "number_of_videos": ("INT", {"default": 2, "min": 2, "max": 50, "step": 1}),
                "output_filename": ("STRING", {"default": "concatenated.mp4"}),
                "use_python_ffmpeg": ("BOOLEAN", {"default": False}),
            },
            "optional": {
                "FFMPEG_CONFIG_JSON": ("STRING", {"forceInput": True}),
            },
            "hidden": {
                **{f"video_path_{i}": ("STRING", {"forceInput": True}) for i in range(1, 51)}
            }
        }

    RETURN_TYPES = ("STRING", "STRING",)
    RETURN_NAMES = ("concat_path", "ffmpeg_command",)
    FUNCTION = "concat_videos"
    OUTPUT_NODE = True
    CATEGORY = "Bjornulf"

    def __init__(self):
        self.work_dir = Path(os.path.abspath("temp_concat"))
        self.output_dir = Path(os.path.abspath("Bjornulf/concat_videos"))
        os.makedirs(self.work_dir, exist_ok=True)
        os.makedirs(self.output_dir, exist_ok=True)

    def concat_videos(self, number_of_videos: int, output_filename: str, 
                     use_python_ffmpeg: bool = False, 
                     FFMPEG_CONFIG_JSON: str = None, **kwargs):
        """
        Concatenate multiple videos using ffmpeg.
        Supports both subprocess and python-ffmpeg methods.
        """
        # Get and validate video paths
        video_paths = [kwargs[f"video_path_{i}"] for i in range(1, number_of_videos + 1) 
                      if f"video_path_{i}" in kwargs]
        
        video_paths = [os.path.abspath(path) for path in video_paths]
        for path in video_paths:
            if not Path(path).exists():
                raise ValueError(f"Video path does not exist: {path}")

        # Ensure output filename has mp4 extension
        output_filename = Path(output_filename).with_suffix('.mp4')
        output_path = self.output_dir / output_filename
        
        # Create concat file with absolute paths
        concat_file = self.work_dir / "concat.txt"
        with open(concat_file, 'w') as f:
            for path in video_paths:
                f.write(f"file '{path}'\n")

        # Default configuration
        config = {
            'ffmpeg': {'path': 'ffmpeg', 'use_python_ffmpeg': use_python_ffmpeg}
        }

        # If FFMPEG_CONFIG_JSON provided, parse and merge with default config
        if FFMPEG_CONFIG_JSON:
            try:
                json_config = json.loads(FFMPEG_CONFIG_JSON)
                # Merge JSON config, giving priority to use_python_ffmpeg from the node input
                config = {**json_config, 'ffmpeg': {**json_config.get('ffmpeg', {}), 'use_python_ffmpeg': use_python_ffmpeg}}
            except json.JSONDecodeError:
                raise ValueError("Invalid FFMPEG_CONFIG_JSON format")

        try:
            # Use python-ffmpeg if enabled
            if config.get('ffmpeg', {}).get('use_python_ffmpeg', False):
                import ffmpeg
                
                # Create input streams
                input_streams = [ffmpeg.input(path) for path in video_paths]
                
                # Set up output stream
                output_kwargs = {}
                
                # Video settings
                video_config = config.get('video', {})
                if video_config.get('codec') and video_config['codec'] != 'None':
                    output_kwargs['vcodec'] = video_config['codec']
                    
                    # Additional video encoding parameters
                    if video_config['codec'] != 'copy':
                        if video_config.get('bitrate'):
                            output_kwargs['video_bitrate'] = video_config['bitrate']
                        if video_config.get('crf') is not None:
                            output_kwargs['crf'] = video_config['crf']
                        if video_config.get('preset') and video_config['preset'] != 'None':
                            output_kwargs['preset'] = video_config['preset']
                
                # Audio settings
                audio_config = config.get('audio', {})
                if audio_config.get('enabled') is False or audio_config.get('codec') == 'None':
                    output_kwargs['an'] = None  # No audio
                elif audio_config.get('codec') and audio_config['codec'] != 'None':
                    output_kwargs['acodec'] = audio_config['codec']
                    if audio_config.get('bitrate'):
                        output_kwargs['audio_bitrate'] = audio_config['bitrate']
                
                # Concatenate and output
                output = ffmpeg.concat(*input_streams)
                output = output.output(str(output_path), **output_kwargs)
                
                # Compile and run the command
                ffmpeg_cmd = output.compile()
                output.run(overwrite_output=True)
                
                return str(output_path), ' '.join(ffmpeg_cmd)

            # Default to subprocess method
            else:
                # Default simple concatenation command
                cmd = [
                    'ffmpeg', '-y',
                    '-f', 'concat',
                    '-safe', '0',
                    '-i', str(concat_file),
                    '-c', 'copy',
                    '-movflags', '+faststart',
                    str(output_path)
                ]

                # If FFMPEG_CONFIG_JSON provided, modify command
                if FFMPEG_CONFIG_JSON:
                    cmd = [
                        config.get('ffmpeg', {}).get('path', 'ffmpeg'), '-y',
                        '-f', 'concat',
                        '-safe', '0',
                        '-i', str(concat_file)
                    ]

                    # Video codec settings
                    video_config = config.get('video', {})
                    if video_config.get('codec') and video_config['codec'] != 'None':
                        cmd.extend(['-c:v', video_config['codec']])
                        
                        # Add encoding parameters if not copying
                        if video_config['codec'] != 'copy':
                            if video_config.get('bitrate'):
                                cmd.extend(['-b:v', video_config['bitrate']])
                            if video_config.get('crf') is not None:
                                cmd.extend(['-crf', str(video_config['crf'])])
                            
                            # Add preset if specified
                            if video_config.get('preset') and video_config['preset'] != 'None':
                                cmd.extend(['-preset', video_config['preset']])
                            
                            # Add pixel format if specified
                            if video_config.get('pixel_format') and video_config['pixel_format'] != 'None':
                                cmd.extend(['-pix_fmt', video_config['pixel_format']])

                    # Audio settings
                    audio_config = config.get('audio', {})
                    if audio_config.get('enabled') is False or audio_config.get('codec') == 'None':
                        cmd.extend(['-an'])
                    elif audio_config.get('codec') and audio_config['codec'] != 'None':
                        cmd.extend(['-c:a', audio_config['codec']])
                        if audio_config.get('bitrate'):
                            cmd.extend(['-b:a', audio_config['bitrate']])

                    cmd.extend(['-movflags', '+faststart', str(output_path)])

                # Run subprocess command
                process = subprocess.run(
                    cmd,
                    check=True,
                    capture_output=True,
                    text=True
                )
                return str(output_path), ' '.join(cmd)
            
        except subprocess.CalledProcessError as e:
            raise RuntimeError(f"FFmpeg error: {e.stderr}")
        except Exception as e:
            raise RuntimeError(f"Error during video concatenation: {str(e)}")

    @classmethod
    def IS_CHANGED(cls, **kwargs):
        return float("NaN")