app.py

"""
Streamlit Web App to Fetch YouTube Transcript and Comments
=========================================================

This Streamlit application provides a simple user interface for retrieving the
transcript (subtitles) and top‑level comments from a YouTube video.  It uses
yt‑dlp under the hood to download the comments and subtitles for a given
video URL.  The resulting transcript and comments are displayed directly in
the browser and can optionally be downloaded as plain text files.

Requirements
------------

To run this application you must have the following tools installed on your
machine:

* **Python 3.8 or newer**
* **yt‑dlp** – a fork of youtube‑dl capable of downloading comments and
  subtitles.  Installation instructions are available in the official
  repository: https://github.com/yt‑dlp/yt‑dlp
* **Streamlit** – used for the web interface.
* **NumPy** – required for audio processing and signal generation. Install via `pip install numpy`.

Once all dependencies are installed you can launch the app with:

```
streamlit run app.py
```

After starting, open the provided local URL in your browser and paste a YouTube
link to retrieve its transcript and comments.

Please note that fetching comments for popular videos can take a few minutes
depending on the number of comments available.
"""

import base64
import datetime
import io
import json
import os
import re
import subprocess
import tempfile
import unicodedata
import uuid
import wave
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional, Any

import langdetect
import numpy as np
import streamlit as st
import yt_dlp
import yt_dlp.version
from st_keyup import st_keyup


@dataclass
class VideoData:
    """A container for storing metadata about a single video."""

    url: str
    title: str
    transcript: str
    comments: List[Dict[str, str]]
    actual_lang: str = ""


def create_slug(title: str) -> str:
    """
    Create a slug filename from the video title.

    Parameters
    ----------
    title : str
        The title of the YouTube video.

    Returns
    -------
    str
        The slugified title.
    """
    # Normalize unicode characters
    title = unicodedata.normalize('NFKD', title).encode('ascii', 'ignore').decode('utf-8')
    # Convert to lowercase
    title = title.lower()
    # Remove non-alphanumeric characters (except hyphens and underscores and dots)
    title = re.sub(r'[^\w\s\.-]', '', title)
    # Replace whitespace with underscores
    title = re.sub(r'[-\s]+', '_', title)
    return title.strip('-_')


def generate_notification_sound() -> bytes:
    """Generate a clean notification beep sound using a simple tone.
    
    Creates a pleasant notification sound with proper ADSR envelope
    to avoid clicks and artifacts.
    
    Returns
    -------
    bytes
        WAV audio data for a short notification beep.
    """
    sample_rate = 44100  # Hz
    duration = 0.4  # seconds - slightly longer for smoother sound
    frequency = 800  # Hz (notification tone)
    
    # Generate time array
    t = np.linspace(0, duration, int(sample_rate * duration))
    
    # Generate pure sine wave
    audio_data = np.sin(2 * np.pi * frequency * t)
    
    # Create ADSR envelope for clean, professional sound
    # Attack: 0.02s, Decay: 0.05s, Sustain: 0.20s, Release: 0.13s
    attack_samples = int(0.02 * sample_rate)
    decay_samples = int(0.05 * sample_rate)
    sustain_samples = int(0.20 * sample_rate)
    release_samples = len(audio_data) - attack_samples - decay_samples - sustain_samples
    
    envelope = np.ones(len(audio_data))
    
    # Attack: linear ramp up
    envelope[:attack_samples] = np.linspace(0, 1, attack_samples)
    
    # Decay: exponential decay to sustain level (0.7)
    decay_curve = np.exp(np.linspace(0, -1.5, decay_samples))
    envelope[attack_samples:attack_samples + decay_samples] = 1 - (1 - 0.7) * (1 - decay_curve)
    
    # Sustain: constant level
    envelope[attack_samples + decay_samples:attack_samples + decay_samples + sustain_samples] = 0.7
    
    # Release: smooth exponential fade out
    release_curve = np.exp(np.linspace(0, -5, release_samples))
    envelope[-release_samples:] = 0.7 * release_curve
    
    # Apply envelope to audio
    audio_data = audio_data * envelope
    
    # Scale to 16-bit integer range with slight reduction to avoid clipping
    audio_data = (audio_data * 30000).astype(np.int16)
    
    # Create WAV file in memory
    byte_io = io.BytesIO()
    with wave.open(byte_io, 'wb') as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(audio_data.tobytes())
    
    return byte_io.getvalue()


def run_yt_dlp(args: List[str]) -> None:
    """Run a yt‑dlp command and raise an exception if it fails.

    The function is separated for easier mocking during tests.  It calls
    subprocess.run with the provided arguments and checks the return code.

    Parameters
    ----------
    args:
        A list of command line arguments to pass directly to yt‑dlp.

    Raises
    ------
    RuntimeError
        If yt‑dlp returns a non‑zero exit status.
    """
    result = subprocess.run(args, capture_output=True, text=True)
    if result.returncode != 0:
        raise RuntimeError(
            f"yt‑dlp failed with status {result.returncode}:\n"
            f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
        )


def parse_srt_contents(contents: str) -> str:
    """Convert an SRT subtitle file into plain text.

    SRT files consist of numbered blocks with timestamps followed by one or
    more lines of subtitle text.  This function strips away the sequence
    numbers and timestamps and joins the remaining text lines into a single
    coherent transcript.

    Parameters
    ----------
    contents:
        Raw contents of an SRT file decoded as UTF‑8.

    Returns
    -------
    str
        The cleaned transcript.
    """
    lines = contents.splitlines()
    transcript_lines: List[str] = []
    for line in lines:
        # Skip blank lines, purely numeric counters and timestamp lines
        if not line.strip():
            continue
        if re.match(r"^\d+$", line.strip()):
            continue
        if "-->" in line:
            continue
        transcript_lines.append(line.strip())
    return "\n".join(transcript_lines)


def format_duration(seconds: Optional[int]) -> str:
    """Format duration in seconds to HH:MM:SS string."""
    if not seconds:
        return "N/A"
    return str(datetime.timedelta(seconds=seconds))


def format_large_number(num: Optional[int]) -> str:
    """Format large numbers (e.g. view counts) to readable strings like 1.5M."""
    if num is None:
        return "N/A"
    if num >= 1_000_000_000:
        return f"{num / 1_000_000_000:.1f}B"
    if num >= 1_000_000:
        return f"{num / 1_000_000:.1f}M"
    if num >= 1_000:
        return f"{num / 1_000:.1f}K"
    return str(num)


def format_file_size(bytes_: Optional[int]) -> str:
    """Format bytes to readable string (KB, MB, GB)."""
    if bytes_ is None:
        return "N/A"
    for unit in ['B', 'KB', 'MB', 'GB']:
        if bytes_ < 1024.0:
            return f"{bytes_:.1f} {unit}"
        bytes_ /= 1024.0
    return f"{bytes_:.1f} TB"


def get_video_info(url: str) -> Optional[Dict[str, Any]]:
    """
    Fetch metadata for a given YouTube URL without downloading the video.

    Parameters
    ----------
    url : str
        The YouTube video URL.

    Returns
    -------
    Optional[Dict[str, Any]]
        The metadata dictionary or None if fetching fails.
    """
    if not url:
        return None

    ydl_opts = {
        'quiet': True,
        'skip_download': True,
        'noplaylist': True,
        # We don't restrict 'extract_flat' because we need full metadata like duration/fps
    }

    try:
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            # download=False ensures we only get metadata
            info = ydl.extract_info(url, download=False)
            return info
    except Exception:
        return None


def fetch_video_data(
    video_url: str,
    work_dir: Path,
    language: str,
    download_transcript: bool,
    download_comments: bool,
    cookies_content: Optional[str] = None,
) -> VideoData:
    """Download all necessary video data using a single yt‑dlp invocation.

    This function fetches the video title, transcript, and comments in a
    single call to yt‑dlp to improve efficiency by avoiding multiple
    invocations.

    Parameters
    ----------
    video_url:
        The full YouTube URL provided by the user.
    work_dir:
        A temporary directory to store downloaded files.
    language:
        The preferred language for the transcript.
    download_transcript:
        A boolean indicating whether to download the transcript.
    download_comments:
        A boolean indicating whether to download the comments.
    cookies_content:
        Optional content of a cookies.txt file to authenticate with YouTube.

    Returns
    -------
    VideoData
        A dataclass containing the title, transcript, and comments.
    """
    video_id_match = re.search(r"(?<=v=)[^&?]+|(?<=youtu.be/)[^?&]+|(?<=shorts/)[^?&]+", video_url)
    if not video_id_match:
        raise ValueError("Impossible d'extraire l'identifiant de la vidéo.")
    video_id = video_id_match.group(0)
    json_path = work_dir / f"{video_id}.info.json"

    cmd = ["yt-dlp", "--skip-download", "-o", str(work_dir / f"{video_id}")]

    if cookies_content:
        cookies_path = work_dir / "cookies.txt"
        with cookies_path.open("w", encoding="utf-8") as f:
            f.write(cookies_content)
        cmd.extend(["--cookies", str(cookies_path)])

    if download_transcript:
        cmd.extend(
            [
                "--write-sub",
                "--write-auto-subs",
                "--sub-format",
                "srt",
                "--sub-lang",
                language,
            ]
        )
    if download_comments:
        cmd.append("--write-comments")

    # If we need comments or a transcript (for the title), get the info json.
    if download_transcript or download_comments:
        cmd.append("--write-info-json")

    cmd.append(video_url)
    run_yt_dlp(cmd)

    # Extract transcript
    transcript = ""
    actual_lang = ""
    if download_transcript:
        possible_files = list(work_dir.glob(f"{video_id}.*.srt"))
        subtitle_path = next(iter(possible_files), None)
        if subtitle_path and subtitle_path.exists():
            with subtitle_path.open("r", encoding="utf-8") as f:
                transcript = parse_srt_contents(f.read())
            # Extract language from filename, e.g., "video_id.fr.srt"
            parts = subtitle_path.name.split(".")
            if len(parts) > 2:
                actual_lang = parts[-2]

    # Extract comments and title
    comments_list: List[Dict[str, str]] = []
    title = ""
    if download_transcript or download_comments:
        if not json_path.exists():
            raise FileNotFoundError(
                f"Fichier JSON introuvable : {json_path}. Assurez-vous que yt‑dlp est correctement installé."
            )
        with json_path.open("r", encoding="utf-8") as f:
            info = json.load(f)
        title = info.get("title", "Titre inconnu")
        if download_comments:
            comments_raw = info.get("comments", [])
            for comment in comments_raw:
                text = comment.get("text") or comment.get("txt", "")
                author = comment.get("author", "Auteur inconnu")
                if text:
                    comments_list.append(
                        {"author": author.strip(), "text": text.strip()}
                    )

    return VideoData(
        url=video_url,
        title=title,
        transcript=transcript,
        comments=comments_list,
        actual_lang=actual_lang,
    )


def main() -> None:
    """Main Streamlit application entry point."""
    st.set_page_config(
        page_title="Transcripteur YouTube", page_icon="🎬", layout="centered"
    )

    if "videos" not in st.session_state:
        st.session_state["videos"] = [
            {"id": str(uuid.uuid4()), "url": "", "download_transcript": True, "download_comments": True}
        ]

    # Ensure all videos have an ID (migration for existing session state)
    for video in st.session_state["videos"]:
        if "id" not in video:
            video["id"] = str(uuid.uuid4())

    st.title("Récupération de transcript et commentaires YouTube")

    st.markdown(
        """
        Entrez un lien **YouTube** ci‑dessous pour obtenir sa transcription et
        ses commentaires.  Le traitement utilise l'outil `yt‑dlp` en interne.
        Veillez donc à ce que celui‑ci soit installé sur votre machine.
        """
    )

    # Use a single loop for all video inputs
    for i, video in enumerate(st.session_state["videos"]):
        st.markdown("---")
        cols = st.columns([0.8, 0.2])

        with cols[0]:
            video_url = st_keyup(
                f"Lien de la vidéo YouTube #{i + 1}",
                debounce=3000,
                value=video.get("url", ""),
                key=f"url_{video['id']}",
            )
        video["url"] = video_url

        # Check if URL has changed or metadata is missing for a non-empty URL
        if video_url and (video_url != video.get("last_fetched_url") or "metadata" not in video):
            info = get_video_info(video_url)
            if info:
                video["metadata"] = info
                # Language detection
                title = info.get("title", "")
                try:
                    detected_lang = langdetect.detect(title)
                    if detected_lang:
                        st.session_state["lang"] = detected_lang
                except langdetect.lang_detect_exception.LangDetectException:
                    pass
            video["last_fetched_url"] = video_url

        # Display metadata if available
        if video.get("metadata"):
            meta = video["metadata"]
            with st.expander("Informations vidéo", expanded=False):
                m_cols = st.columns(3)
                with m_cols[0]:
                    st.write(f"**Titre:** {meta.get('title', 'N/A')}")
                    st.write(f"**Durée:** {format_duration(meta.get('duration'))}")
                    st.write(f"**Vues:** {format_large_number(meta.get('view_count'))}")
                with m_cols[1]:
                    st.write(f"**Résolution:** {meta.get('resolution', 'N/A')}")
                    st.write(f"**FPS:** {meta.get('fps', 'N/A')}")
                    st.write(f"**Date:** {meta.get('upload_date', 'N/A')}")
                with m_cols[2]:
                    # File size is often an approximation in metadata
                    size = meta.get('filesize') or meta.get('filesize_approx')
                    st.write(f"**Taille (approx):** {format_file_size(size)}")
                    st.write(f"**Codec:** {meta.get('vcodec', 'N/A')}")
                    st.write(f"**Ext:** {meta.get('ext', 'N/A')}")

        with cols[1]:
            # Use unique key for delete button
            if st.button("🗑️", key=f"delete_{video['id']}"):
                st.session_state["videos"].pop(i)
                st.experimental_rerun()

        cols = st.columns(2)
        video["download_transcript"] = cols[0].checkbox(
            "Télécharger la transcription",
            value=video["download_transcript"],
            key=f"transcript_{video['id']}",
        )
        video["download_comments"] = cols[1].checkbox(
            "Télécharger les commentaires",
            value=video["download_comments"],
            key=f"comments_{video['id']}",
        )

    # Language selection dropdown - appears once
    lang_options = ["fr", "en", "es", "de", "it"]
    if "lang" in st.session_state and st.session_state["lang"] not in lang_options:
        lang_options.insert(0, st.session_state["lang"])

    try:
        lang_index = lang_options.index(st.session_state.get("lang")) if st.session_state.get("lang") else 0
    except ValueError:
        lang_index = 0

    lang = st.selectbox(
        "Langue des sous‑titres",
        options=lang_options,
        index=lang_index,
        help="Choisissez la langue à privilégier pour les sous‑titres.",
    )

    st.button("Ajouter une autre vidéo", on_click=lambda: st.session_state["videos"].append(
        {"id": str(uuid.uuid4()), "url": "", "download_transcript": True, "download_comments": True}
    ))

    with st.expander("🍪 Cookies / Authentification"):
        st.info("Si vous rencontrez des erreurs 429 ou 'Sign in', collez ici le contenu de votre fichier cookies.txt (format Netscape).")
        cookies_content = st.text_area("Contenu du fichier cookies.txt", height=150)

    if st.button("Récupérer"):
        # Filter out empty URLs
        videos_to_process = [
            v for v in st.session_state["videos"] if v["url"].strip()
        ]
        if not videos_to_process:
            st.error("Merci de fournir au moins une URL valide.")
        else:
            # Check that at least one download option is selected for each video
            valid_selection = True
            for video in videos_to_process:
                if not video["download_transcript"] and not video["download_comments"]:
                    st.error(
                        f"Pour la vidéo {video['url']}, veuillez sélectionner au moins une option de téléchargement."
                    )
                    valid_selection = False
                    break

            if valid_selection:
                try:
                    all_video_data = []
                    with tempfile.TemporaryDirectory() as tmpdir_str:
                        tmpdir = Path(tmpdir_str)
                        for i, video in enumerate(videos_to_process, 1):
                            with st.spinner(
                                f"Traitement de la vidéo {i}/{len(videos_to_process)}..."
                            ):
                                video_data = fetch_video_data(
                                    video["url"],
                                    tmpdir,
                                    lang,
                                    video["download_transcript"],
                                    video["download_comments"],
                                    cookies_content=cookies_content,
                                )
                                all_video_data.append(video_data)

                    # Store results in session state to persist across reruns
                    st.session_state["fetched_results"] = {
                        "all_video_data": all_video_data,
                        "videos_to_process": videos_to_process
                    }

                    st.success("Récupération terminée !")

                    notification_sound = generate_notification_sound()
                    audio_base64 = base64.b64encode(notification_sound).decode()
                    audio_html = f'<audio autoplay><source src="data:audio/wav;base64,{audio_base64}" type="audio/wav"></audio>'
                    st.markdown(audio_html, unsafe_allow_html=True)
                except Exception as exc:
                    st.error(f"Une erreur est survenue : {exc}")

    # Display results if they exist in session state
    if "fetched_results" in st.session_state:
        results = st.session_state["fetched_results"]
        all_video_data = results["all_video_data"]
        videos_to_process = results["videos_to_process"]

        # --- Merged Transcripts ---
        any_transcript_needed = any(
            v["download_transcript"] for v in videos_to_process
        )
        if any_transcript_needed:
            merged_transcript = ""
            for data in all_video_data:
                if data.transcript:
                    merged_transcript += f"--- Transcription pour '{data.title}' ---\n"
                    merged_transcript += f"URL: {data.url}\n\n"
                    merged_transcript += data.transcript + "\n\n"

            st.subheader("Toutes les transcriptions")
            if merged_transcript:
                # Determine filename
                # If only one video has a transcript, use its title for the filename
                videos_with_transcript = [data for data in all_video_data if data.transcript]
                if len(videos_with_transcript) == 1:
                    file_name = f"transcription_{create_slug(videos_with_transcript[0].title)}.txt"
                else:
                    file_name = "transcriptions_fusionnees.txt"

                st.download_button(
                    label="Télécharger toutes les transcriptions",
                    data=merged_transcript,
                    file_name=file_name,
                    mime="text/plain",
                )
                st.text_area(
                    "Transcriptions fusionnées",
                    value=merged_transcript,
                    height=300,
                )
            else:
                st.warning("Aucune transcription n'a été trouvée.")

        # --- Merged Comments ---
        any_comments_needed = any(
            v["download_comments"] for v in videos_to_process
        )
        if any_comments_needed:
            merged_comments = ""
            for data in all_video_data:
                if data.comments:
                    merged_comments += f"--- Commentaires pour '{data.title}' ---\n"
                    merged_comments += f"URL: {data.url}\n\n"
                    for comment in data.comments:
                        merged_comments += (
                            f"Auteur: {comment['author']}\n{comment['text']}\n\n"
                        )

            st.subheader("Tous les commentaires")
            if merged_comments:
                # Determine filename
                # If only one video has comments, use its title for the filename
                videos_with_comments = [data for data in all_video_data if data.comments]
                if len(videos_with_comments) == 1:
                    file_name = f"comments_{create_slug(videos_with_comments[0].title)}.txt"
                else:
                    file_name = "commentaires_fusionnes.txt"

                st.download_button(
                    label="Télécharger tous les commentaires",
                    data=merged_comments,
                    file_name=file_name,
                    mime="text/plain",
                )
                st.text_area(
                    "Commentaires fusionnés",
                    value=merged_comments,
                    height=300,
                )
            else:
                st.warning("Aucun commentaire n'a été trouvé.")

    # Display footer with version
    try:
        with open("version.txt", "r") as f:
            version = f.read().strip()
    except FileNotFoundError:
        version = "development"
    st.markdown(
        f"<div style='text-align: center; color: grey;'>"
        f"Version: {version} | yt-dlp: {yt_dlp.version.__version__}"
        f"</div>",
        unsafe_allow_html=True
    )


if __name__ == "__main__":
    main()