|
|
from fastapi import FastAPI, UploadFile, File |
|
|
from fastapi.responses import FileResponse, JSONResponse |
|
|
import os, subprocess, logging |
|
|
|
|
|
app = FastAPI() |
|
|
|
|
|
JOBS_DIR = "/tmp/jobs" |
|
|
os.makedirs(JOBS_DIR, exist_ok=True) |
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
|
|
|
|
job_status = { |
|
|
"status": "idle", |
|
|
"error": None, |
|
|
} |
|
|
|
|
|
|
|
|
def run_ffmpeg(base_file, overlay_file, output_file): |
|
|
cmd = [ |
|
|
"ffmpeg", "-y", "-i", base_file, "-i", overlay_file, |
|
|
"-filter_complex", "[0:v][1:v]overlay=10:10:format=auto", |
|
|
"-c:a", "copy", output_file |
|
|
] |
|
|
logging.info(f"Running FFmpeg: {' '.join(cmd)}") |
|
|
result = subprocess.run(cmd, capture_output=True, text=True) |
|
|
logging.info(f"FFmpeg stdout: {result.stdout}") |
|
|
logging.info(f"FFmpeg stderr: {result.stderr}") |
|
|
return result.returncode == 0 |
|
|
|
|
|
|
|
|
@app.post("/generate-video/") |
|
|
async def generate_video(video: UploadFile = File(...), overlay: UploadFile = File(...)): |
|
|
base_file = os.path.join(JOBS_DIR, "current_base.mp4") |
|
|
overlay_file = os.path.join(JOBS_DIR, "current_overlay.mp4") |
|
|
output_file = os.path.join(JOBS_DIR, "current_output.mp4") |
|
|
|
|
|
|
|
|
for f in [base_file, overlay_file, output_file]: |
|
|
if os.path.exists(f): |
|
|
os.remove(f) |
|
|
|
|
|
with open(base_file, "wb") as f: |
|
|
f.write(await video.read()) |
|
|
with open(overlay_file, "wb") as f: |
|
|
f.write(await overlay.read()) |
|
|
|
|
|
job_status["status"] = "processing" |
|
|
job_status["error"] = None |
|
|
|
|
|
try: |
|
|
success = run_ffmpeg(base_file, overlay_file, output_file) |
|
|
if success and os.path.exists(output_file): |
|
|
job_status["status"] = "completed" |
|
|
else: |
|
|
job_status["status"] = "failed" |
|
|
job_status["error"] = "FFmpeg failed or output missing." |
|
|
except Exception as e: |
|
|
job_status["status"] = "failed" |
|
|
job_status["error"] = str(e) |
|
|
|
|
|
return {"status": job_status["status"]} |
|
|
|
|
|
|
|
|
@app.get("/status/") |
|
|
async def check_status(): |
|
|
return JSONResponse(job_status) |
|
|
|
|
|
|
|
|
@app.get("/download/") |
|
|
async def download_video(): |
|
|
output_file = os.path.join(JOBS_DIR, "current_output.mp4") |
|
|
if job_status["status"] != "completed" or not os.path.exists(output_file): |
|
|
return JSONResponse({"error": "No completed video to download."}, status_code=404) |
|
|
return FileResponse(output_file, media_type="video/mp4", filename="output.mp4") |
|
|
|
