File size: 2,376 Bytes
ac0b104 c94c34b ac0b104 c94c34b 3a4c1c1 c94c34b ac0b104 c94c34b ac0b104 c94c34b f182c1a c94c34b 3a4c1c1 c94c34b ac0b104 c94c34b ac0b104 c94c34b ac0b104 3a4c1c1 ac0b104 3a4c1c1 ac0b104 3a4c1c1 ac0b104 3a4c1c1 ac0b104 c94c34b ac0b104 c94c34b 3a4c1c1 ac0b104 3a4c1c1 ac0b104 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 |
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")
# Clear old files
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")
|