marijo/movmaker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refine logging and Jamendo docs

Browse source
This commit is contained in:
hbrain 2026-05-25 10:19:45 +00:00
parent 55ee91cae2
commit c42e44231a
3 changed files with 76 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

112
movmaker.py
View file

@ -34,7 +34,7 @@ DATA_NAMES = {"data.txt", "info.txt", "title.txt"}
BASE_DIR = Path(__file__).resolve().parent
FFMPEG = BASE_DIR / "ffmpeg" / "ffmpeg"
FFPROBE = BASE_DIR / "ffmpeg" / "ffprobe"
PIXABAY_ENV = Path.home() / ".config" / "movmaker" / "pixabay.env"
JAMENDO_ENV = Path.home() / ".config" / "movmaker" / "jamendo.env"
LOG_PATH: Path | None = None
@ -120,7 +120,7 @@ def run(cmd: list[str], progress_duration: float | None = None) -> None:
percent = min(100.0, max(0.0, seconds / progress_duration * 100.0))
fps = fields.get("fps", "?")
speed = fields.get("speed", "?")
sys.stdout.write(f"\rRendering {percent:5.1f}% fps={fps} speed={speed} written={format_bytes(written)}")
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")
@ -418,7 +418,7 @@ def infer_place_from_gps(media: list[MediaItem]) -> str | None:
continue
place = reverse_geocode_city_country(*gps)
if place:
print(f"Location from GPS: {place} ({gps[0]:.6f}, {gps[1]:.6f})")
console(f"Location from GPS: {place} ({gps[0]:.6f}, {gps[1]:.6f})")
return place
return None
@ -469,6 +469,14 @@ def display_date(dt: datetime) -> str:
return f"{dt.strftime('%B')} {ordinal_day(dt.day)} {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")
@ -585,66 +593,70 @@ def normalize_clip(item: MediaItem, out: Path, width: int, height: int, fps: int
run(cmd)
def load_pixabay_api_key() -> str | None:
if os.environ.get("PIXABAY_API_KEY"):
return os.environ["PIXABAY_API_KEY"]
if PIXABAY_ENV.exists():
for line in PIXABAY_ENV.read_text(encoding="utf-8", errors="replace").splitlines():
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() == "PIXABAY_API_KEY":
if key.strip() == "JAMENDO_CLIENT_ID":
return value.strip().strip("'\"")
return None
def download_pixabay_music(genre: str, out_dir: Path) -> Path | None:
api_key = load_pixabay_api_key()
if not api_key:
print("No audio files found and no Pixabay API key configured; continuing without music.")
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({
"key": api_key,
"q": genre,
"media_type": "music",
"safesearch": "true",
"per_page": "50",
"client_id": client_id,
"format": "json",
"limit": "50",
"search": genre,
"include": "musicinfo",
"audioformat": "mp32",
})
request = urllib.request.Request(
f"https://pixabay.com/api/audio/?{query}",
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:
print(f"Could not search Pixabay music: {exc}")
console(f"Could not search Jamendo music: {exc}")
return None
hits = [hit for hit in data.get("hits", []) if hit.get("audio")]
if not hits:
print(f"No Pixabay music found for genre/query: {genre}")
results = [track for track in data.get("results", []) if track.get("audio")]
if not results:
console(f"No Jamendo music found for genre/query: {genre}")
return None
hit = random.choice(hits)
audio_url = hit["audio"]
title = hit.get("title") or "pixabay_music"
filename = f"pixabay_{hit.get('id', 'music')}_{slugify(title)}.mp3"
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
print(f"Downloading Pixabay music: {title}")
console(f"Downloading Jamendo music: {title}" + (f" by {artist}" if artist else ""))
try:
urllib.request.urlretrieve(audio_url, out)
except Exception as exc:
print(f"Could not download Pixabay music: {exc}")
console(f"Could not download Jamendo music: {exc}")
return None
attribution = out.with_suffix(".txt")
musicinfo = track.get("musicinfo", {}) if isinstance(track.get("musicinfo"), dict) else {}
attribution.write_text(
"Pixabay music downloaded by movmaker\n"
"Jamendo music downloaded by movmaker\n"
f"Title: {title}\n"
f"Artist/User: {hit.get('user', '')}\n"
f"URL: {hit.get('pageURL', '')}\n"
"Source: Pixabay\n",
f"Artist: {artist}\n"
f"URL: {track.get('shareurl', '')}\n"
f"License: {musicinfo.get('license_ccurl', '')}\n"
"Source: Jamendo\n",
encoding="utf-8",
)
return out
@ -806,7 +818,7 @@ def build_video(
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='{stamp_bold_font}':{textfile_arg(caption_line)}:x=(w-text_w)/2:y={y_expr}:"
f"[{caption_label}]drawtext=fontfile='{title_font}':{textfile_arg(caption_line)}:x=(w-text_w)/2:y={y_expr}:"
f"fontsize={caption_size}:fontcolor=white:borderw=1:bordercolor=black:"
f"alpha='{caption_alpha}':enable='between(t,{start:.3f},{end:.3f})'[{next_label}]"
)
@ -870,11 +882,13 @@ def main(argv: list[str] | None = None) -> int:
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="Pixabay music search query used only when input has no audio files")
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_PATH = args.input / "movmaker.log"
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")
@ -902,7 +916,7 @@ def main(argv: list[str] | None = None) -> int:
video_fade = max(0.0, args.video_fade)
if not audio_files:
downloaded_audio = download_pixabay_music(args.music_genre, args.input)
downloaded_audio = download_jamendo_music(args.music_genre, args.input)
if downloaded_audio:
audio_files.append(downloaded_audio)
@ -921,30 +935,36 @@ def main(argv: list[str] | None = None) -> int:
else:
item.duration = item.duration + (2 * fade)
print(f"Found {len(media)} media file(s), {len(audio_files)} audio file(s).")
print(f"Output: {output}")
print(f"Log: {LOG_PATH}")
console(f"Found {len(media)} media file(s), {len(audio_files)} audio file(s).")
console(f"Output: {output}")
console(f"Log: {LOG_PATH}")
if data.title or data.date or data.place:
print("Title data:", " | ".join(x for x in [data.title, data.date, data.place] if x))
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:
print("Preparing clips...")
for item, out in zip(media, clips):
console("Preparing clips...")
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:
print(f"Preparing clips with {jobs} parallel jobs")
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))
for future in concurrent.futures.as_completed(futures):
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]
@ -953,7 +973,7 @@ def main(argv: list[str] | None = None) -> int:
audio = concat_or_mix_audio(audio_files, tmp / "music.m4a")
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)
print(f"Done: {output}")
console(f"Done: {output}")
return 0