diff --git a/.claude/launch.json b/.claude/launch.json new file mode 100644 index 0000000..da94965 --- /dev/null +++ b/.claude/launch.json @@ -0,0 +1,11 @@ +{ + "version": "0.0.1", + "configurations": [ + { + "name": "pymontage", + "runtimeExecutable": "venv\\Scripts\\python.exe", + "runtimeArgs": ["app.py"], + "port": 5000 + } + ] +} diff --git a/app.py b/app.py index a3730a3..2f76339 100644 --- a/app.py +++ b/app.py @@ -6,6 +6,11 @@ import requests import json import tempfile +import glob +import re +import threading +from PIL import Image +from timeline_renderer import render_project from flask import Flask, render_template, request, send_file, after_this_request, jsonify from video_engine import generate_video_from_web, install_google_font # Import from video_engine.py @@ -14,6 +19,45 @@ app.config['OUTPUT_FOLDER'] = 'temp_outputs' os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) os.makedirs(app.config['OUTPUT_FOLDER'], exist_ok=True) +app.config['MAX_CONTENT_LENGTH'] = 2 * 1024 ** 3 # 2 GB upload cap + +UUID_RE = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$') +MEDIA_EXTS = { + 'video': {'.mp4', '.mov', '.webm', '.mkv'}, + 'audio': {'.mp3', '.wav', '.m4a', '.flac', '.ogg'}, + 'image': {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tif', '.tiff', + '.webp', '.heic', '.heif'}, +} + + +def media_type_for(ext): + for mtype, exts in MEDIA_EXTS.items(): + if ext in exts: + return mtype + return None + + +def resolve_media_path(session, media_id): + """Return the stored file for a media id, or None. Validates both parts.""" + if not UUID_RE.match(session or '') or not re.match(r'^[0-9a-f]{32}$', media_id or ''): + return None + folder = os.path.join(app.config['UPLOAD_FOLDER'], session) + matches = [p for p in glob.glob(os.path.join(folder, media_id + '.*')) + if not p.endswith('.thumb.jpg')] + return matches[0] if matches else None + + +def probe_media(path, mtype): + if mtype == 'image': + with Image.open(path) as im: + return {'duration': 5.0, 'width': im.width, 'height': im.height} + from moviepy.editor import AudioFileClip, VideoFileClip + if mtype == 'video': + with VideoFileClip(path) as clip: + return {'duration': round(clip.duration, 3), + 'width': clip.w, 'height': clip.h} + with AudioFileClip(path) as clip: + return {'duration': round(clip.duration, 3), 'width': 0, 'height': 0} def cleanup_temp_folders(): @@ -123,6 +167,172 @@ def download_font(): return jsonify({'success': True, 'path': path, 'name': name}) return jsonify({'success': False, 'message': 'Could not download font'}), 502 +@app.route('/editor') +def editor(): + return render_template('editor.html') + + +@app.route('/api/media', methods=['POST']) +def upload_media(): + session = request.form.get('session', '') + if not UUID_RE.match(session): + return jsonify({'error': 'Invalid session'}), 400 + file = request.files.get('file') + if not file or not file.filename: + return jsonify({'error': 'No file'}), 400 + ext = os.path.splitext(file.filename)[1].lower() + mtype = media_type_for(ext) + if not mtype: + return jsonify({'error': f'Unsupported file type: {ext}'}), 400 + + folder = os.path.join(app.config['UPLOAD_FOLDER'], session) + os.makedirs(folder, exist_ok=True) + media_id = uuid.uuid4().hex + path = os.path.join(folder, media_id + ext) + file.save(path) + try: + meta = probe_media(path, mtype) + except Exception as e: + os.remove(path) + return jsonify({'error': f'Could not read file: {e}'}), 400 + return jsonify({'id': media_id, 'name': file.filename, 'type': mtype, + 'url': f'/api/media/{session}/{media_id}', **meta}) + + +@app.route('/api/media//') +def get_media(session, media_id): + path = resolve_media_path(session, media_id) + if not path: + return jsonify({'error': 'Not found'}), 404 + return send_file(path, conditional=True) # conditional=True → Range support + + +@app.route('/api/media///thumb') +def get_media_thumb(session, media_id): + path = resolve_media_path(session, media_id) + if not path: + return jsonify({'error': 'Not found'}), 404 + mtype = media_type_for(os.path.splitext(path)[1].lower()) + if mtype == 'audio': + return jsonify({'error': 'No thumbnail for audio'}), 404 + thumb = os.path.join(os.path.dirname(path), media_id + '.thumb.jpg') + if not os.path.exists(thumb): + if mtype == 'image': + with Image.open(path) as im: + im = im.convert('RGB') + im.thumbnail((160, 160)) + im.save(thumb, 'JPEG') + else: + ffmpeg = shutil.which('ffmpeg') + if not ffmpeg: + import imageio_ffmpeg + ffmpeg = imageio_ffmpeg.get_ffmpeg_exe() + import subprocess + subprocess.run([ffmpeg, '-y', '-ss', '0.5', '-i', path, + '-frames:v', '1', '-vf', 'scale=160:-2', thumb], + capture_output=True, timeout=30) + if not os.path.exists(thumb): + return jsonify({'error': 'Thumbnail failed'}), 500 + return send_file(thumb, mimetype='image/jpeg') + + +EXPORT_JOBS = {} +# ponytail: global lock, one export at a time; job queue if concurrent users ever matter +EXPORT_LOCK = threading.Lock() + + +def _validate_export(data): + """Returns (session, project, media_paths) or raises ValueError.""" + session = data.get('session', '') + project = data.get('project') + if not UUID_RE.match(session) or not isinstance(project, dict): + raise ValueError('Invalid session or project') + for key in ('settings', 'media', 'tracks'): + if key not in project: + raise ValueError(f'Project missing "{key}"') + s = project['settings'] + if not (16 <= int(s.get('width', 0)) <= 7680 and + 16 <= int(s.get('height', 0)) <= 4320 and + 1 <= int(s.get('fps', 0)) <= 120): + raise ValueError('Invalid project settings') + used_ids = {c['mediaId'] for t in project['tracks'] for c in t['clips']} + media_paths = {} + for mid in used_ids: + path = resolve_media_path(session, mid) + if not path: + raise ValueError(f'Media {mid} not found on server') + media_paths[mid] = path + return session, project, media_paths + + +@app.route('/api/export', methods=['POST']) +def start_export(): + data = request.get_json(silent=True) or {} + try: + session, project, media_paths = _validate_export(data) + except (ValueError, KeyError, TypeError) as e: + return jsonify({'error': str(e)}), 400 + + if not EXPORT_LOCK.acquire(blocking=False): + return jsonify({'error': 'An export is already running'}), 409 + + job_id = uuid.uuid4().hex + output_path = os.path.join(app.config['OUTPUT_FOLDER'], f'export_{job_id}.mp4') + EXPORT_JOBS[job_id] = {'state': 'running', 'percent': 0, + 'error': None, 'path': output_path} + + def run(): + job = EXPORT_JOBS[job_id] + try: + render_project(project, media_paths, output_path, + progress_cb=lambda p: job.update(percent=int(p * 100))) + job.update(state='done', percent=100) + except Exception as e: + job.update(state='failed', error=str(e)) + if os.path.exists(output_path): + try: + os.remove(output_path) + except OSError: + pass + finally: + EXPORT_LOCK.release() + + threading.Thread(target=run, daemon=True).start() + return jsonify({'job': job_id}) + + +@app.route('/api/export//status') +def export_status(job_id): + job = EXPORT_JOBS.get(job_id) + if not job: + return jsonify({'error': 'Unknown job'}), 404 + return jsonify({'state': job['state'], 'percent': job['percent'], + 'error': job['error']}) + + +@app.route('/api/export//download') +def export_download(job_id): + job = EXPORT_JOBS.get(job_id) + if not job: + return jsonify({'error': 'Unknown job'}), 404 + if job['state'] != 'done': + return jsonify({'error': 'Not finished'}), 409 + path = job['path'] + if not os.path.exists(path): + return jsonify({'error': 'File gone'}), 404 + + @after_this_request + def cleanup(response): + try: + os.remove(path) + except OSError: + pass + EXPORT_JOBS.pop(job_id, None) + return response + + return send_file(path, as_attachment=True, download_name='export.mp4') + + @app.route('/preview', methods=['POST']) def preview(): """Generate a preview video""" diff --git a/docs/superpowers/plans/2026-07-05-editor-timeline.md b/docs/superpowers/plans/2026-07-05-editor-timeline.md new file mode 100644 index 0000000..786c426 --- /dev/null +++ b/docs/superpowers/plans/2026-07-05-editor-timeline.md @@ -0,0 +1,1869 @@ +# PyMontage Editor Phase 1 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add a Premiere-style editor at `/editor` — media pool, multi-track timeline (V2/V1/A1/A2), client-side preview monitor, server-side MoviePy export — driven by one project-JSON model. + +**Architecture:** Browser owns the project JSON (source of truth, undo/redo, localStorage autosave). Flask gains a small API: media upload/streaming and an export job runner. New `timeline_renderer.py` turns project JSON into an MP4 via MoviePy. Frontend is vanilla ES modules + Tailwind CDN, no build step. Spec: `docs/superpowers/specs/2026-07-05-editor-timeline-design.md`. + +**Tech Stack:** Flask 3, MoviePy 1.0.3, Pillow, vanilla JS (ES modules), Tailwind via CDN, Web Audio API for waveforms. + +## Global Constraints + +- No new Python or JS dependencies. No node toolchain. +- Old slideshow UI at `/` stays untouched; editor lives at `/editor`. +- Theme: zinc-900/950 surfaces, accent `#7c3aed`. +- Tracks fixed: V2, V1, A1, A2. First video track in array = topmost at composite. +- One export at a time (global lock → 409 for a second request). +- Upload cap 2 GB; accepted types per spec §3. +- Python tests run with `venv\Scripts\python -m pytest` from repo root (Windows). +- Commit after every task; messages in imperative mood. + +--- + +### Task 1: Timeline renderer (project JSON → MP4) + +**Files:** +- Create: `timeline_renderer.py` +- Test: `test_timeline_renderer.py` + +**Interfaces:** +- Consumes: `video_engine.get_best_video_codec() -> str` +- Produces: `render_project(project: dict, media_paths: dict[str, str], output_path: str, progress_cb: callable | None = None) -> None` (raises `ValueError` on empty timeline). `media_paths` maps media id → absolute file path. Later tasks (export API) call exactly this. + +- [ ] **Step 1: Write the failing test** + +```python +# test_timeline_renderer.py +import os +import pytest +from moviepy.editor import ColorClip, VideoFileClip + +from timeline_renderer import render_project + + +@pytest.fixture +def two_clip_project(tmp_path): + """Two generated color videos placed back-to-back on V1.""" + paths = {} + for mid, color in (("m1", (255, 0, 0)), ("m2", (0, 0, 255))): + p = str(tmp_path / f"{mid}.mp4") + ColorClip((64, 64), color=color, duration=2).write_videofile( + p, fps=10, codec="libx264", audio=False, logger=None) + paths[mid] = p + project = { + "settings": {"width": 128, "height": 72, "fps": 10}, + "media": [ + {"id": "m1", "name": "m1.mp4", "type": "video", "duration": 2.0}, + {"id": "m2", "name": "m2.mp4", "type": "video", "duration": 2.0}, + ], + "tracks": [ + {"id": "V2", "kind": "video", "clips": []}, + {"id": "V1", "kind": "video", "clips": [ + {"id": "c1", "mediaId": "m1", "start": 0.0, "in": 0.0, "out": 1.5, + "effects": {"speed": 1.0, "opacity": 1.0, "filter": None}}, + {"id": "c2", "mediaId": "m2", "start": 1.5, "in": 0.5, "out": 2.0, + "effects": {"speed": 1.0, "opacity": 1.0, "filter": None}}, + ]}, + {"id": "A1", "kind": "audio", "clips": []}, + {"id": "A2", "kind": "audio", "clips": []}, + ], + } + return project, paths + + +def test_render_two_clips(two_clip_project, tmp_path): + project, paths = two_clip_project + out = str(tmp_path / "out.mp4") + percents = [] + render_project(project, paths, out, progress_cb=percents.append) + assert os.path.exists(out) + with VideoFileClip(out) as clip: + assert clip.duration == pytest.approx(2.0, abs=0.2) # 1.5 + 0.5 + assert clip.size == [128, 72] + assert percents and percents[-1] > 0.5 + + +def test_empty_timeline_raises(two_clip_project, tmp_path): + project, paths = two_clip_project + for t in project["tracks"]: + t["clips"] = [] + with pytest.raises(ValueError): + render_project(project, paths, str(tmp_path / "x.mp4")) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `venv\Scripts\python -m pytest test_timeline_renderer.py -v` +Expected: FAIL / ERROR with `ModuleNotFoundError: No module named 'timeline_renderer'` + +- [ ] **Step 3: Write the implementation** + +```python +# timeline_renderer.py +"""Render a project JSON (see spec §1) to an MP4 with MoviePy.""" +from moviepy.editor import (AudioFileClip, ColorClip, CompositeAudioClip, + CompositeVideoClip, ImageClip, VideoFileClip, vfx) +from proglog import ProgressBarLogger + +from video_engine import get_best_video_codec + + +class _JobLogger(ProgressBarLogger): + """Forwards MoviePy's frame-writing progress to a 0..1 callback.""" + + def __init__(self, cb): + super().__init__() + self._cb = cb + + def bars_callback(self, bar, attr, value, old_value=None): + total = self.bars[bar].get("total") + if attr == "index" and total: + self._cb(value / total) + + +def _build_clip(spec, media, path, size): + effects = spec.get("effects") or {} + speed = effects.get("speed") or 1.0 + opacity = effects.get("opacity", 1.0) + if media["type"] == "audio": + clip = AudioFileClip(path).subclip(spec["in"], spec["out"]) + return clip.set_start(spec["start"]) + if media["type"] == "image": + clip = ImageClip(path).set_duration((spec["out"] - spec["in"]) / speed) + else: + clip = VideoFileClip(path).subclip(spec["in"], spec["out"]) + if speed != 1.0: + clip = clip.fx(vfx.speedx, speed) + scale = min(size[0] / clip.w, size[1] / clip.h) # letterbox: fit inside frame + clip = clip.resize(scale).set_position("center") + if opacity is not None and opacity < 1.0: + clip = clip.set_opacity(opacity) + return clip.set_start(spec["start"]) + + +def render_project(project, media_paths, output_path, progress_cb=None): + settings = project["settings"] + size = (settings["width"], settings["height"]) + media_by_id = {m["id"]: m for m in project["media"]} + + video_clips, audio_clips = [], [] + video_tracks = [t for t in project["tracks"] if t["kind"] == "video"] + for track in reversed(video_tracks): # composite order: later = on top, first track topmost + for spec in track["clips"]: + m = media_by_id[spec["mediaId"]] + video_clips.append(_build_clip(spec, m, media_paths[m["id"]], size)) + for track in project["tracks"]: + if track["kind"] != "audio": + continue + for spec in track["clips"]: + m = media_by_id[spec["mediaId"]] + audio_clips.append(_build_clip(spec, m, media_paths[m["id"]], size)) + + if not video_clips and not audio_clips: + raise ValueError("Timeline is empty") + + end = max(c.end for c in video_clips + audio_clips) + base = ColorClip(size, color=(0, 0, 0), duration=end) + video = CompositeVideoClip([base] + video_clips, size=size).set_duration(end) + audio_parts = list(audio_clips) + if video.audio is not None: + audio_parts.insert(0, video.audio) + if audio_parts: + video = video.set_audio(CompositeAudioClip(audio_parts).set_duration(end)) + + video.write_videofile( + output_path, + fps=settings["fps"], + codec=get_best_video_codec(), + audio_codec="aac", + threads=4, + logger=_JobLogger(progress_cb) if progress_cb else None, + ) + video.close() +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `venv\Scripts\python -m pytest test_timeline_renderer.py -v` +Expected: 2 passed (takes ~30–60 s; MoviePy renders twice) + +- [ ] **Step 5: Commit** + +```bash +git add timeline_renderer.py test_timeline_renderer.py +git commit -m "feat: add timeline renderer (project JSON to MP4)" +``` + +--- + +### Task 2: Media API (upload, probe, stream, thumbnail) + +**Files:** +- Modify: `app.py` (add routes after the existing `/fonts/download` route, ~line 125) +- Test: `test_media_api.py` + +**Interfaces:** +- Produces: + - `POST /api/media` — multipart `file` + form `session` (UUID string). Returns `{id, name, type, duration, width, height, url}` (width/height 0 for audio). 400 on bad session/type. + - `GET /api/media//` — streams file with Range support. + - `GET /api/media///thumb` — JPEG thumbnail (404 for audio). + - Helper `resolve_media_path(session, media_id) -> str | None` used by Task 3. + +- [ ] **Step 1: Write the failing test** + +```python +# test_media_api.py +import io +import uuid + +import pytest +from PIL import Image + +from app import app + + +@pytest.fixture +def client(): + app.config["TESTING"] = True + return app.test_client() + + +def _png_bytes(): + buf = io.BytesIO() + Image.new("RGB", (320, 200), (255, 0, 0)).save(buf, format="PNG") + buf.seek(0) + return buf + + +def test_upload_and_stream_image(client): + session = str(uuid.uuid4()) + resp = client.post("/api/media", data={ + "session": session, + "file": (_png_bytes(), "red.png"), + }) + assert resp.status_code == 200 + meta = resp.get_json() + assert meta["type"] == "image" + assert meta["width"] == 320 and meta["height"] == 200 + assert meta["duration"] == 5.0 + + got = client.get(meta["url"]) + assert got.status_code == 200 + + partial = client.get(meta["url"], headers={"Range": "bytes=0-9"}) + assert partial.status_code == 206 # Range support required for