#!/usr/bin/env python3
"""Process media for portfolio site.
Usage:
python3 process.py # thumbs + hls + manifest
python3 process.py --force # regenerate everything from scratch
python3 process.py thumbs # just thumbnails
python3 process.py hls # just HLS segments
python3 process.py manifest # just manifest
Walks images/<collection>/ and creates thumbs/ subdirectories.
Walks videos/<collection>/ and creates thumbs/ poster frames + HLS segments.
Never modifies original files.
"""
import json
import os
import subprocess
import sys
from PIL import Image
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
VIDEO_EXTENSIONS = {".mp4", ".mov", ".webm", ".avi", ".mkv"}
IGNORE_DIRS = {"thumbs", "old", "_archive"}
DEFAULTS = {
"thumbnail": 800,
"quality": 85,
"video_thumbnail_time": 2,
}
def load_config(path="site.json"):
config = dict(DEFAULTS)
if os.path.exists(path):
with open(path) as f:
data = json.load(f)
proc = data.get("processing", {})
for key in config:
if key in proc:
config[key] = int(proc[key])
return config
def load_site_config(path="site.json"):
if os.path.exists(path):
with open(path) as f:
return json.load(f)
return {}
# ---------------------------------------------------------------------------
# Thumbnails
# ---------------------------------------------------------------------------
def make_thumb(src, thumb_path, max_edge, quality, force=False):
"""Create a thumbnail from src image. Never modifies the original."""
if not force:
src_mtime = os.path.getmtime(src)
if os.path.exists(thumb_path) and os.path.getmtime(thumb_path) >= src_mtime:
return False # already up to date
try:
img = Image.open(src)
img.load()
except Exception as e:
print(f" WARNING: Could not open {src}: {e}", file=sys.stderr)
return False
# Convert to RGB if needed
if img.mode == "RGB":
pass
elif img.mode in ("RGBA", "LA", "PA"):
bg = Image.new("RGB", img.size, (0, 0, 0))
if img.mode != "RGBA":
img = img.convert("RGBA")
bg.paste(img, mask=img.split()[3])
img = bg
elif img.mode == "P":
rgba = img.convert("RGBA")
bg = Image.new("RGB", rgba.size, (0, 0, 0))
bg.paste(rgba, mask=rgba.split()[3])
img = bg
else:
img = img.convert("RGB")
# Resize (never upscale)
w, h = img.size
longest = max(w, h)
if longest > max_edge:
ratio = max_edge / longest
img = img.resize((int(w * ratio), int(h * ratio)), Image.LANCZOS)
img.save(thumb_path, "JPEG", quality=quality, optimize=True)
return True
def make_video_thumb(src, thumb_path, thumbnail_time, force=False):
"""Extract a poster frame from a video. Never modifies the original."""
if not force:
src_mtime = os.path.getmtime(src)
if os.path.exists(thumb_path) and os.path.getmtime(thumb_path) >= src_mtime:
return False
for seek_args in [["-ss", str(thumbnail_time)], []]:
try:
subprocess.run(
["ffmpeg", "-y", "-i", src] + seek_args +
["-vframes", "1", "-q:v", "2", thumb_path],
capture_output=True, check=True
)
except subprocess.CalledProcessError:
pass
if os.path.exists(thumb_path) and os.path.getsize(thumb_path) > 0:
return True
print(f" WARNING: Could not extract thumbnail from {src}", file=sys.stderr)
return False
def get_video_resolution(path):
"""Get video height in pixels using ffprobe."""
try:
result = subprocess.run([
"ffprobe", "-v", "error",
"-select_streams", "v:0",
"-show_entries", "stream=height",
"-of", "default=noprint_wrappers=1:nokey=1",
path
], capture_output=True, text=True, check=True)
return int(result.stdout.strip())
except (subprocess.CalledProcessError, ValueError):
return 0
HLS_LEVELS = [
{"name": "480p", "height": 480, "bitrate": "1000k", "maxrate": "1200k", "bufsize": "2000k", "profile": "baseline"},
{"name": "720p", "height": 720, "bitrate": "2500k", "maxrate": "3000k", "bufsize": "5000k", "profile": "main"},
{"name": "1080p", "height": 1080, "bitrate": "5000k", "maxrate": "6000k", "bufsize": "10000k", "profile": "high"},
]
def generate_hls(src, hls_dir, force=False):
"""Generate HLS segments at multiple quality levels for a video.
Creates master.m3u8 + per-level playlists and .ts segments.
Only generates quality levels the source resolution supports (no upscaling).
Never modifies the original file.
"""
master_path = os.path.join(hls_dir, "master.m3u8")
if not force and os.path.exists(master_path):
src_mtime = os.path.getmtime(src)
if os.path.getmtime(master_path) >= src_mtime:
return False # already up to date
src_height = get_video_resolution(src)
if src_height == 0:
print(f" WARNING: Could not determine resolution of {src}", file=sys.stderr)
return False
# Pick quality levels that don't upscale
levels = [l for l in HLS_LEVELS if l["height"] <= src_height]
if not levels:
# Source is smaller than 480p — just do one level at native res
levels = [{"name": f"{src_height}p", "height": src_height, "bitrate": "1000k",
"maxrate": "1200k", "bufsize": "2000k", "profile": "baseline"}]
os.makedirs(hls_dir, exist_ok=True)
# Generate each quality level
for level in levels:
playlist = os.path.join(hls_dir, f"{level['name']}.m3u8")
segment_pattern = os.path.join(hls_dir, f"{level['name']}_%03d.ts")
try:
subprocess.run([
"ffmpeg", "-y", "-i", src,
"-vf", f"scale=-2:{level['height']}",
"-c:v", "libx264", "-profile:v", level["profile"],
"-b:v", level["bitrate"], "-maxrate", level["maxrate"],
"-bufsize", level["bufsize"],
"-c:a", "aac", "-b:a", "128k", "-ac", "2",
"-hls_time", "6",
"-hls_list_size", "0",
"-hls_segment_filename", segment_pattern,
"-f", "hls",
playlist
], capture_output=True, check=True)
except subprocess.CalledProcessError as e:
print(f" WARNING: HLS {level['name']} failed for {src}: {e.stderr.decode()[:200]}", file=sys.stderr)
continue
# Write master playlist
bandwidth_map = {"480p": 1200000, "720p": 3000000, "1080p": 6000000}
with open(master_path, "w") as f:
f.write("#EXTM3U\n")
for level in levels:
bw = bandwidth_map.get(level["name"], 1200000)
res_w = int(level["height"] * 16 / 9) # assume 16:9
# Make width even
if res_w % 2 != 0:
res_w += 1
f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={bw},RESOLUTION={res_w}x{level['height']}\n")
f.write(f"{level['name']}.m3u8\n")
return True
def generate_all_hls(base_dir=".", force=False):
"""Walk all video collections and generate HLS segments."""
videos_dir = os.path.join(base_dir, "videos")
if not os.path.isdir(videos_dir):
print("No videos/ directory found")
return 0, 0
total = 0
skipped = 0
for coll in sorted(os.listdir(videos_dir)):
coll_dir = os.path.join(videos_dir, coll)
if not os.path.isdir(coll_dir) or coll in IGNORE_DIRS:
continue
for fname in sorted(os.listdir(coll_dir)):
fpath = os.path.join(coll_dir, fname)
if not os.path.isfile(fpath):
continue
ext = os.path.splitext(fname)[1].lower()
if ext not in VIDEO_EXTENSIONS:
continue
basename = os.path.splitext(fname)[0]
hls_dir = os.path.join(coll_dir, basename)
if generate_hls(fpath, hls_dir, force):
total += 1
print(f" HLS: videos/{coll}/{fname}")
else:
skipped += 1
print(f"HLS: {total} generated, {skipped} up-to-date")
return total, skipped
def generate_thumbs(base_dir=".", config=None, force=False):
"""Walk all collections in images/ and videos/, generate thumbs/.
Never touches original files.
"""
if config is None:
config = load_config()
max_edge = config["thumbnail"]
quality = config["quality"]
video_time = config["video_thumbnail_time"]
total = 0
skipped = 0
# Image thumbnails
images_dir = os.path.join(base_dir, "images")
if os.path.isdir(images_dir):
for coll in sorted(os.listdir(images_dir)):
coll_dir = os.path.join(images_dir, coll)
if not os.path.isdir(coll_dir) or coll in IGNORE_DIRS:
continue
thumbs_dir = os.path.join(coll_dir, "thumbs")
os.makedirs(thumbs_dir, exist_ok=True)
for fname in sorted(os.listdir(coll_dir)):
fpath = os.path.join(coll_dir, fname)
if not os.path.isfile(fpath):
continue
ext = os.path.splitext(fname)[1].lower()
if ext not in IMAGE_EXTENSIONS:
continue
thumb_path = os.path.join(thumbs_dir, fname)
if make_thumb(fpath, thumb_path, max_edge, quality, force):
total += 1
print(f" Thumb: images/{coll}/{fname}")
else:
skipped += 1
# Video thumbnails
videos_dir = os.path.join(base_dir, "videos")
if os.path.isdir(videos_dir):
for coll in sorted(os.listdir(videos_dir)):
coll_dir = os.path.join(videos_dir, coll)
if not os.path.isdir(coll_dir) or coll in IGNORE_DIRS:
continue
thumbs_dir = os.path.join(coll_dir, "thumbs")
os.makedirs(thumbs_dir, exist_ok=True)
for fname in sorted(os.listdir(coll_dir)):
fpath = os.path.join(coll_dir, fname)
if not os.path.isfile(fpath):
continue
ext = os.path.splitext(fname)[1].lower()
if ext not in VIDEO_EXTENSIONS:
continue
basename = os.path.splitext(fname)[0]
thumb_path = os.path.join(thumbs_dir, basename + ".jpg")
if make_video_thumb(fpath, thumb_path, video_time, force):
total += 1
print(f" Thumb: videos/{coll}/{fname}")
else:
skipped += 1
print(f"Thumbnails: {total} generated, {skipped} up-to-date")
return total, skipped
# ---------------------------------------------------------------------------
# Manifest
# ---------------------------------------------------------------------------
def get_video_duration(path):
try:
result = subprocess.run([
"ffprobe", "-v", "error",
"-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1",
path
], capture_output=True, text=True, check=True)
return float(result.stdout.strip())
except (subprocess.CalledProcessError, ValueError):
return 0.0
def generate_manifest(base_dir=".", site_config=None):
if site_config is None:
site_config = load_site_config()
collections_config = site_config.get("collections", {})
order = collections_config.get("order", [])
ignore = set(collections_config.get("ignore", []))
ignore.add("thumbs")
collection_names = set()
images_dir = os.path.join(base_dir, "images")
videos_dir = os.path.join(base_dir, "videos")
if os.path.isdir(images_dir):
for name in os.listdir(images_dir):
if os.path.isdir(os.path.join(images_dir, name)) and name not in ignore:
collection_names.add(name)
if os.path.isdir(videos_dir):
for name in os.listdir(videos_dir):
if os.path.isdir(os.path.join(videos_dir, name)) and name not in ignore:
collection_names.add(name)
ordered = [c for c in order if c in collection_names]
remaining = sorted(collection_names - set(ordered))
sorted_collections = ordered + remaining
collections = []
items = []
for coll_name in sorted_collections:
coll_images_dir = os.path.join(images_dir, coll_name)
coll_videos_dir = os.path.join(videos_dir, coll_name)
# Images
if os.path.isdir(coll_images_dir):
md_files = set()
for f in os.listdir(coll_images_dir):
if f.endswith(".md"):
md_files.add(os.path.splitext(f)[0])
for fname in sorted(os.listdir(coll_images_dir)):
if fname == "thumbs" or not os.path.isfile(os.path.join(coll_images_dir, fname)):
continue
ext = os.path.splitext(fname)[1].lower()
if ext not in IMAGE_EXTENSIONS:
continue
basename = os.path.splitext(fname)[0]
items.append({
"type": "image",
"collection": coll_name,
"name": fname,
"full": f"images/{coll_name}/{fname}",
"thumb": f"images/{coll_name}/thumbs/{fname}",
"hasInfo": basename in md_files,
"mdUrl": f"images/{coll_name}/{basename}.md",
})
# Videos
if os.path.isdir(coll_videos_dir):
for fname in sorted(os.listdir(coll_videos_dir)):
if fname == "thumbs" or not os.path.isfile(os.path.join(coll_videos_dir, fname)):
continue
ext = os.path.splitext(fname)[1].lower()
if ext not in VIDEO_EXTENSIONS:
continue
basename = os.path.splitext(fname)[0]
duration = get_video_duration(os.path.join(coll_videos_dir, fname))
item = {
"type": "video",
"collection": coll_name,
"name": fname,
"full": f"videos/{coll_name}/{fname}",
"thumb": f"videos/{coll_name}/thumbs/{basename}.jpg",
"duration": round(duration, 2),
}
# Check for HLS master playlist
hls_master = os.path.join(coll_videos_dir, basename, "master.m3u8")
if os.path.exists(hls_master):
item["hls"] = f"videos/{coll_name}/{basename}/master.m3u8"
items.append(item)
collections.append(coll_name)
return {"collections": collections, "items": items}
def write_manifest(base_dir="."):
site_config = load_site_config()
manifest = generate_manifest(base_dir, site_config)
path = os.path.join(base_dir, "manifest.json")
with open(path, "w") as f:
json.dump(manifest, f, indent=2)
f.write("\n")
print(f"Manifest: {len(manifest['collections'])} collections, {len(manifest['items'])} items")
return manifest
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
args = sys.argv[1:]
force = "--force" in args
args = [a for a in args if a != "--force"]
cmd = args[0] if args else "all"
if cmd == "thumbs":
generate_thumbs(force=force)
elif cmd == "hls":
generate_all_hls(force=force)
elif cmd == "manifest":
write_manifest()
elif cmd == "all":
generate_thumbs(force=force)
generate_all_hls(force=force)
write_manifest()
elif cmd in ("-h", "--help", "help"):
print(__doc__)
else:
print(f"Unknown command: {cmd}")
print("Usage: python3 process.py [thumbs|hls|manifest|all] [--force]")
sys.exit(1)
if __name__ == "__main__":
main()