663 lines
24 KiB
Python
663 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""Render titled daily video manifests into crossfaded MP4 videos."""
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from fractions import Fraction
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
# --- CONFIGURATION ---
|
|
XFADE_DURATION = 1.5 # Crossfade duration between every pair of segments (seconds)
|
|
FADE_IN_OUT_DURATION = 1.5 # Fade duration at the beginning and end (seconds)
|
|
TITLE_START = 3.0 # Time at which the title starts appearing (seconds)
|
|
TITLE_DURATION = 5.0 # Full-opacity hold time between the title fades (seconds)
|
|
TITLE_FADE_DURATION = 0.75 # Title fade-in and fade-out duration (seconds)
|
|
DEFAULT_OUTPUT_FPS = Fraction(30, 1)
|
|
TITLE_FONT_SIZE = 50
|
|
TITLE_FONT_FAMILY = "MV Boli"
|
|
TITLE_FONT_FILE = r"C:\Windows\Fonts\MVBOLI.TTF"
|
|
TITLE_MARGIN_RIGHT = 72
|
|
TITLE_MARGIN_BOTTOM = 60
|
|
TITLE_LINE_SPACING = 10
|
|
TITLE_RENDERING_PADDING_SPACES = 2 # Extra invisible line width for italic glyph overhang
|
|
PREFER_HARDWARE_ENCODING = True
|
|
HARDWARE_ENCODER = "h264_nvenc"
|
|
HARDWARE_PRESET = "p4"
|
|
HARDWARE_CQ = "18"
|
|
SOFTWARE_ENCODER = "libx264"
|
|
SOFTWARE_PRESET = "medium"
|
|
SOFTWARE_CRF = "18"
|
|
|
|
|
|
@dataclass
|
|
class Segment:
|
|
input_index: int
|
|
start: float
|
|
end: float
|
|
|
|
@property
|
|
def duration(self):
|
|
return self.end - self.start
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VideoProfile:
|
|
width: int
|
|
height: int
|
|
fps: Optional[Fraction]
|
|
|
|
|
|
def parse_frame_rate(stream):
|
|
for key in ("avg_frame_rate", "r_frame_rate"):
|
|
value = stream.get(key)
|
|
if not value:
|
|
continue
|
|
try:
|
|
frame_rate = Fraction(value)
|
|
except (ValueError, ZeroDivisionError):
|
|
continue
|
|
if frame_rate > 0:
|
|
return frame_rate
|
|
return None
|
|
|
|
|
|
def probe_media(path, ffprobe):
|
|
result = subprocess.run(
|
|
[ffprobe, "-v", "error", "-show_format", "-show_streams", "-of", "json", str(path)],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
data = json.loads(result.stdout)
|
|
video_stream = next(
|
|
(
|
|
stream for stream in data.get("streams", [])
|
|
if stream.get("codec_type") == "video"
|
|
and not stream.get("disposition", {}).get("attached_pic", 0)
|
|
),
|
|
None,
|
|
)
|
|
if video_stream is None:
|
|
raise ValueError("no video stream")
|
|
|
|
duration_values = [data.get("format", {}).get("duration")]
|
|
duration_values.append(video_stream.get("duration"))
|
|
duration = None
|
|
for value in duration_values:
|
|
try:
|
|
candidate = float(value)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if math.isfinite(candidate) and candidate > 0:
|
|
duration = candidate
|
|
break
|
|
if duration is None:
|
|
raise ValueError("no positive duration")
|
|
has_audio = any(stream.get("codec_type") == "audio" for stream in data.get("streams", []))
|
|
|
|
try:
|
|
width = int(video_stream["width"])
|
|
height = int(video_stream["height"])
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise ValueError("video stream has no valid resolution") from exc
|
|
if width <= 0 or height <= 0:
|
|
raise ValueError("video stream has no valid resolution")
|
|
|
|
rotation = video_stream.get("tags", {}).get("rotate")
|
|
for side_data in video_stream.get("side_data_list", []):
|
|
if "rotation" in side_data:
|
|
rotation = side_data["rotation"]
|
|
break
|
|
try:
|
|
rotated_degrees = round(float(rotation or 0))
|
|
except (TypeError, ValueError):
|
|
rotated_degrees = 0
|
|
if abs(rotated_degrees) % 180 == 90:
|
|
width, height = height, width
|
|
|
|
return duration, has_audio, VideoProfile(width, height, parse_frame_rate(video_stream))
|
|
|
|
|
|
def read_json(path):
|
|
"""Read strict JSON and common JSON5-style LosslessCut project files."""
|
|
raw = path.read_text(encoding="utf-8-sig")
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
# LosslessCut project files may use JSON5 syntax. Normalize comments,
|
|
# single-quoted strings, unquoted keys, and trailing commas.
|
|
without_comments = strip_json5_comments(raw)
|
|
converted = convert_single_quoted_strings(without_comments)
|
|
converted = re.sub(r"([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:", r'\1"\2":', converted)
|
|
converted = re.sub(r",\s*([}\]])", r"\1", converted)
|
|
return json.loads(converted)
|
|
|
|
|
|
def strip_json5_comments(value):
|
|
result = []
|
|
quote = None
|
|
index = 0
|
|
while index < len(value):
|
|
char = value[index]
|
|
if quote:
|
|
result.append(char)
|
|
if char == "\\" and index + 1 < len(value):
|
|
index += 1
|
|
result.append(value[index])
|
|
elif char == quote:
|
|
quote = None
|
|
elif char in ("'", '"'):
|
|
quote = char
|
|
result.append(char)
|
|
elif char == "/" and index + 1 < len(value) and value[index + 1] == "/":
|
|
newline = value.find("\n", index + 2)
|
|
if newline == -1:
|
|
break
|
|
result.append("\n")
|
|
index = newline
|
|
elif char == "/" and index + 1 < len(value) and value[index + 1] == "*":
|
|
end = value.find("*/", index + 2)
|
|
if end == -1:
|
|
raise ValueError("unterminated comment in JSON5 project")
|
|
result.extend("\n" for line in value[index:end + 2].splitlines()[1:])
|
|
index = end + 1
|
|
else:
|
|
result.append(char)
|
|
index += 1
|
|
return "".join(result)
|
|
|
|
|
|
def convert_single_quoted_strings(value):
|
|
result = []
|
|
index = 0
|
|
while index < len(value):
|
|
if value[index] == '"':
|
|
result.append(value[index])
|
|
index += 1
|
|
while index < len(value):
|
|
char = value[index]
|
|
result.append(char)
|
|
index += 1
|
|
if char == "\\" and index < len(value):
|
|
result.append(value[index])
|
|
index += 1
|
|
elif char == '"':
|
|
break
|
|
continue
|
|
if value[index] != "'":
|
|
result.append(value[index])
|
|
index += 1
|
|
continue
|
|
result.append('"')
|
|
index += 1
|
|
while index < len(value):
|
|
char = value[index]
|
|
if char == "'":
|
|
result.append('"')
|
|
index += 1
|
|
break
|
|
if char == "\\" and index + 1 < len(value):
|
|
following = value[index + 1]
|
|
if following == "'":
|
|
result.append("'")
|
|
elif following == '"':
|
|
result.append('\\"')
|
|
else:
|
|
result.extend(("\\", following))
|
|
index += 2
|
|
continue
|
|
if char == '"':
|
|
result.append('\\"')
|
|
else:
|
|
result.append(char)
|
|
index += 1
|
|
else:
|
|
raise ValueError("unterminated single-quoted string in JSON5 project")
|
|
return "".join(result)
|
|
|
|
|
|
def resolve_manifest_item(value, manifest_path):
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError("manifest video entry must be a non-empty path string")
|
|
|
|
path = Path(value).expanduser()
|
|
if not path.is_absolute():
|
|
path = manifest_path.parent / path
|
|
path = path.resolve()
|
|
if path.suffix.lower() != ".llc":
|
|
return path, None
|
|
|
|
project = read_json(path)
|
|
if not isinstance(project, dict):
|
|
raise ValueError(f"LosslessCut project is not an object: {path}")
|
|
media_name = project.get("mediaFileName")
|
|
if not isinstance(media_name, str) or not media_name.strip():
|
|
raise ValueError(f"LosslessCut project has no mediaFileName: {path}")
|
|
media_path = Path(media_name).expanduser()
|
|
if not media_path.is_absolute():
|
|
media_path = path.parent / media_path
|
|
media_path = media_path.resolve()
|
|
cut_segments = project.get("cutSegments") or []
|
|
if not isinstance(cut_segments, list):
|
|
raise ValueError(f"LosslessCut cutSegments must be an array: {path}")
|
|
return media_path, cut_segments
|
|
|
|
|
|
def load_manifest_segments(manifest_path, videos, ffprobe):
|
|
input_paths = []
|
|
input_audio = []
|
|
input_profiles = []
|
|
segments = []
|
|
|
|
for input_index, value in enumerate(videos):
|
|
media_path, cut_segments = resolve_manifest_item(value, manifest_path)
|
|
if not media_path.is_file():
|
|
raise ValueError(f"video file does not exist: {media_path}")
|
|
duration, has_audio, profile = probe_media(media_path, ffprobe)
|
|
fps_text = f"{float(profile.fps):g} fps" if profile.fps is not None else "unknown fps"
|
|
print(f" Probe {media_path.name}: {profile.width}x{profile.height}, {fps_text}")
|
|
input_paths.append(media_path)
|
|
input_audio.append(has_audio)
|
|
input_profiles.append(profile)
|
|
|
|
if cut_segments is None or not cut_segments:
|
|
segments.append(Segment(input_index, 0.0, duration))
|
|
continue
|
|
|
|
# LosslessCut cutSegments are the time ranges to export. Markers without
|
|
# an end time do not contain media and are ignored.
|
|
for segment in cut_segments:
|
|
if not isinstance(segment, dict) or segment.get("end") is None:
|
|
continue
|
|
try:
|
|
start = float(segment.get("start", 0))
|
|
end = float(segment["end"])
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"invalid cut segment in {value}: {segment!r}") from exc
|
|
start = max(0.0, start)
|
|
end = min(duration, end)
|
|
if not math.isfinite(start) or not math.isfinite(end) or end <= start:
|
|
continue
|
|
segments.append(Segment(input_index, start, end))
|
|
|
|
if not segments:
|
|
raise ValueError("manifest has no renderable video segments")
|
|
if len(segments) > 1:
|
|
for segment in segments:
|
|
if segment.duration <= XFADE_DURATION:
|
|
path = input_paths[segment.input_index]
|
|
raise ValueError(
|
|
f"segment in {path} is {segment.duration:.2f}s; every segment must be longer "
|
|
f"than the {XFADE_DURATION:.2f}s crossfade"
|
|
)
|
|
maximum_resolution = max(
|
|
input_profiles,
|
|
key=lambda profile: (profile.width * profile.height, profile.width, profile.height),
|
|
)
|
|
maximum_fps = max(
|
|
(profile.fps for profile in input_profiles if profile.fps is not None),
|
|
default=DEFAULT_OUTPUT_FPS,
|
|
)
|
|
return (
|
|
input_paths,
|
|
input_audio,
|
|
segments,
|
|
maximum_resolution.width,
|
|
maximum_resolution.height,
|
|
maximum_fps,
|
|
)
|
|
|
|
|
|
def measure_title_widths(lines):
|
|
"""Estimate relative line widths so separate FFmpeg lines stay left-aligned."""
|
|
try:
|
|
import tkinter as tk
|
|
from tkinter import font as tkfont
|
|
except ImportError:
|
|
tk = None
|
|
|
|
if tk is not None:
|
|
root = None
|
|
try:
|
|
root = tk.Tk()
|
|
root.withdraw()
|
|
title_font = tkfont.Font(root=root, family=TITLE_FONT_FAMILY, size=-TITLE_FONT_SIZE)
|
|
return tuple(title_font.measure(line) for line in lines)
|
|
except tk.TclError:
|
|
# Keep rendering available in headless environments without Tk.
|
|
pass
|
|
finally:
|
|
if root is not None:
|
|
root.destroy()
|
|
return tuple(math.ceil(len(line) * TITLE_FONT_SIZE * 0.62) for line in lines)
|
|
|
|
|
|
def make_filter_graph(
|
|
input_audio,
|
|
segments,
|
|
date_file_name,
|
|
title_file_name,
|
|
title_lines,
|
|
output_width,
|
|
output_height,
|
|
output_fps,
|
|
):
|
|
width = max(2, output_width + output_width % 2)
|
|
height = max(2, output_height + output_height % 2)
|
|
fps = f"{output_fps.numerator}/{output_fps.denominator}"
|
|
filters = []
|
|
|
|
for segment_index, segment in enumerate(segments):
|
|
source = segment.input_index
|
|
start = segment.start
|
|
end = segment.end
|
|
filters.append(
|
|
f"[{source}:v:0]trim=start={start:.6f}:end={end:.6f},"
|
|
f"setpts=PTS-STARTPTS,"
|
|
f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
|
|
f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:color=black,"
|
|
f"setsar=1,fps={fps},format=yuv420p,settb=AVTB[v{segment_index}]"
|
|
)
|
|
if input_audio[source]:
|
|
filters.append(
|
|
f"[{source}:a:0]atrim=start={start:.6f}:end={end:.6f},"
|
|
f"asetpts=PTS-STARTPTS,aresample=48000,"
|
|
f"aformat=sample_fmts=fltp:channel_layouts=stereo[a{segment_index}]"
|
|
)
|
|
else:
|
|
filters.append(
|
|
f"anullsrc=r=48000:cl=stereo,atrim=duration={segment.duration:.6f}[a{segment_index}]"
|
|
)
|
|
|
|
current_video = "v0"
|
|
current_audio = "a0"
|
|
output_duration = segments[0].duration
|
|
for segment_index in range(1, len(segments)):
|
|
offset = output_duration - XFADE_DURATION
|
|
out_video = f"xv{segment_index}"
|
|
out_audio = f"xa{segment_index}"
|
|
filters.append(
|
|
f"[{current_video}][v{segment_index}]xfade=transition=fade:"
|
|
f"duration={XFADE_DURATION:.3f}:offset={offset:.6f}[{out_video}]"
|
|
)
|
|
filters.append(
|
|
f"[{current_audio}][a{segment_index}]acrossfade=d={XFADE_DURATION:.3f}[{out_audio}]"
|
|
)
|
|
current_video = out_video
|
|
current_audio = out_audio
|
|
output_duration += segments[segment_index].duration - XFADE_DURATION
|
|
|
|
fade_duration = min(FADE_IN_OUT_DURATION, output_duration / 2)
|
|
fade_out_start = max(0.0, output_duration - fade_duration)
|
|
filters.append(
|
|
f"[{current_video}]fade=t=in:st=0:d={fade_duration:.6f},"
|
|
f"fade=t=out:st={fade_out_start:.6f}:d={fade_duration:.6f}[fadedv]"
|
|
)
|
|
filters.append(
|
|
f"[{current_audio}]afade=t=in:st=0:d={fade_duration:.6f},"
|
|
f"afade=t=out:st={fade_out_start:.6f}:d={fade_duration:.6f}[fadeda]"
|
|
)
|
|
|
|
title_fade = min(TITLE_FADE_DURATION, TITLE_DURATION / 2)
|
|
title_fade_out_start = TITLE_START + title_fade + TITLE_DURATION
|
|
title_end = title_fade_out_start + title_fade
|
|
alpha = (
|
|
f"if(lt(t,{TITLE_START:.3f}),0,"
|
|
f"if(lt(t,{TITLE_START + title_fade:.3f}),(t-{TITLE_START:.3f})/{title_fade:.3f},"
|
|
f"if(lt(t,{title_fade_out_start:.3f}),1,"
|
|
f"if(lt(t,{title_end:.3f}),({title_end:.3f}-t)/{title_fade:.3f},0))))"
|
|
)
|
|
date_width, title_width = measure_title_widths(title_lines)
|
|
padding_width = measure_title_widths((" " * TITLE_RENDERING_PADDING_SPACES,))[0]
|
|
if title_width >= date_width:
|
|
date_x = f"w-text_w-{TITLE_MARGIN_RIGHT + title_width - date_width}+{padding_width}"
|
|
title_x = f"w-text_w-{TITLE_MARGIN_RIGHT}+{padding_width}"
|
|
else:
|
|
date_x = f"w-text_w-{TITLE_MARGIN_RIGHT}+{padding_width}"
|
|
title_x = f"w-text_w-{TITLE_MARGIN_RIGHT + date_width - title_width}+{padding_width}"
|
|
font_option = f"fontfile='{escape_filter_path(TITLE_FONT_FILE)}':" if TITLE_FONT_FILE else ""
|
|
title_style = (
|
|
f"{font_option}expansion=none:fontsize={TITLE_FONT_SIZE}:fontcolor=white:"
|
|
"borderw=2:bordercolor=black:shadowcolor=black@0.8:shadowx=2:shadowy=2:"
|
|
"fix_bounds=1:"
|
|
)
|
|
date_y = f"h-text_h-{TITLE_MARGIN_BOTTOM + TITLE_FONT_SIZE + TITLE_LINE_SPACING}"
|
|
title_y = f"h-text_h-{TITLE_MARGIN_BOTTOM}"
|
|
filters.append(
|
|
f"[fadedv]drawtext=textfile={date_file_name}:{title_style}"
|
|
f"x={date_x}:y={date_y}:alpha='{alpha}',"
|
|
f"drawtext=textfile={title_file_name}:{title_style}"
|
|
f"x={title_x}:y={title_y}:alpha='{alpha}'[outv]"
|
|
)
|
|
filters.append("[fadeda]anull[outa]")
|
|
return ";".join(filters), output_duration
|
|
|
|
|
|
def make_title_text(manifest_path, title):
|
|
try:
|
|
manifest_date = date.fromisoformat(manifest_path.stem)
|
|
date_text = f"{manifest_date:%B} {manifest_date.day}, {manifest_date.year}"
|
|
except ValueError:
|
|
date_text = manifest_path.stem
|
|
title_text = title.strip().replace("\r", " ").replace("\n", " ")
|
|
return date_text, title_text
|
|
|
|
|
|
def escape_filter_path(value):
|
|
return str(value).replace("\\", "/").replace(":", "\\:").replace("'", "\\'")
|
|
|
|
|
|
def ffmpeg_has_encoder(ffmpeg, encoder):
|
|
result = subprocess.run(
|
|
[ffmpeg, "-hide_banner", "-encoders"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
return re.search(rf"\b{re.escape(encoder)}\b", result.stdout) is not None
|
|
|
|
|
|
def ffmpeg_command(ffmpeg, input_paths, graph, output_path, overwrite, encoder):
|
|
command = [ffmpeg, "-hide_banner", "-y" if overwrite else "-n"]
|
|
for path in input_paths:
|
|
command.extend(["-i", str(path)])
|
|
command.extend([
|
|
"-filter_complex", graph,
|
|
"-map", "[outv]",
|
|
"-map", "[outa]",
|
|
])
|
|
if encoder == HARDWARE_ENCODER:
|
|
command.extend([
|
|
"-c:v", HARDWARE_ENCODER,
|
|
"-preset", HARDWARE_PRESET,
|
|
"-cq", HARDWARE_CQ,
|
|
"-b:v", "0",
|
|
])
|
|
else:
|
|
command.extend([
|
|
"-c:v", SOFTWARE_ENCODER,
|
|
"-preset", SOFTWARE_PRESET,
|
|
"-crf", SOFTWARE_CRF,
|
|
])
|
|
command.extend([
|
|
"-pix_fmt", "yuv420p",
|
|
"-c:a", "aac",
|
|
"-b:a", "192k",
|
|
"-ar", "48000",
|
|
"-movflags", "+faststart",
|
|
str(output_path),
|
|
])
|
|
return command
|
|
|
|
|
|
def render_manifest(manifest_path, output_path, ffmpeg, ffprobe, overwrite, use_hardware):
|
|
manifest = read_json(manifest_path)
|
|
if not isinstance(manifest, dict):
|
|
print(f"Skip {manifest_path.name}: manifest is not an object")
|
|
return "skipped"
|
|
title = manifest.get("title")
|
|
if not isinstance(title, str) or not title.strip():
|
|
print(f"Skip {manifest_path.name}: title is blank")
|
|
return "skipped"
|
|
title = title.strip()
|
|
videos = manifest.get("videos")
|
|
if not isinstance(videos, list) or not videos:
|
|
raise ValueError("manifest needs a non-empty 'videos' array")
|
|
if output_path.exists() and not overwrite:
|
|
print(f"Skip {manifest_path.name}: output already exists ({output_path}); pass --overwrite to replace it")
|
|
return "skipped"
|
|
|
|
(
|
|
input_paths,
|
|
input_audio,
|
|
segments,
|
|
output_width,
|
|
output_height,
|
|
output_fps,
|
|
) = load_manifest_segments(manifest_path, videos, ffprobe)
|
|
text_files = []
|
|
temporary_output = None
|
|
try:
|
|
date_text, title_text = make_title_text(manifest_path, title)
|
|
for prefix, value in ((".video_date_", date_text), (".video_title_", title_text)):
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
newline="",
|
|
prefix=prefix,
|
|
suffix=".txt",
|
|
dir=output_path.parent,
|
|
delete=False,
|
|
) as text_file:
|
|
text_file.write(value + (" " * TITLE_RENDERING_PADDING_SPACES))
|
|
text_files.append(Path(text_file.name))
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
prefix=".render_",
|
|
suffix=output_path.suffix,
|
|
dir=output_path.parent,
|
|
delete=False,
|
|
) as output_file:
|
|
temporary_output = Path(output_file.name)
|
|
temporary_output.unlink()
|
|
|
|
graph, duration = make_filter_graph(
|
|
input_audio,
|
|
segments,
|
|
text_files[0].name,
|
|
text_files[1].name,
|
|
(date_text, title_text),
|
|
output_width,
|
|
output_height,
|
|
output_fps,
|
|
)
|
|
print(
|
|
f"Render {manifest_path.name}: {len(input_paths)} video(s), {len(segments)} segment(s), "
|
|
f"output {output_width}x{output_height} at {float(output_fps):g} fps, ~{duration:.1f}s"
|
|
)
|
|
if use_hardware:
|
|
hardware_command = ffmpeg_command(
|
|
ffmpeg, input_paths, graph, temporary_output, overwrite, HARDWARE_ENCODER
|
|
)
|
|
try:
|
|
subprocess.run(hardware_command, cwd=output_path.parent, check=True)
|
|
except subprocess.CalledProcessError:
|
|
temporary_output.unlink(missing_ok=True)
|
|
print(
|
|
f"NVENC failed for {manifest_path.name}; retrying with {SOFTWARE_ENCODER}.",
|
|
file=sys.stderr,
|
|
)
|
|
software_command = ffmpeg_command(
|
|
ffmpeg, input_paths, graph, temporary_output, overwrite, SOFTWARE_ENCODER
|
|
)
|
|
subprocess.run(software_command, cwd=output_path.parent, check=True)
|
|
else:
|
|
software_command = ffmpeg_command(
|
|
ffmpeg, input_paths, graph, temporary_output, overwrite, SOFTWARE_ENCODER
|
|
)
|
|
subprocess.run(software_command, cwd=output_path.parent, check=True)
|
|
temporary_output.replace(output_path)
|
|
finally:
|
|
for text_file in text_files:
|
|
text_file.unlink(missing_ok=True)
|
|
if temporary_output is not None:
|
|
temporary_output.unlink(missing_ok=True)
|
|
print(f"Saved {output_path}")
|
|
return "rendered"
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"manifest_dir",
|
|
type=Path,
|
|
nargs="?",
|
|
default=Path("out"),
|
|
help="folder containing daily .json manifests (default: ./out)",
|
|
)
|
|
parser.add_argument("--output-dir", type=Path, help="where to save MP4s (default: MANIFEST_DIR/rendered)")
|
|
parser.add_argument("--overwrite", action="store_true", help="replace existing rendered MP4 files")
|
|
parser.add_argument("--ffmpeg", default="ffmpeg", help="FFmpeg executable (default: ffmpeg on PATH)")
|
|
parser.add_argument("--ffprobe", default="ffprobe", help="ffprobe executable (default: ffprobe on PATH)")
|
|
args = parser.parse_args()
|
|
|
|
manifest_dir = args.manifest_dir.expanduser().resolve()
|
|
if not manifest_dir.is_dir():
|
|
parser.error("manifest_dir must be an existing directory")
|
|
output_dir = (args.output_dir.expanduser().resolve() if args.output_dir else manifest_dir / "rendered")
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
ffmpeg = shutil.which(args.ffmpeg)
|
|
if ffmpeg is None:
|
|
candidate = Path(args.ffmpeg).expanduser()
|
|
if not candidate.is_file():
|
|
parser.error(f"FFmpeg executable not found: {args.ffmpeg}")
|
|
ffmpeg = str(candidate.resolve())
|
|
ffprobe = shutil.which(args.ffprobe)
|
|
if ffprobe is None:
|
|
candidate = Path(args.ffprobe).expanduser()
|
|
if not candidate.is_file():
|
|
parser.error(f"ffprobe executable not found: {args.ffprobe}")
|
|
ffprobe = str(candidate.resolve())
|
|
|
|
use_hardware = PREFER_HARDWARE_ENCODING and ffmpeg_has_encoder(ffmpeg, HARDWARE_ENCODER)
|
|
if use_hardware:
|
|
print(f"Video encoder: {HARDWARE_ENCODER} (software fallback: {SOFTWARE_ENCODER})")
|
|
else:
|
|
print(f"Video encoder: {SOFTWARE_ENCODER} (NVENC unavailable or disabled)")
|
|
|
|
rendered = skipped = failed = 0
|
|
manifests = sorted(path for path in manifest_dir.glob("*.json") if path.is_file())
|
|
if not manifests:
|
|
print(f"No .json manifests found in {manifest_dir}")
|
|
return 0
|
|
for manifest_path in manifests:
|
|
output_path = output_dir / f"{manifest_path.stem}.mp4"
|
|
try:
|
|
result = render_manifest(
|
|
manifest_path, output_path, ffmpeg, ffprobe, args.overwrite, use_hardware
|
|
)
|
|
if result == "rendered":
|
|
rendered += 1
|
|
else:
|
|
skipped += 1
|
|
except (OSError, subprocess.CalledProcessError, json.JSONDecodeError, ValueError) as exc:
|
|
failed += 1
|
|
print(f"Failed {manifest_path.name}: {exc}", file=sys.stderr)
|
|
|
|
print(f"Finished: {rendered} rendered, {skipped} skipped, {failed} failed")
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|