BackendJunsen / app /services /youtube_downloader.py
2nzi's picture
update backend with video upload on HF
923cd30 verified
import yt_dlp
import io
import asyncio
from typing import List
import os
async def download_youtube_video(url: str) -> bytes:
try:
print(f"[DEBUG] Téléchargement de {url}")
# Créer un dossier temporaire pour le téléchargement
temp_dir = "temp_downloads"
os.makedirs(temp_dir, exist_ok=True)
# Configuration de yt-dlp
ydl_opts = {
'format': 'bestvideo[ext=mp4][vcodec^=avc1]+bestaudio[ext=m4a]/mp4',
'merge_output_format': 'mp4',
'quiet': True,
'no_warnings': True,
'outtmpl': os.path.join(temp_dir, '%(title)s.%(ext)s'),
}
def custom_download(d):
if d['status'] == 'finished':
print(f"[DEBUG] Téléchargement terminé pour {url}")
ydl_opts['progress_hooks'] = [custom_download]
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
# Télécharger la vidéo
info = ydl.extract_info(url, download=True)
video_path = ydl.prepare_filename(info)
# Lire le fichier en bytes
with open(video_path, 'rb') as f:
content = f.read()
# Nettoyer le fichier temporaire
os.remove(video_path)
return content
except Exception as e:
print(f"[ERROR] Erreur téléchargement YouTube: {str(e)}")
raise Exception(f"Erreur téléchargement YouTube: {str(e)}")
finally:
# Nettoyer le dossier temporaire si vide
try:
os.rmdir(temp_dir)
except:
pass
def parse_urls(content: str) -> List[str]:
urls = [url.strip() for url in content.replace(';', '\n').split('\n')]
return [url for url in urls if url and ('youtube.com' in url or 'youtu.be' in url)]
async def test_download():
test_url = "https://www.youtube.com/watch?v=ksIfusO_DgA" # URL de test de surf
try:
content = await download_youtube_video(test_url)
print(f"[SUCCESS] Vidéo téléchargée avec succès! Taille: {len(content)} bytes")
return True
except Exception as e:
print(f"[ERROR] Test échoué: {str(e)}")
return False
def test_parse_urls():
test_content = """
https://www.youtube.com/watch?v=ksIfusO_DgA&ab_channel=NobodySurf%3ASurfingVideos
https://www.youtube.com/watch?v=GHDLZwccg8E&ab_channel=RIPITUP
"""
urls = parse_urls(test_content)
print(f"[DEBUG] URLs trouvées: {urls}")
return len(urls) == 2
if __name__ == "__main__":
# Test de parse_urls
print("\nTest de parse_urls:")
if test_parse_urls():
print("[SUCCESS] Test parse_urls réussi!")
else:
print("[ERROR] Test parse_urls échoué")
# Test de download_youtube_video
print("\nTest de download_youtube_video:")
asyncio.run(test_download())