forked from LiveCarta/ContentGeneration
102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Merge videos/output_n.mp4 with audios/output_n.mp3 into merged/merged_n.mp4."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from src.logging_config import configure_logging
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
DEFAULT_BASE_DIR = SCRIPT_DIR.parents[1]
|
|
DEFAULT_VIDEOS_DIR = DEFAULT_BASE_DIR / "videos"
|
|
DEFAULT_AUDIOS_DIR = DEFAULT_BASE_DIR / "audios"
|
|
DEFAULT_OUTPUT_DIR = DEFAULT_BASE_DIR / "merged"
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def shot_number(path: Path) -> int:
|
|
match = re.search(r"output_(\d+)\.mp4$", path.name)
|
|
return int(match.group(1)) if match else -1
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--videos-dir", type=Path, default=DEFAULT_VIDEOS_DIR)
|
|
parser.add_argument("--audios-dir", type=Path, default=DEFAULT_AUDIOS_DIR)
|
|
parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
|
|
parser.add_argument(
|
|
"--allow-missing-audio",
|
|
action="store_true",
|
|
help="If set, create merged output from video only when audio is missing.",
|
|
)
|
|
parser.add_argument("--log-level", default="INFO")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
configure_logging(args.log_level)
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
videos = sorted(args.videos_dir.glob("output_*.mp4"), key=shot_number)
|
|
if not videos:
|
|
LOGGER.warning("No videos found in %s", args.videos_dir)
|
|
return 1
|
|
|
|
for video in videos:
|
|
num = shot_number(video)
|
|
audio = args.audios_dir / f"output_{num}.mp3"
|
|
output = args.output_dir / f"merged_{num}.mp4"
|
|
|
|
if output.exists():
|
|
LOGGER.info("Already exists; skipped shot %s", num)
|
|
continue
|
|
|
|
if not audio.exists():
|
|
if args.allow_missing_audio:
|
|
LOGGER.warning(
|
|
"No audio found for shot %s (%s); using video-only output",
|
|
num,
|
|
audio,
|
|
)
|
|
shutil.copy2(video, output)
|
|
LOGGER.info("Done (video-only): %s", output)
|
|
continue
|
|
|
|
LOGGER.warning("No audio found for shot %s (%s); skipped", num, audio)
|
|
continue
|
|
|
|
LOGGER.info("Merging shot %s: %s + %s -> %s", num, video, audio, output)
|
|
subprocess.run(
|
|
[
|
|
"ffmpeg",
|
|
"-i",
|
|
str(video),
|
|
"-i",
|
|
str(audio),
|
|
"-c:v",
|
|
"copy",
|
|
"-c:a",
|
|
"aac",
|
|
"-shortest",
|
|
"-y",
|
|
str(output),
|
|
],
|
|
check=True,
|
|
)
|
|
LOGGER.info("Done: %s", output)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|