Youtube_Downloader/downloader.py

from __future__ import annotations

import json
import os
import subprocess
import sys
import threading
from pathlib import Path
from typing import Callable


# ------------------------------------------------------------
# Gestion des chemins
# ------------------------------------------------------------
# En mode développement :
#   APP_DIR = dossier du projet
#   RESOURCE_DIR = dossier du projet
#
# En mode .exe PyInstaller :
#   APP_DIR = dossier où se trouve YoutubeDownloader.exe
#   RESOURCE_DIR = dossier temporaire/interne utilisé par PyInstaller
# ------------------------------------------------------------

if getattr(sys, "frozen", False):
    APP_DIR = Path(sys.executable).resolve().parent
    RESOURCE_DIR = Path(getattr(sys, "_MEIPASS", APP_DIR))
else:
    APP_DIR = Path(__file__).resolve().parent
    RESOURCE_DIR = APP_DIR


BASE_DIR = APP_DIR
BIN_DIR = RESOURCE_DIR / "bin"

# Les téléchargements iront dans le dossier Downloads de l'utilisateur.
# C'est plus propre pour une app installée.
DOWNLOADS_DIR = Path.home() / "Downloads" / "Youtube_Downloader"

SETTINGS_FILE = RESOURCE_DIR / "settings.json"

YTDLP_EXE = BIN_DIR / "yt-dlp.exe"
FFMPEG_EXE = BIN_DIR / "ffmpeg.exe"
FFPROBE_EXE = BIN_DIR / "ffprobe.exe"


def load_settings() -> dict:
    """
    Charge les paramètres depuis settings.json.
    Si le fichier est absent ou cassé, on utilise des valeurs par défaut.
    """
    default_settings = {
        "default_output_folder": "downloads",
        "default_video_quality": "720",
        "default_audio_quality": "192K",
    }

    if not SETTINGS_FILE.exists():
        return default_settings

    try:
        with open(SETTINGS_FILE, "r", encoding="utf-8") as file:
            settings = json.load(file)

        return {
            "default_output_folder": settings.get(
                "default_output_folder",
                default_settings["default_output_folder"],
            ),
            "default_video_quality": settings.get(
                "default_video_quality",
                default_settings["default_video_quality"],
            ),
            "default_audio_quality": settings.get(
                "default_audio_quality",
                default_settings["default_audio_quality"],
            ),
        }

    except Exception:
        return default_settings


def validate_tools() -> list[str]:
    """
    Vérifie que yt-dlp.exe et ffmpeg.exe existent dans le dossier bin.
    """
    errors = []

    if not YTDLP_EXE.exists():
        errors.append("yt-dlp.exe est introuvable dans le dossier bin.")

    if not FFMPEG_EXE.exists():
        errors.append("ffmpeg.exe est introuvable dans le dossier bin.")

    # ffprobe est utile mais pas toujours obligatoire.
    # On ne bloque pas l'application s'il est absent.
    return errors


def clean_filename(filename: str) -> str:
    """
    Nettoie le nom du fichier pour éviter les caractères interdits sous Windows.
    """
    forbidden_chars = '<>:"/\\|?*'
    cleaned = "".join("_" if char in forbidden_chars else char for char in filename)
    cleaned = cleaned.strip()

    if not cleaned:
        return "telechargement"

    return cleaned


def resolve_output_folder(output_folder: str) -> Path:
    """
    Prépare le dossier de sortie.
    Si le chemin est relatif, il est créé dans le dossier de l'application.
    """
    if not output_folder.strip():
        folder = DOWNLOADS_DIR
    else:
        folder = Path(output_folder)

    if not folder.is_absolute():
        folder = BASE_DIR / folder

    folder.mkdir(parents=True, exist_ok=True)
    return folder


def build_command(
    url: str,
    mode: str,
    output_folder: str,
    filename: str,
    start_time: str = "",
    end_time: str = "",
    video_quality: str = "720",
    audio_quality: str = "192K",
) -> list[str]:
    """
    Construit la commande yt-dlp selon le mode choisi.
    """

    if not url.strip():
        raise ValueError("L'URL YouTube est obligatoire.")

    filename = clean_filename(filename)
    folder = resolve_output_folder(output_folder)

    output_template = str(folder / f"{filename}.%(ext)s")

    cmd = [
        str(YTDLP_EXE),
        url.strip(),
        "--newline",
        "--no-playlist",
        "--ffmpeg-location",
        str(BIN_DIR),
    ]

    # Gestion des extraits
    if mode in ["video_extract", "audio_extract"]:
        if not start_time.strip() or not end_time.strip():
            raise ValueError("Pour un extrait, il faut indiquer le début et la fin.")

        cmd += [
            "--download-sections",
            f"*{start_time.strip()}-{end_time.strip()}",
        ]

    # Vidéo MP4
    if mode in ["video_full", "video_extract"]:
        if video_quality == "best":
            format_selector = "bv*[ext=mp4]+ba[ext=m4a]/b[ext=mp4]/best"
        else:
            format_selector = (
                f"bv*[height<={video_quality}][ext=mp4]+ba[ext=m4a]/"
                f"b[height<={video_quality}][ext=mp4]/best"
            )

        cmd += [
            "-f",
            format_selector,
            "--merge-output-format",
            "mp4",
        ]

    # Audio MP3
    elif mode in ["audio_full", "audio_extract"]:
        cmd += [
            "-f",
            "ba/bestaudio",
            "-x",
            "--audio-format",
            "mp3",
            "--audio-quality",
            audio_quality,
        ]

    else:
        raise ValueError("Mode de téléchargement inconnu.")

    cmd += [
        "-o",
        output_template,
    ]

    return cmd


class CommandRunner:
    """
    Lance yt-dlp dans un thread séparé pour éviter de bloquer l'interface.
    """

    def __init__(self):
        self.process: subprocess.Popen | None = None
        self.cancelled = False

    def run(
        self,
        command: list[str],
        on_line: Callable[[str], None],
        on_finish: Callable[[bool, str], None],
    ) -> None:
        self.cancelled = False

        def worker():
            try:
                creation_flags = 0

                if os.name == "nt":
                    creation_flags = subprocess.CREATE_NO_WINDOW

                self.process = subprocess.Popen(
                    command,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    text=True,
                    encoding="utf-8",
                    errors="replace",
                    creationflags=creation_flags,
                )

                if self.process.stdout:
                    for line in self.process.stdout:
                        on_line(line.rstrip())

                return_code = self.process.wait()

                if self.cancelled:
                    on_finish(False, "Téléchargement annulé.")
                elif return_code == 0:
                    on_finish(True, "Téléchargement terminé avec succès.")
                else:
                    on_finish(False, f"Erreur pendant le téléchargement. Code : {return_code}")

            except FileNotFoundError:
                on_finish(False, "Impossible de lancer yt-dlp ou FFmpeg.")
            except Exception as error:
                on_finish(False, f"Erreur : {error}")

        thread = threading.Thread(target=worker, daemon=True)
        thread.start()

    def cancel(self) -> None:
        self.cancelled = True

        if self.process and self.process.poll() is None:
            try:
                self.process.terminate()
            except Exception:
                pass