File size: 2,994 Bytes
923cd30
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
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())