1031 lines
38 KiB
Python
Executable file
1031 lines
38 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"
|
|
JAMENDO_ENV = Path.home() / ".config" / "movmaker" / "jamendo.env"
|
|
LOG_PATH: Path | None = None
|
|
|
|
|
|
def tool_path(local_path: Path, fallback: str) -> str:
|
|
return str(local_path) if local_path.exists() else fallback
|
|
|
|
|
|
@dataclass
|
|
class MovieData:
|
|
title: str = ""
|
|
date: str = ""
|
|
place: str = ""
|
|
captions: dict[str, str] | None = None
|
|
|
|
|
|
@dataclass
|
|
class MediaItem:
|
|
path: Path
|
|
kind: str # image or video
|
|
duration: float
|
|
|
|
|
|
def log_message(message: str) -> None:
|
|
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]:
|
|
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:
|
|
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_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_date(path: Path) -> datetime:
|
|
ext = path.suffix.lower()
|
|
dt = None
|
|
if ext in IMAGE_EXTS:
|
|
try:
|
|
dt = exif_image_date(path)
|
|
except Exception:
|
|
dt = None
|
|
if dt is None:
|
|
dt = ffprobe_creation_date(path)
|
|
if dt is None:
|
|
dt = datetime.fromtimestamp(path.stat().st_mtime)
|
|
return dt
|
|
|
|
|
|
def ascii_safe(value: str) -> str:
|
|
replacements = {
|
|
"æ": "ae", "Æ": "Ae",
|
|
"ø": "o", "Ø": "O",
|
|
"å": "aa", "Å": "Aa",
|
|
"đ": "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:
|
|
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 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 natural_key(path: Path) -> list[object]:
|
|
# Simple natural-ish sort: IMG_2 before IMG_10.
|
|
import re
|
|
parts = re.split(r"(\d+)", path.name.lower())
|
|
return [int(p) if p.isdigit() else p for p in parts]
|
|
|
|
|
|
def scan_input(input_dir: Path, image_duration: float) -> tuple[list[MediaItem], list[Path], MovieData]:
|
|
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 sorted(input_dir.iterdir(), key=natural_key):
|
|
if not path.is_file():
|
|
continue
|
|
ext = path.suffix.lower()
|
|
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}")
|
|
|
|
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={})
|
|
|
|
lines = [line.strip() for line in path.read_text(encoding="utf-8", errors="replace").splitlines()]
|
|
lines = [line for line in lines if line and not line.startswith("#")]
|
|
|
|
data = MovieData(captions={})
|
|
plain: list[str] = []
|
|
|
|
for line in lines:
|
|
if ":" in line:
|
|
key, value = line.split(":", 1)
|
|
key = key.strip()
|
|
value = value.strip()
|
|
low = key.lower()
|
|
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 Path(key).suffix.lower() in IMAGE_EXTS | VIDEO_EXTS:
|
|
data.captions[key] = value
|
|
else:
|
|
plain.append(line)
|
|
else:
|
|
plain.append(line)
|
|
|
|
# Simple mode: first three non-key lines are title/location/date.
|
|
if plain:
|
|
data.title = data.title or plain[0]
|
|
if len(plain) > 1:
|
|
data.place = data.place or plain[1]
|
|
if len(plain) > 2:
|
|
data.date = data.date or plain[2]
|
|
|
|
return data
|
|
|
|
|
|
def drawtext_escape(text: str) -> str:
|
|
# Escape for ffmpeg drawtext text=... without surrounding quotes.
|
|
return (
|
|
text.replace("\\", "\\\\")
|
|
.replace(":", "\\:")
|
|
.replace(",", "\\,")
|
|
.replace("%", "\\%")
|
|
)
|
|
|
|
|
|
def normalize_clip(item: MediaItem, out: Path, width: int, height: int, fps: int, preset: str, crf: int) -> None:
|
|
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 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:
|
|
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:
|
|
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,
|
|
output: Path,
|
|
data: MovieData,
|
|
creation_date: datetime,
|
|
captions: list[str],
|
|
fade: float,
|
|
audio_fade: float,
|
|
audio_fade_in: float,
|
|
video_fade: float,
|
|
width: int,
|
|
height: int,
|
|
fps: int,
|
|
preset: str,
|
|
crf: int,
|
|
text_dir: Path,
|
|
) -> None:
|
|
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)]
|
|
|
|
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 = 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
|
|
if video_fade > 0:
|
|
fade_in_start = intro_duration
|
|
fade_out_start = max(0.0, total_duration - video_fade)
|
|
fade_label = "video_faded"
|
|
filters.append(
|
|
f"[{current}]fade=t=in:st={fade_in_start:.3f}:d={video_fade:.3f},fade=t=out:st={fade_out_start:.3f}:d={video_fade:.3f}[{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)
|
|
title_alpha = "if(lt(t\\,5)\\,1\\,if(lt(t\\,6)\\,6-t\\,0))"
|
|
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}':enable='between(t,0,6)'[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}':enable='between(t,0,6)'[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}':enable='between(t,0,6)'[{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}':enable='between(t,0,6)'[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}':enable='between(t,0,6)'[{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 = [line.strip() for line in caption.split("|") if line.strip()]
|
|
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_lines = [x for x in [data.title, data.date, data.place] if x]
|
|
stamp_text = drawtext_escape(" | ".join(stamp_lines))
|
|
stamp_left_label = "stamp_left"
|
|
if stamp_text:
|
|
filters.append(
|
|
f"[{caption_label}]drawtext=fontfile='{stamp_font}':{textfile_arg(' | '.join(stamp_lines))}:x=24:y=h-text_h-18:"
|
|
f"fontsize={stamp_size}:fontcolor=white:borderw=1:bordercolor=black[{stamp_left_label}]"
|
|
)
|
|
else:
|
|
filters.append(f"[{caption_label}]copy[{stamp_left_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}]"
|
|
)
|
|
|
|
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,
|
|
"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 audio:
|
|
audio_index = len(clips)
|
|
audio_fade_out_start = max(0.0, total_duration - audio_fade)
|
|
audio_filter = f"afade=t=in:st=0:d={audio_fade_in:.3f},afade=t=out:st={audio_fade_out_start:.3f}:d={audio_fade:.3f}"
|
|
cmd += ["-map", f"{audio_index}:a:0", "-af", audio_filter, "-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:
|
|
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=6.0, help="seconds each image is visible")
|
|
parser.add_argument("--fade", type=float, default=3.0, help="crossfade duration in seconds")
|
|
parser.add_argument("--audio-fade", type=float, default=10.0, help="music fade-out duration in seconds")
|
|
parser.add_argument("--audio-fade-in", type=float, default=2.0, help="music fade-in duration in seconds")
|
|
parser.add_argument("--video-fade", type=float, default=1.5, help="video fade-in/fade-out duration in seconds")
|
|
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 = 24
|
|
preset = "ultrafast"
|
|
crf = 28
|
|
|
|
media, audio_files, data = scan_input(args.input, args.image_duration)
|
|
first_media_date = media_date(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)
|
|
|
|
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 * 2) + 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]
|
|
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))
|
|
build_video(clips, durations, audio, output, data, first_media_date, captions, fade, audio_fade, audio_fade_in, video_fade, width, height, fps, preset, crf, tmp)
|
|
|
|
console(f"Done: {output}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|