Spaces:
Sleeping
Sleeping
File size: 4,155 Bytes
c2781b0 |
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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
# main.py
import gradio as gr
from fastapi import FastAPI, File, UploadFile, Form
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
import os
import tempfile
import shutil
import traceback
import logging
from merge_utils import merge_videos_and_audios # move your merge logic to merge_utils.py for clean code!
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# --- FASTAPI SETUP ---
fastapi_app = FastAPI(title="Video & Audio Merger API")
# CORS for n8n to call the API
fastapi_app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # You can restrict to your domain if needed
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# FastAPI endpoint to accept binary files from n8n
@fastapi_app.post("/merge")
async def merge_endpoint(
files: list[UploadFile] = File(...),
orig_vol: float = Form(1.0),
music_vol: float = Form(0.5),
):
try:
logger.info(f"Received {len(files)} files from n8n")
# Save files to temp dir
temp_dir = tempfile.mkdtemp()
saved_files = []
for upload_file in files:
filename = os.path.basename(upload_file.filename)
file_path = os.path.join(temp_dir, filename)
with open(file_path, "wb") as f:
content = await upload_file.read()
f.write(content)
saved_files.append(file_path)
logger.info(f"Saved file: {file_path}")
# Split files into video and audio
video_files = [f for f in saved_files if f.lower().endswith(".mp4")]
audio_files = [f for f in saved_files if f.lower().endswith((".mp3", ".wav"))]
# Run merge
output_path = merge_videos_and_audios(
video_files=video_files,
audio_files=audio_files,
orig_vol=orig_vol,
music_vol=music_vol,
temp_dir=temp_dir,
)
# Return result as path or error
if isinstance(output_path, str) and output_path.startswith("Error"):
return {"success": False, "error": output_path}
# Move output to persistent location (optional)
public_output_dir = "./outputs"
os.makedirs(public_output_dir, exist_ok=True)
final_output_path = os.path.join(public_output_dir, os.path.basename(output_path))
shutil.move(output_path, final_output_path)
logger.info(f"Final output moved to {final_output_path}")
return {
"success": True,
"output_url": f"/outputs/{os.path.basename(final_output_path)}"
}
except Exception as e:
error_msg = f"Exception: {str(e)}\n{traceback.format_exc()}"
logger.error(error_msg)
return {"success": False, "error": error_msg}
# --- GRADIO SETUP ---
with gr.Blocks(title="Video and Audio Merger UI") as gradio_app:
gr.Markdown("## Video and Audio Merger")
file_input = gr.File(label="Upload Files", type="binary", file_count="multiple")
orig_vol_input = gr.Slider(minimum=0.0, maximum=1.0, value=1.0, step=0.05, label="Original Video Audio Volume")
music_vol_input = gr.Slider(minimum=0.0, maximum=1.0, value=0.5, step=0.05, label="Background Audio Volume")
output_video = gr.Video(label="Merged Video")
output_audio = gr.Audio(label="Merged Audio")
merge_button = gr.Button("Merge Files")
def gradio_merge_files(file_binaries, orig_vol, music_vol):
# You can call your existing function here (same as before)
pass # Keep your old gradio_merge_files() here!
merge_button.click(
fn=gradio_merge_files,
inputs=[file_input, orig_vol_input, music_vol_input],
outputs=[output_video, output_audio],
)
# --- COMBINE BOTH ---
if __name__ == "__main__":
import threading
# Run FastAPI server in background
def run_fastapi():
uvicorn.run(fastapi_app, host="0.0.0.0", port=8000)
threading.Thread(target=run_fastapi).start()
# Run Gradio app on another port (HuggingFace will pick this)
gradio_app.queue(api_open=True)
gradio_app.launch(server_port=7860, share=True)
|