`build_hangover_montage.py` creates a video montage from images and videos, synced to a chosen audio track. It processes images (portrait or landscape) by resizing, cropping, or adding a blurred background. Videos are either kept at their original length, sped up, or split into chunks. The final montage is adjusted to the audio's duration and sa…

build_hangover_montage.py

import os
import random
import logging
from typing import List, Tuple, Optional
from pathlib import Path

import numpy as np
from PIL import Image, ImageFilter

# Try to enable HEIC support if pillow-heif is available
try:
	import pillow_heif
	pillow_heif.register_heif_opener()
	HEIC_SUPPORTED = True
except Exception:
	HEIC_SUPPORTED = False

from moviepy.editor import (
	ImageClip,
	VideoFileClip,
	AudioFileClip,
	concatenate_videoclips,
	CompositeVideoClip,
	vfx,
	ColorClip,
)

# ==========================
# Configuration (tweak here)
# ==========================
MEDIA_DIR = Path("media")
# Accept either exact provided name or the existing file in workspace
AUDIO_CANDIDATES = [Path("rightround.mp3"), Path("right round.mp3")]  # will pick the first that exists
OUTPUT_FILE = Path("hangover_montage.mp4")

FINAL_SIZE: Tuple[int, int] = (1080, 1920)  # (width, height)
FPS: int = 24

# Memory saver: reduce concurrency/threads, use chain concat, drop per-clip audio
MEMORY_SAVER: bool = True

# Image durations to choose from (seconds)
IMAGE_DURATIONS: List[float] = [0.7 , 0.5]

# Landscape handling mode for both images and videos: "crop" or "blur"
LANDSCAPE_MODE: str = "blur"

# Video rules (seconds)
VIDEO_SHORT_MAX: float = 15.0
VIDEO_MEDIUM_MAX: float = 40.0
VIDEO_LONG_MIN: float = 60.0

# Speed-up factors for medium videos
VIDEO_SPEED_OPTIONS: List[float] = [2.0, 3.0]

# Chunking for long videos (seconds)
VIDEO_CHUNK_MIN: float = 5.0
VIDEO_CHUNK_MAX: float = 10.0
MAX_CHUNKS_PER_VIDEO: int = 6  # safety cap

# Expected counts (from user info). Only used for logging, not enforcement.
EXPECTED_MEDIA_COUNT = 81
EXPECTED_VIDEO_COUNT = 8

# Random seed for reproducibility if desired (set to None for pure random)
RANDOM_SEED: Optional[int] = None  # e.g., 42

# Concat method based on memory mode
CONCAT_METHOD: str = "chain" if MEMORY_SAVER else "compose"

# ==========================
# Logging setup
# ==========================
logging.basicConfig(
	level=logging.INFO,
	format="[%(levelname)s] %(message)s",
)
logger = logging.getLogger("hangover_montage")

if RANDOM_SEED is not None:
	random.seed(RANDOM_SEED)
	np.random.seed(RANDOM_SEED)

# ==========================
# Helpers
# ==========================
IMAGE_EXTS = {".jpg", ".jpeg", ".heic", ".HEIC", ".JPG", ".JPEG"}
VIDEO_EXTS = {".mp4", ".mov", ".MP4", ".MOV"}


def find_audio_file() -> Path:
	for candidate in AUDIO_CANDIDATES:
		if candidate.exists():
			logger.info(f"Audio selected: {candidate}")
			return candidate
	raise FileNotFoundError(
		"No audio file found. Place 'rightround.mp3' in project root (or 'right round.mp3')."
	)


def list_media_files(media_dir: Path) -> Tuple[List[Path], List[Path]]:
	images: List[Path] = []
	videos: List[Path] = []
	for entry in sorted(media_dir.iterdir()):
		if not entry.is_file():
			continue
		ext = entry.suffix
		if ext in IMAGE_EXTS:
			images.append(entry)
		elif ext in VIDEO_EXTS:
			videos.append(entry)
	return images, videos


def load_image_array(path: Path) -> np.ndarray:
	# Use PIL (with pillow-heif registered if available) to read any supported format
	with Image.open(path) as im:
		im = im.convert("RGB")
		return np.array(im)


def blur_clip(clip: ImageClip | VideoFileClip, radius: int = 25) -> ImageClip | VideoFileClip:
    """Apply Gaussian blur. To save memory, if clip is an ImageClip, blur once statically.
    For VideoFileClip, fall back to per-frame blur (slower but memory-safe)."""
    if isinstance(clip, ImageClip):
        # Static image: blur one frame and reuse
        frame = clip.get_frame(0)
        img = Image.fromarray(frame)
        blurred = np.array(img.filter(ImageFilter.GaussianBlur(radius=radius)))
        return ImageClip(blurred).set_duration(clip.duration)
    else:
        def _blur_frame(frame: np.ndarray) -> np.ndarray:
            img = Image.fromarray(frame)
            return np.array(img.filter(ImageFilter.GaussianBlur(radius=radius)))
        return clip.fl_image(_blur_frame)


def make_blur_background(base_clip: ImageClip | VideoFileClip, final_size: Tuple[int, int]) -> CompositeVideoClip:
	# Create a blurred background by resizing to cover, then blurring, then overlaying original centered
	target_w, target_h = final_size
	bg = base_clip.resize(lambda t: max(target_w / base_clip.w, target_h / base_clip.h))
	# Apply Gaussian blur via PIL to ensure compatibility across MoviePy versions
	bg = blur_clip(bg, radius=30).set_position("center").set_duration(base_clip.duration)

	# Foreground: fit within final frame respecting the rule (fit width for landscape per requirement)
	# We'll fit foreground by width if base is landscape else by height
	if base_clip.w >= base_clip.h:
		fg = base_clip.resize(width=target_w)
	else:
		fg = base_clip.resize(height=target_h)
	fg = fg.set_position("center").set_duration(base_clip.duration)

	bg_color = ColorClip(size=final_size, color=(0, 0, 0), duration=base_clip.duration)
	composite = CompositeVideoClip([bg_color, bg, fg], size=final_size)
	return composite


def resize_and_crop_to_fill(clip: ImageClip | VideoFileClip, final_size: Tuple[int, int]) -> ImageClip | VideoFileClip:
	# Scale so the frame is fully covered, then center-crop
	target_w, target_h = final_size
	scale = max(target_w / clip.w, target_h / clip.h)
	scaled = clip.resize(scale)
	x1 = (scaled.w - target_w) // 2
	y1 = (scaled.h - target_h) // 2
	return scaled.crop(x1=x1, y1=y1, x2=x1 + target_w, y2=y1 + target_h)


def resize_portrait_fit_height(clip: ImageClip | VideoFileClip, final_size: Tuple[int, int]) -> CompositeVideoClip | ImageClip | VideoFileClip:
	# For portrait sources: scale to fit height. If width != target, pad with blurred background for exact size
	target_w, target_h = final_size
	resized = clip.resize(height=target_h)
	if resized.w == target_w:
		return resized
	# Pad using blurred background for clean look
	return make_blur_background(resized, final_size)


def build_image_clip(path: Path, final_size: Tuple[int, int]) -> ImageClip | CompositeVideoClip:
	arr = load_image_array(path)
	base = ImageClip(arr)
	orientation = "portrait" if base.h >= base.w else "landscape"
	duration = float(random.choice(IMAGE_DURATIONS))

	logger.info(f"Image: {path.name} | {orientation} | duration={duration:.2f}s")

	if orientation == "portrait":
		processed = resize_portrait_fit_height(base.set_duration(duration), final_size)
		logger.info(f"  -> portrait handling: fit height")
	else:
		if LANDSCAPE_MODE == "crop":
			processed = resize_and_crop_to_fill(base.set_duration(duration), final_size)
			logger.info("  -> landscape handling: crop center to fill")
		else:
			processed = make_blur_background(base.set_duration(duration), final_size)
			logger.info("  -> landscape handling: blur-fill background")

	# Ensure exact size
	processed = processed.resize(newsize=final_size)
	return processed


def decide_video_treatment(duration: float) -> Tuple[str, Optional[float], Optional[float]]:
	"""Return (action, speed_factor, chunk_length)
	action in {"keep", "speed", "split"}
	"""
	if duration <= VIDEO_SHORT_MAX:
		return "keep", None, None
	if duration <= VIDEO_MEDIUM_MAX:
		factor = float(random.choice(VIDEO_SPEED_OPTIONS))
		return "speed", factor, None
	if duration >= VIDEO_LONG_MIN:
		chunk_len = float(random.uniform(VIDEO_CHUNK_MIN, VIDEO_CHUNK_MAX))
		return "split", None, chunk_len
	# default to keep for anything between (40,60)
	return "keep", None, None


def build_video_clips(path: Path, final_size: Tuple[int, int]) -> List[CompositeVideoClip | VideoFileClip | ImageClip]:
	clips: List[CompositeVideoClip | VideoFileClip | ImageClip] = []
	# Keep reader alive for the lifetime of the returned clips; close later in main
	src = VideoFileClip(str(path))
	orientation = "portrait" if src.h >= src.w else "landscape"
	logger.info(f"Video: {path.name} | {orientation} | original_duration={src.duration:.2f}s")

	action, speed_factor, chunk_len = decide_video_treatment(src.duration)
	if action == "keep":
		logger.info("  -> keep duration as-is")
		base = src
	elif action == "speed":
		logger.info(f"  -> speed up by x{speed_factor}")
		base = src.fx(vfx.speedx, factor=speed_factor)
	else:
		assert action == "split"
		# Prepare chunks list
		n_chunks_estimate = int(max(1, min(MAX_CHUNKS_PER_VIDEO, src.duration // max(1.0, chunk_len))))
		starts = sorted(random.sample(
			[float(t) for t in np.linspace(0, max(0.0, src.duration - chunk_len), max(n_chunks_estimate, 1)).tolist()],
			k=max(1, min(n_chunks_estimate, 5)),
		)) if src.duration > chunk_len else [0.0]
		chunked: List[VideoFileClip] = []
		for s in starts:
			end = min(src.duration, s + chunk_len)
			if end - s <= 0.2:
				continue
			chunk = src.subclip(s, end)
			chunked.append(chunk)
		logger.info(f"  -> split into {len(chunked)} chunk(s), each ~{chunk_len:.1f}s")
		bases = chunked

	# Now normalize size/orientation for either a single base clip or multiple chunks
	def normalize_one(c: VideoFileClip | ImageClip) -> CompositeVideoClip | VideoFileClip | ImageClip:
		if orientation == "portrait":
			processed = resize_portrait_fit_height(c, final_size)
			logger.info("  -> portrait handling: scale to 1080x1920 height fit")
		else:
			if LANDSCAPE_MODE == "crop":
				processed = resize_and_crop_to_fill(c, final_size)
				logger.info("  -> landscape handling: crop center to fill")
			else:
				processed = make_blur_background(c, final_size)
				logger.info("  -> landscape handling: blur-fill background")
		return processed.resize(newsize=final_size)

	if action == "split":
		for c in bases:  # type: ignore[name-defined]
			clips.append(normalize_one(c))
	else:
		clips.append(normalize_one(base))  # type: ignore[arg-type]

	# Caller will close src when done by walking clips and closing readers
	return clips


def build_all_clips(images: List[Path], videos: List[Path], final_size: Tuple[int, int]) -> List[CompositeVideoClip | VideoFileClip | ImageClip]:
	prepared: List[CompositeVideoClip | VideoFileClip | ImageClip] = []
	for img in images:
		try:
			prepared.append(build_image_clip(img, final_size))
		except Exception as e:
			logger.exception(f"Failed to process image {img}: {e}")
	for vid in videos:
		try:
			prepared.extend(build_video_clips(vid, final_size))
		except Exception as e:
			logger.exception(f"Failed to process video {vid}: {e}")
	return prepared


def assemble_montage(clips: List[CompositeVideoClip | VideoFileClip | ImageClip], audio_path: Path, fps: int, final_size: Tuple[int, int]) -> CompositeVideoClip:
    audio = AudioFileClip(str(audio_path))
    target_duration = audio.duration
    logger.info(f"Target montage duration (audio): {target_duration:.2f}s")

    random.shuffle(clips)

    selected: List[CompositeVideoClip | VideoFileClip | ImageClip] = []
    cumulative = 0.0
    for clip in clips:
        remaining = target_duration - cumulative
        if remaining <= 0.05:
            break
        clip_duration = float(clip.duration)
        start_time = cumulative
        logger.info(f"Queue clip start @ {start_time:.2f}s | len={clip_duration:.2f}s")
        if clip_duration > remaining:
            # Trim this clip to exactly fit the remaining duration
            try:
                trimmed = clip.subclip(0, max(0.01, remaining))
            except Exception:
                # If subclip is not supported on this clip type, fallback to set_duration
                trimmed = clip.set_duration(max(0.01, remaining))
            selected.append(trimmed)
            cumulative += float(trimmed.duration)
            break
        else:
            selected.append(clip)
            cumulative += clip_duration

    if not selected:
        raise RuntimeError("No clips prepared for montage.")

    montage = concatenate_videoclips(selected, method=CONCAT_METHOD)
    # Ensure exact final duration by padding/trimming if off by a tiny epsilon
    if montage.duration < target_duration - 0.02:
        pad = ColorClip(size=final_size, color=(0, 0, 0), duration=target_duration - montage.duration)
        montage = concatenate_videoclips([montage, pad.set_fps(fps)], method=CONCAT_METHOD)
    elif montage.duration > target_duration + 0.02:
        try:
            montage = montage.subclip(0, target_duration)
        except Exception:
            montage = montage.set_duration(target_duration)

    montage = montage.set_audio(audio.set_duration(target_duration))
    return montage


def main():
    logger.info("Starting Hangover-style montage creation...")
    if not MEDIA_DIR.exists():
        raise FileNotFoundError(f"Media directory not found: {MEDIA_DIR}")

    audio_path = find_audio_file()

    images, videos = list_media_files(MEDIA_DIR)
    logger.info(f"Found media: {len(images) + len(videos)} total | {len(images)} images | {len(videos)} videos")
    if (len(images) + len(videos)) != EXPECTED_MEDIA_COUNT:
        logger.warning(
            f"Expected ~{EXPECTED_MEDIA_COUNT} media files (per user note), found {len(images) + len(videos)}."
        )
    if len(videos) != EXPECTED_VIDEO_COUNT:
        logger.warning(
            f"Expected ~{EXPECTED_VIDEO_COUNT} videos (per user note), found {len(videos)}."
        )
    if any(p.suffix.lower() == ".heic" for p in images) and not HEIC_SUPPORTED:
        logger.warning(
            "HEIC images detected but pillow-heif not available. Install 'pillow-heif' to load HEIC: pip install pillow-heif"
        )

    clips = build_all_clips(images, videos, FINAL_SIZE)
    logger.info(f"Prepared {len(clips)} clip(s) after processing (including video chunks).")

    montage = assemble_montage(clips, audio_path, FPS, FINAL_SIZE)
    logger.info(f"Final montage length: {montage.duration:.2f}s | Writing to {OUTPUT_FILE}")

    # Favor stability over speed in memory saver mode
    write_kwargs = dict(
        fps=FPS,
        codec="libx264",
        preset="slow" if MEMORY_SAVER else "medium",
        bitrate="4000k" if MEMORY_SAVER else "6000k",
        audio_codec="aac",
        threads=1 if MEMORY_SAVER else (os.cpu_count() or 4),
        temp_audiofile=str(OUTPUT_FILE.with_suffix(".temp-audio.m4a")),
        remove_temp=True,
    )
    montage.write_videofile(str(OUTPUT_FILE), **write_kwargs)

    # Proactively close any readers to free resources
    try:
        for c in clips:
            if hasattr(c, "reader") and getattr(c, "reader") is not None:
                try:
                    c.reader.close()
                except Exception:
                    pass
            if hasattr(c, "audio") and getattr(c, "audio") is not None and hasattr(c.audio, "reader"):
                try:
                    c.audio.reader.close_proc()
                except Exception:
                    pass
    except Exception:
        pass

    logger.info("Done. Saved hangover_montage.mp4")


if __name__ == "__main__":
	main()