| import os |
| import requests |
| from fastapi import FastAPI, HTTPException, Query |
| from fastapi.responses import JSONResponse, FileResponse |
| from fastapi.staticfiles import StaticFiles |
| import yt_dlp |
| from yt_dlp import YoutubeDL |
| from urllib.parse import urlparse, parse_qs |
| from pydantic import BaseModel |
| import assemblyai as aai |
| import uvicorn |
| import tempfile |
| import imageio |
| import argparse |
| import uuid |
| import subprocess |
| import shutil |
|
|
|
|
| app = FastAPI() |
|
|
|
|
| STATIC_DIR = os.path.join(os.getcwd(), "static") |
| if not os.path.exists(STATIC_DIR): |
| os.makedirs(STATIC_DIR) |
|
|
|
|
| app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") |
|
|
|
|
| API_KEY = os.getenv("ASSEMBLYAI_API_KEY") |
| if not API_KEY: |
| raise ValueError("ASSEMBLYAI_API_KEY not set in OS secrets") |
|
|
| aai.settings.api_key = API_KEY |
| transcriber = aai.Transcriber() |
|
|
|
|
| YOUTUBE_COOKIES = os.getenv("YOUTUBE_COOKIES", "") |
|
|
| def create_cookie_file(): |
| """ |
| Create a temporary cookie file from environment variable |
| Expected format: Netscape format cookies as a multi-line string |
| """ |
| if not YOUTUBE_COOKIES: |
| return None |
| |
| try: |
|
|
| cookie_file = tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) |
| |
| |
| cookie_file.write(YOUTUBE_COOKIES) |
| cookie_file.flush() |
| cookie_file.close() |
| |
| return cookie_file.name |
| except Exception as e: |
| print(f"Warning: Failed to create cookie file: {e}") |
| return None |
|
|
| def get_ydl_opts(): |
| """ |
| Get YoutubeDL options with cookies from environment variables |
| """ |
| base_opts = { |
| 'quiet': True, |
| 'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', |
| } |
| |
|
|
| cookie_file = create_cookie_file() |
| if cookie_file: |
| base_opts['cookiefile'] = cookie_file |
| |
| return base_opts |
|
|
| def cleanup_cookie_file(ydl_opts): |
| """ |
| Clean up temporary cookie file if it was created |
| """ |
| if 'cookiefile' in ydl_opts: |
| try: |
| os.unlink(ydl_opts['cookiefile']) |
| except Exception: |
| pass |
|
|
| def extract_video_info(url: str): |
| ydl_opts = { |
| **get_ydl_opts(), |
| 'extract_flat': False, |
| } |
|
|
| with yt_dlp.YoutubeDL(ydl_opts) as ydl: |
| try: |
| info = ydl.extract_info(url, download=False) |
| title = info.get('title', 'Unknown Title') |
| formats = info.get('formats', []) |
| |
| |
| thumbnails = info.get('thumbnails', []) |
| best_thumbnail = None |
| max_width = 0 |
| |
| for thumb in thumbnails: |
| width = thumb.get('width', 0) |
| if width > max_width: |
| max_width = width |
| best_thumbnail = thumb.get('url') |
| |
|
|
| default_thumbnail = info.get('thumbnail', None) |
|
|
| |
| audio_video_formats = [ |
| { |
| 'format_id': fmt['format_id'], |
| 'resolution': fmt.get('resolution', 'N/A'), |
| 'filesize': fmt.get('filesize') or fmt.get('filesize_approx'), |
| 'url': fmt['url'] |
| } |
| for fmt in formats if fmt.get('acodec') != 'none' and fmt.get('vcodec') != 'none' |
| ] |
|
|
| |
| highest_audio = max( |
| (fmt for fmt in formats if fmt.get('acodec') != 'none' and fmt.get('vcodec') == 'none' and fmt.get('ext') == 'm4a'), |
| key=lambda x: x.get('abr', 0), |
| default=None |
| ) |
|
|
| highest_audio_info = { |
| 'format_id': highest_audio['format_id'], |
| 'bitrate': highest_audio.get('abr', 'Unknown'), |
| 'filesize': highest_audio.get('filesize') or highest_audio.get('filesize_approx'), |
| 'url': highest_audio['url'] |
| } if highest_audio else None |
|
|
| result = { |
| "title": title, |
| "formats": audio_video_formats, |
| "highest_audio": highest_audio_info, |
| "thumbnail": best_thumbnail or default_thumbnail |
| } |
| |
| |
| cleanup_cookie_file(ydl_opts) |
| |
| return result |
| except Exception as e: |
| cleanup_cookie_file(ydl_opts) |
| raise HTTPException(status_code=500, detail=f"Error extracting video info: {str(e)}") |
|
|
| def search_video_by_title(title: str): |
| ydl_opts = { |
| **get_ydl_opts(), |
| 'extract_flat': True, |
| 'force_generic_extractor': True, |
| } |
|
|
| with YoutubeDL(ydl_opts) as ydl: |
| try: |
| search_results = ydl.extract_info(f"ytsearch:{title}", download=False) |
| if not search_results or 'entries' not in search_results: |
| cleanup_cookie_file(ydl_opts) |
| raise HTTPException(status_code=404, detail="No videos found for the given title.") |
|
|
| |
| first_entry = search_results['entries'][0] |
| result = first_entry['url'] |
| |
| |
| cleanup_cookie_file(ydl_opts) |
| |
| return result |
| except Exception as e: |
| cleanup_cookie_file(ydl_opts) |
| raise HTTPException(status_code=500, detail=f"Error searching video by title: {str(e)}") |
|
|
| def extract_playlist_id(url: str) -> str: |
| """ |
| Extracts the playlist ID from a YouTube URL. |
| """ |
| parsed_url = urlparse(url) |
| query_params = parse_qs(parsed_url.query) |
| playlist_id = query_params.get('list', [None])[0] |
| if not playlist_id: |
| raise HTTPException(status_code=400, detail="No playlist ID found in the URL.") |
| return playlist_id |
|
|
| def extract_playlist_video_urls(playlist_url: str): |
| """ |
| Extracts video URLs and metadata from a YouTube playlist. |
| """ |
| |
| if "watch?v=" in playlist_url and "list=" in playlist_url: |
| playlist_id = extract_playlist_id(playlist_url) |
| playlist_url = f"https://www.youtube.com/playlist?list={playlist_id}" |
|
|
| ydl_opts = { |
| **get_ydl_opts(), |
| 'extract_flat': True, |
| 'force_generic_extractor': True, |
| 'extractor_args': { |
| 'youtubetab': {'skip': ['authcheck']} |
| } |
| } |
|
|
| with YoutubeDL(ydl_opts) as ydl: |
| try: |
| info = ydl.extract_info(playlist_url, download=False) |
| if not info or 'entries' not in info: |
| raise HTTPException(status_code=404, detail="No videos found in the playlist.") |
|
|
| video_urls = [ |
| { |
| 'title': entry.get('title', 'Unknown Title'), |
| 'url': entry.get('url') or entry.get('webpage_url', 'N/A'), |
| 'thumbnail': entry.get('thumbnail', None) |
| } |
| for entry in info['entries'] if entry |
| ] |
|
|
| return { |
| "playlist_title": info.get('title', 'Unknown Playlist'), |
| "playlist_thumbnail": info.get('thumbnail', None), |
| "video_urls": video_urls |
| } |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=f"Error extracting playlist video URLs: {str(e)}") |
|
|
| def download_and_merge_video(url: str, resolution: str): |
| """ |
| Downloads video and audio separately and merges with ffmpeg. |
| Returns the path to the merged file and thumbnail URL. |
| |
| If the specified resolution (720p/1080p) isn't available, falls back to best available quality. |
| """ |
| try: |
| |
| file_id = str(uuid.uuid4()) |
| output_dir = os.path.join(STATIC_DIR, file_id) |
| os.makedirs(output_dir, exist_ok=True) |
| |
| |
| common_opts = get_ydl_opts() |
| |
| |
| with YoutubeDL({**common_opts, 'noplaylist': True}) as ydl: |
| info = ydl.extract_info(url, download=False) |
| title = info.get('title', 'video') |
| |
| |
| thumbnails = info.get('thumbnails', []) |
| best_thumbnail = None |
| max_width = 0 |
| |
| for thumb in thumbnails: |
| width = thumb.get('width', 0) |
| if width > max_width: |
| max_width = width |
| best_thumbnail = thumb.get('url') |
| |
| if not best_thumbnail: |
| best_thumbnail = info.get('thumbnail', None) |
| |
| target_height = int(resolution.replace("p", "")) |
| actual_resolution = resolution |
| |
| title = "".join(c for c in title if c.isalnum() or c in [' ', '-', '_']).strip() |
| title = title.replace(' ', '_') |
| |
| output_path = os.path.join(STATIC_DIR, f"{title}_{resolution}_{file_id}.mp4") |
| |
| video_path = os.path.join(output_dir, f"{file_id}_video.mp4") |
| audio_path = os.path.join(output_dir, f"{file_id}_audio.m4a") |
| |
| |
| if resolution == "720p": |
| |
| video_format = ( |
| "bestvideo[height<=720][ext=mp4]/bestvideo[height<=720][ext=webm]/" |
| "bestvideo[height<=720]/best[height<=720][ext=mp4]/best[height<=720][ext=webm]/" |
| "best[height<=720]/bestvideo[height<=800]/best[height<=800]/" |
| "bestvideo/best" |
| ) |
| height_filter = 720 |
| else: |
|
|
| video_format = ( |
| "bestvideo[height<=1080][ext=mp4]/bestvideo[height<=1080][ext=webm]/" |
| "bestvideo[height<=1080]/best[height<=1080][ext=mp4]/best[height<=1080][ext=webm]/" |
| "best[height<=1080]/bestvideo[height<=1200]/best[height<=1200]/" |
| "bestvideo/best" |
| ) |
| height_filter = 1080 |
| |
| |
| video_opts = { |
| **common_opts, |
| 'format': video_format, |
| 'outtmpl': video_path, |
| 'format_sort': ['res', 'ext:mp4:m4a', 'ext:webm'], |
| } |
| |
| |
| video_downloaded = False |
| fallback_formats = [ |
| video_format, |
| f"best[height<={height_filter}]", |
| "bestvideo/best", |
| "best" |
| ] |
| |
| for fmt in fallback_formats: |
| try: |
| video_opts['format'] = fmt |
| with YoutubeDL(video_opts) as ydl: |
| ydl.download([url]) |
| video_downloaded = True |
| break |
| except Exception as e: |
| print(f"Video download failed with format '{fmt}': {e}") |
| continue |
| |
| if not video_downloaded: |
| raise Exception("Failed to download video with any format") |
| |
| |
| audio_downloaded = False |
| temp_audio_path = os.path.join(output_dir, f"{file_id}_temp_audio") |
| |
| |
| audio_strategies = [ |
| |
| { |
| 'format': 'bestaudio/best', |
| 'outtmpl': temp_audio_path, |
| 'postprocessors': [{ |
| 'key': 'FFmpegExtractAudio', |
| 'preferredcodec': 'm4a', |
| 'preferredquality': '192', |
| }], |
| }, |
| |
| { |
| 'format': 'bestaudio[ext=m4a]/bestaudio[ext=aac]/bestaudio[ext=mp3]/bestaudio', |
| 'outtmpl': audio_path, |
| }, |
| |
| { |
| 'format': 'bestaudio', |
| 'outtmpl': audio_path, |
| }, |
| |
| { |
| 'format': 'best', |
| 'outtmpl': temp_audio_path, |
| 'postprocessors': [{ |
| 'key': 'FFmpegExtractAudio', |
| 'preferredcodec': 'm4a', |
| 'preferredquality': '192', |
| }], |
| } |
| ] |
| |
| for i, strategy in enumerate(audio_strategies): |
| try: |
| audio_opts = {**common_opts, **strategy} |
| with YoutubeDL(audio_opts) as ydl: |
| ydl.download([url]) |
| |
|
|
| if 'postprocessors' in strategy: |
| audio_files = [f for f in os.listdir(output_dir) if 'temp_audio' in f] |
| if audio_files: |
| actual_audio_path = os.path.join(output_dir, audio_files[0]) |
| shutil.move(actual_audio_path, audio_path) |
| audio_downloaded = True |
| break |
| else: |
| |
| if os.path.exists(audio_path): |
| audio_downloaded = True |
| break |
| |
| except Exception as e: |
| print(f"Audio download strategy {i+1} failed: {e}") |
| continue |
| |
| if not audio_downloaded: |
| raise Exception("Failed to download audio with any strategy") |
| |
| |
| probe_cmd = [ |
| 'ffprobe', |
| '-v', 'error', |
| '-select_streams', 'v:0', |
| '-show_entries', 'stream=width,height', |
| '-of', 'csv=p=0', |
| video_path |
| ] |
| |
| try: |
| probe_result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) |
| width, height = map(int, probe_result.stdout.strip().split(',')) |
| actual_resolution = f"{height}p" |
| except Exception: |
| |
| pass |
| |
| |
| cmd = [ |
| 'ffmpeg', |
| '-i', video_path, |
| '-i', audio_path, |
| '-c:v', 'copy', |
| '-c:a', 'aac', |
| '-map', '0:v:0', |
| '-map', '1:a:0', |
| output_path |
| ] |
| |
|
|
| subprocess.run(cmd, check=True) |
| |
| |
| if actual_resolution != resolution: |
| new_output_path = os.path.join(STATIC_DIR, f"{title}_{actual_resolution}_{file_id}.mp4") |
| os.rename(output_path, new_output_path) |
| output_path = new_output_path |
| |
| |
| shutil.rmtree(output_dir) |
| |
|
|
| relative_path = os.path.basename(output_path) |
| return relative_path, best_thumbnail |
| |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=f"Error processing video: {str(e)}") |
|
|
| import os |
| import uuid |
| import shutil |
| import requests |
| from mutagen.mp3 import MP3 |
| from mutagen.mp4 import MP4 |
| from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB |
| from yt_dlp import YoutubeDL |
|
|
| def download_audio(url: str, audio_format: str): |
| """ |
| Downloads audio from a video URL in the specified format and adds metadata. |
| Returns the path to the downloaded audio file. |
| |
| Parameters: |
| - url: URL of the video |
| - audio_format: Format of the audio (mp3 or m4a) |
| """ |
| try: |
| file_id = str(uuid.uuid4()) |
| output_dir = os.path.join(STATIC_DIR, file_id) |
| os.makedirs(output_dir, exist_ok=True) |
| |
| common_opts = get_ydl_opts() |
| |
| |
| with YoutubeDL({**common_opts, 'noplaylist': True}) as ydl: |
| info = ydl.extract_info(url, download=False) |
| title = info.get('title', 'audio') |
| uploader = info.get('uploader', 'Unknown Artist') |
| channel = info.get('channel', uploader) |
| description = info.get('description', '') |
| |
| |
| thumbnails = info.get('thumbnails', []) |
| best_thumbnail = None |
| max_width = 0 |
| |
| for thumb in thumbnails: |
| width = thumb.get('width', 0) |
| if width > max_width: |
| max_width = width |
| best_thumbnail = thumb.get('url') |
| |
| if not best_thumbnail: |
| best_thumbnail = info.get('thumbnail', None) |
| |
| |
| clean_title = "".join(c for c in title if c.isalnum() or c in [' ', '-', '_']).strip() |
| clean_title = clean_title.replace(' ', '_') |
| |
| |
| temp_audio_path = os.path.join(output_dir, f"{file_id}_temp.{audio_format}") |
| output_path = os.path.join(STATIC_DIR, f"{clean_title}_{audio_format}_{file_id}.{audio_format}") |
| thumbnail_path = os.path.join(output_dir, f"{file_id}_thumb.jpg") |
| |
| |
| thumbnail_data = None |
| if best_thumbnail: |
| try: |
| response = requests.get(best_thumbnail, timeout=10) |
| if response.status_code == 200: |
| with open(thumbnail_path, 'wb') as f: |
| f.write(response.content) |
| thumbnail_data = response.content |
| except Exception as e: |
| print(f"Failed to download thumbnail: {e}") |
| |
| |
| if audio_format == "mp3": |
| format_option = "bestaudio/best" |
| postprocessors = [{ |
| 'key': 'FFmpegExtractAudio', |
| 'preferredcodec': 'mp3', |
| 'preferredquality': '192', |
| }] |
| else: |
| format_option = "bestaudio[ext=m4a]/bestaudio" |
| postprocessors = [{ |
| 'key': 'FFmpegExtractAudio', |
| 'preferredcodec': 'm4a', |
| 'preferredquality': '192', |
| }] |
| |
| audio_opts = { |
| **common_opts, |
| 'format': format_option, |
| 'outtmpl': temp_audio_path, |
| 'postprocessors': postprocessors, |
| } |
| |
| with YoutubeDL(audio_opts) as ydl: |
| ydl.download([url]) |
| |
| |
| downloaded_files = os.listdir(output_dir) |
| audio_files = [f for f in downloaded_files if f.endswith(('.mp3', '.m4a'))] |
| |
| if not audio_files: |
| raise Exception("No audio file was downloaded") |
| |
| downloaded_file = os.path.join(output_dir, audio_files[0]) |
| shutil.move(downloaded_file, output_path) |
| |
| |
| add_metadata(output_path, audio_format, title, uploader, channel, thumbnail_data) |
| |
| |
| shutil.rmtree(output_dir) |
| |
| relative_path = os.path.basename(output_path) |
| return relative_path, best_thumbnail |
| |
| except Exception as e: |
| |
| if os.path.exists(output_dir): |
| shutil.rmtree(output_dir) |
| raise HTTPException(status_code=500, detail=f"Error processing audio: {str(e)}") |
|
|
|
|
| def add_metadata(file_path: str, audio_format: str, title: str, artist: str, album: str, thumbnail_data: bytes = None): |
| """ |
| Adds metadata to the audio file including thumbnail as cover art. |
| |
| Parameters: |
| - file_path: Path to the audio file |
| - audio_format: Format of the audio (mp3 or m4a) |
| - title: Track title |
| - artist: Artist name |
| - album: Album name (using channel name) |
| - thumbnail_data: Binary data of the thumbnail image |
| """ |
| try: |
| if audio_format == "mp3": |
| |
| audio = MP3(file_path, ID3=ID3) |
| |
| |
| if audio.tags is None: |
| audio.add_tags() |
| |
| |
| audio.tags.add(TIT2(encoding=3, text=title)) |
| audio.tags.add(TPE1(encoding=3, text=artist)) |
| audio.tags.add(TALB(encoding=3, text=album)) |
| |
| |
| if thumbnail_data: |
| audio.tags.add(APIC( |
| encoding=3, |
| mime='image/jpeg', |
| type=3, |
| desc='Cover', |
| data=thumbnail_data |
| )) |
| |
| audio.save() |
| |
| elif audio_format == "m4a": |
| |
| audio = MP4(file_path) |
| |
| |
| audio['\xa9nam'] = [title] |
| audio['\xa9ART'] = [artist] |
| audio['\xa9alb'] = [album] |
| |
|
|
| if thumbnail_data: |
| audio['covr'] = [thumbnail_data] |
| |
| audio.save() |
| |
| except Exception as e: |
| print(f"Failed to add metadata: {e}") |
|
|
|
|
| @app.get("/download-audio") |
| async def download_audio_endpoint( |
| query: str = Query(..., description="URL or title of the video"), |
| format: str = Query("mp3", description="Audio format (mp3 or m4a)") |
| ): |
| try: |
| |
| if format not in ["mp3", "m4a"]: |
| raise HTTPException(status_code=400, detail="Format must be either mp3 or m4a") |
| |
| |
| parsed_url = urlparse(query) |
| if parsed_url.scheme and parsed_url.netloc: |
| video_url = query |
| else: |
| video_url = search_video_by_title(query) |
| |
| |
| output_file, thumbnail_url = download_audio(video_url, format) |
| |
| |
| file_url = f"/static/{output_file}" |
| server_url = os.getenv("SERVER_URL", "https://akane710-y2mateee.hf.space") |
| download_url = f"{server_url}{file_url}" |
| |
| return JSONResponse(content={ |
| "download_url": download_url, |
| "filename": output_file, |
| "thumbnail": thumbnail_url |
| }) |
| |
| except HTTPException as e: |
| raise e |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
|
|
| @app.get("/video-info") |
| async def get_video_info(query: str = Query(..., description="URL or title of the video")): |
| try: |
| |
| parsed_url = urlparse(query) |
| if parsed_url.scheme and parsed_url.netloc: |
| video_url = query |
| else: |
| video_url = search_video_by_title(query) |
|
|
| video_data = extract_video_info(video_url) |
| return JSONResponse(content=video_data) |
| except HTTPException as e: |
| raise e |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
|
|
| @app.get("/download-video") |
| async def download_video( |
| query: str = Query(..., description="URL or title of the video"), |
| resolution: str = Query("720p", description="Video resolution (720p or 1080p)") |
| ): |
| try: |
| |
| if resolution not in ["720p", "1080p"]: |
| raise HTTPException(status_code=400, detail="Resolution must be either 720p or 1080p") |
| |
|
|
| parsed_url = urlparse(query) |
| if parsed_url.scheme and parsed_url.netloc: |
| video_url = query |
| else: |
| video_url = search_video_by_title(query) |
| |
| |
| output_file, thumbnail_url = download_and_merge_video(video_url, resolution) |
| |
|
|
| actual_resolution = resolution |
| if "_best_" in output_file: |
| actual_resolution = "best quality available" |
| elif "_" in output_file: |
| parts = output_file.split("_") |
| for part in parts: |
| if part.endswith("p") and part[:-1].isdigit(): |
| actual_resolution = part |
| break |
| |
| |
| file_url = f"/static/{output_file}" |
| server_url = os.getenv("SERVER_URL", "https://akane710-y2mateee.hf.space") |
| download_url = f"{server_url}{file_url}" |
| |
| return JSONResponse(content={ |
| "download_url": download_url, |
| "filename": output_file, |
| "thumbnail": thumbnail_url, |
| "actual_resolution": actual_resolution |
| }) |
| |
| except HTTPException as e: |
| raise e |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
|
|
| @app.get("/direct-download/{filename}") |
| async def get_file(filename: str): |
| file_path = os.path.join(STATIC_DIR, filename) |
| if not os.path.exists(file_path): |
| raise HTTPException(status_code=404, detail="File not found") |
| |
| return FileResponse( |
| path=file_path, |
| filename=filename, |
| media_type="video/mp4" |
| ) |
|
|
|
|
| class TranscriptionRequest(BaseModel): |
| audio_url: str |
|
|
| @app.get("/transcribe") |
| async def transcribe_audio(audio_url: str = Query(..., description="Publicly accessible URL for audio file")): |
| try: |
| config = aai.TranscriptionConfig( |
| speaker_labels=True, |
| speech_model=aai.SpeechModel.nano, |
| language_detection=True |
| ) |
| transcript = transcriber.transcribe(audio_url, config) |
|
|
| if transcript.status == aai.TranscriptStatus.error: |
| raise HTTPException(status_code=500, detail=f"Transcription failed: {transcript.error}") |
|
|
| return {"transcription": transcript.text} |
|
|
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
|
|
| @app.get("/playlist-download-urls") |
| async def get_playlist_video_urls(playlist_url: str = Query(..., description="URL of the YouTube playlist")): |
| try: |
| result = extract_playlist_video_urls(playlist_url) |
| return JSONResponse(content=result) |
| except HTTPException as e: |
| raise e |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
| async def process_and_upload_webp(url: str) -> str: |
| try: |
|
|
| parsed_url = urlparse(url) |
| if not parsed_url.scheme or not parsed_url.netloc: |
| raise ValueError("Invalid URL") |
|
|
|
|
| response = requests.get(url) |
| if response.status_code != 200: |
| raise HTTPException(status_code=400, detail="Failed to download the file") |
|
|
| |
| with tempfile.NamedTemporaryFile(suffix=".webp", delete=False) as webp_file: |
| webp_file.write(response.content) |
| webp_file_path = webp_file.name |
|
|
|
|
| output_path = webp_file_path.replace(".webp", ".mp4") |
| reader = imageio.get_reader(webp_file_path) |
| writer = imageio.get_writer(output_path, fps=30) |
| for frame in reader: |
| writer.append_data(frame) |
| writer.close() |
|
|
| |
| with open(output_path, 'rb') as mp4_file: |
| upload_response = requests.post( |
| 'https://uguu.se/upload', |
| files={"files[]": mp4_file} |
| ) |
| if upload_response.status_code != 200: |
| raise HTTPException(status_code=500, detail="Failed to upload the file") |
|
|
| file_url = upload_response.json()["files"][0]["url"] |
|
|
|
|
| os.remove(webp_file_path) |
| os.remove(output_path) |
|
|
| return file_url |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
|
|
| @app.get("/convert") |
| async def convert_webp_to_mp4(url: str): |
| file_url = await process_and_upload_webp(url) |
| return JSONResponse(content={"url": file_url}) |
|
|
|
|
| def cli(): |
| parser = argparse.ArgumentParser(description="Convert and upload WEBP to MP4.") |
| parser.add_argument("--url", type=str, required=True, help="URL of the .webp file.") |
| args = parser.parse_args() |
|
|
|
|
| import asyncio |
| result = asyncio.run(process_and_upload_webp(args.url)) |
| print(f"Uploaded file link: {result}") |
|
|
|
|
| if __name__ == "__main__": |
| uvicorn.run(app, host="0.0.0.0", port=8000) |
