src/notebooks/youtube-demos.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dazbo's YouTube and Video Demos\n",
    "\n",
    "## Overview\n",
    "\n",
    "Examples of how to work with YouTube videos using Python. Here I'll demonstrate:\n",
    "\n",
    "- How to [download videos and extract audio](#downloading-videos-and-extracting-audio)\n",
    "- How to [transcribe audio to text using a speech-to-text API](#extracting-audio-using-python-speech-recognition)\n",
    "- How to [extract existing transcripts and translate](#extract-existing-transcripts-from-videos)\n",
    "\n",
    "**To run this notebook, first execute the cells in the [Setup](#Setup) section, as described below.** Then you can experiment with any of the subsequent cells.\n",
    "\n",
    "A few useful notes:\n",
    "\n",
    "- The source for this notebook source lives in my GitHub repo, <a href=\"https://github.com/derailed-dash/youtube-and-video\" target=\"_blank\">Youtube-and-Video</a>.\n",
    "- Check out further guidance - including tips on how to run the notebook, in the project's `README.md`.\n",
    "- For example, you could...\n",
    "  - Run the notebook locally, in your own Jupyter environment.\n",
    "  - Run the notebook in a cloud-based Jupyter environment, with no setup required on your part! For example, with **Google Colab**: <br><br><a href=\"https://colab.research.google.com/github/derailed-dash/youtube-and-video/blob/main/src/notebooks/youtube-demos.ipynb\" target=\"_blank\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Google Colab\"/></a><br><br>It looks like this:<br><br><img src=\"static/images/collab-view.png\" width=\"640px\"></img>\n",
    "- For more ways to run Jupyter Notebooks, check out [my guide](https://medium.com/python-in-plain-english/five-ways-to-run-jupyter-labs-and-notebooks-23209f71e5c0).\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Packages\n",
    "\n",
    "First, let's install any dependent packages:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install --upgrade --no-cache-dir python-dotenv dazbo-commons pytubefix moviepy yt_dlp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import IPython\n",
    "from IPython.display import display\n",
    "from IPython.core.display import Markdown\n",
    "\n",
    "import logging\n",
    "import re\n",
    "import io\n",
    "import sys\n",
    "from pathlib import Path\n",
    "from dataclasses import dataclass\n",
    "import dazbo_commons as dc\n",
    "from dotenv import load_dotenv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Colab requires an older version of Ipykernel\n",
    "if not \"google.colab\" in sys.modules:\n",
    "    pass\n",
    "    %pip install --upgrade --no-cache-dir ipykernel\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Logging\n",
    "\n",
    "Now we'll setup logging. Here I'm using coloured logging from my [dazbo-commons](https://pypi.org/project/dazbo-commons/) package. Feel free to change the logging level."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Setup logging\n",
    "APP_NAME=\"dazbo-yt-demos\"\n",
    "logger = dc.retrieve_console_logger(APP_NAME)\n",
    "logger.setLevel(logging.DEBUG)\n",
    "logger.info(\"Logger initialised.\")\n",
    "logger.debug(\"DEBUG level logging enabled.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### File Locations\n",
    "\n",
    "Here we initialise some file path locations, e.g. an output folder."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "locations = dc.get_locations(APP_NAME)\n",
    "for attribute, value in vars(locations).items():\n",
    "    logger.debug(f\"{attribute}: {value}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Utility Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_filename(filename):\n",
    "    \"\"\" Create a clean filename by removing unallowed characters. \"\"\"\n",
    "    pattern = r'[^a-zA-Z0-9._\\s-]'\n",
    "    return  re.sub(pattern, '_', filename)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Install Additional Packages You May Need\n",
    "\n",
    "You can run the cell below, but it may not work on your OS.\n",
    "\n",
    "So you might need to install packages manually, e.g.\n",
    "\n",
    "<table>\n",
    "  <col style=\"width:10%\">\n",
    "  <col style=\"width:45%\">\n",
    "  <col style=\"width:45%\"> <!-- Adjust as needed or remove for auto-sizing -->\n",
    "  <tr>\n",
    "    <th>Package</th>\n",
    "    <th>Purpose</th>\n",
    "    <th>Install Command</th>\n",
    "  </tr>\n",
    "  <tr>\n",
    "    <td><a href=\"https://ffmpeg.org/\">ffmpeg</a></td>\n",
    "    <td>A useful utility for video and audio format conversion. Many Python libraries use it. It will not generally be used by this notebook, but if you run into errors requiring ffmpeg, you will want to run this section.</td>\n",
    "    <td>Linux: <code>sudo apt install ffmpeg</code><br>Windows: <code>winget install ffmpeg</code></td>\n",
    "  </tr>\n",
    "  <tr>\n",
    "    <td><a href=\"https://xiph.org/flac/download.html\">FLAC</a></td>\n",
    "    <td>The Python <code>speech_recognition</code> library uses the FLAC utility to convert audio files into a format that can be processed for speech recognition.</td>\n",
    "    <td>Linux: <code>sudo apt install flac</code><br>Windows: Download the latest</td>\n",
    "  </tr>\n",
    "</table>\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import platform\n",
    "import subprocess\n",
    "\n",
    "def run_command(command):\n",
    "    \"\"\"Run a shell command and print its output in real-time.\"\"\"\n",
    "    process = subprocess.Popen(\n",
    "        command, \n",
    "        shell=True, \n",
    "        stdout=subprocess.PIPE, \n",
    "        stderr=subprocess.PIPE\n",
    "    )\n",
    "    \n",
    "    # Read and print the output line by line\n",
    "    if process.stdout is not None:\n",
    "        for line in iter(process.stdout.readline, b''):\n",
    "            logger.info(line.decode().strip())\n",
    "        process.stdout.close()\n",
    "        \n",
    "    process.wait()\n",
    "    \n",
    "def install_software(appname: str):\n",
    "    os_name = platform.system()\n",
    "    logger.info(f\"Installing {appname} on {os_name}...\")\n",
    "    \n",
    "    # Mapping operating systems to their respective installation commands\n",
    "    command_map = {\n",
    "        \"Windows\": f\"winget install {appname} --silent --no-upgrade\",\n",
    "        \"Linux\": f\"apt -qq -y install {appname}\",\n",
    "        \"Darwin\": f\"brew install {appname}\"\n",
    "    }\n",
    "    command = command_map.get(os_name)\n",
    "    if command:\n",
    "        run_command(command)\n",
    "        logger.info(f\"Done.\")\n",
    "    else:\n",
    "        logger.error(f\"Unsupported operating system: {os_name}\")\n",
    "\n",
    "def check_installed(app_exec: str) -> bool:    \n",
    "    appname, *arg = app_exec.split()\n",
    "    arg = \" \".join(arg)\n",
    "    logger.debug(f\"Checking if {appname} is installed\")\n",
    "    \n",
    "    try:\n",
    "        output = subprocess.check_output([appname, arg], stderr=subprocess.STDOUT)\n",
    "        logger.debug(f\"{appname} version: {output.decode().strip()}\")\n",
    "        logger.debug(f\"{appname} is already installed.\")\n",
    "        return True\n",
    "    except (subprocess.CalledProcessError, FileNotFoundError):\n",
    "        logger.debug(f\"{appname} is not installed or absent from path.\")\n",
    "        \n",
    "    return False\n",
    "\n",
    "apps = [ (\"ffmpeg\", \"ffmpeg -version\"),\n",
    "         (\"flac\", \"flac --version\") ]\n",
    "          \n",
    "for app_install, app_exec in apps:\n",
    "    if not check_installed(app_exec):\n",
    "        install_software(app_install)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we'll check `ffmpeg` has been installed.\n",
    "\n",
    "On Windows, this may not have been added to your path. If so, you can check your default install location using `winget --info`, and then add it to your path."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logger.info(\"Note that installed applications may not be immediately available after first installing.\\n\" \\\n",
    "            \"It may be necessary to relaunch the notebook environment.\")\n",
    "\n",
    "!ffmpeg -version"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Videos to Work With\n",
    "\n",
    "We start by defining a list of videos to test our application with, along with a function that takes a full YouTube URL and returns just the id portion.\n",
    "\n",
    "I’ve used these videos because…\n",
    "\n",
    "- The first is the fantastic [Burning Bridges](https://www.youtube.com/watch?v=udRAIF6MOm8) by Sigrid. The video has no embedded transcript.\n",
    "- The second is the beautiful song [I Believe](https://www.youtube.com/watch?v=CiTn4j7gVvY) by Melissa Hollick. It’s one of my favourite songs of all time. When I get a migraine, I turn off the lights, and listen to this to feel better! And for those who enjoy gaming, this song is the end titles to the amazing Wolfenstein: New Order game. This video has an embedded transcript.\n",
    "- Then we have a short [Jim Carey speech](https://www.youtube.com/watch?v=nLgHNu2N3JU), which gives us dialog without music or other ambient noise. It has an embedded transcript.\n",
    "- And finally, a [Ukrainian song](https://www.youtube.com/watch?v=d4N82wPpdg8) from Eurovision 2024, by Jerry Heil and Alyona Alyona. This gives us an opportunity to test translation. It also has an embedded transcript."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Videos to download\n",
    "urls = [\n",
    "    \"https://www.youtube.com/watch?v=udRAIF6MOm8\",  # Sigrid - Burning Bridges (English)\n",
    "    \"https://www.youtube.com/watch?v=CiTn4j7gVvY\",  # Melissa Hollick - I Believe (English)\n",
    "    \"https://www.youtube.com/watch?v=nLgHNu2N3JU\",  # Jim Carey - Motivational speech (English)\n",
    "    \"https://www.youtube.com/watch?v=d4N82wPpdg8\",  # Jerry Heil & Alyona Alyona - Teresa & Maria (Ukrainian)\n",
    "]\n",
    "\n",
    "def get_video_id(url: str) -> str:\n",
    "    \"\"\" Return the video ID, which is the part after 'v=' \"\"\"\n",
    "    return url.split(\"v=\")[-1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Downloading Videos and Extracting Audio\n",
    "\n",
    "Here I'll demonstrate a few different Python libraries for working with YouTube videos."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Option 1 - With PyTubeFix\n",
    "\n",
    "Here I'll use the [pytubefix](https://github.com/JuanBindez/pytubefix) library to download YouTube videos, and then to download mp3 audio-only streams as files.\n",
    "\n",
    "This library is a community-maintained fork of `pytube`. It was created to provide quick fixes for issues that the official pytube library faced, particularly when YouTube's updates break `pytube`.\n",
    "\n",
    "Pros:\n",
    "\n",
    "- The library is very easy to use.\n",
    "- We can work with video, audio, channels, playlists, and even search and filter.\n",
    "- It is [well documented](https://pytubefix.readthedocs.io/en/latest/).\n",
    "- It can be used from the command line, with its simple CLI.\n",
    "- It is VERY FAST!\n",
    "\n",
    "Cons:\n",
    "\n",
    "- Does not offer some of the more sophisticated capabilities that are offered by `yt_dlp`.\n",
    "- It does not appear to set mp3 headers correctly. The mp3s are actually encoded as mp4a. I don't think this is a problem, but it's worth bearing in mind!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "from pytubefix import YouTube\n",
    "from pytubefix.cli import on_progress\n",
    "\n",
    "output_locn = f\"{locations.output_dir}/pytubefix\"\n",
    "\n",
    "def process_yt_videos():\n",
    "    for i, url in enumerate(urls):\n",
    "        logger.info(f\"Downloads progress: {i+1}/{len(urls)}\")\n",
    "\n",
    "        try:\n",
    "            yt = YouTube(url, on_progress_callback=on_progress)\n",
    "            logger.info(f\"Getting: {yt.title}\")\n",
    "            video_stream = yt.streams.get_highest_resolution()\n",
    "            if not video_stream:\n",
    "                raise Exception(\"Stream not available.\")\n",
    "            \n",
    "            # YouTube resource titles may contain special characters which \n",
    "            # can't be used when saving the file. So we need to clean the filename.\n",
    "            cleaned = clean_filename(yt.title)\n",
    "            \n",
    "            video_output = f\"{output_locn}/{cleaned}.mp4\"\n",
    "            logger.info(f\"Downloading video {cleaned}.mp4 ...\")\n",
    "            video_stream.download(output_path=output_locn, filename=f\"{cleaned}.mp4\")\n",
    "        \n",
    "            logger.info(f\"Creating audio...\")\n",
    "            audio_stream = yt.streams.get_audio_only()\n",
    "            audio_stream.download(output_path=output_locn, filename=cleaned, mp3=True)\n",
    "            \n",
    "            logger.info(\"Done\")\n",
    "            \n",
    "        except Exception as e:        \n",
    "            logger.error(f\"Error processing URL '{url}'.\")\n",
    "            logger.debug(f\"The cause was: {e}\") \n",
    "            \n",
    "    logger.info(f\"Downloads finished. See files in {output_locn}.\")\n",
    "    \n",
    "process_yt_videos()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Option 2 - PyTubeFix and MoviePy\n",
    "\n",
    "Here I'm doing the same as before, but I'm extracting the audio using the Python [MoviePy](https://github.com/Zulko/moviepy) library. This is a powerful video and audio editing library. \n",
    "\n",
    "Pros:\n",
    "\n",
    "- We can extract audio as mp3 with correct headers.\n",
    "- It is [well documented](https://zulko.github.io/moviepy/).\n",
    "- It is powerful.\n",
    "\n",
    "Cons:\n",
    "\n",
    "- It is slower to extract the audio than using `pytubefix` alone."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "from pytubefix import YouTube\n",
    "from pytubefix.cli import on_progress\n",
    "from moviepy.editor import VideoFileClip\n",
    "\n",
    "output_locn = f\"{locations.output_dir}/pytubefix_with_moviepy\"\n",
    "\n",
    "def process_yt_videos():\n",
    "    for i, url in enumerate(urls):\n",
    "        logger.info(f\"Downloads progress: {i+1}/{len(urls)}\")\n",
    "\n",
    "        try:\n",
    "            yt = YouTube(url, on_progress_callback=on_progress)\n",
    "            logger.info(f\"Getting: {yt.title}\")\n",
    "            video_stream = yt.streams.get_highest_resolution()\n",
    "            if not video_stream:\n",
    "                raise Exception(\"Stream not available.\")\n",
    "            \n",
    "            # YouTube resource titles may contain special characters which \n",
    "            # can't be used when saving the file. So we need to clean the filename.\n",
    "            cleaned = clean_filename(yt.title)\n",
    "\n",
    "            video_output = f\"{output_locn}/{cleaned}.mp4\"\n",
    "            logger.info(f\"Downloading video {cleaned}.mp4 ...\")\n",
    "            video_stream.download(output_path=output_locn, filename=f\"{cleaned}\")\n",
    "        \n",
    "            logger.info(f\"Creating audio...\")\n",
    "            video_clip = VideoFileClip(video_output) # purely to give us access to methods\n",
    "            assert video_clip.audio is not None\n",
    "            video_clip.audio.write_audiofile(f\"{output_locn}/{cleaned}.mp3\")\n",
    "            video_clip.close()\n",
    "            \n",
    "            logger.info(\"Done\")\n",
    "            \n",
    "        except Exception as e:        \n",
    "            logger.error(f\"Error processing URL '{url}'.\")\n",
    "            logger.debug(f\"The cause was: {e}\") \n",
    "            \n",
    "    logger.info(f\"Downloads finished. See files in {output_locn}.\")\n",
    "    \n",
    "process_yt_videos()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Option 3 - With YT_DLP\n",
    "\n",
    "I wanted to try the other popular YouTube package: [yt-dlp](https://pypi.org/project/yt-dlp/). The [repo](https://github.com/yt-dlp/yt-dlp) is a fork of the now unmaintained `youtube-dl`. \n",
    "\n",
    "Pros:\n",
    "\n",
    "- It is very powerful, with far more options and features than `pytubefix`.\n",
    "- It can be installed as a standalone command-line executable, or as a pip-installable Python package.\n",
    "- Sets mp3 headers properly!\n",
    "- It has some powerful and network proxy settings. This can be useful if, for example, you are trying to download videos that are geo-restricted.\n",
    "\n",
    "Cons:\n",
    "\n",
    "- It is more complicated to use.\n",
    "- The documentation is complex and somewhat hard to understand. And there's no real Python-specific documentation.\n",
    "- It depends on having ffmpeg installed for some use cases.\n",
    "- It is significantly slower that `pytubefix` for performing video download and audio extraction.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import yt_dlp\n",
    "\n",
    "output_locn = f\"{locations.output_dir}/yt_dlp\"\n",
    "\n",
    "def process_yt_videos():\n",
    "    for i, url in enumerate(urls):\n",
    "        logger.info(f\"Downloads progress: {i+1}/{len(urls)}\")\n",
    "\n",
    "        try:\n",
    "            # Options for downloading the video\n",
    "            video_opts = {\n",
    "                'format': 'best',  # Download the best quality video\n",
    "                'outtmpl': f'{output_locn}/%(title)s.%(ext)s',  # Save video in output directory\n",
    "            }\n",
    "            \n",
    "            # Download the video\n",
    "            with yt_dlp.YoutubeDL(video_opts) as ydl:\n",
    "                logger.info(\"Downloading video...\")\n",
    "                ydl.download([url])\n",
    "            \n",
    "            # Options for extracting audio and saving as MP3\n",
    "            audio_opts = {\n",
    "                'format': 'bestaudio',  # Download the best quality audio\n",
    "                'outtmpl': f'{output_locn}/%(title)s.%(ext)s',  # Save audio in output directory\n",
    "                'postprocessors': [{\n",
    "                    'key': 'FFmpegExtractAudio',\n",
    "                    'preferredcodec': 'mp3',\n",
    "                }],\n",
    "            }\n",
    "            \n",
    "            # Download and extract audio\n",
    "            with yt_dlp.YoutubeDL(audio_opts) as ydl:\n",
    "                logger.info(\"Extracting and saving audio as MP3...\")\n",
    "                ydl.download([url])\n",
    "            \n",
    "        except Exception as e:        \n",
    "            logger.error(f\"Error processing URL '{url}'.\")\n",
    "            logger.debug(f\"The cause was: {e}\") \n",
    "            \n",
    "    logger.info(f\"Downloads finished. Check out files at {output_locn}.\")\n",
    "    \n",
    "process_yt_videos()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Conclusion\n",
    "\n",
    "If you:\n",
    "\n",
    "- Want to just download the videos and/or audio in the simplest and fastest way possible, then go with [Option 1](#option-1---with-pytubefix).\n",
    "- Want to download the videos and/or audio and then carry out some sort of manipulation or conversion of the media, go with [Option 2](#option-2---pytubefix-and-moviepy).\n",
    "- If you want out-of-the-box proxy configuration, e.g. to bypass geo-restrictions, then go with [Option 3](#option-3---with-yt_dlp)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transcribing Audio to Text"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Extracting Audio Using Python Speech Recognition\n",
    "\n",
    "The Python `speech_recognition` package has a number of built in `Recognizer` implementations. Here I'm using the [Google Web Speech API](https://wicg.github.io/speech-api/) `Recognizer`, which has its default API key hard coded into the Python `speech_recognition` library. It is free, but has some limitations. For example, it only allows a max of 60s segments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install --upgrade --no-cache-dir pydub SpeechRecognition ffmpeg-python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import speech_recognition as sr\n",
    "from pydub import AudioSegment\n",
    "import ffmpeg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def divide_chunks(sound, segment_size_secs=60):\n",
    "    \"\"\" Split audio file into 60s chunks \"\"\"\n",
    "    \n",
    "    segment_size_ms = segment_size_secs*1000\n",
    "    for start_idx in range(0, len(sound), segment_size_ms):\n",
    "        # Yield a chunk of audio data from start_idx to start_idx + segment_size_ms\n",
    "        yield sound[start_idx:start_idx + segment_size_ms]\n",
    "\n",
    "def transcribe_audio():\n",
    "    \"\"\" Use Speech Recognition API with Google Web Speech API\n",
    "    to convert audio dialog to text \"\"\"\n",
    "    recogniser = sr.Recognizer()        \n",
    "    for mp3_file in Path(output_locn).glob(f'*.mp3'):\n",
    "        transcribe_audio_file(recogniser, mp3_file)\n",
    "\n",
    "def transcribe_audio_file(recogniser, mp3_file, language=\"en-US\"):\n",
    "    logger.info(f\"Converting {mp3_file}...\")\n",
    "    try:\n",
    "        audio = AudioSegment.from_file(mp3_file)\n",
    "        # If AudioSegment is not working - e.g. due to broken mp3 headers - we\n",
    "        # can use ffmpeg as a workaround. However, it's a lot slower.\n",
    "        # ffmpeg.input(mp3_file).output(wav_file).run() # Convert with ffmpeg\n",
    "        # logger.info(f\"Successfully converted {mp3_file} to {wav_file}.\")\n",
    "        # audio = AudioSegment.from_wav(wav_file) # Read the audio\n",
    "\n",
    "        segments = list(divide_chunks(audio, segment_size_secs=60)) # split the wav into 60s segments     \n",
    "        transcription_extracts = {}\n",
    "        for index, chunk in enumerate(segments):\n",
    "            with io.BytesIO() as wav_io:\n",
    "                chunk.export(wav_io, format='wav')\n",
    "                wav_io.seek(0)  # Move to the start of the BytesIO object before reading from it\n",
    "                        \n",
    "                with sr.AudioFile(wav_io) as source:\n",
    "                    audio_data = recogniser.record(source)\n",
    "\n",
    "                try:\n",
    "                    extracted = recogniser.recognize_google(audio_data, language=language)\n",
    "                    logger.debug(f\"Chunk {index} extracted.\")\n",
    "                    transcription_extracts[index] = extracted\n",
    "                except sr.UnknownValueError:\n",
    "                        # Log the unknown value error and continue\n",
    "                    logger.warning(f\"Chunk {index}: Could not understand the audio. Maybe it was empty.\")\n",
    "            \n",
    "        logger.info(\"Extract:\")\n",
    "        for idx, extract in transcription_extracts.items():\n",
    "            logger.info(f\"{idx}: {extract}\")\n",
    "\n",
    "    except ffmpeg.Error as e:\n",
    "        logger.error(f\"FFmpeg failed to convert {mp3_file}: {str(e)}\")\n",
    "    except Exception as e:\n",
    "        logger.error(\"Unexpected error.\", exc_info=True)\n",
    "            \n",
    "transcribe_audio()\n",
    "logger.info(\"Done\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Results\n",
    "\n",
    "It's a bit flakey!  Sometimes it runs, but sometimes the API returns errors and fails to run.\n",
    "\n",
    "When the API does run...\n",
    "\n",
    "- It fails to transcribe the Ukrainian song. Not too surprising, since this API does not detect language automatically, and defaults to recognising English.\n",
    "- It does an amazing job with the Jim Carey speech.\n",
    "- It is partially successful when transcribing songs.\n",
    "\n",
    "### Conclusions\n",
    "\n",
    "It's not great!  It's pretty good if there's no background sound or ambient noise.  But it's pretty poor when working with songs. And it seems unreliable.\n",
    "\n",
    "### Transcribing Ukrainian\n",
    "\n",
    "Let's try and transcribe from the Ukrainian song:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def transcribe_ua():\n",
    "    recogniser = sr.Recognizer()\n",
    "    for mp3_file in Path(output_locn).glob(f'alyona*.mp3'):\n",
    "        transcribe_audio_file(recogniser, mp3_file, language=\"uk-UA\")\n",
    "        \n",
    "transcribe_ua()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Results\n",
    "\n",
    "Partial success.  But overall... Not great!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extract Existing Transcripts from Videos\n",
    "\n",
    "Now I'm going to use the [youtube-transcript-api](https://github.com/jdepoix/youtube-transcript-api) to extract existing transcripts from YouTube videos. Not only will it return the transcript, but it can also be used to translate those to translate those transcripts into other languages.  So now I can download my Ukrainian song, and see both the Ukrainian transcript and the English translation. This is pretty awesome!\n",
    "\n",
    "However, some videos do not contain transcripts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install --upgrade --no-cache-dir youtube_transcript_api"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import youtube_transcript_api as yt_api\n",
    "from pytubefix import YouTube\n",
    "from pytubefix.cli import on_progress\n",
    "\n",
    "def get_transcripts():\n",
    "    \"\"\" Extract existing transcript data from videos \"\"\"\n",
    "    for url in urls:\n",
    "        try: # Just so we can get the video title\n",
    "            yt = YouTube(url, on_progress_callback=on_progress)\n",
    "        except Exception as e:        \n",
    "            logger.error(f\"Error processing URL '{url}'.\")\n",
    "            logger.debug(f\"The cause was: {e}\") \n",
    "            continue\n",
    "        \n",
    "        logger.info(f\"Processing '{yt.title}'...\")\n",
    "        video_id = get_video_id(url)\n",
    "        \n",
    "        try:\n",
    "            # By default, we get a list of 1: only get the preferred language transcript\n",
    "            transcript_list = yt_api.YouTubeTranscriptApi.list_transcripts(video_id)\n",
    "        except Exception as e:\n",
    "            logger.error(f\"Unable to extract transcript for '{yt.title}'.\")\n",
    "            logger.debug(e)\n",
    "            continue\n",
    "        \n",
    "        # iterate over all available transcripts\n",
    "        for transcript in transcript_list:\n",
    "            # The Transcript object provides metadata properties. Here are some...\n",
    "            properties = {\n",
    "                \"video_id\": transcript.video_id,\n",
    "                \"language\": transcript.language,\n",
    "                \"language_code\": transcript.language_code,\n",
    "                \"is_generated\": transcript.is_generated,  # Whether it has been manually created or generated by YouTube\n",
    "                \"is_translatable\": transcript.is_translatable,  # Whether this transcript can be translated or not\n",
    "                \"translation_languages\": transcript.translation_languages,\n",
    "            }\n",
    "            \n",
    "            for prop, value in properties.items():\n",
    "                logger.info(f\"{prop}: {value}\")\n",
    "\n",
    "            # Fetch the actual transcript data\n",
    "            transcript_data = transcript.fetch() # returns a list of dicts\n",
    "            logger.info(f\"Raw transcript:\\n{transcript_data}\") \n",
    "            \n",
    "            processed_transcript = process_transcript(transcript_data)\n",
    "            logger.info(f\"Processed transcript:\\n{processed_transcript}\")\n",
    "            \n",
    "            # Translate to en if we can\n",
    "            if (transcript.language_code != \"en\" and \n",
    "                    transcript.is_translatable and \n",
    "                    any(lang['language_code'] == 'en' for lang in transcript.translation_languages)):\n",
    "                transcript_data = transcript.translate('en').fetch() # translate to en\n",
    "                processed_transcript = process_transcript(transcript_data)\n",
    "                logger.info(f\"Processed translated transcript:\\n{processed_transcript}\")\n",
    "\n",
    "def process_transcript(transcript_data):\n",
    "    \"\"\" Get all entries that are of type 'text' and NOT starting with [ \"\"\"\n",
    "    return \"\\n\".join([entry['text'] for entry in transcript_data \n",
    "                                     if entry['text'][0] != \"[\"])\n",
    "                \n",
    "get_transcripts()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How cool is this!?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Adding Google Cloud Smarts\n",
    "\n",
    "Let's integrate some Google Cloud Vertex AI smarts. Start by installing the **Google Cloud Vertex AI SDK for Python**. \n",
    "\n",
    "From [Introduction to the Vertex AI SDK for Python](https://cloud.google.com/vertex-ai/docs/python-sdk/use-vertex-ai-python-sdk#sdk-vs-client-library):\n",
    "\n",
    "When you install the Vertex AI SDK for Python (`google.cloud.aiplatform`), the Vertex AI Python client library (`google.cloud.aiplatform.gapic`) is also installed. The Vertex AI SDK and the Vertex AI Python client library provide similar functionality with different levels of granularity. The Vertex AI SDK operates at a higher level of abstraction than the client library and is suitable for most common data science workflows. If you need lower-level functionality, then use the Vertex AI Python client library."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Install Vertex AI SDK for Python and Vertex Generative AI SDK for Python\n",
    "%pip install --upgrade google-cloud-aiplatform google-generativeai"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.cloud import aiplatform # Google Cloud Vertex AI SDK for Python\n",
    "# import vertexai   # Google Cloud Vertex Generative AI SDK for Python\n",
    "import google.generativeai as genai  # Google Gemini API (GenAI)\n",
    "from vertexai.generative_models import GenerativeModel\n",
    "\n",
    "import sys\n",
    "from getpass import getpass\n",
    "\n",
    "# If we're running Google Colab, authenticate\n",
    "if \"google.colab\" in sys.modules:\n",
    "    from google.colab import auth\n",
    "    auth.authenticate_user()\n",
    "\n",
    "# Retrieve PROJECT_ID and other variables from any .env we can find\n",
    "try:\n",
    "    dc.get_envs_from_file()\n",
    "except ValueError as e:\n",
    "    logger.error(f\"Problem reading env file:\\n{e}\")\n",
    "\n",
    "env_vars = [\"PROJECT_ID\", \"REGION\"] # The vars we want to retrieve\n",
    "for env_var in env_vars:\n",
    "    if not os.getenv(env_var):\n",
    "        gcp_project_id = '' # @param {type: \"string\"}\n",
    "        # If not retrieved from .env we'll need to input the value\n",
    "        os.environ[env_var] = getpass(f\"Enter {env_var}: \")\n",
    "\n",
    "    # Set Python variable of the same name as the env var, e.g. PROJECT_ID\n",
    "    globals()[env_var] = os.environ[env_var]\n",
    "    val = globals()[env_var]\n",
    "    if env_var == \"PROJECT_ID\":\n",
    "        logger.info(f\"{env_var} retrieved: {val[-3:]}\")\n",
    "    else:\n",
    "        logger.info(f\"{env_var} retrieved: {val}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Only run the next cell if you want to manually clear the environment variables and then input new values. In this scenario, you'll also want to comment out any variables in your .env file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Optionally run this if we want to clear env vars\n",
    "for env_var in env_vars:\n",
    "    if env_var in os.environ:\n",
    "        del os.environ[env_var]\n",
    "        logger.info(f\"Cleared environment variable: {env_var}\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}