Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, UploadFile, File, Form | |
| from fastapi.responses import FileResponse | |
| import tempfile | |
| import shutil | |
| import os | |
| import logging | |
| # Import your existing merge functions here | |
| # from your_module import merge_videos_and_audios # if separated, or define inline | |
| app = FastAPI() | |
| logger = logging.getLogger("uvicorn.error") | |
| async def merge_endpoint( | |
| files: list[UploadFile] = File(...), | |
| orig_vol: float = Form(1.0), | |
| music_vol: float = Form(0.5), | |
| ): | |
| temp_dir = tempfile.mkdtemp() | |
| try: | |
| saved_files = [] | |
| for uploaded_file in files: | |
| # Save each uploaded file to disk | |
| file_location = os.path.join(temp_dir, uploaded_file.filename) | |
| with open(file_location, "wb") as out_file: | |
| content = await uploaded_file.read() | |
| out_file.write(content) | |
| saved_files.append(file_location) | |
| if len(saved_files) < 2: | |
| return {"error": "Please upload at least 2 video/audio files."} | |
| # Call your existing merge function | |
| output_path = merge_videos_and_audios( | |
| 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"))], | |
| orig_vol=orig_vol, | |
| music_vol=music_vol, | |
| temp_dir=temp_dir, | |
| ) | |
| if isinstance(output_path, str) and output_path.startswith("Error"): | |
| return {"error": output_path} | |
| # Return the merged file with appropriate media type | |
| media_type = "video/mp4" if output_path.lower().endswith(".mp4") else "audio/mpeg" | |
| file_name = os.path.basename(output_path) | |
| return FileResponse(output_path, media_type=media_type, filename=file_name) | |
| finally: | |
| # Clean up uploaded files after response is sent | |
| shutil.rmtree(temp_dir) |