463 lines
20 KiB
Python
463 lines
20 KiB
Python
"""
|
||
视频合并节点
|
||
1. CreateAndSaveVideo: 从图片创建视频并保存，返回路径
|
||
2. ConcatVideoFiles: 使用 FFmpeg 合并多个视频文件（多行文本输入路径）
|
||
3. SaveAudioGetPath: 保存音频并返回文件路径
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import subprocess
|
||
import tempfile
|
||
import folder_paths
|
||
from fractions import Fraction
|
||
from comfy.cli_args import args
|
||
CATEGORY_TYPE = "🎈LAOGOU/Group"
|
||
|
||
class LG_CreateAndSaveVideo:
|
||
"""
|
||
从图片创建视频并保存到文件，返回文件路径
|
||
"""
|
||
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"images": ("IMAGE",),
|
||
"fps": ("FLOAT", {"default": 30.0, "min": 1.0, "max": 120.0, "step": 1.0}),
|
||
"filename_prefix": ("STRING", {"default": "video/segment"}),
|
||
},
|
||
"optional": {
|
||
"audio": ("AUDIO",),
|
||
},
|
||
"hidden": {
|
||
"prompt": "PROMPT",
|
||
"extra_pnginfo": "EXTRA_PNGINFO"
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING",)
|
||
RETURN_NAMES = ("file_path",)
|
||
FUNCTION = "create_and_save"
|
||
CATEGORY = CATEGORY_TYPE
|
||
DESCRIPTION = "从图片创建视频并保存，返回文件路径"
|
||
|
||
def create_and_save(self, images, fps, filename_prefix, audio=None, prompt=None, extra_pnginfo=None):
|
||
try:
|
||
from comfy_api.latest._input_impl.video_types import VideoFromComponents
|
||
from comfy_api.latest._util.video_types import VideoComponents, VideoContainer, VideoCodec
|
||
except ImportError:
|
||
from comfy_api.input_impl import VideoFromComponents
|
||
from comfy_api._util.video_types import VideoComponents, VideoContainer, VideoCodec
|
||
|
||
# 创建视频
|
||
components = VideoComponents(images=images, audio=audio, frame_rate=Fraction(fps))
|
||
video = VideoFromComponents(components)
|
||
|
||
# 保存
|
||
width, height = images.shape[2], images.shape[1]
|
||
output_dir = folder_paths.get_output_directory()
|
||
|
||
full_output_folder, filename, counter, subfolder, _ = folder_paths.get_save_image_path(
|
||
filename_prefix, output_dir, width, height
|
||
)
|
||
|
||
saved_metadata = None
|
||
if not args.disable_metadata:
|
||
metadata = {}
|
||
if extra_pnginfo is not None:
|
||
metadata.update(extra_pnginfo)
|
||
if prompt is not None:
|
||
metadata["prompt"] = prompt
|
||
if len(metadata) > 0:
|
||
saved_metadata = metadata
|
||
|
||
file = f"{filename}_{counter:05}_.mp4"
|
||
file_path = os.path.join(full_output_folder, file)
|
||
|
||
video.save_to(
|
||
file_path,
|
||
format=VideoContainer.MP4,
|
||
codec=VideoCodec.H264,
|
||
metadata=saved_metadata
|
||
)
|
||
|
||
return (file_path,)
|
||
|
||
|
||
class LG_ConcatVideoFiles:
|
||
"""
|
||
使用 FFmpeg 合并多个视频文件
|
||
支持字符串列表输入或多行文本输入
|
||
可选添加音频轨道
|
||
"""
|
||
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"video_paths": ("STRING", {"forceInput": True}),
|
||
"filename_prefix": ("STRING", {"default": "video/merged"}),
|
||
},
|
||
"optional": {
|
||
"reencode": ("BOOLEAN", {"default": False,
|
||
"tooltip": "False=直接拼接(快)，True=重新编码(兼容性好)"}),
|
||
"audio_path": ("STRING", {"forceInput": True,
|
||
"tooltip": "可选的音频文件路径，将替换或添加到合并后的视频"}),
|
||
"audio_mode": (["replace", "mix"], {"default": "replace",
|
||
"tooltip": "replace=替换原音频，mix=混合原音频和新音频"}),
|
||
"audio_volume": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 2.0, "step": 0.1,
|
||
"tooltip": "音频音量，1.0为原始音量"}),
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING",)
|
||
RETURN_NAMES = ("output_path",)
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "concat_files"
|
||
CATEGORY = CATEGORY_TYPE
|
||
INPUT_IS_LIST = True
|
||
DESCRIPTION = "使用 FFmpeg 合并多个视频文件，支持列表输入，可选添加音频"
|
||
|
||
def concat_files(self, video_paths, filename_prefix, reencode=None, audio_path=None, audio_mode=None, audio_volume=None):
|
||
# 处理其他参数（因为 INPUT_IS_LIST=True，所有参数都是列表）
|
||
filename_prefix = filename_prefix[0] if isinstance(filename_prefix, list) else filename_prefix
|
||
reencode = reencode[0] if isinstance(reencode, list) and reencode else False
|
||
audio_path = audio_path[0] if isinstance(audio_path, list) and audio_path else None
|
||
audio_mode = audio_mode[0] if isinstance(audio_mode, list) and audio_mode else "replace"
|
||
audio_volume = audio_volume[0] if isinstance(audio_volume, list) and audio_volume else 1.0
|
||
|
||
# 展平并处理路径列表
|
||
paths = []
|
||
for item in video_paths:
|
||
if isinstance(item, list):
|
||
# 嵌套列表
|
||
for p in item:
|
||
if isinstance(p, str) and p.strip():
|
||
paths.append(p.strip())
|
||
elif isinstance(item, str):
|
||
# 可能是多行文本或单个路径
|
||
if '\n' in item:
|
||
paths.extend([p.strip() for p in item.split('\n') if p.strip()])
|
||
elif item.strip():
|
||
paths.append(item.strip())
|
||
|
||
if len(paths) == 0:
|
||
raise ValueError("没有输入任何视频路径")
|
||
|
||
# 验证文件存在
|
||
for p in paths:
|
||
if not os.path.exists(p):
|
||
raise FileNotFoundError(f"视频文件不存在: {p}")
|
||
|
||
# 验证音频文件存在，不存在则跳过音频处理（不报错）
|
||
if audio_path and not os.path.exists(audio_path):
|
||
print(f"[ConcatVideoFiles] 警告：音频文件不存在，跳过音频合并: {audio_path}")
|
||
audio_path = None
|
||
|
||
output_dir = folder_paths.get_output_directory()
|
||
full_output_folder, filename, counter, subfolder, _ = folder_paths.get_save_image_path(
|
||
filename_prefix, output_dir, 0, 0
|
||
)
|
||
output_file = os.path.join(full_output_folder, f"{filename}_{counter:05}_.mp4")
|
||
|
||
# 创建 FFmpeg concat 列表
|
||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||
for p in paths:
|
||
escaped_path = p.replace("'", "'\\''")
|
||
f.write(f"file '{escaped_path}'\n")
|
||
concat_file = f.name
|
||
|
||
try:
|
||
if audio_path:
|
||
# 有音频输入时，需要两步处理：先合并视频，再添加音频
|
||
# 创建临时文件用于中间结果
|
||
temp_video = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False).name
|
||
|
||
# 第一步：合并视频片段
|
||
if reencode:
|
||
cmd1 = [
|
||
'ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', concat_file,
|
||
'-c:v', 'libx264', '-preset', 'medium', '-crf', '23',
|
||
'-an', # 不包含音频
|
||
'-movflags', '+faststart',
|
||
temp_video
|
||
]
|
||
else:
|
||
cmd1 = [
|
||
'ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', concat_file,
|
||
'-c:v', 'copy', '-an',
|
||
'-movflags', '+faststart',
|
||
temp_video
|
||
]
|
||
|
||
result = subprocess.run(cmd1, capture_output=True, text=True)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f"FFmpeg 合并视频错误:\n{result.stderr}")
|
||
|
||
# 第二步：添加音频
|
||
if audio_mode == "mix":
|
||
# 混合模式：先合并带原音频的视频，再混合新音频
|
||
# 重新合并带音频的视频
|
||
temp_video_with_audio = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False).name
|
||
if reencode:
|
||
cmd_audio = [
|
||
'ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', concat_file,
|
||
'-c:v', 'libx264', '-preset', 'medium', '-crf', '23',
|
||
'-c:a', 'aac', '-b:a', '128k',
|
||
'-movflags', '+faststart',
|
||
temp_video_with_audio
|
||
]
|
||
else:
|
||
cmd_audio = [
|
||
'ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', concat_file,
|
||
'-c', 'copy',
|
||
'-movflags', '+faststart',
|
||
temp_video_with_audio
|
||
]
|
||
result = subprocess.run(cmd_audio, capture_output=True, text=True)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f"FFmpeg 合并音频错误:\n{result.stderr}")
|
||
|
||
# 混合两个音频
|
||
cmd2 = [
|
||
'ffmpeg', '-y',
|
||
'-i', temp_video_with_audio,
|
||
'-i', audio_path,
|
||
'-filter_complex', f'[0:a][1:a]amix=inputs=2:duration=first:dropout_transition=2,volume={audio_volume}[a]',
|
||
'-map', '0:v', '-map', '[a]',
|
||
'-c:v', 'copy',
|
||
'-c:a', 'aac', '-b:a', '128k',
|
||
'-movflags', '+faststart',
|
||
'-shortest',
|
||
output_file
|
||
]
|
||
try:
|
||
result = subprocess.run(cmd2, capture_output=True, text=True)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f"FFmpeg 混合音频错误:\n{result.stderr}")
|
||
finally:
|
||
if os.path.exists(temp_video_with_audio):
|
||
os.unlink(temp_video_with_audio)
|
||
else:
|
||
# 替换模式：直接用新音频替换
|
||
volume_filter = f'volume={audio_volume}' if audio_volume != 1.0 else None
|
||
cmd2 = [
|
||
'ffmpeg', '-y',
|
||
'-i', temp_video,
|
||
'-i', audio_path,
|
||
'-map', '0:v', '-map', '1:a',
|
||
'-c:v', 'copy',
|
||
]
|
||
if volume_filter:
|
||
cmd2.extend(['-af', volume_filter])
|
||
cmd2.extend([
|
||
'-c:a', 'aac', '-b:a', '128k',
|
||
'-movflags', '+faststart',
|
||
'-shortest',
|
||
output_file
|
||
])
|
||
result = subprocess.run(cmd2, capture_output=True, text=True)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f"FFmpeg 添加音频错误:\n{result.stderr}")
|
||
|
||
# 清理临时文件
|
||
if os.path.exists(temp_video):
|
||
os.unlink(temp_video)
|
||
else:
|
||
# 无音频输入，原有逻辑
|
||
if reencode:
|
||
cmd = [
|
||
'ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', concat_file,
|
||
'-c:v', 'libx264', '-preset', 'medium', '-crf', '23',
|
||
'-c:a', 'aac', '-b:a', '128k',
|
||
'-movflags', '+faststart',
|
||
output_file
|
||
]
|
||
else:
|
||
cmd = [
|
||
'ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', concat_file,
|
||
'-c', 'copy',
|
||
'-movflags', '+faststart',
|
||
output_file
|
||
]
|
||
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f"FFmpeg 错误:\n{result.stderr}")
|
||
finally:
|
||
if os.path.exists(concat_file):
|
||
os.unlink(concat_file)
|
||
|
||
return {"ui": {"images": [{"filename": os.path.basename(output_file), "subfolder": subfolder, "type": "output"}], "animated": (True,)},
|
||
"result": (output_file,)}
|
||
|
||
|
||
class LG_SaveAudioGetPath:
|
||
"""
|
||
保存音频文件并返回文件路径
|
||
支持 FLAC、MP3、OPUS 格式
|
||
音频为空或无效时返回空字符串，不会中断工作流
|
||
"""
|
||
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"filename_prefix": ("STRING", {"default": "audio/ComfyUI"}),
|
||
"format": (["flac", "mp3", "opus"], {"default": "flac"}),
|
||
},
|
||
"optional": {
|
||
"audio": ("AUDIO",),
|
||
"quality": (["64k", "96k", "128k", "192k", "320k", "V0"], {"default": "128k",
|
||
"tooltip": "MP3/OPUS 比特率，FLAC 忽略此参数"}),
|
||
},
|
||
"hidden": {
|
||
"prompt": "PROMPT",
|
||
"extra_pnginfo": "EXTRA_PNGINFO"
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING",)
|
||
RETURN_NAMES = ("file_path",)
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "save_audio"
|
||
CATEGORY = CATEGORY_TYPE
|
||
DESCRIPTION = "保存音频文件并返回文件路径，音频无效时返回空字符串"
|
||
|
||
def save_audio(self, filename_prefix, format="flac", audio=None, quality="128k", prompt=None, extra_pnginfo=None):
|
||
# 检查音频是否有效
|
||
if audio is None:
|
||
print("[SaveAudioGetPath] 警告：音频输入为空，跳过保存")
|
||
return {"ui": {"audio": []}, "result": ("",)}
|
||
|
||
# 检查音频数据是否有效
|
||
try:
|
||
if "waveform" not in audio or audio["waveform"] is None:
|
||
print("[SaveAudioGetPath] 警告：音频数据无效（无 waveform），跳过保存")
|
||
return {"ui": {"audio": []}, "result": ("",)}
|
||
|
||
if audio["waveform"].numel() == 0:
|
||
print("[SaveAudioGetPath] 警告：音频数据为空，跳过保存")
|
||
return {"ui": {"audio": []}, "result": ("",)}
|
||
except Exception as e:
|
||
print(f"[SaveAudioGetPath] 警告：检查音频数据时出错 ({e})，跳过保存")
|
||
return {"ui": {"audio": []}, "result": ("",)}
|
||
|
||
import json
|
||
from io import BytesIO
|
||
import av
|
||
|
||
try:
|
||
import torchaudio
|
||
TORCH_AUDIO_AVAILABLE = True
|
||
except:
|
||
TORCH_AUDIO_AVAILABLE = False
|
||
|
||
try:
|
||
output_dir = folder_paths.get_output_directory()
|
||
full_output_folder, filename, counter, subfolder, _ = folder_paths.get_save_image_path(
|
||
filename_prefix, output_dir
|
||
)
|
||
|
||
# 构建元数据
|
||
metadata = {}
|
||
if not args.disable_metadata:
|
||
if prompt is not None:
|
||
metadata["prompt"] = json.dumps(prompt)
|
||
if extra_pnginfo is not None:
|
||
for x in extra_pnginfo:
|
||
metadata[x] = json.dumps(extra_pnginfo[x])
|
||
|
||
# Opus 支持的采样率
|
||
OPUS_RATES = [8000, 12000, 16000, 24000, 48000]
|
||
|
||
results = []
|
||
file_paths = []
|
||
|
||
for batch_number, waveform in enumerate(audio["waveform"].cpu()):
|
||
# 检查单个 waveform 是否有效
|
||
if waveform.numel() == 0:
|
||
print(f"[SaveAudioGetPath] 警告：批次 {batch_number} 的音频为空，跳过")
|
||
continue
|
||
|
||
filename_with_batch_num = filename.replace("%batch_num%", str(batch_number))
|
||
file = f"{filename_with_batch_num}_{counter:05}_.{format}"
|
||
output_path = os.path.join(full_output_folder, file)
|
||
|
||
sample_rate = audio["sample_rate"]
|
||
|
||
# 处理 Opus 采样率要求
|
||
if format == "opus":
|
||
if sample_rate > 48000:
|
||
sample_rate = 48000
|
||
elif sample_rate not in OPUS_RATES:
|
||
for rate in sorted(OPUS_RATES):
|
||
if rate > sample_rate:
|
||
sample_rate = rate
|
||
break
|
||
if sample_rate not in OPUS_RATES:
|
||
sample_rate = 48000
|
||
|
||
# 重采样
|
||
if sample_rate != audio["sample_rate"]:
|
||
if not TORCH_AUDIO_AVAILABLE:
|
||
print("[SaveAudioGetPath] 警告：torchaudio 不可用，无法重采样，跳过保存")
|
||
return {"ui": {"audio": []}, "result": ("",)}
|
||
waveform = torchaudio.functional.resample(waveform, audio["sample_rate"], sample_rate)
|
||
|
||
# 创建输出
|
||
output_buffer = BytesIO()
|
||
output_container = av.open(output_buffer, mode="w", format=format)
|
||
|
||
# 设置元数据
|
||
for key, value in metadata.items():
|
||
output_container.metadata[key] = value
|
||
|
||
layout = "mono" if waveform.shape[0] == 1 else "stereo"
|
||
|
||
# 设置输出流
|
||
if format == "opus":
|
||
out_stream = output_container.add_stream("libopus", rate=sample_rate, layout=layout)
|
||
bit_rates = {"64k": 64000, "96k": 96000, "128k": 128000, "192k": 192000, "320k": 320000}
|
||
out_stream.bit_rate = bit_rates.get(quality, 128000)
|
||
elif format == "mp3":
|
||
out_stream = output_container.add_stream("libmp3lame", rate=sample_rate, layout=layout)
|
||
if quality == "V0":
|
||
out_stream.codec_context.qscale = 1
|
||
else:
|
||
bit_rates = {"64k": 64000, "96k": 96000, "128k": 128000, "192k": 192000, "320k": 320000}
|
||
out_stream.bit_rate = bit_rates.get(quality, 128000)
|
||
else: # flac
|
||
out_stream = output_container.add_stream("flac", rate=sample_rate, layout=layout)
|
||
|
||
frame = av.AudioFrame.from_ndarray(
|
||
waveform.movedim(0, 1).reshape(1, -1).float().numpy(),
|
||
format="flt",
|
||
layout=layout,
|
||
)
|
||
frame.sample_rate = sample_rate
|
||
frame.pts = 0
|
||
output_container.mux(out_stream.encode(frame))
|
||
output_container.mux(out_stream.encode(None)) # Flush
|
||
output_container.close()
|
||
|
||
# 写入文件
|
||
output_buffer.seek(0)
|
||
with open(output_path, "wb") as f:
|
||
f.write(output_buffer.getbuffer())
|
||
|
||
results.append({"filename": file, "subfolder": subfolder, "type": "output"})
|
||
file_paths.append(output_path)
|
||
counter += 1
|
||
|
||
# 返回第一个文件的路径（通常只有一个）
|
||
return {"ui": {"audio": results}, "result": (file_paths[0] if file_paths else "",)}
|
||
|
||
except Exception as e:
|
||
print(f"[SaveAudioGetPath] 警告：保存音频时出错 ({e})，返回空路径")
|
||
return {"ui": {"audio": []}, "result": ("",)}
|
||
|
||
|