Spaces:
Sleeping
Sleeping
| """ | |
| FastAPI Server for Scene Selection and Video Composition | |
| Dynamically generates video from manifest without hardcoded labels | |
| """ | |
| from fastapi import FastAPI, HTTPException, File, UploadFile, Form | |
| from fastapi.responses import FileResponse | |
| from pydantic import BaseModel | |
| from typing import List, Dict, Any, Optional | |
| import os | |
| import json | |
| import shutil | |
| from pathlib import Path | |
| import subprocess | |
| import sys | |
| # ───────────────────────────────────────────────────────────────────────── | |
| # Pydantic Models | |
| # ───────────────────────────────────────────────────────────────────────── | |
| class SceneRequest(BaseModel): | |
| """Scene metadata from manifest""" | |
| label: str | |
| image_query: str | |
| class ManifestRequest(BaseModel): | |
| """Manifest JSON format from client""" | |
| title: str | |
| scenes: List[SceneRequest] | |
| class VideoResponse(BaseModel): | |
| """Response after video generation""" | |
| status: str | |
| message: str | |
| output_path: Optional[str] = None | |
| size_mb: Optional[float] = None | |
| duration_s: Optional[float] = None | |
| # ───────────────────────────────────────────────────────────────────────── | |
| # FastAPI App | |
| # ───────────────────────────────────────────────────────────────────────── | |
| app = FastAPI( | |
| title="TrendClip Video Composer", | |
| description="Dynamic video composition from manifest without hardcoded configs", | |
| version="2.0" | |
| ) | |
| BASE_DIR = Path(__file__).parent | |
| CANDIDATES_DIR = BASE_DIR / "candidates" | |
| SELECTED_DIR = BASE_DIR / "selected" | |
| RENDERS_DIR = BASE_DIR / "renders" | |
| # Ensure directories exist | |
| CANDIDATES_DIR.mkdir(exist_ok=True) | |
| SELECTED_DIR.mkdir(exist_ok=True) | |
| RENDERS_DIR.mkdir(exist_ok=True) | |
| # ───────────────────────────────────────────────────────────────────────── | |
| # Scene Configuration Generation | |
| # ───────────────────────────────────────────────────────────────────────── | |
| # Intro scene config (4.7s, 95pt font) | |
| INTRO_CONFIG = { | |
| "duration_s": 4.7, | |
| "motion": {"type": "slow_push_in", "scale_start": 1.0, "scale_end": 1.08}, | |
| "text": {"type": "center_stroke_pop", "entry_frame": 2, "hold_frames": 125, "font_size": 95, "align": "center"}, | |
| "grade": {"crush_blacks": 15, "contrast": 1.15}, | |
| "transition": {"type": "hard_cut", "frames": 1}, | |
| } | |
| # Scene templates for subsequent scenes (2.3s, 110pt font) | |
| SCENES_TEMPLATES = [ | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "snap_zoom", "scale_start": 1.0, "scale_end": 1.12}, | |
| "text": {"type": "center_pop", "entry_frame": 0, "hold_frames": 69, "font_size": 110, "align": "center"}, | |
| "grade": {"warm_tint": True, "lift_mids": 10}, | |
| "transition": {"type": "whip_pan_right", "frames": 4}, | |
| }, | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "static"}, | |
| "text": {"type": "center_fade_pop", "entry_frame": 2, "hold_frames": 66, "font_size": 110, "align": "center"}, | |
| "grade": {"desaturate": True, "lift_blacks": 5}, | |
| "transition": {"type": "whip_pan_right", "frames": 4}, | |
| }, | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "static"}, | |
| "text": {"type": "center_fade_pop", "entry_frame": 2, "hold_frames": 66, "font_size": 110, "align": "center"}, | |
| "grade": {"cool_tint": True, "highlights": -15}, | |
| "transition": {"type": "whip_pan_right", "frames": 4}, | |
| }, | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "static"}, | |
| "text": {"type": "center_fade_pop", "entry_frame": 2, "hold_frames": 66, "font_size": 110, "align": "center"}, | |
| "grade": {"soft_pink": True, "lift_mids": 15}, | |
| "transition": {"type": "whip_pan_right", "frames": 4}, | |
| }, | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "static"}, | |
| "text": {"type": "center_fade_pop", "entry_frame": 2, "hold_frames": 66, "font_size": 110, "align": "center"}, | |
| "grade": {"indoor_warm": True, "lift_shadows": 8}, | |
| "transition": {"type": "whip_pan_right", "frames": 4}, | |
| }, | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "static"}, | |
| "text": {"type": "center_fade_pop", "entry_frame": 2, "hold_frames": 66, "font_size": 110, "align": "center"}, | |
| "grade": {"teal_orange": True, "crush_blacks": 10}, | |
| "transition": {"type": "whip_pan_right", "frames": 4}, | |
| }, | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "static"}, | |
| "text": {"type": "center_fade_pop", "entry_frame": 2, "hold_frames": 66, "font_size": 110, "align": "center"}, | |
| "grade": {"dark_moody": True, "crush_blacks": 20, "desaturate": 15}, | |
| "transition": {"type": "whip_pan_right", "frames": 4}, | |
| }, | |
| { | |
| "duration_s": 2.3, | |
| "motion": {"type": "static"}, | |
| "text": {"type": "center_fade_pop", "entry_frame": 2, "hold_frames": 66, "font_size": 110, "align": "center"}, | |
| "grade": {"warm_indoor": True, "soft_glow": True, "lift_mids": 12}, | |
| "transition": {"type": "end_fade_black", "frames": 30}, | |
| }, | |
| ] | |
| def generate_scene_config(manifest: ManifestRequest) -> list: | |
| """Generate SCENE_CONFIG from manifest, extracting and uppercasing labels.""" | |
| config = [] | |
| for idx, scene in enumerate(manifest.scenes): | |
| # Extract label and convert to UPPERCASE for captions | |
| label = scene.label.upper() | |
| scene_cfg = { | |
| "idx": idx, | |
| "label": label, | |
| } | |
| if idx == 0: | |
| # First scene is intro | |
| scene_cfg.update(INTRO_CONFIG) | |
| else: | |
| # Use templated config for subsequent scenes | |
| template_idx = min(idx - 1, len(SCENES_TEMPLATES) - 1) | |
| scene_cfg.update(SCENES_TEMPLATES[template_idx]) | |
| config.append(scene_cfg) | |
| return config | |
| # ───────────────────────────────────────────────────────────────────────── | |
| # Endpoints | |
| # ───────────────────────────────────────────────────────────────────────── | |
| # ───────────────────────────────────────────────────────────────────────── | |
| # Endpoints | |
| # ───────────────────────────────────────────────────────────────────────── | |
| async def generate_video( | |
| manifest: str = Form(...), | |
| files: List[UploadFile] = File(...) | |
| ): | |
| """ | |
| Full pipeline: Upload candidates + manifest, get MP4 back. | |
| Form parameters: | |
| - manifest: JSON string with title and scenes | |
| - files: Uploaded image files for each scene | |
| Returns: MP4 video file | |
| """ | |
| try: | |
| # Parse manifest | |
| manifest_data = json.loads(manifest) | |
| manifest_req = ManifestRequest(**manifest_data) | |
| # Step 1: Clean and prepare upload directory | |
| upload_dir = BASE_DIR / "upload_candidates" | |
| if upload_dir.exists(): | |
| shutil.rmtree(upload_dir) | |
| upload_dir.mkdir(exist_ok=True) | |
| # Step 2: Save uploaded files organized by scene index | |
| uploaded_count = 0 | |
| for file in files: | |
| if file.filename: | |
| # Save to upload_candidates folder | |
| file_path = upload_dir / file.filename | |
| with open(file_path, "wb") as f: | |
| f.write(await file.read()) | |
| uploaded_count += 1 | |
| if uploaded_count != len(manifest_req.scenes): | |
| raise Exception( | |
| f"Expected {len(manifest_req.scenes)} files, received {uploaded_count}" | |
| ) | |
| # Step 3: Select scenes (evaluate and copy to selected/) | |
| select_result = await _select_scenes(manifest_req, upload_dir) | |
| if select_result["status"] != "success": | |
| raise Exception(select_result.get("message", "Scene selection failed")) | |
| # Step 4: Compose video with manifest data | |
| compose_result = await _compose(manifest_req) | |
| if compose_result["status"] != "success": | |
| raise Exception(compose_result.get("message", "Composition failed")) | |
| # Step 5: Return the MP4 file | |
| output_file = Path(compose_result["output_path"]) | |
| if not output_file.exists(): | |
| raise Exception("Output file not found") | |
| return FileResponse( | |
| path=output_file, | |
| media_type="video/mp4", | |
| filename="video.mp4" | |
| ) | |
| except json.JSONDecodeError as e: | |
| raise HTTPException(status_code=400, detail=f"Invalid manifest JSON: {str(e)}") | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def select_scenes(manifest: ManifestRequest): | |
| """ | |
| Endpoint for just scene selection (without upload). | |
| Assumes candidates are already in candidates/ folder. | |
| """ | |
| try: | |
| result = await _select_scenes(manifest, CANDIDATES_DIR) | |
| return result | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def compose(manifest: ManifestRequest): | |
| """ | |
| Endpoint for just composition. | |
| Assumes selected images are already in selected/ folder. | |
| """ | |
| try: | |
| result = await _compose(manifest) | |
| return result | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def _select_scenes(manifest: ManifestRequest, source_dir: Path): | |
| """ | |
| Internal: Select scenes from source directory. | |
| Copies best image from each scene folder to selected/ folder. | |
| """ | |
| try: | |
| # Clean and recreate selected directory | |
| if SELECTED_DIR.exists(): | |
| shutil.rmtree(SELECTED_DIR) | |
| SELECTED_DIR.mkdir(exist_ok=True) | |
| selected_count = 0 | |
| # For each scene, find and copy its image | |
| for i, scene in enumerate(manifest.scenes): | |
| # Try multiple naming conventions | |
| scene_folder = None | |
| for pattern in [f"scene_{i}", f"scene_{i:02d}", f"{i}", f"scene{i}"]: | |
| potential = source_dir / pattern | |
| if potential.exists(): | |
| scene_folder = potential | |
| break | |
| # If no folder, look for files named with scene index | |
| if scene_folder is None: | |
| images = list(source_dir.glob(f"*scene*{i}*")) + list( | |
| source_dir.glob(f"{i:02d}*") | |
| ) | |
| if images: | |
| dest = SELECTED_DIR / f"scene_{i:02d}.jpg" | |
| shutil.copy2(images[0], dest) | |
| selected_count += 1 | |
| continue | |
| # If folder found, get best image | |
| if scene_folder and scene_folder.is_dir(): | |
| images = sorted( | |
| list(scene_folder.glob("*.jpg")) + list(scene_folder.glob("*.png")), | |
| key=lambda p: p.stat().st_size, | |
| reverse=True | |
| ) | |
| if images: | |
| dest = SELECTED_DIR / f"scene_{i:02d}.jpg" | |
| shutil.copy2(images[0], dest) | |
| selected_count += 1 | |
| if selected_count == 0: | |
| raise Exception( | |
| f"No images found in {source_dir}. " | |
| "Expected scene_0/, scene_1/, etc. folders or numbered files." | |
| ) | |
| return { | |
| "status": "success", | |
| "message": f"Selected {selected_count}/{len(manifest.scenes)} scenes", | |
| "selected_count": selected_count, | |
| "selected_dir": str(SELECTED_DIR), | |
| } | |
| except Exception as e: | |
| return { | |
| "status": "error", | |
| "message": str(e), | |
| } | |
| async def _compose(manifest: ManifestRequest): | |
| """ | |
| Internal: Compose video from manifest and selected images. | |
| """ | |
| try: | |
| # Verify selected directory has images | |
| selected_images = sorted(SELECTED_DIR.glob("scene_*.jpg")) | |
| if len(selected_images) != len(manifest.scenes): | |
| raise Exception( | |
| f"Expected {len(manifest.scenes)} selected images, found {len(selected_images)}" | |
| ) | |
| # Generate dynamic SCENE_CONFIG from manifest | |
| scene_config = generate_scene_config(manifest) | |
| # Save config as JSON for composer to use | |
| config_json = { | |
| "title": manifest.title, | |
| "scenes": scene_config, | |
| } | |
| config_path = BASE_DIR / "manifest_config.json" | |
| with open(config_path, "w") as f: | |
| json.dump(config_json, f, indent=2) | |
| # Call composer_v2.py with the config | |
| env = os.environ.copy() | |
| env["COMPOSER_MANIFEST_CONFIG"] = str(config_path) | |
| env["PYTHONIOENCODING"] = "utf-8" | |
| result = subprocess.run( | |
| [sys.executable, str(BASE_DIR / "composer_v2.py")], | |
| cwd=str(BASE_DIR), | |
| env=env, | |
| capture_output=True, | |
| text=True, | |
| ) | |
| # Filter out numpy warnings from stderr | |
| error_msg = result.stderr or result.stdout | |
| if error_msg: | |
| lines = error_msg.split("\n") | |
| # Filter out numpy and importlib warnings | |
| errors = [ | |
| line for line in lines | |
| if line.strip() and not any( | |
| skip in line for skip in [ | |
| "numpy", | |
| "Numpy", | |
| "MINGW-W64", | |
| "CRASHES ARE TO BE EXPECTED", | |
| "RuntimeWarning", | |
| "importlib", | |
| "getlimits", | |
| "nextafter", | |
| "ld(", | |
| "exp2", | |
| "log10", | |
| ] | |
| ) | |
| ] | |
| if errors and result.returncode != 0: | |
| error_summary = "\n".join(errors[-5:]) # Last 5 meaningful lines | |
| raise Exception(f"Composer failed: {error_summary}") | |
| # Verify output file exists | |
| output_file = RENDERS_DIR / "sunset_reel.mp4" | |
| if not output_file.exists(): | |
| raise Exception("Output video file not created") | |
| size_mb = output_file.stat().st_size / (1024 * 1024) | |
| duration_s = len(scene_config) * 2.3 + 2.4 # Approximate | |
| return { | |
| "status": "success", | |
| "message": "Video composed successfully", | |
| "output_path": str(output_file), | |
| "size_mb": round(size_mb, 1), | |
| "duration_s": round(duration_s, 1), | |
| } | |
| except Exception as e: | |
| return { | |
| "status": "error", | |
| "message": str(e), | |
| } | |
| async def health_check(): | |
| """Health check endpoint""" | |
| return { | |
| "status": "healthy", | |
| "service": "TrendClip Video Composer", | |
| "version": "2.0", | |
| } | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |