1473 lines
56 KiB
Python
Executable file
1473 lines
56 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Simple flat-folder photo/video movie maker using ffmpeg.
|
|
|
|
Usage:
|
|
python movmaker.py input --out-dir .
|
|
|
|
Put images, videos, audio files, and optionally data.txt in one directory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import concurrent.futures
|
|
import json
|
|
import math
|
|
import os
|
|
import random
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unicodedata
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff"}
|
|
VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".avi", ".webm", ".m4v"}
|
|
AUDIO_EXTS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg"}
|
|
DATA_NAMES = {"data.txt", "info.txt", "title.txt"}
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
FFMPEG = BASE_DIR / "ffmpeg" / "ffmpeg"
|
|
FFPROBE = BASE_DIR / "ffmpeg" / "ffprobe"
|
|
INTRO_LOGO_NAME = "intro.png"
|
|
DEFAULT_INTRO_LOGO = BASE_DIR / "img" / "moto_travel.png"
|
|
JAMENDO_ENV = Path.home() / ".config" / "movmaker" / "jamendo.env"
|
|
LOG_PATH: Path | None = None
|
|
|
|
INTRO_LOGO_TARGET_HEIGHT_RATIO = 0.99
|
|
INTRO_LOGO_EXTRA_HOLD = 6.0
|
|
INTRO_LOGO_FADE_OFFSET = 3.0
|
|
INTRO_LOGO_DEFAULT_ALPHA = 0.25
|
|
INTRO_TITLE_START_OFFSET = 0.0
|
|
INTRO_TITLE_FADE_LENGTH = 1.0
|
|
INTRO_TITLE_VISIBLE = 9.0
|
|
INTRO_TITLE_FADE_IN_ENABLED = False
|
|
DEFAULT_IMAGE_DURATION = 6.0
|
|
DEFAULT_CROSSFADE_DURATION = 3.0
|
|
FIRST_IMAGE_DURATION_MULTIPLIER = 2.0
|
|
DEFAULT_AUDIO_FADE_IN = 2.0
|
|
DEFAULT_AUDIO_FADE_OUT = 10.0
|
|
DEFAULT_VIDEO_FADE = 1.0
|
|
INTRO_VIDEO_FADE_ENABLED = False
|
|
CLIP_AUDIO_FADE = 0.75
|
|
CLIP_AUDIO_SIDECHAIN_THRESHOLD = 0.04
|
|
CLIP_AUDIO_SIDECHAIN_RATIO = 8.0
|
|
CLIP_AUDIO_SIDECHAIN_ATTACK = 5.0
|
|
CLIP_AUDIO_SIDECHAIN_RELEASE = 250.0
|
|
|
|
|
|
ORDER_FILENAME = "order.json"
|
|
|
|
def load_order_json(directory: Path) -> list[str] | None:
|
|
order_file = directory / ORDER_FILENAME
|
|
if order_file.exists():
|
|
try:
|
|
return json.loads(order_file.read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
console(f"Error reading {ORDER_FILENAME}: {e}")
|
|
return None
|
|
|
|
|
|
def tool_path(local_path: Path, fallback: str) -> str:
|
|
"""Return preferred executable path for bundled or system tools."""
|
|
return str(local_path) if local_path.exists() else fallback
|
|
|
|
|
|
@dataclass
|
|
class MovieData:
|
|
title: str = ""
|
|
date: str = ""
|
|
place: str = ""
|
|
description: str = ""
|
|
captions: dict[str, str] | None = None
|
|
video_audio: set[str] | None = None
|
|
|
|
|
|
@dataclass
|
|
class MediaItem:
|
|
path: Path
|
|
kind: str # image or video
|
|
duration: float
|
|
|
|
|
|
def log_message(message: str) -> None:
|
|
"""Append a diagnostic line to the current render log if it is enabled."""
|
|
if LOG_PATH:
|
|
with LOG_PATH.open("a", encoding="utf-8") as log:
|
|
log.write(message.rstrip() + "\n")
|
|
|
|
|
|
def ffmpeg_progress_cmd(cmd: list[str], progress: bool) -> list[str]:
|
|
"""Inject ffmpeg-friendly progress flags when progress monitoring is requested."""
|
|
if not cmd or Path(cmd[0]).name != "ffmpeg":
|
|
return cmd
|
|
out = [cmd[0], "-hide_banner", "-nostats"]
|
|
rest = cmd[1:]
|
|
if rest and rest[0] == "-y":
|
|
out.append("-y")
|
|
rest = rest[1:]
|
|
if progress:
|
|
out += ["-progress", "pipe:1"]
|
|
return out + rest
|
|
|
|
|
|
def format_bytes(value: int) -> str:
|
|
size = float(max(0, value))
|
|
for unit in ["B", "KB", "MB", "GB"]:
|
|
if size < 1024 or unit == "GB":
|
|
return f"{size:.1f}{unit}"
|
|
size /= 1024
|
|
return f"{size:.1f}GB"
|
|
|
|
|
|
def run(cmd: list[str], progress_duration: float | None = None) -> None:
|
|
"""Execute a subprocess while streaming progress and teeing output to the log.
|
|
|
|
Args:
|
|
cmd: argv array ready for subprocess.run.
|
|
progress_duration: Expected output duration in seconds used to estimate progress; omit for instant commands.
|
|
"""
|
|
progress = progress_duration is not None and progress_duration > 0 and Path(cmd[0]).name == "ffmpeg"
|
|
run_cmd = ffmpeg_progress_cmd(cmd, progress)
|
|
log_message("+ " + shlex.join(run_cmd))
|
|
if not LOG_PATH:
|
|
subprocess.run(run_cmd, check=True)
|
|
return
|
|
|
|
with LOG_PATH.open("a", encoding="utf-8") as log:
|
|
if not progress:
|
|
subprocess.run(run_cmd, stdout=log, stderr=log, check=True)
|
|
return
|
|
|
|
process = subprocess.Popen(run_cmd, stdout=subprocess.PIPE, stderr=log, text=True, bufsize=1)
|
|
fields: dict[str, str] = {}
|
|
assert process.stdout is not None
|
|
for raw_line in process.stdout:
|
|
line = raw_line.strip()
|
|
log.write(line + "\n")
|
|
if "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
fields[key] = value
|
|
if key in {"out_time_ms", "total_size", "speed", "fps", "progress"}:
|
|
try:
|
|
seconds = int(fields.get("out_time_ms", "0")) / 1_000_000
|
|
except ValueError:
|
|
seconds = 0.0
|
|
try:
|
|
written = int(fields.get("total_size", "0"))
|
|
except ValueError:
|
|
written = 0
|
|
percent = min(100.0, max(0.0, seconds / progress_duration * 100.0))
|
|
fps = fields.get("fps", "?")
|
|
speed = fields.get("speed", "?")
|
|
sys.stdout.write(f"\r[{timestamp()}] Rendering {percent:5.1f}% fps={fps} speed={speed} written={format_bytes(written)}")
|
|
sys.stdout.flush()
|
|
rc = process.wait()
|
|
sys.stdout.write("\n")
|
|
if rc != 0:
|
|
raise subprocess.CalledProcessError(rc, run_cmd)
|
|
|
|
|
|
def capture(cmd: list[str]) -> str:
|
|
return subprocess.check_output(cmd, text=True).strip()
|
|
|
|
|
|
def require_tool(name: str) -> None:
|
|
try:
|
|
executable = tool_path(FFMPEG, "ffmpeg") if name == "ffmpeg" else tool_path(FFPROBE, "ffprobe") if name == "ffprobe" else name
|
|
subprocess.run([executable, "-version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)
|
|
except Exception:
|
|
raise SystemExit(f"Error: required tool not found: {name}")
|
|
|
|
|
|
def parse_resolution(value: str) -> tuple[int, int]:
|
|
try:
|
|
w, h = value.lower().split("x", 1)
|
|
return int(w), int(h)
|
|
except Exception:
|
|
raise argparse.ArgumentTypeError("resolution must look like 1920x1080")
|
|
|
|
|
|
def ffprobe_duration(path: Path) -> float:
|
|
try:
|
|
out = capture([
|
|
tool_path(FFPROBE, "ffprobe"), "-v", "error",
|
|
"-show_entries", "format=duration",
|
|
"-of", "json",
|
|
str(path),
|
|
])
|
|
data = json.loads(out)
|
|
return max(0.1, float(data["format"]["duration"]))
|
|
except Exception:
|
|
return 3.0
|
|
|
|
|
|
def ffprobe_resolution(path: Path) -> tuple[int, int] | None:
|
|
try:
|
|
out = capture([
|
|
tool_path(FFPROBE, "ffprobe"), "-v", "error",
|
|
"-select_streams", "v:0",
|
|
"-show_entries", "stream=width,height",
|
|
"-of", "json",
|
|
str(path),
|
|
])
|
|
data = json.loads(out)
|
|
for stream in data.get("streams", []):
|
|
width = int(stream.get("width") or 0)
|
|
height = int(stream.get("height") or 0)
|
|
if width > 0 and height > 0:
|
|
return width, height
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def ffprobe_creation_date(path: Path) -> datetime | None:
|
|
try:
|
|
out = capture([
|
|
tool_path(FFPROBE, "ffprobe"), "-v", "error",
|
|
"-show_entries", "format_tags=creation_time:stream_tags=creation_time",
|
|
"-of", "json",
|
|
str(path),
|
|
])
|
|
data = json.loads(out)
|
|
candidates: list[str] = []
|
|
if isinstance(data.get("format"), dict):
|
|
candidates.append(data["format"].get("tags", {}).get("creation_time", ""))
|
|
for stream in data.get("streams", []):
|
|
candidates.append(stream.get("tags", {}).get("creation_time", ""))
|
|
for value in candidates:
|
|
if not value:
|
|
continue
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def image_tiff_data(path: Path) -> bytes | None:
|
|
data = path.read_bytes()[:1024 * 1024]
|
|
if data.startswith(b"\xff\xd8"):
|
|
pos = 2
|
|
while pos + 4 <= len(data):
|
|
if data[pos] != 0xFF:
|
|
break
|
|
marker = data[pos + 1]
|
|
pos += 2
|
|
if marker in {0xD8, 0xD9}:
|
|
continue
|
|
if pos + 2 > len(data):
|
|
break
|
|
size = int.from_bytes(data[pos:pos + 2], "big")
|
|
segment = data[pos + 2:pos + size]
|
|
pos += size
|
|
if marker == 0xE1 and segment.startswith(b"Exif\x00\x00"):
|
|
return segment[6:]
|
|
return None
|
|
return data
|
|
|
|
|
|
def exif_image_date(path: Path) -> datetime | None:
|
|
"""Read common JPEG/TIFF EXIF date tags using only the stdlib."""
|
|
tiff = image_tiff_data(path)
|
|
if tiff is None:
|
|
return None
|
|
|
|
if tiff[:2] == b"II":
|
|
endian = "little"
|
|
elif tiff[:2] == b"MM":
|
|
endian = "big"
|
|
else:
|
|
return None
|
|
|
|
def u16(off: int) -> int:
|
|
return int.from_bytes(tiff[off:off + 2], endian)
|
|
|
|
def u32(off: int) -> int:
|
|
return int.from_bytes(tiff[off:off + 4], endian)
|
|
|
|
def read_ascii(offset: int, count: int) -> str:
|
|
return tiff[offset:offset + count].split(b"\x00", 1)[0].decode("ascii", "ignore").strip()
|
|
|
|
def parse_ifd(ifd_off: int) -> tuple[str | None, int | None]:
|
|
if ifd_off <= 0 or ifd_off + 2 > len(tiff):
|
|
return None, None
|
|
entries = u16(ifd_off)
|
|
found_date = None
|
|
exif_ifd = None
|
|
for i in range(entries):
|
|
off = ifd_off + 2 + i * 12
|
|
if off + 12 > len(tiff):
|
|
break
|
|
tag = u16(off)
|
|
typ = u16(off + 2)
|
|
count = u32(off + 4)
|
|
value = u32(off + 8)
|
|
if tag in {0x0132, 0x9003, 0x9004} and typ == 2:
|
|
found_date = read_ascii(value, count) if count > 4 else tiff[off + 8:off + 8 + count].split(b"\x00", 1)[0].decode("ascii", "ignore")
|
|
elif tag == 0x8769:
|
|
exif_ifd = value
|
|
return found_date, exif_ifd
|
|
|
|
try:
|
|
if u16(2) != 42:
|
|
return None
|
|
date_text, exif_ifd = parse_ifd(u32(4))
|
|
if exif_ifd:
|
|
exif_date, _ = parse_ifd(exif_ifd)
|
|
date_text = exif_date or date_text
|
|
if date_text:
|
|
return datetime.strptime(date_text[:19], "%Y:%m:%d %H:%M:%S")
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def exif_image_gps(path: Path) -> tuple[float, float] | None:
|
|
tiff = image_tiff_data(path)
|
|
if tiff is None:
|
|
return None
|
|
if tiff[:2] == b"II":
|
|
endian = "little"
|
|
elif tiff[:2] == b"MM":
|
|
endian = "big"
|
|
else:
|
|
return None
|
|
|
|
def u16(off: int) -> int:
|
|
return int.from_bytes(tiff[off:off + 2], endian)
|
|
|
|
def u32(off: int) -> int:
|
|
return int.from_bytes(tiff[off:off + 4], endian)
|
|
|
|
def rational(off: int) -> float:
|
|
num = u32(off)
|
|
den = u32(off + 4)
|
|
return num / den if den else 0.0
|
|
|
|
def ascii_value(value: int, count: int, entry_off: int) -> str:
|
|
raw = tiff[entry_off + 8:entry_off + 8 + count] if count <= 4 else tiff[value:value + count]
|
|
return raw.split(b"\x00", 1)[0].decode("ascii", "ignore").strip()
|
|
|
|
def rationals(value: int, count: int) -> list[float]:
|
|
return [rational(value + i * 8) for i in range(count)]
|
|
|
|
try:
|
|
if u16(2) != 42:
|
|
return None
|
|
ifd0 = u32(4)
|
|
entries = u16(ifd0)
|
|
gps_ifd = None
|
|
for i in range(entries):
|
|
off = ifd0 + 2 + i * 12
|
|
if off + 12 > len(tiff):
|
|
break
|
|
if u16(off) == 0x8825:
|
|
gps_ifd = u32(off + 8)
|
|
break
|
|
if gps_ifd is None or gps_ifd + 2 > len(tiff):
|
|
return None
|
|
gps_entries = u16(gps_ifd)
|
|
lat_ref = lon_ref = ""
|
|
lat = lon = None
|
|
for i in range(gps_entries):
|
|
off = gps_ifd + 2 + i * 12
|
|
if off + 12 > len(tiff):
|
|
break
|
|
tag = u16(off)
|
|
typ = u16(off + 2)
|
|
count = u32(off + 4)
|
|
value = u32(off + 8)
|
|
if tag == 1 and typ == 2:
|
|
lat_ref = ascii_value(value, count, off)
|
|
elif tag == 2 and typ == 5:
|
|
lat = rationals(value, count)
|
|
elif tag == 3 and typ == 2:
|
|
lon_ref = ascii_value(value, count, off)
|
|
elif tag == 4 and typ == 5:
|
|
lon = rationals(value, count)
|
|
if not lat or not lon or len(lat) < 3 or len(lon) < 3:
|
|
return None
|
|
lat_dec = lat[0] + lat[1] / 60 + lat[2] / 3600
|
|
lon_dec = lon[0] + lon[1] / 60 + lon[2] / 3600
|
|
if lat_ref.upper() == "S":
|
|
lat_dec = -lat_dec
|
|
if lon_ref.upper() == "W":
|
|
lon_dec = -lon_dec
|
|
return lat_dec, lon_dec
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def ffprobe_gps(path: Path) -> tuple[float, float] | None:
|
|
try:
|
|
out = capture([
|
|
tool_path(FFPROBE, "ffprobe"), "-v", "error",
|
|
"-show_format", "-show_streams",
|
|
"-of", "json",
|
|
str(path),
|
|
])
|
|
data = json.loads(out)
|
|
except Exception:
|
|
return None
|
|
|
|
values: list[str] = []
|
|
def collect(obj: object) -> None:
|
|
if isinstance(obj, dict):
|
|
tags = obj.get("tags")
|
|
if isinstance(tags, dict):
|
|
for key, value in tags.items():
|
|
if "location" in key.lower() or "gps" in key.lower():
|
|
values.append(str(value))
|
|
for value in obj.values():
|
|
collect(value)
|
|
elif isinstance(obj, list):
|
|
for value in obj:
|
|
collect(value)
|
|
collect(data)
|
|
for value in values:
|
|
match = re.search(r"([+-]\d+(?:\.\d+)?)([+-]\d+(?:\.\d+)?)", value)
|
|
if match:
|
|
return float(match.group(1)), float(match.group(2))
|
|
return None
|
|
|
|
|
|
def media_gps(path: Path) -> tuple[float, float] | None:
|
|
if path.suffix.lower() in IMAGE_EXTS:
|
|
gps = exif_image_gps(path)
|
|
if gps:
|
|
return gps
|
|
return ffprobe_gps(path)
|
|
|
|
|
|
def reverse_geocode_city_country(lat: float, lon: float) -> str | None:
|
|
last_country = ""
|
|
for zoom in range(18, 2, -1):
|
|
query = urllib.parse.urlencode({
|
|
"format": "jsonv2",
|
|
"lat": f"{lat:.8f}",
|
|
"lon": f"{lon:.8f}",
|
|
"zoom": str(zoom),
|
|
"addressdetails": "1",
|
|
})
|
|
request = urllib.request.Request(
|
|
f"https://nominatim.openstreetmap.org/reverse?{query}",
|
|
headers={"User-Agent": "movmaker/1.0 (reverse geocoding for local movie metadata)"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
data = json.loads(response.read().decode("utf-8"))
|
|
address = data.get("address", {})
|
|
country = address.get("country", "")
|
|
last_country = country or last_country
|
|
city = next((address.get(key) for key in ("city", "town", "village") if address.get(key)), "")
|
|
if city:
|
|
return ", ".join(x for x in [city, country] if x)
|
|
except Exception:
|
|
continue
|
|
return last_country or None
|
|
|
|
|
|
def infer_place_from_gps(media: list[MediaItem]) -> str | None:
|
|
for item in media:
|
|
gps = media_gps(item.path)
|
|
if not gps:
|
|
continue
|
|
place = reverse_geocode_city_country(*gps)
|
|
if place:
|
|
console(f"Location from GPS: {place} ({gps[0]:.6f}, {gps[1]:.6f})")
|
|
return place
|
|
return None
|
|
|
|
|
|
def media_embedded_date(path: Path) -> datetime | None:
|
|
ext = path.suffix.lower()
|
|
if ext in IMAGE_EXTS:
|
|
try:
|
|
dt = exif_image_date(path)
|
|
if dt is not None:
|
|
return dt
|
|
except Exception:
|
|
pass
|
|
return None
|
|
if ext in VIDEO_EXTS:
|
|
return ffprobe_creation_date(path)
|
|
return None
|
|
|
|
|
|
def parse_date_from_filename(path: Path) -> datetime | None:
|
|
# Match exactly YYYYMMDD[_-]HHMMSS as a standalone numeric chunk.
|
|
m = re.search(r'(?:^|[^0-9])(\d{8})[_-]?(\d{6})(?:[^0-9]|$)', path.name)
|
|
if not m:
|
|
return None
|
|
full_ts = f"{m.group(1)}{m.group(2)}"
|
|
try:
|
|
return datetime.strptime(full_ts, "%Y%m%d%H%M%S")
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def media_date(path: Path) -> datetime:
|
|
# 1) Embedded metadata (EXIF/ffprobe)
|
|
dt = media_embedded_date(path)
|
|
|
|
# 2) Filename parsing
|
|
if dt is None:
|
|
dt = parse_date_from_filename(path)
|
|
|
|
# 3) Filesystem datetime fallback
|
|
if dt is None:
|
|
dt = datetime.fromtimestamp(path.stat().st_mtime)
|
|
|
|
if dt.tzinfo is not None:
|
|
dt = dt.astimezone().replace(tzinfo=None)
|
|
return dt
|
|
|
|
|
|
def ascii_safe(value: str) -> str:
|
|
"""Return a human-readable ASCII approximation of Scandinavian/Slavic letters."""
|
|
replacements = {
|
|
# Danish letters
|
|
"æ": "ae", "Æ": "Ae",
|
|
"ø": "o", "Ø": "O",
|
|
"å": "aa", "Å": "Aa",
|
|
# German letters
|
|
"ä": "ae", "Ä": "Ae",
|
|
"ö": "oe", "Ö": "Oe",
|
|
"ü": "ue", "Ü": "Ue",
|
|
"ß": "ss", "ẞ": "SS",
|
|
# Croatian letters
|
|
"č": "c", "Č": "C",
|
|
"ć": "c", "Ć": "C",
|
|
"ž": "z", "Ž": "Z",
|
|
"š": "s", "Š": "S",
|
|
"đ": "d", "Đ": "D",
|
|
}
|
|
value = "".join(replacements.get(ch, ch) for ch in value)
|
|
return unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
|
|
|
|
|
|
def slugify(value: str) -> str:
|
|
"""Convert an arbitrary string into a filesystem-safe slug."""
|
|
value = ascii_safe(value)
|
|
value = value.strip().lower()
|
|
value = re.sub(r"[^a-z0-9]+", "_", value)
|
|
value = value.strip("_")
|
|
return value or "movie"
|
|
|
|
|
|
def ordinal_day(day: int) -> str:
|
|
if 10 <= day % 100 <= 20:
|
|
suffix = "th"
|
|
else:
|
|
suffix = {1: "st", 2: "nd", 3: "rd"}.get(day % 10, "th")
|
|
return f"{day}{suffix}"
|
|
|
|
|
|
def display_date(dt: datetime) -> str:
|
|
return f"{dt.strftime('%B')} {dt.year}"
|
|
|
|
|
|
def prepare_clip_stamps(media: list[MediaItem], data: MovieData) -> list[str]:
|
|
"""Build per-clip stamp strings containing title/date/place metadata."""
|
|
geocode_cache: dict[tuple[float, float], str | None] = {}
|
|
stamps: list[str] = []
|
|
for item in media:
|
|
embedded_date = media_embedded_date(item.path)
|
|
date_text = display_date(embedded_date) if embedded_date else data.date
|
|
place_text = data.place
|
|
gps = media_gps(item.path)
|
|
if gps:
|
|
key = (round(gps[0], 4), round(gps[1], 4))
|
|
if key not in geocode_cache:
|
|
geocode_cache[key] = reverse_geocode_city_country(*gps)
|
|
place_text = geocode_cache[key] or place_text
|
|
stamp = " | ".join(x for x in [data.title, date_text, place_text] if x)
|
|
stamps.append(stamp)
|
|
if stamp:
|
|
console(f"Clip stamp: {item.path.name}: {ascii_safe(stamp)}")
|
|
return stamps
|
|
|
|
|
|
def timestamp() -> str:
|
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def console(message: str) -> None:
|
|
print(f"[{timestamp()}] {ascii_safe(message)}")
|
|
|
|
|
|
def default_output_path(first_media: Path, data: MovieData, out_dir: Path) -> Path:
|
|
date = media_date(first_media).strftime("%Y%m%d")
|
|
title = slugify(data.title or first_media.parent.name or "movie")
|
|
return out_dir / f"{date}_{title}.mp4"
|
|
|
|
|
|
def scan_input(input_dir: Path, image_duration: float) -> tuple[list[MediaItem], list[Path], MovieData]:
|
|
"""Detect media/audio/data files inside the input directory and build metadata."""
|
|
if not input_dir.is_dir():
|
|
raise SystemExit(f"Input directory does not exist: {input_dir}")
|
|
|
|
media: list[MediaItem] = []
|
|
audio: list[Path] = []
|
|
data_file: Path | None = None
|
|
|
|
for path in input_dir.iterdir():
|
|
if not path.is_file():
|
|
continue
|
|
ext = path.suffix.lower()
|
|
if path.name.lower() == ORDER_FILENAME:
|
|
continue
|
|
if path.name.lower() in DATA_NAMES:
|
|
data_file = path
|
|
elif ext in IMAGE_EXTS:
|
|
media.append(MediaItem(path=path, kind="image", duration=image_duration))
|
|
elif ext in VIDEO_EXTS:
|
|
media.append(MediaItem(path=path, kind="video", duration=ffprobe_duration(path)))
|
|
elif ext in AUDIO_EXTS:
|
|
audio.append(path)
|
|
|
|
if not media:
|
|
raise SystemExit(f"No images or videos found in {input_dir}")
|
|
|
|
order = load_order_json(input_dir)
|
|
if order:
|
|
# Sort based on order.json, append missing files at the end
|
|
media_map = {item.path.name: item for item in media}
|
|
sorted_media = []
|
|
for name in order:
|
|
if name in media_map:
|
|
sorted_media.append(media_map.pop(name))
|
|
# Add remaining
|
|
sorted_media.extend(media_map.values())
|
|
media = sorted_media
|
|
else:
|
|
# Datetime-only ordering (no filename-order fallback).
|
|
# Python sort is stable, so equal datetimes keep discovery order.
|
|
media.sort(key=lambda item: media_date(item.path))
|
|
|
|
return media, audio, parse_data_file(data_file)
|
|
|
|
|
|
def parse_data_file(path: Path | None) -> MovieData:
|
|
if not path or not path.exists():
|
|
return MovieData(captions={})
|
|
|
|
data = MovieData(captions={})
|
|
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
i = 0
|
|
while i < len(lines):
|
|
line_no = i + 1
|
|
line = lines[i].strip()
|
|
if not line or line.startswith("#"):
|
|
i += 1
|
|
continue
|
|
if ":" not in line:
|
|
raise SystemExit(f"Invalid data file syntax in {path} line {line_no}: every entry must be key: value")
|
|
key, value = line.split(":", 1)
|
|
key = key.strip()
|
|
value = value.strip()
|
|
low = key.lower()
|
|
if value == "|" and low in {"description", "desc", "synopsis"}:
|
|
block: list[str] = []
|
|
i += 1
|
|
while i < len(lines):
|
|
raw = lines[i]
|
|
if raw.strip() and not raw.startswith((" ", "\t")):
|
|
break
|
|
if raw.startswith(" "):
|
|
raw = raw[2:]
|
|
elif raw.startswith((" ", "\t")):
|
|
raw = raw[1:]
|
|
block.append(raw.rstrip())
|
|
i += 1
|
|
data.description = "\n".join(block).strip()
|
|
continue
|
|
if low in {"title", "name"}:
|
|
data.title = value
|
|
elif low in {"date", "dates", "when"}:
|
|
data.date = value
|
|
elif low in {"place", "location", "where"}:
|
|
data.place = value
|
|
elif low in {"description", "desc", "synopsis"}:
|
|
data.description = value
|
|
elif low.endswith(" audio"):
|
|
file_key = key[:-6].strip()
|
|
if file_key and value.lower() in {"1", "yes", "true", "on"}:
|
|
if data.video_audio is None:
|
|
data.video_audio = set()
|
|
data.video_audio.add(file_key)
|
|
elif key:
|
|
data.captions[key] = value
|
|
i += 1
|
|
|
|
return data
|
|
|
|
|
|
def strip_unsupported_caption_chars(text: str) -> str:
|
|
"""Remove characters that bundled caption fonts cannot render reliably.
|
|
|
|
FFmpeg drawtext does not provide dependable emoji fallback for our bundled
|
|
caption fonts, so emoji/symbol glyphs become unknown-character boxes in the
|
|
final video. Keep normal letters, numbers, punctuation, spacing, and marks;
|
|
drop symbols such as emoji.
|
|
"""
|
|
cleaned = []
|
|
for ch in text:
|
|
cat = unicodedata.category(ch)
|
|
if cat.startswith("C"):
|
|
continue
|
|
if cat.startswith("S"):
|
|
continue
|
|
cleaned.append(ch)
|
|
return re.sub(r"\s+", " ", "".join(cleaned)).strip()
|
|
|
|
|
|
def drawtext_escape(text: str) -> str:
|
|
# Escape for ffmpeg drawtext text=... without surrounding quotes.
|
|
return (
|
|
text.replace("\\", "\\\\")
|
|
.replace(":", "\\:")
|
|
.replace(",", "\\,")
|
|
.replace("%", "\\%")
|
|
)
|
|
|
|
|
|
def render_intro_logo_frame(logo: Path, out: Path, width: int, height: int, alpha: float) -> None:
|
|
"""Build a single black frame with the intro logo centered and scaled.
|
|
|
|
This implementation first renders a temporary PNG of the logo with the
|
|
requested alpha applied, then overlays that precomposed PNG onto a black
|
|
background. This avoids passing problematic options to the overlay filter
|
|
and works reliably across ffmpeg builds.
|
|
|
|
Args:
|
|
logo: Source PNG to overlay.
|
|
out: Destination PNG path for the composed frame.
|
|
width: Target video width.
|
|
height: Target video height.
|
|
alpha: Desired opacity between 0 and 1.
|
|
"""
|
|
if width <= 0 or height <= 0:
|
|
raise ValueError("intro logo frame must have positive width/height")
|
|
alpha = min(1.0, max(0.0, alpha))
|
|
target_logo_height = max(1, int(round(height * INTRO_LOGO_TARGET_HEIGHT_RATIO)))
|
|
|
|
ffmpeg_exe = tool_path(FFMPEG, "ffmpeg")
|
|
|
|
# Prepare a temporary logo PNG with the requested alpha applied.
|
|
tmp_logo = out.with_name(out.stem + "_alpha.png")
|
|
cmd_logo = [
|
|
ffmpeg_exe, "-y",
|
|
"-i", str(logo),
|
|
"-vf", (
|
|
f"scale=w={width}:h={target_logo_height}:force_original_aspect_ratio=decrease,"
|
|
f"format=rgba,colorchannelmixer=aa={alpha:.3f}"
|
|
),
|
|
"-frames:v", "1",
|
|
"-pix_fmt", "rgba",
|
|
str(tmp_logo),
|
|
]
|
|
run(cmd_logo)
|
|
|
|
# Overlay the precomposed logo onto a black background and write the single frame.
|
|
cmd = [
|
|
ffmpeg_exe, "-y",
|
|
"-f", "lavfi", "-i", f"color=c=black:s={width}x{height}",
|
|
"-i", str(tmp_logo),
|
|
"-filter_complex", "[0:v][1:v]overlay=x=(W-w)/2:y=(H-h)/2:format=auto[vout]",
|
|
"-map", "[vout]",
|
|
"-frames:v", "1",
|
|
"-pix_fmt", "rgba",
|
|
"-f", "image2",
|
|
"-update", "1",
|
|
str(out),
|
|
]
|
|
try:
|
|
run(cmd)
|
|
finally:
|
|
try:
|
|
if tmp_logo.exists():
|
|
tmp_logo.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def normalize_clip(item: MediaItem, out: Path, width: int, height: int, fps: int, preset: str, crf: int) -> None:
|
|
"""Transcode one media item into a padded, FPS-normalized H.264 clip."""
|
|
vf = (
|
|
f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
|
|
f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black,"
|
|
f"fps={fps},format=yuv420p,setsar=1"
|
|
)
|
|
|
|
if item.kind == "image":
|
|
cmd = [
|
|
tool_path(FFMPEG, "ffmpeg"), "-y",
|
|
"-loop", "1", "-t", f"{item.duration:.3f}", "-i", str(item.path),
|
|
"-an", "-vf", vf,
|
|
"-c:v", "libx264", "-preset", preset, "-crf", str(crf),
|
|
str(out),
|
|
]
|
|
else:
|
|
cmd = [
|
|
tool_path(FFMPEG, "ffmpeg"), "-y",
|
|
"-i", str(item.path),
|
|
"-an", "-vf", vf,
|
|
"-c:v", "libx264", "-preset", preset, "-crf", str(crf),
|
|
str(out),
|
|
]
|
|
run(cmd)
|
|
|
|
|
|
def media_has_audio(path: Path) -> bool:
|
|
try:
|
|
out = capture([
|
|
tool_path(FFPROBE, "ffprobe"), "-v", "error",
|
|
"-select_streams", "a:0",
|
|
"-show_entries", "stream=index",
|
|
"-of", "csv=p=0",
|
|
str(path),
|
|
])
|
|
return bool(out.strip())
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def audio_metadata(path: Path) -> tuple[str, str]:
|
|
try:
|
|
out = capture([
|
|
tool_path(FFPROBE, "ffprobe"), "-v", "error",
|
|
"-show_entries", "format_tags=artist,title",
|
|
"-of", "json",
|
|
str(path),
|
|
])
|
|
tags = json.loads(out).get("format", {}).get("tags", {})
|
|
artist = tags.get("artist") or tags.get("ARTIST") or ""
|
|
title = tags.get("title") or tags.get("TITLE") or ""
|
|
return str(artist), str(title)
|
|
except Exception:
|
|
return "", ""
|
|
|
|
|
|
def load_jamendo_client_id() -> str | None:
|
|
if os.environ.get("JAMENDO_CLIENT_ID"):
|
|
return os.environ["JAMENDO_CLIENT_ID"]
|
|
if JAMENDO_ENV.exists():
|
|
for line in JAMENDO_ENV.read_text(encoding="utf-8", errors="replace").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
if key.strip() == "JAMENDO_CLIENT_ID":
|
|
return value.strip().strip("'\"")
|
|
return None
|
|
|
|
|
|
def download_jamendo_music(genre: str, out_dir: Path) -> Path | None:
|
|
"""Download a random Jamendo track matching the requested genre."""
|
|
client_id = load_jamendo_client_id()
|
|
if not client_id:
|
|
console("No audio files found and no Jamendo client ID configured; continuing without music.")
|
|
return None
|
|
query = urllib.parse.urlencode({
|
|
"client_id": client_id,
|
|
"format": "json",
|
|
"limit": "50",
|
|
"search": genre,
|
|
"include": "musicinfo licenses",
|
|
"audioformat": "mp32",
|
|
"ccnc": "true",
|
|
"ccnd": "false",
|
|
})
|
|
request = urllib.request.Request(
|
|
f"https://api.jamendo.com/v3.0/tracks/?{query}",
|
|
headers={"User-Agent": "movmaker/1.0"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=20) as response:
|
|
data = json.loads(response.read().decode("utf-8"))
|
|
except Exception as exc:
|
|
console(f"Could not search Jamendo music: {exc}")
|
|
return None
|
|
|
|
results = []
|
|
for track in data.get("results", []):
|
|
license_url = str(track.get("license_ccurl", "")).lower()
|
|
if not track.get("audio"):
|
|
continue
|
|
if track.get("audiodownload_allowed") is False:
|
|
continue
|
|
if "creativecommons.org" not in license_url or "/nd" in license_url:
|
|
continue
|
|
results.append(track)
|
|
if not results:
|
|
console(f"No Jamendo music found for genre/query: {genre}")
|
|
return None
|
|
track = random.choice(results)
|
|
audio_url = track["audio"]
|
|
title = track.get("name") or "jamendo_music"
|
|
artist = track.get("artist_name", "")
|
|
filename = f"jamendo_{track.get('id', 'music')}_{slugify(title)}.mp3"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
out = out_dir / filename
|
|
console(f"Downloading Jamendo music: {title}" + (f" by {artist}" if artist else ""))
|
|
try:
|
|
urllib.request.urlretrieve(audio_url, out)
|
|
except Exception as exc:
|
|
console(f"Could not download Jamendo music: {exc}")
|
|
return None
|
|
if title or artist:
|
|
console("Audio metadata: " + " - ".join(x for x in [artist, title] if x))
|
|
log_dir = out_dir / ".logs"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
attribution = log_dir / f"{out.stem}.txt"
|
|
musicinfo = track.get("musicinfo", {}) if isinstance(track.get("musicinfo"), dict) else {}
|
|
attribution.write_text(
|
|
"Jamendo music downloaded by movmaker\n"
|
|
f"Title: {title}\n"
|
|
f"Artist: {artist}\n"
|
|
f"URL: {track.get('shareurl', '')}\n"
|
|
f"License: {track.get('license_ccurl', '')}\n"
|
|
"Source: Jamendo\n",
|
|
encoding="utf-8",
|
|
)
|
|
return out
|
|
|
|
|
|
def concat_or_mix_audio(audio_files: list[Path], out: Path) -> Path | None:
|
|
"""Concatenate or mix multiple audio tracks into one temp file."""
|
|
if not audio_files:
|
|
return None
|
|
if len(audio_files) == 1:
|
|
return audio_files[0]
|
|
|
|
list_file = out.with_suffix(".txt")
|
|
def esc(p: Path) -> str:
|
|
return str(p.resolve()).replace("'", "'\\''")
|
|
list_file.write_text("".join(f"file '{esc(p)}'\n" for p in audio_files), encoding="utf-8")
|
|
run([
|
|
tool_path(FFMPEG, "ffmpeg"), "-y", "-f", "concat", "-safe", "0", "-i", str(list_file),
|
|
"-vn", "-c:a", "aac", "-b:a", "192k", str(out),
|
|
])
|
|
return out
|
|
|
|
|
|
def build_video(
|
|
clips: list[Path],
|
|
durations: list[float],
|
|
audio: Path | None,
|
|
clip_audio_paths: list[Path | None],
|
|
output: Path,
|
|
data: MovieData,
|
|
creation_date: datetime,
|
|
captions: list[str],
|
|
clip_stamps: list[str],
|
|
fade: float,
|
|
audio_fade: float,
|
|
audio_fade_in: float,
|
|
video_fade: float,
|
|
intro_logo: Path | None,
|
|
intro_logo_alpha: float,
|
|
width: int,
|
|
height: int,
|
|
fps: int,
|
|
preset: str,
|
|
crf: int,
|
|
text_dir: Path,
|
|
video_intro_duration: float | None = None,
|
|
intro_visible_lead: float = 0.0,
|
|
skip_video_fade_out: bool = False,
|
|
) -> None:
|
|
"""Assemble normalized clips, overlays, and audio into the final MP4."""
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
text_file_count = 0
|
|
def textfile_arg(text: str) -> str:
|
|
nonlocal text_file_count
|
|
path = text_dir / f"drawtext_{text_file_count:04d}.txt"
|
|
text_file_count += 1
|
|
path.write_text(text, encoding="utf-8")
|
|
return f"textfile='{path}':expansion=none"
|
|
|
|
cmd = [tool_path(FFMPEG, "ffmpeg"), "-y"]
|
|
for clip in clips:
|
|
cmd += ["-i", str(clip)]
|
|
if audio:
|
|
cmd += ["-stream_loop", "-1", "-i", str(audio)]
|
|
clip_audio_input_start = len(clips) + (1 if audio else 0)
|
|
for clip_audio in clip_audio_paths:
|
|
if clip_audio is not None:
|
|
cmd += ["-i", str(clip_audio)]
|
|
|
|
filters: list[str] = []
|
|
n = len(clips)
|
|
|
|
total_duration = durations[0]
|
|
transition_offsets: list[float] = []
|
|
if n == 1:
|
|
current = "0:v"
|
|
else:
|
|
current = "v01"
|
|
offset = max(0.0, durations[0] - fade)
|
|
transition_offsets.append(offset)
|
|
filters.append(f"[0:v][1:v]xfade=transition=fade:duration={fade:.3f}:offset={offset:.3f}[{current}]")
|
|
total_duration = durations[0] + durations[1] - fade
|
|
for i in range(2, n):
|
|
next_label = f"v{i:02d}"
|
|
offset = max(0.0, total_duration - fade)
|
|
transition_offsets.append(offset)
|
|
filters.append(f"[{current}][{i}:v]xfade=transition=fade:duration={fade:.3f}:offset={offset:.3f}[{next_label}]")
|
|
current = next_label
|
|
total_duration += durations[i] - fade
|
|
|
|
intro_duration = max(0.0, video_intro_duration) if video_intro_duration is not None else (audio_fade_in if audio_fade_in > 0 else 0.0)
|
|
if intro_duration > 0:
|
|
intro_label = "intro_padded"
|
|
filters.append(
|
|
f"[{current}]tpad=start_duration={intro_duration:.3f}:start_mode=add:color=black[{intro_label}]"
|
|
)
|
|
current = intro_label
|
|
total_duration += intro_duration
|
|
|
|
|
|
fade_label = current
|
|
apply_fade_in = video_fade > 0 and INTRO_VIDEO_FADE_ENABLED
|
|
apply_fade_out = video_fade > 0 and not skip_video_fade_out
|
|
if apply_fade_in or apply_fade_out:
|
|
fade_in_start = intro_duration
|
|
fade_out_start = max(0.0, total_duration - video_fade)
|
|
fade_label = "video_faded"
|
|
fade_parts = []
|
|
if apply_fade_in:
|
|
fade_parts.append(f"fade=t=in:st={fade_in_start:.3f}:d={video_fade:.3f}")
|
|
if apply_fade_out:
|
|
fade_parts.append(f"fade=t=out:st={fade_out_start:.3f}:d={video_fade:.3f}")
|
|
filters.append(f"[{current}]" + ",".join(fade_parts) + f"[{fade_label}]")
|
|
current = fade_label
|
|
|
|
title_lines = [x for x in [data.title, data.date, data.place] if x]
|
|
final_label = "vout"
|
|
title_label = "titled"
|
|
title_font = str(Path.home() / ".local/share/fonts/google/BebasNeue-Regular.ttf")
|
|
stamp_font = "/usr/share/fonts/truetype/noto/NotoSans-Bold.ttf"
|
|
caption_font = "/usr/share/fonts/truetype/ibm-plex/IBMPlexSans-SemiBold.ttf"
|
|
if title_lines:
|
|
title_font_size = max(60, height // 10)
|
|
subtitle_font_size = max(40, height // 18)
|
|
main_title = data.title or title_lines[0]
|
|
subtitle = " | ".join(x for x in [data.date, data.place] if x)
|
|
fade_in_start = INTRO_TITLE_START_OFFSET
|
|
fade_in_enabled = INTRO_TITLE_FADE_IN_ENABLED and INTRO_TITLE_FADE_LENGTH > 0
|
|
fade_out_enabled = INTRO_TITLE_FADE_LENGTH > 0
|
|
fade_in_end = fade_in_start + (INTRO_TITLE_FADE_LENGTH if fade_in_enabled else 0.0)
|
|
visible_start = fade_in_end if fade_in_enabled else fade_in_start
|
|
visible_end = visible_start + INTRO_TITLE_VISIBLE
|
|
fade_out_end = visible_end + (INTRO_TITLE_FADE_LENGTH if fade_out_enabled else 0.0)
|
|
if fade_in_enabled:
|
|
title_alpha = (
|
|
f"if(lt(t\\,{fade_in_start:.3f})\\,0\\,"
|
|
f"if(lt(t\\,{fade_in_end:.3f})\\,(t-{fade_in_start:.3f})/{INTRO_TITLE_FADE_LENGTH:.3f}\\,"
|
|
f"if(lt(t\\,{visible_end:.3f})\\,1\\,"
|
|
f"if(lt(t\\,{fade_out_end:.3f})\\,({fade_out_end:.3f}-t)/{INTRO_TITLE_FADE_LENGTH:.3f}\\,0))))"
|
|
)
|
|
else:
|
|
if fade_out_enabled:
|
|
title_alpha = (
|
|
f"if(lt(t\\,{fade_in_start:.3f})\\,0\\,"
|
|
f"if(lt(t\\,{visible_end:.3f})\\,1\\,"
|
|
f"if(lt(t\\,{fade_out_end:.3f})\\,({fade_out_end:.3f}-t)/{INTRO_TITLE_FADE_LENGTH:.3f}\\,0)))"
|
|
)
|
|
else:
|
|
title_alpha = (
|
|
f"if(lt(t\\,{fade_in_start:.3f})\\,0\\,"
|
|
f"if(lt(t\\,{visible_end:.3f})\\,1\\,0))"
|
|
)
|
|
title_enable_end = fade_out_end if fade_out_enabled else visible_end
|
|
title_enable = f"enable='between(t,{fade_in_start:.3f},{title_enable_end:.3f})'"
|
|
if subtitle:
|
|
title_y = f"(h-text_h)/2-{title_font_size // 2}"
|
|
filters.append(
|
|
f"[{current}]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2:y={title_y}:"
|
|
f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:"
|
|
f"alpha='{title_alpha}':{title_enable}[title0]"
|
|
)
|
|
filters.append(
|
|
f"[title0]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2+1:y={title_y}:"
|
|
f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:"
|
|
f"alpha='{title_alpha}':{title_enable}[title0b]"
|
|
)
|
|
filters.append(
|
|
f"[title0b]drawtext=fontfile='{title_font}':{textfile_arg(subtitle)}:x=(w-text_w)/2:y=(h-text_h)/2+{subtitle_font_size}:"
|
|
f"fontsize={subtitle_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=2:shadowy=2:"
|
|
f"alpha='{title_alpha}':{title_enable}[{title_label}]"
|
|
)
|
|
else:
|
|
filters.append(
|
|
f"[{current}]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2:y=(h-text_h)/2:"
|
|
f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:"
|
|
f"alpha='{title_alpha}':{title_enable}[title0]"
|
|
)
|
|
filters.append(
|
|
f"[title0]drawtext=fontfile='{title_font}':{textfile_arg(main_title)}:x=(w-text_w)/2+1:y=(h-text_h)/2:"
|
|
f"fontsize={title_font_size}:fontcolor=white:borderw=0:shadowcolor=black@0.75:shadowx=3:shadowy=3:"
|
|
f"alpha='{title_alpha}':{title_enable}[{title_label}]"
|
|
)
|
|
else:
|
|
filters.append(f"[{current}]copy[{title_label}]")
|
|
|
|
caption_label = title_label
|
|
caption_size = max(24, int((subtitle_font_size if title_lines else max(40, height // 18)) * 0.7))
|
|
for idx, caption in enumerate(captions):
|
|
if not caption:
|
|
continue
|
|
if n == 1:
|
|
start = intro_duration
|
|
end = total_duration
|
|
elif idx == 0:
|
|
start = intro_duration
|
|
end = intro_duration + transition_offsets[0]
|
|
elif idx == n - 1:
|
|
start = intro_duration + transition_offsets[idx - 1] + fade
|
|
end = total_duration
|
|
else:
|
|
start = intro_duration + transition_offsets[idx - 1] + fade
|
|
end = intro_duration + transition_offsets[idx]
|
|
if end <= start:
|
|
continue
|
|
caption_fade = min(1.0, max(0.0, (end - start) / 2))
|
|
caption_alpha = (
|
|
f"if(lt(t\\,{start + caption_fade:.3f})\\,(t-{start:.3f})/{caption_fade:.3f}\\,"
|
|
f"if(lt(t\\,{end - caption_fade:.3f})\\,1\\,({end:.3f}-t)/{caption_fade:.3f}))"
|
|
) if caption_fade > 0 else "1"
|
|
caption_lines = []
|
|
for line in caption.split("|"):
|
|
clean_line = strip_unsupported_caption_chars(line.strip())
|
|
if clean_line:
|
|
caption_lines.append(clean_line)
|
|
line_gap = int(caption_size * 1.2)
|
|
block_height = line_gap * (len(caption_lines) - 1)
|
|
for line_idx, caption_line in enumerate(caption_lines):
|
|
y_expr = f"h-text_h-62-{block_height - line_idx * line_gap}"
|
|
next_label = f"caption{idx}_{line_idx}"
|
|
filters.append(
|
|
f"[{caption_label}]drawtext=fontfile='{caption_font}':{textfile_arg(caption_line)}:x=(w-text_w)/2:y={y_expr}:"
|
|
f"fontsize={caption_size}:fontcolor=0xF2F2EE:borderw=0:shadowcolor=black@0.80:shadowx=3:shadowy=3:"
|
|
f"alpha='{caption_alpha}':enable='between(t,{start:.3f},{end:.3f})'[{next_label}]"
|
|
)
|
|
caption_label = next_label
|
|
|
|
stamp_size = max(16, height // 58)
|
|
stamp_left_label = caption_label
|
|
for idx, stamp_text in enumerate(clip_stamps):
|
|
if not stamp_text:
|
|
continue
|
|
if n == 1:
|
|
start = intro_duration
|
|
end = total_duration
|
|
elif idx == 0:
|
|
start = intro_duration
|
|
end = intro_duration + transition_offsets[0] + fade / 2
|
|
elif idx == n - 1:
|
|
start = intro_duration + transition_offsets[idx - 1] + fade / 2
|
|
end = total_duration
|
|
else:
|
|
start = intro_duration + transition_offsets[idx - 1] + fade / 2
|
|
end = intro_duration + transition_offsets[idx] + fade / 2
|
|
if end <= start:
|
|
continue
|
|
next_label = f"stamp_left_{idx}"
|
|
filters.append(
|
|
f"[{stamp_left_label}]drawtext=fontfile='{stamp_font}':{textfile_arg(stamp_text)}:x=24:y=h-text_h-18:"
|
|
f"fontsize={stamp_size}:fontcolor=white:borderw=1:bordercolor=black:enable='between(t,{start:.3f},{end:.3f})'[{next_label}]"
|
|
)
|
|
stamp_left_label = next_label
|
|
filters.append(
|
|
f"[{stamp_left_label}]drawtext=fontfile='{stamp_font}':{textfile_arg('Bubulescu.Org')}:x=w-text_w-24:y=h-text_h-18:"
|
|
f"fontsize={stamp_size}:fontcolor=white:borderw=1:bordercolor=black[{final_label}]"
|
|
)
|
|
|
|
bg_audio_label: str | None = None
|
|
if audio:
|
|
audio_index = len(clips)
|
|
audio_fade_out_start = max(0.0, total_duration - audio_fade)
|
|
filters.append(
|
|
f"[{audio_index}:a:0]atrim=0:{total_duration:.3f},asetpts=PTS-STARTPTS,"
|
|
f"afade=t=in:st=0:d={audio_fade_in:.3f},afade=t=out:st={audio_fade_out_start:.3f}:d={audio_fade:.3f}[bg_audio]"
|
|
)
|
|
bg_audio_label = "bg_audio"
|
|
clip_audio_labels: list[str] = []
|
|
clip_audio_input = clip_audio_input_start
|
|
for idx, clip_audio in enumerate(clip_audio_paths):
|
|
if clip_audio is None:
|
|
continue
|
|
if n == 1:
|
|
start = intro_duration
|
|
end = total_duration
|
|
elif idx == 0:
|
|
start = intro_duration
|
|
end = intro_duration + transition_offsets[0]
|
|
elif idx == n - 1:
|
|
start = intro_duration + transition_offsets[idx - 1] + fade
|
|
end = total_duration
|
|
else:
|
|
start = intro_duration + transition_offsets[idx - 1] + fade
|
|
end = intro_duration + transition_offsets[idx]
|
|
segment = max(0.0, end - start)
|
|
if segment <= 0:
|
|
clip_audio_input += 1
|
|
continue
|
|
fade_len = min(CLIP_AUDIO_FADE, segment / 2)
|
|
delay_ms = max(0, int(start * 1000))
|
|
label = f"clip_audio_{idx}"
|
|
filters.append(
|
|
f"[{clip_audio_input}:a:0]atrim=0:{segment:.3f},asetpts=PTS-STARTPTS,"
|
|
f"afade=t=in:st=0:d={fade_len:.3f},afade=t=out:st={max(0.0, segment - fade_len):.3f}:d={fade_len:.3f},"
|
|
f"adelay={delay_ms}:all=1[{label}]"
|
|
)
|
|
clip_audio_labels.append(label)
|
|
clip_audio_input += 1
|
|
clip_mix_label: str | None = None
|
|
if clip_audio_labels:
|
|
if len(clip_audio_labels) == 1:
|
|
clip_mix_label = clip_audio_labels[0]
|
|
else:
|
|
inputs = "".join(f"[{name}]" for name in clip_audio_labels)
|
|
filters.append(inputs + f"amix=inputs={len(clip_audio_labels)}:duration=longest:normalize=0[clip_mix]")
|
|
clip_mix_label = "clip_mix"
|
|
|
|
final_bg_label = bg_audio_label
|
|
# Pad clip audio so it doesn't terminate the filter chain early
|
|
if clip_mix_label:
|
|
filters.append(f"[{clip_mix_label}]apad=whole_dur={total_duration:.3f}[clip_padded]")
|
|
final_clip_label = "clip_padded"
|
|
else:
|
|
final_clip_label = clip_mix_label
|
|
|
|
if final_bg_label and final_clip_label:
|
|
sc_label = f"{final_clip_label}_sc"
|
|
mix_label = f"{final_clip_label}_main"
|
|
filters.append(f"[{final_clip_label}]asplit=2[{sc_label}][{mix_label}]")
|
|
filters.append(
|
|
f"[{final_bg_label}][{sc_label}]sidechaincompress=threshold={CLIP_AUDIO_SIDECHAIN_THRESHOLD}:"
|
|
f"ratio={CLIP_AUDIO_SIDECHAIN_RATIO}:attack={CLIP_AUDIO_SIDECHAIN_ATTACK}:"
|
|
f"release={CLIP_AUDIO_SIDECHAIN_RELEASE}[bg_ducked]"
|
|
)
|
|
final_bg_label = "bg_ducked"
|
|
final_clip_label = mix_label
|
|
|
|
if final_bg_label and final_clip_label:
|
|
filters.append(f"[{final_bg_label}][{final_clip_label}]amix=inputs=2:duration=longest:normalize=0[aout]")
|
|
elif final_bg_label:
|
|
filters.append(f"[{final_bg_label}]anull[aout]")
|
|
elif final_clip_label:
|
|
filters.append(f"[{final_clip_label}]anull[aout]")
|
|
|
|
cmd += ["-filter_complex", ";".join(filters), "-map", f"[{final_label}]"]
|
|
metadata = {
|
|
"title": data.title,
|
|
"date": data.date,
|
|
"location": data.place,
|
|
"place": data.place,
|
|
"com.apple.quicktime.location.name": data.place,
|
|
"description": data.description,
|
|
"synopsis": data.description,
|
|
"artist": "Bubulescu.Org",
|
|
"comment": "Created with movmaker",
|
|
"creation_time": creation_date.isoformat(),
|
|
}
|
|
for key, value in metadata.items():
|
|
if value:
|
|
cmd += ["-metadata", f"{key}={value}"]
|
|
|
|
if final_bg_label or final_clip_label:
|
|
cmd += ["-map", "[aout]", "-c:a", "aac", "-b:a", "192k", "-shortest"]
|
|
else:
|
|
cmd += ["-an"]
|
|
|
|
cmd += ["-c:v", "libx264", "-preset", preset, "-crf", str(crf), "-r", str(fps), "-pix_fmt", "yuv420p", "-movflags", "+faststart+use_metadata_tags", str(output)]
|
|
run(cmd, total_duration)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
"""CLI entry point for movmaker."""
|
|
parser = argparse.ArgumentParser(description="Make a video from a flat folder of images/videos/audio.")
|
|
parser.add_argument("input", type=Path, help="flat input directory")
|
|
parser.add_argument("--out-dir", type=Path, default=Path("."), help="directory for auto-named output videos")
|
|
parser.add_argument("--image-duration", type=float, default=DEFAULT_IMAGE_DURATION, help="seconds each image is visible")
|
|
parser.add_argument("--fade", type=float, default=DEFAULT_CROSSFADE_DURATION, help="crossfade duration in seconds")
|
|
parser.add_argument("--audio-fade", type=float, default=DEFAULT_AUDIO_FADE_OUT, help="music fade-out duration in seconds")
|
|
parser.add_argument("--audio-fade-in", type=float, default=DEFAULT_AUDIO_FADE_IN, help="music fade-in duration in seconds")
|
|
parser.add_argument("--video-fade", type=float, default=DEFAULT_VIDEO_FADE, help="video fade-in/fade-out duration in seconds")
|
|
parser.add_argument("--intro-logo", type=Path, default=DEFAULT_INTRO_LOGO, help="PNG logo shown centered over the black intro")
|
|
parser.add_argument("--intro-logo-alpha", type=float, default=INTRO_LOGO_DEFAULT_ALPHA, help="opacity for the intro logo, from 0 to 1")
|
|
parser.add_argument("--resolution", type=parse_resolution, default=(1920, 1080), help="output resolution, e.g. 1920x1080")
|
|
parser.add_argument("--fps", type=int, default=30, help="output frames per second")
|
|
parser.add_argument("--preset", default="veryfast", help="x264 speed preset, e.g. ultrafast, superfast, veryfast")
|
|
parser.add_argument("--crf", type=int, default=20, help="x264 quality; lower is better/larger, higher is faster/smaller")
|
|
parser.add_argument("--jobs", type=int, default=1, help="parallel clip preparation jobs; try 2 on Raspberry Pi")
|
|
parser.add_argument("--preview", action="store_true", help="fast preview: 1280x720, 24 fps, ultrafast, CRF 28")
|
|
parser.add_argument("--music-genre", default="cinematic punk rock", help="Jamendo music search query used only when input has no audio files")
|
|
args = parser.parse_args(argv)
|
|
|
|
global LOG_PATH
|
|
log_dir = args.input / ".logs"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
LOG_PATH = log_dir / f"movmaker_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
|
|
LOG_PATH.write_text("movmaker log\n", encoding="utf-8")
|
|
|
|
require_tool("ffmpeg")
|
|
require_tool("ffprobe")
|
|
|
|
width, height = args.resolution
|
|
fps = args.fps
|
|
preset = args.preset
|
|
crf = args.crf
|
|
if args.preview:
|
|
width, height = (1280, 720)
|
|
fps = 20
|
|
preset = "ultrafast"
|
|
crf = 28
|
|
|
|
media, audio_files, data = scan_input(args.input, args.image_duration)
|
|
first_media_date = media_date(media[0].path)
|
|
first_media_resolution = ffprobe_resolution(media[0].path)
|
|
if not data.date:
|
|
data.date = display_date(first_media_date)
|
|
if not data.place:
|
|
data.place = infer_place_from_gps(media) or ""
|
|
fade = min(args.fade, max(0.0, args.image_duration - 0.1))
|
|
audio_fade = max(0.0, args.audio_fade)
|
|
audio_fade_in = max(0.0, args.audio_fade_in)
|
|
video_fade = max(0.0, args.video_fade)
|
|
intro_logo_alpha = min(1.0, max(0.0, args.intro_logo_alpha))
|
|
intro_logo_path = args.intro_logo if args.intro_logo.exists() else DEFAULT_INTRO_LOGO if DEFAULT_INTRO_LOGO.exists() else None
|
|
|
|
if not audio_files:
|
|
downloaded_audio = download_jamendo_music(args.music_genre, args.input)
|
|
if downloaded_audio:
|
|
audio_files.append(downloaded_audio)
|
|
|
|
output = default_output_path(media[0].path, data, args.out_dir)
|
|
|
|
# Image duration means fully visible time, not including crossfades.
|
|
# Middle images need both a fade-in and fade-out added to their clip length.
|
|
if len(media) > 1:
|
|
for idx, item in enumerate(media):
|
|
if item.kind != "image":
|
|
continue
|
|
if idx == 0:
|
|
item.duration = max(item.duration, (args.image_duration * FIRST_IMAGE_DURATION_MULTIPLIER) + fade)
|
|
elif idx == len(media) - 1:
|
|
item.duration = item.duration + fade
|
|
else:
|
|
item.duration = item.duration + (2 * fade)
|
|
|
|
console(f"Found {len(media)} media file(s), {len(audio_files)} audio file(s).")
|
|
console(f"Audio: {audio_files[0] if audio_files else 'none'}")
|
|
if audio_files:
|
|
artist, title = audio_metadata(audio_files[0])
|
|
if artist or title:
|
|
console("Audio metadata: " + " - ".join(x for x in [artist, title] if x))
|
|
console(f"Output: {output}")
|
|
console(f"Log: {LOG_PATH}")
|
|
if data.title or data.date or data.place:
|
|
console("Title data: " + " | ".join(x for x in [data.title, data.date, data.place] if x))
|
|
|
|
with tempfile.TemporaryDirectory(prefix="movmaker-") as tmp_name:
|
|
tmp = Path(tmp_name)
|
|
jobs = max(1, args.jobs)
|
|
clips = [tmp / f"clip_{idx:04d}.mp4" for idx in range(len(media))]
|
|
if jobs == 1:
|
|
total_clips = len(media)
|
|
for done, (item, out) in enumerate(zip(media, clips), start=1):
|
|
percent = done / total_clips * 100
|
|
console(f"Preparing clips {done}/{total_clips} ({percent:.1f}%)")
|
|
log_message(f"Preparing {item.kind}: {item.path.name}")
|
|
normalize_clip(item, out, width, height, fps, preset, crf)
|
|
else:
|
|
console(f"Preparing clips with {jobs} parallel jobs")
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as executor:
|
|
futures = []
|
|
for item, out in zip(media, clips):
|
|
log_message(f"Preparing {item.kind}: {item.path.name}")
|
|
futures.append(executor.submit(normalize_clip, item, out, width, height, fps, preset, crf))
|
|
total_clips = len(futures)
|
|
for done, future in enumerate(concurrent.futures.as_completed(futures), start=1):
|
|
future.result()
|
|
percent = done / total_clips * 100
|
|
console(f"Preparing clips {done}/{total_clips} ({percent:.1f}%)")
|
|
|
|
# Use probed normalized duration for accuracy.
|
|
durations = [ffprobe_duration(out) for out in clips]
|
|
captions = [data.captions.get(item.path.name, data.captions.get(str(item.path), "")) for item in media]
|
|
video_audio = data.video_audio or set()
|
|
clip_audio_paths = []
|
|
for item in media:
|
|
use_clip_audio = item.kind == "video" and (item.path.name in video_audio or str(item.path) in video_audio)
|
|
if use_clip_audio and media_has_audio(item.path):
|
|
clip_audio_paths.append(item.path)
|
|
console(f"Using video file audio: {item.path.name}")
|
|
elif use_clip_audio:
|
|
clip_audio_paths.append(None)
|
|
console(f"Video file has no audio stream: {item.path.name}")
|
|
else:
|
|
clip_audio_paths.append(None)
|
|
clip_stamps = prepare_clip_stamps(media, data)
|
|
audio = concat_or_mix_audio(audio_files, tmp / "music.m4a")
|
|
if len(audio_files) > 1:
|
|
console(f"Audio: {audio if audio else 'none'}")
|
|
if audio:
|
|
artist, title = audio_metadata(audio)
|
|
if artist or title:
|
|
console("Audio metadata: " + " - ".join(x for x in [artist, title] if x))
|
|
video_intro_duration = audio_fade_in
|
|
intro_visible_lead = 0.0
|
|
intro_clip_path: Path | None = None
|
|
intro_clip_duration = 0.0
|
|
if intro_logo_path is not None and audio_fade_in > 0:
|
|
if first_media_resolution is None:
|
|
console("Warning: Could not determine first media resolution; skipping intro logo frame")
|
|
else:
|
|
intro_width, intro_height = first_media_resolution
|
|
intro_image = tmp / "clip_intro_logo.png"
|
|
intro_clip = tmp / "clip_intro_logo.mp4"
|
|
console("Preparing intro logo frame")
|
|
render_intro_logo_frame(intro_logo_path, intro_image, intro_width, intro_height, intro_logo_alpha)
|
|
intro_item = MediaItem(path=intro_image, kind="image", duration=audio_fade_in + INTRO_LOGO_EXTRA_HOLD)
|
|
normalize_clip(intro_item, intro_clip, width, height, fps, preset, crf)
|
|
clips = [intro_clip, *clips]
|
|
durations = [intro_item.duration, *durations]
|
|
intro_clip_path = intro_clip
|
|
intro_clip_duration = intro_item.duration
|
|
captions = ["", *captions]
|
|
clip_audio_paths = [None, *clip_audio_paths]
|
|
clip_stamps = ["", *clip_stamps]
|
|
intro_visible_lead = min(intro_item.duration, INTRO_LOGO_FADE_OFFSET)
|
|
video_intro_duration = 0.0
|
|
outro_added = False
|
|
if intro_clip_path is not None:
|
|
clips.append(intro_clip_path)
|
|
durations.append(intro_clip_duration)
|
|
captions.append("")
|
|
clip_audio_paths.append(None)
|
|
clip_stamps.append("")
|
|
outro_added = True
|
|
build_video(clips, durations, audio, clip_audio_paths, output, data, first_media_date, captions, clip_stamps, fade, audio_fade, audio_fade_in, video_fade, args.intro_logo, intro_logo_alpha, width, height, fps, preset, crf, tmp, video_intro_duration, intro_visible_lead=intro_visible_lead, skip_video_fade_out=outro_added)
|
|
|
|
console(f"Done: {output}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|