# # app.py # import os # import uuid # import shutil # import tempfile # import asyncio # from pathlib import Path # from fastapi import FastAPI, File, UploadFile, Form, HTTPException # from fastapi.responses import FileResponse, JSONResponse # from fastapi.middleware.cors import CORSMiddleware # # local imports # from task_queue import TaskQueue, TaskStatus # from core.pipeline import process_image_pipeline # your real pipeline # # App config # APP_NAME = "manim_render_service" # TMP_ROOT = Path(tempfile.gettempdir()) / APP_NAME # TASKS_DIR = TMP_ROOT / "tasks" # OUTPUTS_DIR = TMP_ROOT / "outputs" # TASKS_DIR.mkdir(parents=True, exist_ok=True) # OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) # # instantiate queue (file-backed) # queue = TaskQueue(base_dir=TMP_ROOT, max_workers=os.cpu_count() or 2) # app = FastAPI(title="Manim Render Service") # app.add_middleware( # CORSMiddleware, # allow_origins=["*"], # change in prod # allow_credentials=True, # allow_methods=["*"], # allow_headers=["*"], # ) # @app.on_event("startup") # async def startup_event(): # # start background workers (non-blocking) # await queue.start(processor=process_image_pipeline) # @app.on_event("shutdown") # async def shutdown_event(): # await queue.stop() # def _make_task_dir(task_id: str) -> Path: # p = TASKS_DIR / task_id # p.mkdir(parents=True, exist_ok=True) # return p # def _secure_filename(filename: str) -> str: # # Minimal safe filename normalizer # return "".join(c for c in filename if c.isalnum() or c in "._-").strip("_") # @app.post("/render", status_code=202) # async def submit_render( # image: UploadFile = File(...), # style: str = Form("fade-in"), # quality: str = Form("final"), # preview or final # ): # # Basic validation # if image.content_type.split("/")[0] != "image": # raise HTTPException(status_code=400, detail="Uploaded file must be an image.") # task_id = uuid.uuid4().hex # task_dir = _make_task_dir(task_id) # # Save upload to tmp task directory # safe_name = _secure_filename(image.filename or f"{task_id}.png") # uploaded_path = task_dir / safe_name # try: # with uploaded_path.open("wb") as f: # content = await image.read() # # Limit size for safety (example: 25 MB) # if len(content) > 25 * 1024 * 1024: # raise HTTPException(status_code=413, detail="File too large (max 25MB).") # f.write(content) # finally: # await image.close() # # Compose metadata # meta = { # "task_id": task_id, # "input_image": str(uploaded_path), # "style": style, # "quality": quality, # "task_dir": str(task_dir), # } # # Enqueue the task # queue.enqueue(meta) # return JSONResponse({"task_id": task_id, "status": "queued"}) # @app.get("/status/{task_id}") # async def status(task_id: str): # st = queue.get_status(task_id) # if st is None: # raise HTTPException(status_code=404, detail="Task not found") # return JSONResponse({"task_id": task_id, "status": st.name}) # @app.get("/result/{task_id}") # async def result(task_id: str): # info = queue.get_task_info(task_id) # if info is None: # raise HTTPException(status_code=404, detail="Task not found") # status = queue.get_status(task_id) # if status != TaskStatus.COMPLETED: # return JSONResponse({"task_id": task_id, "status": status.name}) # output_path = Path(info.get("output_path", "")) # if not output_path.exists(): # raise HTTPException(status_code=404, detail="Output not found on disk") # return FileResponse(path=str(output_path), filename=output_path.name, media_type="video/mp4") # @app.delete("/task/{task_id}") # async def delete_task(task_id: str): # info = queue.get_task_info(task_id) # if info: # # attempt cleanup # task_dir = Path(info.get("task_dir", "")) # if task_dir.exists(): # shutil.rmtree(task_dir, ignore_errors=True) # queue.remove_task(task_id) # return JSONResponse({"task_id": task_id, "status": "removed"}) # else: # raise HTTPException(status_code=404, detail="Task not found") # # app.py # import os # import uuid # import shutil # import tempfile # import asyncio # from pathlib import Path # from fastapi import FastAPI, File, UploadFile, Form, HTTPException # from fastapi.responses import FileResponse, JSONResponse # from fastapi.middleware.cors import CORSMiddleware # import subprocess # # local imports # from task_queue import TaskQueue, TaskStatus # from core.pipeline import process_image_pipeline # your real pipeline # from fastapi.responses import FileResponse, JSONResponse, Response # from fastapi.responses import JSONResponse, Response, FileResponse # from fastapi import HTTPException # from pathlib import Path # import base64 # import base64 # from pathlib import Path # from moviepy.video.io.VideoFileClip import VideoFileClip # # -------------------------------------------------- # # Logging Setup # # -------------------------------------------------- # import logging # logging.basicConfig( # level=logging.DEBUG, # format="๐ŸŒ€ [%(asctime)s] [%(levelname)s] %(message)s", # datefmt="%H:%M:%S", # ) # logger = logging.getLogger("manim_render_service") # # -------------------------------------------------- # # App config # # -------------------------------------------------- # APP_NAME = "manim_render_service" # TMP_ROOT = Path("tmp")/ APP_NAME # TASKS_DIR = TMP_ROOT / "tasks" # OUTPUTS_DIR = TMP_ROOT / "outputs" # TASKS_DIR.mkdir(parents=True, exist_ok=True) # OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) # queue = TaskQueue(base_dir=TMP_ROOT, max_workers=os.cpu_count() or 2) # app = FastAPI(title="Manim Render Service") # app.add_middleware( # CORSMiddleware, # allow_origins=["*"], # allow_credentials=True, # allow_methods=["*"], # allow_headers=["*"], # ) # # -------------------------------------------------- # # Lifecycle Events # # -------------------------------------------------- # @app.on_event("startup") # async def startup_event(): # logger.info("๐Ÿš€ Starting up backend...") # logger.debug(f"Temporary root: {TMP_ROOT}") # await queue.start(processor=process_image_pipeline) # logger.info("โœ… Queue system initialized and worker started.") # @app.on_event("shutdown") # async def shutdown_event(): # logger.info("๐Ÿงน Shutting down backend...") # await queue.stop() # logger.info("๐Ÿ›‘ Queue stopped gracefully.") # # -------------------------------------------------- # # Helpers # # -------------------------------------------------- # def _make_task_dir(task_id: str) -> Path: # p = TASKS_DIR / task_id # p.mkdir(parents=True, exist_ok=True) # logger.debug(f"๐Ÿ“ Created task directory: {p}") # return p # def _secure_filename(filename: str) -> str: # safe = "".join(c for c in filename if c.isalnum() or c in "._-").strip("_") # logger.debug(f"๐Ÿ”’ Secured filename: {filename} โ†’ {safe}") # return safe # # -------------------------------------------------- # # Routes # # -------------------------------------------------- # @app.post("/render", status_code=202) # async def submit_render( # image: UploadFile = File(...), # style: str = Form("fade-in"), # quality: str = Form("final"), # ): # logger.info(f"๐Ÿ“จ Received new render request | style={style}, quality={quality}") # logger.debug(f"Uploaded file info: {image.filename}, type={image.content_type}") # if image.content_type.split("/")[0] != "image": # logger.error("โŒ Invalid file type, not an image.") # raise HTTPException(status_code=400, detail="Uploaded file must be an image.") # task_id = uuid.uuid4().hex # task_dir = _make_task_dir(task_id) # logger.info(f"๐Ÿ†” Generated Task ID: {task_id}") # safe_name = _secure_filename(image.filename or f"{task_id}.png") # uploaded_path = task_dir / safe_name # logger.debug(f"๐Ÿ—‚ Saving upload to {uploaded_path}") # try: # with uploaded_path.open("wb") as f: # content = await image.read() # logger.debug(f"๐Ÿ“ฆ File size: {len(content)/1024:.2f} KB") # if len(content) > 25 * 1024 * 1024: # logger.warning("โš ๏ธ Upload too large (>25MB). Rejecting.") # raise HTTPException(status_code=413, detail="File too large (max 25MB).") # f.write(content) # finally: # await image.close() # logger.debug("๐Ÿ“ Image file closed after writing.") # meta = { # "task_id": task_id, # "input_image": str(uploaded_path), # "style": style, # "quality": quality, # "task_dir": str(task_dir), # } # logger.debug(f"๐Ÿงพ Task metadata: {meta}") # queue.enqueue(meta) # logger.info(f"๐Ÿ“ค Task {task_id} successfully enqueued.") # return JSONResponse({"task_id": task_id, "status": "queued"}) # @app.get("/status/{task_id}") # async def status(task_id: str): # logger.debug(f"Status check for task: {task_id}") # st = queue.get_status(task_id) # if st is None: # logger.warning(f"Task {task_id} not found") # raise HTTPException(status_code=404, detail="Task not found") # # Get additional info # task_info = queue.get_task_info(task_id) # logger.info(f"Task {task_id} status: {st.name} | info: {task_info}") # return JSONResponse({ # "task_id": task_id, # "status": st.name, # "details": task_info # }) # @app.get("/result/{task_id}") # async def result(task_id: str): # logger.debug(f"๐Ÿ“ฆ Fetching result for task: {task_id}") # info = queue.get_task_info(task_id) # if info is None: # logger.warning(f"โš ๏ธ Task info not found for {task_id}") # raise HTTPException(status_code=404, detail="Task not found") # status = queue.get_status(task_id) # logger.debug(f"๐Ÿ“ˆ Task {task_id} current status: {status.name}") # if status != TaskStatus.COMPLETED: # logger.info(f"โณ Task {task_id} still in progress ({status.name})") # return JSONResponse({"task_id": task_id, "status": status.name}) # info = queue.get_task_info(task_id) # output_path = Path(info.get("output_path", "")) # MOV path # if not output_path.exists(): # logger.error(f"โŒ Output file missing for task {task_id}") # raise HTTPException(status_code=404, detail="Output not found") # # Convert MOV to optimized WEBM with reduced file size # webm_path = output_path.with_suffix(".webm") # if output_path.suffix.lower() == ".mov" and not webm_path.exists(): # try: # logger.info(f"๐ŸŽž๏ธ Converting .mov โ†’ .webm (optimized for size)...") # cmd = [ # "ffmpeg", # "-y", # "-i", str(output_path), # "-c:v", "libvpx-vp9", # "-pix_fmt", "yuva420p", # keep alpha channel # "-b:v", "2M", # Reduced bitrate (from 4M to 2M) - keeps quality but smaller file # "-maxrate", "2.5M", # "-bufsize", "5M", # "-auto-alt-ref", "0", # "-cpu-used", "4", # Speed up encoding # "-tile-columns", "2", # Enable parallelization # "-tile-rows", "2", # str(webm_path) # ] # subprocess.run(cmd, check=True, capture_output=True) # # Log file sizes # mov_size = output_path.stat().st_size / (1024 * 1024) # webm_size = webm_path.stat().st_size / (1024 * 1024) # logger.info(f"โœ… Converted successfully โ†’ {webm_path}") # logger.info(f"๐Ÿ“Š File sizes - MOV: {mov_size:.2f}MB โ†’ WEBM: {webm_size:.2f}MB (reduction: {((1 - webm_size/mov_size) * 100):.1f}%)") # except Exception as e: # logger.error(f"โš ๏ธ MOVโ†’WEBM conversion failed: {e}") # raise HTTPException(status_code=500, detail=f"Conversion failed: {e}") # # Read only WEBM as bytes # webm_bytes = webm_path.read_bytes() # webm_size = len(webm_bytes) / (1024 * 1024) # logger.info(f"โœ… Sending WEBM only ({webm_size:.2f}MB) for task {task_id}") # return JSONResponse({ # "task_id": task_id, # "status": "COMPLETED", # "results": [ # { # "format": "webm", # "data": base64.b64encode(webm_bytes).decode("utf-8"), # }, # ], # }) # @app.delete("/task/{task_id}") # async def delete_task(task_id: str): # logger.info(f"๐Ÿ—‘ Request to delete task: {task_id}") # info = queue.get_task_info(task_id) # if info: # task_dir = Path(info.get("task_dir", "")) # if task_dir.exists(): # logger.debug(f"๐Ÿงน Removing directory: {task_dir}") # shutil.rmtree(task_dir, ignore_errors=True) # queue.remove_task(task_id) # logger.info(f"โœ… Task {task_id} removed successfully.") # return JSONResponse({"task_id": task_id, "status": "removed"}) # else: # logger.warning(f"โš ๏ธ Task {task_id} not found for deletion.") # raise HTTPException(status_code=404, detail="Task not found") # @app.get("/") # def home(): # return {"status": "Your Manim backend is running!"} import os import uuid import shutil import tempfile import asyncio from pathlib import Path from fastapi import FastAPI, File, UploadFile, Form, HTTPException from fastapi.responses import FileResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware import subprocess import logging import base64 # local imports from task_queue import TaskQueue, TaskStatus from core.pipeline import process_image_pipeline # -------------------------------------------------- # Logging Setup # -------------------------------------------------- logging.basicConfig( level=logging.DEBUG, format="๐ŸŒ€ [%(asctime)s] [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger("manim_render_service") # -------------------------------------------------- # App config # -------------------------------------------------- APP_NAME = "manim_render_service" TMP_ROOT = Path("tmp") / APP_NAME TASKS_DIR = TMP_ROOT / "tasks" OUTPUTS_DIR = TMP_ROOT / "outputs" TASKS_DIR.mkdir(parents=True, exist_ok=True) OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) queue = TaskQueue(base_dir=TMP_ROOT, max_workers=os.cpu_count() or 2) app = FastAPI(title="Manim Render Service") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # -------------------------------------------------- # Lifecycle Events # -------------------------------------------------- @app.on_event("startup") async def startup_event(): logger.info("๐Ÿš€ Starting up backend...") logger.debug(f"Temporary root: {TMP_ROOT}") await queue.start(processor=process_image_pipeline) logger.info("โœ… Queue system initialized and worker started.") @app.on_event("shutdown") async def shutdown_event(): logger.info("๐Ÿงน Shutting down backend...") await queue.stop() logger.info("๐Ÿ›‘ Queue stopped gracefully.") # -------------------------------------------------- # Helpers # -------------------------------------------------- def _make_task_dir(task_id: str) -> Path: p = TASKS_DIR / task_id p.mkdir(parents=True, exist_ok=True) logger.debug(f"๐Ÿ“ Created task directory: {p}") return p def _secure_filename(filename: str) -> str: safe = "".join(c for c in filename if c.isalnum() or c in "._-").strip("_") logger.debug(f"๐Ÿ”’ Secured filename: {filename} โ†’ {safe}") return safe # -------------------------------------------------- # Routes # -------------------------------------------------- @app.post("/render", status_code=202) async def submit_render( image: UploadFile = File(...), style: str = Form("fade-in"), quality: str = Form("final"), ): logger.info(f"๐Ÿ“จ Received new render request | style={style}, quality={quality}") logger.debug(f"Uploaded file info: {image.filename}, type={image.content_type}") if image.content_type.split("/")[0] != "image": logger.error("โŒ Invalid file type, not an image.") raise HTTPException(status_code=400, detail="Uploaded file must be an image.") task_id = uuid.uuid4().hex task_dir = _make_task_dir(task_id) logger.info(f"๐Ÿ†” Generated Task ID: {task_id}") safe_name = _secure_filename(image.filename or f"{task_id}.png") uploaded_path = task_dir / safe_name logger.debug(f"๐Ÿ—‚ Saving upload to {uploaded_path}") try: with uploaded_path.open("wb") as f: content = await image.read() logger.debug(f"๐Ÿ“ฆ File size: {len(content)/1024:.2f} KB") if len(content) > 25 * 1024 * 1024: logger.warning("โš ๏ธ Upload too large (>25MB). Rejecting.") raise HTTPException(status_code=413, detail="File too large (max 25MB).") f.write(content) finally: await image.close() logger.debug("๐Ÿ“ Image file closed after writing.") meta = { "task_id": task_id, "input_image": str(uploaded_path), "style": style, "quality": quality, "task_dir": str(task_dir), } logger.debug(f"๐Ÿงพ Task metadata: {meta}") queue.enqueue(meta) logger.info(f"๐Ÿ“ค Task {task_id} successfully enqueued.") return JSONResponse({"task_id": task_id, "status": "queued"}) @app.get("/status/{task_id}") async def status(task_id: str): logger.debug(f"Status check for task: {task_id}") st = queue.get_status(task_id) if st is None: logger.warning(f"Task {task_id} not found") raise HTTPException(status_code=404, detail="Task not found") # Get additional info task_info = queue.get_task_info(task_id) logger.info(f"Task {task_id} status: {st.name} | info: {task_info}") return JSONResponse({ "task_id": task_id, "status": st.name, "details": task_info }) @app.get("/result/{task_id}") async def result(task_id: str): logger.debug(f"๐Ÿ“ฆ Fetching result for task: {task_id}") info = queue.get_task_info(task_id) if info is None: logger.warning(f"โš ๏ธ Task info not found for {task_id}") raise HTTPException(status_code=404, detail="Task not found") status = queue.get_status(task_id) logger.debug(f"๐Ÿ“ˆ Task {task_id} current status: {status.name}") if status != TaskStatus.COMPLETED: logger.info(f"โณ Task {task_id} still in progress ({status.name})") return JSONResponse({"task_id": task_id, "status": status.name}) output_path = Path(info.get("output_path", "")) if not output_path.exists(): logger.error(f"โŒ Output file missing for task {task_id}") raise HTTPException(status_code=404, detail="Output not found") # Convert MOV to optimized WEBM webm_path = output_path.with_suffix(".webm") if output_path.suffix.lower() == ".mov" and not webm_path.exists(): try: logger.info(f"๐ŸŽž๏ธ Converting .mov โ†’ .webm (optimized)...") cmd = [ "ffmpeg", "-y", "-i", str(output_path), "-c:v", "libvpx-vp9", "-pix_fmt", "yuva420p", "-b:v", "2M", "-maxrate", "2.5M", "-bufsize", "5M", "-auto-alt-ref", "0", "-cpu-used", "4", "-tile-columns", "2", "-tile-rows", "2", str(webm_path) ] subprocess.run(cmd, check=True, capture_output=True) mov_size = output_path.stat().st_size / (1024 * 1024) webm_size = webm_path.stat().st_size / (1024 * 1024) logger.info(f"โœ… Converted: MOV {mov_size:.2f}MB โ†’ WEBM {webm_size:.2f}MB ({((1 - webm_size/mov_size) * 100):.1f}% smaller)") except Exception as e: logger.error(f"โš ๏ธ Conversion failed: {e}") raise HTTPException(status_code=500, detail=f"Conversion failed: {e}") if not webm_path.exists(): logger.error(f"โŒ WEBM file not found after conversion: {webm_path}") raise HTTPException(status_code=500, detail="WEBM file generation failed") webm_size = webm_path.stat().st_size / (1024 * 1024) logger.info(f"โœ… Streaming WEBM ({webm_size:.2f}MB) for task {task_id}") # Stream the file directly instead of base64 encoding return FileResponse( path=webm_path, media_type="video/webm", filename=f"{task_id}.webm" ) @app.delete("/task/{task_id}") async def delete_task(task_id: str): logger.info(f"๐Ÿ—‘ Request to delete task: {task_id}") info = queue.get_task_info(task_id) if info: task_dir = Path(info.get("task_dir", "")) if task_dir.exists(): logger.debug(f"๐Ÿงน Removing directory: {task_dir}") shutil.rmtree(task_dir, ignore_errors=True) queue.remove_task(task_id) logger.info(f"โœ… Task {task_id} removed successfully.") return JSONResponse({"task_id": task_id, "status": "removed"}) else: logger.warning(f"โš ๏ธ Task {task_id} not found for deletion.") raise HTTPException(status_code=404, detail="Task not found") @app.get("/") def home(): return {"status": "Your Manim backend is running!"}