Spaces:
Sleeping
Sleeping
| """Scene clips (Ken Burns) -> xfade chain -> subtitles + narration + ducked music + SFX -> final MP4.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import random | |
| import threading | |
| import time | |
| import traceback | |
| from pathlib import Path | |
| from . import config | |
| from .storage import ServiceError, job_dir, media_duration, read_json, run, scene_file, write_json | |
| AUDIO_EXT = (".mp3", ".wav", ".ogg", ".m4a", ".flac") | |
| _render_slots = threading.Semaphore(config.RENDER_CONCURRENCY) | |
| MOTIONS = ["zoom_in", "pan_right", "zoom_out", "pan_up", "pan_left", "pan_down"] | |
| def motion_for(text: str, idx: int) -> str: | |
| t = (text or "").lower().replace("-", " ") | |
| table = [ | |
| ("zoom_out", ("zoom out", "pull back", "pull out", "dolly out", "reveal")), | |
| ("zoom_in", ("zoom in", "push in", "dolly in", "close up", "closeup")), | |
| ("pan_left", ("pan left", "left")), | |
| ("pan_right", ("pan right", "right", "track")), | |
| ("pan_up", ("tilt up", "pan up", "rise", "upward", "crane up")), | |
| ("pan_down", ("tilt down", "pan down", "descend", "downward")), | |
| ("static", ("static", "still", "locked")), | |
| ] | |
| for motion, keys in table: | |
| if any(k in t for k in keys): | |
| return motion | |
| return MOTIONS[idx % len(MOTIONS)] | |
| def _zoompan(motion: str, frames: int) -> str: | |
| z = config.KEN_BURNS_ZOOM | |
| n = max(1, frames - 1) | |
| cx, cy = "iw/2-(iw/zoom/2)", "ih/2-(ih/zoom/2)" | |
| if motion == "zoom_out": | |
| zz, x, y = f"{1 + z}-{z}*on/{n}", cx, cy | |
| elif motion == "pan_left": | |
| zz, x, y = f"{1 + z}", f"(iw-iw/zoom)*(1-on/{n})", cy | |
| elif motion == "pan_right": | |
| zz, x, y = f"{1 + z}", f"(iw-iw/zoom)*on/{n}", cy | |
| elif motion == "pan_up": | |
| zz, x, y = f"{1 + z}", cx, f"(ih-ih/zoom)*(1-on/{n})" | |
| elif motion == "pan_down": | |
| zz, x, y = f"{1 + z}", cx, f"(ih-ih/zoom)*on/{n}" | |
| elif motion == "static": | |
| zz, x, y = f"{1 + z * 0.25}+{z * 0.1}*on/{n}", cx, cy # near-static, avoids a frozen look | |
| else: # zoom_in | |
| zz, x, y = f"1+{z}*on/{n}", cx, cy | |
| return f"zoompan=z='{zz}':x='{x}':y='{y}':d={frames}:s={config.WIDTH}x{config.HEIGHT}:fps={config.FPS}" | |
| def render_scene_clip(image: Path, out: Path, seconds: float, motion: str) -> None: | |
| frames = max(2, int(round(seconds * config.FPS))) | |
| # upscale 2x before zoompan: removes the integer-pixel "jitter" zoompan has at native size | |
| w2, h2 = config.WIDTH * 2, config.HEIGHT * 2 | |
| vf = (f"scale={w2}:{h2}:force_original_aspect_ratio=increase:flags=lanczos,crop={w2}:{h2}," | |
| f"{_zoompan(motion, frames)},setsar=1,format=yuv420p") | |
| run(["ffmpeg", "-y", "-v", "error", "-i", str(image), "-vf", vf, "-frames:v", str(frames), | |
| "-r", str(config.FPS), "-c:v", "libx264", "-preset", config.X264_PRESET, "-crf", str(config.CRF), | |
| "-pix_fmt", "yuv420p", str(out)], timeout=600) | |
| def _pick(folder: Path, seed: str) -> Path | None: | |
| if not folder.is_dir(): | |
| return None | |
| files = sorted(p for p in folder.rglob("*") if p.suffix.lower() in AUDIO_EXT) | |
| if not files: | |
| return None | |
| return random.Random(seed).choice(files) | |
| def choose_music(job_id: str, mood: str | None, genre: str | None, explicit: str | None) -> Path | None: | |
| if explicit: | |
| p = Path(explicit) | |
| if not p.is_absolute(): | |
| p = config.AUDIO_LIBRARY / "music" / explicit | |
| return p if p.exists() else None | |
| mapping = json.loads((config.AUDIO_LIBRARY / "mood_map.json").read_text()) \ | |
| if (config.AUDIO_LIBRARY / "mood_map.json").exists() else {} | |
| candidates = [] | |
| for key in (mood, genre): | |
| if key: | |
| k = key.lower().strip() | |
| candidates += [mapping.get(k, k), k] | |
| candidates.append("default") | |
| for folder in candidates: | |
| found = _pick(config.AUDIO_LIBRARY / "music" / folder, job_id) | |
| if found: | |
| return found | |
| return None | |
| def choose_sfx(name: str | None, seed: str) -> Path | None: | |
| if not name or name.lower() in ("none", "null", ""): | |
| return None | |
| root = config.AUDIO_LIBRARY / "sfx" | |
| if not root.is_dir(): | |
| return None | |
| key = name.lower().replace(" ", "_") | |
| files = sorted(p for p in root.rglob("*") if p.suffix.lower() in AUDIO_EXT) | |
| matches = [p for p in files if key in p.stem.lower() or p.parent.name.lower() == key] | |
| if not matches: | |
| for token in key.split("_"): | |
| if len(token) > 3: | |
| matches = [p for p in files if token in p.stem.lower() or token == p.parent.name.lower()] | |
| if matches: | |
| break | |
| return random.Random(seed + key).choice(matches) if matches else None | |
| def _status(job_id: str, **data) -> None: | |
| path = job_dir(job_id) / "render_status.json" | |
| cur = json.loads(path.read_text()) if path.exists() else {} | |
| cur.update(data, updated_at=time.time()) | |
| path.write_text(json.dumps(cur, indent=2)) | |
| def get_status(job_id: str) -> dict: | |
| base = job_dir(job_id) | |
| path = base / "render_status.json" | |
| status = json.loads(path.read_text()) if path.exists() else {"status": "not_started"} | |
| status["job_id"] = job_id | |
| status["files"] = { | |
| "story": (base / "story.json").exists(), | |
| "scenes": (base / "scenes.json").exists(), | |
| "images": len(list((base / "images").glob("scene_*.png"))), | |
| "audio": len(list((base / "audio").glob("scene_*.wav"))), | |
| "narration": (base / "narration.wav").exists(), | |
| "subtitles": (base / "subtitles" / "subtitles.ass").exists(), | |
| "video": (base / "final" / "video.mp4").exists(), | |
| "thumbnail": (base / "final" / "thumbnail.jpg").exists(), | |
| } | |
| return status | |
| def render(job_id: str, opts: dict) -> dict: | |
| base = job_dir(job_id) | |
| scenes = read_json(job_id, "scenes")["scenes"] | |
| timing = read_json(job_id, "timing") | |
| by_num = {t["scene_number"]: t for t in timing["scenes"]} | |
| n = len(scenes) | |
| if n == 0 or n > config.MAX_SCENES: | |
| raise ServiceError(f"scene count {n} outside 1..{config.MAX_SCENES}", stage="scenes") | |
| tr = float(opts.get("transition_seconds", config.TRANSITION_SECONDS)) | |
| transition = opts.get("transition", config.TRANSITION) | |
| # 1) per-scene Ken Burns clips; each non-final clip is extended by `tr` so xfade overlaps keep sync | |
| clips, durations = [], [] | |
| for i, sc in enumerate(scenes): | |
| num = sc["scene_number"] | |
| img = scene_file(job_id, "image", num) | |
| if not img.exists(): | |
| raise ServiceError(f"image for scene {num} missing", 409, stage="visuals") | |
| d = by_num[num]["duration"] | |
| length = d + (tr if i < n - 1 else 0) | |
| out = scene_file(job_id, "clip", num) | |
| _status(job_id, status="rendering", step=f"scene {num}/{n}") | |
| render_scene_clip(img, out, length, motion_for(sc.get("camera_motion", ""), i)) | |
| clips.append(out) | |
| durations.append(d) | |
| # 2) video graph: xfade chain + subtitles | |
| inputs, vf = [], [] | |
| for c in clips: | |
| inputs += ["-i", str(c)] | |
| if n == 1: | |
| vf.append("[0:v]null[vx]") | |
| else: | |
| prev, offset = "[0:v]", 0.0 | |
| for i in range(1, n): | |
| offset += durations[i - 1] | |
| label = "[vx]" if i == n - 1 else f"[x{i}]" | |
| vf.append(f"{prev}[{i}:v]xfade=transition={transition}:duration={tr}:offset={offset:.3f}{label}") | |
| prev = label | |
| ass = base / "subtitles" / "subtitles.ass" | |
| if ass.exists() and opts.get("burn_subtitles", True): | |
| ass_escaped = str(ass).replace("\\", "/").replace(":", "\\:") | |
| vf.append(f"[vx]ass='{ass_escaped}':fontsdir='{config.FONTS_DIR}',format=yuv420p[vout]") | |
| else: | |
| vf.append("[vx]format=yuv420p[vout]") | |
| # 3) audio graph: narration (+ ducked music) (+ SFX at scene starts) -> loudnorm | |
| total = sum(durations) | |
| idx = n | |
| inputs += ["-i", str(base / "narration.wav")] | |
| nar = idx | |
| idx += 1 | |
| af = [f"[{nar}:a]aresample=48000,apad,atrim=0:{total:.3f},asplit=2[nar][key]"] | |
| mix = ["[nar]"] | |
| music = choose_music(job_id, opts.get("mood"), opts.get("genre"), opts.get("music_file")) | |
| if music: | |
| inputs += ["-stream_loop", "-1", "-i", str(music)] | |
| m = idx | |
| idx += 1 | |
| vol = float(opts.get("music_volume", config.MUSIC_VOLUME)) | |
| af.append(f"[{m}:a]aresample=48000,aformat=channel_layouts=mono,atrim=0:{total:.3f},volume={vol}," | |
| f"afade=t=in:d=0.8,afade=t=out:st={max(0, total - 1.5):.3f}:d=1.5[mus]") | |
| af.append("[mus][key]sidechaincompress=threshold=0.03:ratio=8:attack=15:release=350[duck]") | |
| mix.append("[duck]") | |
| else: | |
| af.append("[key]anullsink") | |
| sfx_used = [] | |
| for sc in scenes: | |
| sfx = choose_sfx(sc.get("sfx"), job_id) | |
| if not sfx: | |
| continue | |
| inputs += ["-i", str(sfx)] | |
| s = idx | |
| idx += 1 | |
| start_ms = int(by_num[sc["scene_number"]]["start"] * 1000) | |
| af.append(f"[{s}:a]aresample=48000,aformat=channel_layouts=mono,atrim=0:4,afade=t=out:st=3:d=1," | |
| f"volume={config.SFX_VOLUME},adelay={start_ms}|{start_ms}[s{s}]") | |
| mix.append(f"[s{s}]") | |
| sfx_used.append({"scene_number": sc["scene_number"], "file": sfx.name}) | |
| if len(mix) > 1: | |
| af.append(f"{''.join(mix)}amix=inputs={len(mix)}:duration=first:normalize=0[mixed]") | |
| else: | |
| af.append("[nar]anull[mixed]") | |
| af.append(f"[mixed]loudnorm=I={config.TARGET_LUFS}:TP=-1.5:LRA=11,aresample=48000[aout]") | |
| final = base / "final" / "video.mp4" | |
| tmp_final = base / "final" / "video.tmp.mp4" | |
| _status(job_id, status="rendering", step="final encode") | |
| run(["ffmpeg", "-y", "-v", "error", *inputs, | |
| "-filter_complex", ";".join(vf + af), "-map", "[vout]", "-map", "[aout]", | |
| "-t", f"{total:.3f}", "-r", str(config.FPS), | |
| "-c:v", "libx264", "-profile:v", "high", "-preset", config.X264_PRESET, "-crf", str(config.CRF), | |
| "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2", | |
| "-movflags", "+faststart", str(tmp_final)], timeout=1800) | |
| tmp_final.replace(final) | |
| manifest = {"video_path": str(final), "duration": round(media_duration(final), 3), "scenes": n, | |
| "music": music.name if music else None, "sfx": sfx_used, "transition": transition, | |
| "size_bytes": final.stat().st_size, | |
| "sha256": hashlib.sha256(final.read_bytes()).hexdigest()} | |
| write_json(job_id, "render_manifest", manifest) | |
| return manifest | |
| def start_render(job_id: str, opts: dict) -> dict: | |
| job_dir(job_id) | |
| cur = get_status(job_id) | |
| if cur.get("status") == "rendering" and time.time() - cur.get("updated_at", 0) < 900: | |
| return {"job_id": job_id, "status": "rendering", "message": "render already in progress"} | |
| _status(job_id, status="queued", error=None, started_at=time.time()) | |
| def worker(): | |
| with _render_slots: | |
| try: | |
| _status(job_id, status="rendering", step="starting") | |
| manifest = render(job_id, opts) | |
| _status(job_id, status="completed", step="done", **manifest) | |
| except ServiceError as exc: | |
| _status(job_id, status="failed", error=exc.message, failed_stage=exc.stage or "render") | |
| except Exception as exc: # noqa: BLE001 | |
| _status(job_id, status="failed", error=f"{exc}\n{traceback.format_exc()[-1500:]}", | |
| failed_stage="render") | |
| threading.Thread(target=worker, daemon=True).start() | |
| return {"job_id": job_id, "status": "queued"} | |