src/notebooks/youtube-demos-with-google-ai.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dazbo's YouTube and Video Demos - with Google AI\n",
    "\n",
    "## Overview\n",
    "\n",
    "Welcome to notebook #2 in this tutorial guide. The overall series covers:\n",
    "\n",
    "1. Starting with an idea. Here, the goal is to work with vidoes, which could be on YouTube. We want to be able to download videos, extract audio, transcribe, translate, and potentially summarise the content.\n",
    "1. Experimenting on this idea, using a Jupyter notebook, with Python.\n",
    "1. Trying a few libraries and a couple of classical AI models. \n",
    "1. Building a solution that makes use of Google Gemini multiomodal GenAI.\n",
    "1. Turning the notebook into a web application, using Streamlit.\n",
    "1. Packaging the application as a container.\n",
    "1. Finally, hosting the application on Google Cloud's serverless Cloud Run service.\n",
    "\n",
    "The code and notebooks are intended to be supplemented by these walkthroughs:\n",
    "\n",
    "1. [Downloading YouTube Videos, Extracting Audio, and Generating Transcripts with Python and Jupyter Notebooks](https://medium.com/python-in-plain-english/downloading-youtube-videos-extracting-audio-and-generating-transcripts-with-python-and-jupyter-c3068f82bbe0)\n",
    "1. [YouTube Video Downloader with Generative AI: Run Anywhere, Transcribe and Translate](https://python.plainenglish.io/youtube-video-downloader-with-generative-ai-and-python-run-anywhere-transcribe-and-translate-dec2e593dd58)\n",
    "1. [Building and Running an AI YouTube and Video Processing as a Python Streamlit Web Application, on Serverless Google Cloud Run](https://medium.com/google-cloud/running-ai-youtube-and-video-processing-as-a-python-streamlit-web-application-and-hosting-on-748aae8e54b4)\n",
    "\n",
    "Additionally, you will find supporting READMEs and scripts in my [GitHub repo](https://github.com/derailed-dash/youtube-and-video).\n",
    "\n",
    "In the previous notebook I demonstrated:\n",
    "\n",
    "- Multiple methods for downloading videos and extracting audio\n",
    "- How to transcribe audio to text using a free speech-to-text API\n",
    "- How to extract existing transcripts and translate to different languages\n",
    "\n",
    "## This Notebook\n",
    "\n",
    "In this part we'll strip out parts of the first notebook we don't need, and add some smarts using Google technology:\n",
    "\n",
    "- Using the Google Video Intelligence API to provide more reliable and more accurate trancription.\n",
    "- Using Google Gemini Generative AI to transcribe, translate and summarise video content.\n",
    "- How to build your Jupyter notebook so it can run locally, in Google Colab, or in Google Vertex AI Workbench.\n",
    "\n",
    "## How to Launch and Run this Notebook\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",
    "- You can run the notebook locally, or with any of the options shown below.\n",
    "- Check out further guidance - including tips on how to run the notebook - in the project's `README.md`.\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",
    "\n",
    "**When running 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"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<table width=\"800px\" style=\"border-collapse: collapse;\">\n",
    "  <td style=\"text-align: center\">\n",
    "    <a href=\"https://colab.research.google.com/github/derailed-dash/youtube-and-video/blob/main/src/notebooks/youtube-demos-with-google-ai.ipynb\">\n",
    "      <img src=\"https://cloud.google.com/ml-engine/images/colab-logo-32px.png\" alt=\"Google Colaboratory logo\"><br>Run in Colab\n",
    "    </a>\n",
    "  </td>\n",
    "  <td style=\"text-align: center\">\n",
    "    <a href=\"https://github.com/derailed-dash/youtube-and-video/blob/main/src/notebooks/youtube-demos-with-google-ai.ipynb\">\n",
    "      <img src=\"https://cloud.google.com/ml-engine/images/github-logo-32px.png\" alt=\"GitHub logo\"><br>View on GitHub\n",
    "    </a>\n",
    "  </td>\n",
    "  <td style=\"text-align: center\">\n",
    "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://raw.githubusercontent.com/derailed-dash/youtube-and-video/blob/main/src/notebooks/youtube-demos-with-google-ai.ipynb\">\n",
    "      <img src=\"https://lh3.googleusercontent.com/UiNooY4LUgW_oTvpsNhPpQzsstV5W8F7rYgxgGBD85cWJoLmrOzhVs_ksK_vgx40SHs7jCqkTkCk=e14-rj-sc0xffffff-h130-w32\" alt=\"Vertex AI logo\"><br> Open in Vertex AI Workbench\n",
    "    </a>\n",
    "  </td>\n",
    "</table>"
   ]
  },
  {
   "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 \\\n",
    "                                      dazbo-commons \\\n",
    "                                      pytubefix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import display, Markdown\n",
    "\n",
    "import io\n",
    "import logging\n",
    "import os\n",
    "import re\n",
    "import sys\n",
    "from pathlib import Path\n",
    "from dataclasses import dataclass\n",
    "from dotenv import load_dotenv\n",
    "import dazbo_commons as dc"
   ]
  },
  {
   "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"
   ]
  },
  {
   "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",
    "    cleaned_name = re.sub(pattern, '_', filename).replace(\"_ _\", \"_\").replace(\"__\", \"_\")\n",
    "    return  cleaned_name"
   ]
  },
  {
   "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",
    "    \"https://www.youtube.com/shorts/41iWg91yFv0\",   # Rick Astley short\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]\n",
    "\n",
    "output_locn = f\"{locations.output_dir}/pytubefix\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Install Additional Packages You May Need\n",
    "\n",
    "**Bear in mind that `nodejs` is required by the `pytubefix` library, to prevent this appliction being detected as a bot.**\n",
    "\n",
    "You can run the cell below, but it may not work on your environment. 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",
    "    <tr>\n",
    "    <td>nodejs</td>\n",
    "    <td>The pytubefix library can automatically create YouTube PO tokens, but this relies on nodejs being installed.</td>\n",
    "    <td>Linux: <code>sudo apt install nodejs</code><br>Windows: <code>winget install node.js</code></td>\n",
    "  </tr>\n",
    "</table>\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",
    "         (\"nodejs\" , \"node --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": [
    "## Downloading Videos and Extracting Audio\n",
    "\n",
    "Let's 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`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "from pytubefix import YouTube\n",
    "from pytubefix.cli import on_progress\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",
    "            # YouTube now requires the PO token to be passed in the requet\n",
    "            # The library will automatically generate a PO token, \n",
    "            # but nodejs must be installed to do so.\n",
    "            yt = YouTube(url, on_progress_callback=on_progress, client=\"WEB\")\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",
    "            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=f\"{cleaned}.mp3\")\n",
    "            \n",
    "            logger.info(\"Done\")\n",
    "            \n",
    "        except Exception as e:        \n",
    "            logger.error(f\"Error processing URL '{url}'.\")\n",
    "            logger.error(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": [
    "## Extract Existing Transcripts from Videos\n",
    "\n",
    "Here I'm using 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!"
   ]
  },
  {
   "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, client=\"WEB\")\n",
    "        except Exception as e:        \n",
    "            logger.error(f\"Error processing URL '{url}'.\")\n",
    "            logger.error(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.error(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!?\n",
    "\n",
    "Alas, some videos don't have pre-existing transcripts. So let's see if we can improve our transcription capability using some Google Cloud AI..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Adding Google Cloud Intelligence\n",
    "\n",
    "We're going to leverage Google Cloud APIs. In order to leverage these Google services, you'll need to have first created a Google Cloud project. So, if you haven't already, create your project, attach it to a billing account, and then come back.\n",
    "\n",
    "### How to Consume Google Cloud Services from your Notebook\n",
    "\n",
    "Then, in order to give your notebook access to the Google Cloud APIs, you broadly have three options:\n",
    "\n",
    "1. You can build and run your notebook locally.\n",
    "1. You can build and run your notebook in Google Colab.\n",
    "1. You can build and run your notebook in the Google Vertex AI Workbench environment.\n",
    "\n",
    "Let's look at the options..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Local Notebook\n",
    "\n",
    "For local development - e.g. a Jupyter notebook running in your own machine - you will need to:\n",
    "\n",
    "1. Have the Google Cloud `gcloud CLI` installed. See instructions [here](https://cloud.google.com/sdk/docs/install).\n",
    "2. Authenticate to `gcloud`, so we can externally run `gcloud` commands from the notebook.\n",
    "3. Set your quota project, and set your Application Default Credentials (ADC) by authenticating to your gcloud environment.\n",
    "\n",
    "```bash\n",
    "# From your terminal...\n",
    "export PROJECT_ID = <your project>\n",
    "gcloud auth login # authenticate to gcloud\n",
    "gcloud auth application-default login # set up ADC\n",
    "gcloud auth application-default set-quota-project $PROJECT_ID\n",
    "gcloud config set project $PROJECT_ID\n",
    "```\n",
    "\n",
    "4. Use [Application Default Credentials](https://cloud.google.com/docs/authentication/application-default-credentials) from within the notebook.\n",
    "\n",
    "```python\n",
    "from google.auth import default\n",
    "credentials, _ = default()\n",
    "\n",
    "PROJECT = !gcloud config get-value project\n",
    "PROJECT_ID = PROJECT[0]\n",
    "REGION = \"europe-west2\"\n",
    "\n",
    "# Now use whatever Google services...\n",
    "import vertexai\n",
    "vertexai.init(project=PROJECT_ID, location=REGION)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Google Colab\n",
    "\n",
    "This is a great way to create and run Jupyter notebooks in the Cloud, and it makes them super-easy to share.\n",
    "\n",
    "The great thing about this approach is that Colab provides native integration to authenticate your user account and provide your Google project details to the Colab environment.\n",
    "\n",
    "Notes:\n",
    "\n",
    "- No don't need to install Google `gcloud CLI` locally. It is pre-installed in the environment.\n",
    "- You can share notebooks using Google Drive, with Drive-based access control.\n",
    "- There are limitations for notebook size, and for notebook runtime instance size.\n",
    "\n",
    "Check out [this guide](https://github.com/GoogleCloudPlatform/devrel-demos/blob/main/other/colab/Using%20Google%20Cloud%20from%20Colab.ipynb).\n",
    "\n",
    "For example, in your notebook:\n",
    "\n",
    "```python\n",
    "import sys\n",
    "\n",
    "# First, set PROJECT_ID and REGION variables from environment variables or secrets\n",
    "# Then...\n",
    "PROJECT_ID = str(os.environ.get(\"PROJECT_ID\"))\n",
    "REGION = str(os.environ.get(\"REGION\"))\n",
    "!gcloud config set project $PROJECT_ID\n",
    "\n",
    "# Check if we're running in the Colab environment, and if so\n",
    "# Use Colab native authentication to Google Cloud\n",
    "if \"google.colab\" in sys.modules:\n",
    "    from google.colab import auth\n",
    "    auth.authenticate_user()\n",
    "    !gcloud auth application-default login # set ADC\n",
    "\n",
    "credentials, _ = google.auth.default()\n",
    "\n",
    "# Now use your Google services...\n",
    "import vertexai\n",
    "vertexai.init(\n",
    "    project=PROJECT_ID,\n",
    "    location=REGION,\n",
    "    credentials=credentials\n",
    ")\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Vertex AI Workbench\n",
    "\n",
    "[Vertex AI Workbench](https://cloud.google.com/vertex-ai/docs/workbench/introduction) is Google's most powerful managed enterprise Jupyter notebook hosting service. Is is fully-integrated with the Google Cloud and Vertex AI ecosystem.\n",
    "\n",
    "Notes:\n",
    "\n",
    "- The gcloud CLI is pre-installed in the environment.\n",
    "- The JupyterLab environment is pre-installed.\n",
    "- Access control and sharing is managed by Google Cloud IAM, rather than Google Drive.\n",
    "- Because it is natively integrated with the Google Cloud environment, you don't need to provide any credentials or authenticate. You just need to provide Google project ID and region to any services that require this information. E.g.\n",
    "\n",
    "```python\n",
    "# First, set PROJECT_ID and REGION variables from environment variables or secrets\n",
    "# Then...\n",
    "PROJECT_ID = str(os.environ.get(\"PROJECT_ID\"))\n",
    "REGION = str(os.environ.get(\"REGION\"))\n",
    "\n",
    "# Go ahead and use your Google services...\n",
    "import vertexai\n",
    "vertexai.init(project=PROJECT_ID, location=REGION)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Retrieve Environment Variables\n",
    "\n",
    "I put my environment variables in a `.env` file at the top level of my project. It looks like this...\n",
    "\n",
    "```bash\n",
    "PYTHONPATH=src;src/notebooks\n",
    "PROJECT_ID=my-project-id\n",
    "REGION=my-region\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "from getpass import getpass\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",
    "if not (PROJECT_ID := os.getenv(\"PROJECT_ID\")):\n",
    "    PROJECT_ID = input(f\"Enter PROJECT_ID: \")\n",
    "\n",
    "if not (REGION := os.getenv(\"REGION\")):\n",
    "    REGION = input(f\"Enter REGION: \")\n",
    "    \n",
    "if not (GEMINI_API_KEY := os.getenv(\"GEMINI_API_KEY\")):\n",
    "    GEMINI_API_KEY = getpass(f\"Enter Gemini API Key:\")\n",
    "\n",
    "logger.info(f\"{PROJECT_ID=}\")\n",
    "logger.info(f\"{REGION=}\")\n",
    "logger.info(f\"GEMINI_API_KEY={GEMINI_API_KEY[:5]}...{GEMINI_API_KEY[-5:]}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Clear Environment Variables\n",
    "\n",
    "**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": [
    "# Only run this if we want to clear env vars\n",
    "del os.environ[\"PROJECT_ID\"]\n",
    "del os.environ[\"REGION\"]\n",
    "del os.environ[\"GEMINI_API_KEY\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### A Notebook We Can Run in ANY Environment\n",
    "\n",
    "Let's engineer the notebook to be agnostic of where it is hosted."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install --upgrade google-auth \\\n",
    "                       google-auth-oauthlib \\\n",
    "                       google-auth-httplib2 \\\n",
    "                       google-cloud-storage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.auth import default\n",
    "from google.auth.exceptions import DefaultCredentialsError\n",
    "from google.cloud import storage # Enable \n",
    "\n",
    "# If we're running Google Colab, authenticate\n",
    "if \"google.colab\" in sys.modules:\n",
    "    from google.colab import auth # type: ignore\n",
    "    auth.authenticate_user()\n",
    "    !gcloud auth application-default login # set ADC\n",
    "else: # If you're not running in a local dev CLI, from your terminal...\n",
    "    # export PROJECT_ID=<your project ID>\n",
    "    # gcloud auth login\n",
    "    # gcloud auth application-default login\n",
    "    # gcloud auth application-default set-quota-project $PROJECT_ID\n",
    "    pass\n",
    "\n",
    "try:\n",
    "    credentials, _ = default() # Retrieve ADC\n",
    "    !gcloud config set project $PROJECT_ID\n",
    "except DefaultCredentialsError as e:\n",
    "    logger.error(e)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Video Transcription Using the Video Intelligence API\n",
    "\n",
    "Recall that in the previous notebook, I tried to perform audio trascription using the Python `speech_recognition` package, and the built-in [Google Web Speech API](https://wicg.github.io/speech-api/) `Recognizer`. It wasn't great!\n",
    "\n",
    "So now let's use Google's [Video Intelligence API](https://cloud.google.com/video-intelligence/docs) to perform transcription..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install --upgrade google-cloud-videointelligence"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.cloud import videointelligence\n",
    "\n",
    "DEFAULT_MIN_CONFIDENCE=0.6\n",
    "\n",
    "def transcribe_video(video, minimum_confidence=DEFAULT_MIN_CONFIDENCE, src_language=\"en-US\"):\n",
    "    logger.info(f\"Processing {video.name}...\")\n",
    "    \n",
    "    video_client = videointelligence.VideoIntelligenceServiceClient()\n",
    "\n",
    "    # This API can do loads of things. Here I'll tell it to do speech transcription.\n",
    "    features = [videointelligence.Feature.SPEECH_TRANSCRIPTION]\n",
    "    config = videointelligence.SpeechTranscriptionConfig(\n",
    "        language_code=src_language, enable_automatic_punctuation=True\n",
    "    )\n",
    "    video_context = videointelligence.VideoContext(speech_transcription_config=config)\n",
    "\n",
    "    try:\n",
    "        with io.open(video, \"rb\") as file:\n",
    "            input_content = file.read()\n",
    "            \n",
    "        operation = video_client.annotate_video(\n",
    "            request={\n",
    "                \"features\": features,\n",
    "                \"input_content\": input_content, # for lcoal\n",
    "                # \"input_uri\": path, # for objects in GCS\n",
    "                \"video_context\": video_context,\n",
    "            }\n",
    "        )\n",
    "\n",
    "        result = operation.result(timeout=600)\n",
    "\n",
    "        # There is only one annotation_result per video.\n",
    "        annotation_results = result.annotation_results[0]\n",
    "        complete_transcript = \"\"\n",
    "        for speech_transcription in annotation_results.speech_transcriptions:\n",
    "            # Each SpeechTranscription can contain multiple alternatives.\n",
    "            # Each alternative is a different possible transcription and has its own confidence score.\n",
    "            # They are ordered in terms of accuracy. So we really only need the first.\n",
    "            part = speech_transcription.alternatives[0]\n",
    "            if part.confidence < minimum_confidence:\n",
    "                logger.debug(f\"Ignoring transcript alternative with confidence of {part.confidence}.\")\n",
    "                continue\n",
    "                \n",
    "            logger.debug(\"Part transcript: {}\".format(part.transcript))\n",
    "            logger.debug(\"Part confidence: {}\\n\".format(part.confidence))\n",
    "            complete_transcript += part.transcript.strip() + \"\\n\"\n",
    "                \n",
    "        return complete_transcript.strip()\n",
    "    except Exception as e:\n",
    "       logger.error(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It will take a few minutes to process these videos..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logger.debug(f\"Looking for videos in {output_locn}...\")\n",
    "for video in Path(output_locn).glob(f'*.mp4'):\n",
    "    # Fortunately, this API natively supports mp4 without any conversion\n",
    "    transcript = transcribe_video(video)\n",
    "    if transcript:\n",
    "        logger.info(f\"Transcript:\\n{transcript}\")\n",
    "    else:\n",
    "        logger.warning(f\"Unable to retrieve a transcript with confidence.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Conclusions\n",
    "\n",
    "It's pretty good!\n",
    "\n",
    "- It's reliable and doesn't give random _pipe_ errors.\n",
    "- It transcribes with higher accuracy than the Python `speech_recognition` package using the Google Web Speech API `Recognizer`. In particular, it does a much better job with the \"I Believe\" track.\n",
    "- We don't need to split the video into chunks.\n",
    "- The API provides an estimate of transcription accuracy. And we can use this to filter out transcriptions that we don't want to keep.\n",
    "\n",
    "Some minor issues:\n",
    "\n",
    "- It takes a long time to process each video.\n",
    "- It doesn't automatically detect the source language. So it fails with the Ukrainian music video.\n",
    "\n",
    "Let's now try and translate the Ukrainian song by passing in the language code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "video = next(Path(output_locn).glob(f'alyona*.mp4'), None)\n",
    "\n",
    "transcript = transcribe_video(video, src_language=\"uk-UA\")\n",
    "if transcript:\n",
    "    logger.info(f\"Transcript:\\n{transcript}\")\n",
    "else:\n",
    "    logger.warning(f\"Unable to retrieve a transcript with confidence.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Not great. Oh well!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### What Next?\n",
    "\n",
    "Now I'm going to:\n",
    "\n",
    "- See if we can use a Google Gemini generative AI model to extract, transcribe and translate for us.\n",
    "- Use Gemini to do some summarising."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Vertex Gemini Generative AI\n",
    "\n",
    "Let's integrate some generative AI!\n",
    "\n",
    "There are a couple of APIs we can use to do this.  Here are some pointers...\n",
    "\n",
    "| API | Requires | Use Cases |\n",
    "|-----|----------|-----------|\n",
    "| [Google Vertex AI SDK](https://cloud.google.com/vertex-ai/docs/python-sdk/use-vertex-ai-python-sdk) | A Google project | Working with Google Vertex services, including Gemini models. |\n",
    "| [Google Gemini API](https://ai.google.dev/api?lang=python) | [An API key](https://medium.com/r/?url=https%3A%2F%2Faistudio.google.com%2Fapp%2Fapikey). You don't need a Google Cloud project! | Prototyping and experimentation. |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Working with Google Cloud Storage (GCS)\n",
    "\n",
    "When working with many Google Vertex AI APIs it can be convenient to store files i GCS. So it will be useful to create a GCS bucket in our project (if we haven't already done so). Of course, you can do that in the Cloud Console, but here I'll show you how to do that from the notebook.\n",
    "\n",
    "Before you proceed, don't forget to enable the Cloud Storage API (`storage.googleapis.com`)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create / Check the Bucket\n",
    "\n",
    "If we need a bucket for any of our application, we can create one using the code here. Let's set the lifecycle policy so that files are deleted automatically after 3 days."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This cell is to create a policy for our bucket when we create it. You will only ever need to run this once."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile lifecycle-policy.json\n",
    "{\n",
    "  \"rule\": [\n",
    "    {\n",
    "      \"action\": {\"type\": \"Delete\"},\n",
    "      \"condition\": {\"age\": 3}\n",
    "    }\n",
    "  ]\n",
    "}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "BUCKET_NAME = f\"{PROJECT_ID}-bucket\"\n",
    "BUCKET_URI = f\"gs://{BUCKET_NAME}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "bucket_check = !gcloud storage ls $BUCKET_URI # type: ignore\n",
    "bucket_exists = True\n",
    "for line in bucket_check:\n",
    "    if \"404\" in line:\n",
    "        bucket_exists = False\n",
    "        break\n",
    "        \n",
    "if not bucket_exists:\n",
    "    logger.info(f\"Creating bucket {BUCKET_URI}\")\n",
    "    ! gcloud storage buckets create {BUCKET_URI} --location={REGION}\n",
    "    \n",
    "    # Set bucket so files are automatically deleted\n",
    "    logger.info(f\"Setting lifecycle policy for {BUCKET_URI}\")\n",
    "    ! gcloud storage buckets update {BUCKET_URI} --lifecycle-file=lifecycle-policy.json\n",
    "else:\n",
    "    logger.info(f\"{BUCKET_URI} already exists.\")\n",
    "\n",
    "logger.info(f\"{BUCKET_NAME=}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Utility Function to Upload Files to GCS\n",
    "\n",
    "Now let's create a utility function for uploading files from local storage to GCS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.cloud import storage\n",
    "\n",
    "def upload_to_gcs(bucket_name: str, src_file_name, dest_name):\n",
    "    \"\"\" Upload a file to a GCS bucket. \"\"\"\n",
    "    try:\n",
    "        storage_client = storage.Client()\n",
    "        bucket = storage_client.bucket(bucket_name)\n",
    "        \n",
    "        # Destination blob name\n",
    "        blob_name = dest_name \n",
    "        blob = bucket.blob(blob_name)\n",
    "        logger.info(f\"Uploading {src_file_name} to gs://{bucket.name}/{blob_name}\")\n",
    "        blob.upload_from_filename(src_file_name)\n",
    "        # we could also upload a BytesIO e.g.\n",
    "        # blobl.upload_from_file(src_file)\n",
    "        \n",
    "        return f\"gs://{bucket}/{blob_name}\" # Return the full GCS URI\n",
    "    except Exception as e:\n",
    "        logger.exception(f\"Error uploading {src_file_name} to GCS: {e}\")\n",
    "        return\n",
    "\n",
    "def get_gcs_uris(bucket_name:str, glob:str=None) -> list[str]:\n",
    "    \"\"\"\n",
    "    Retrieves bucket URIs for files in a specified folder, \n",
    "    optionally matching a wildcard glob.\n",
    "\n",
    "    Args:\n",
    "        bucket_name: The name of the GCS bucket.\n",
    "        glob: A wildcard string to match filenames (e.g., '*.mp4').\n",
    "\n",
    "    Returns:\n",
    "        A list of bucket URIs for matching files.\n",
    "    \"\"\"\n",
    "    logger.debug(f\"Listing blobs in {bucket_name}/{glob if glob else ''}...\")\n",
    "    client = storage.Client() # Uses ADC so we don't have to pass in credentials\n",
    "\n",
    "    blobs = client.list_blobs(bucket_or_name=bucket_name, \n",
    "                              match_glob=glob)\n",
    "    return [f\"gs://{bucket_name}/{blob.name}\" for blob in blobs]\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Upload our Videos to GCS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "folder = \"video_files\"\n",
    "logger.info(f\"Upload .mp4 files to a bucket folder called {folder}...\")\n",
    "for file in Path(output_locn).glob(f'*.mp4'):\n",
    "    response = upload_to_gcs(bucket=BUCKET_NAME, \n",
    "                             src_file_name=file,\n",
    "                             dest_name=f\"{folder}/{file.name}\")\n",
    "    \n",
    "folder = \"audio_files\"\n",
    "logger.info(f\"Upload .mp3/.m4a files to a bucket folder called {folder}...\")\n",
    "for ext in ('*.mp3', '*.m4a'):\n",
    "    for file in Path(output_locn).glob(ext):\n",
    "        response = upload_to_gcs(bucket=BUCKET_NAME, \n",
    "                                src_file_name=file,\n",
    "                                dest_name=f\"{folder}/{file.name}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Test retrieval\n",
    "matched_files = get_gcs_uris(BUCKET_NAME)\n",
    "for file_uri in matched_files:\n",
    "    logger.info(file_uri)\n",
    "    \n",
    "sigrid_video_uri = get_gcs_uris(BUCKET_NAME, glob=\"video_files/Sigrid*.mp4\")[0]\n",
    "logger.info(f\"{sigrid_video_uri=}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using the Vertex AI SDK\n",
    "\n",
    "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\n",
    "%pip install --upgrade google-cloud-aiplatform \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point, if this is the first time installing the Vertex AI SDK in your notebook, you might want to **restart your kernel / runtime.**\n",
    "\n",
    "Now let's load a model. Gemini 1.5 Flash (`gemini-1.5-flash`) is a multimodal model that supports multimodal prompts. You can include text, image(s), and video in your prompt requests and get text or code responses."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import vertexai # Google Cloud Vertex Generative AI SDK for Python\n",
    "from vertexai.generative_models import GenerationConfig, GenerativeModel, Part\n",
    "from google.api_core.exceptions import ResourceExhausted"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "vertexai.init(project=PROJECT_ID, location=REGION)\n",
    "model = GenerativeModel(\"gemini-1.5-flash-002\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Test with a Simple Prompt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = model.generate_content(\"Write a story about a silly black and white cat called Mycroft\")\n",
    "display(Markdown(f\"### A Story About Mycroft\\n\\n{response.text}\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Process a Video\n",
    "\n",
    "Let's upload a video to the model and ask it some questions. We'll also ask it to transcribe.  This might take a minute or two, so be patient!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The pro-vision model gave me nonsense lyrics\n",
    "# model = GenerativeModel(\"gemini-1.0-pro-vision\")\n",
    "\n",
    "try:\n",
    "    # See if already defined\n",
    "    sigrid_video_uri # type: ignore\n",
    "except NameError:\n",
    "    sigrid_video_uri = get_gcs_uris(BUCKET_NAME, glob=\"video_files/Sigrid*.mp4\")[0]\n",
    "    \n",
    "logger.info(f\"{sigrid_video_uri=}\")\n",
    "\n",
    "video = Part.from_uri(\n",
    "    uri=sigrid_video_uri,\n",
    "    mime_type=\"video/mp4\",\n",
    ")\n",
    "\n",
    "# But we could also do this...\n",
    "# video = Part.from_data(data=video_bytes_io, mime_type=\"video/mp4\")\n",
    "\n",
    "prompt = \"\"\"\n",
    "What is shown in this video?\n",
    "Who is the artist?\n",
    "What are the lyrics?\n",
    "Can you summarise them?\n",
    "What is the meaning of the lyrics?\n",
    "\"\"\"\n",
    "\n",
    "contents = [prompt, video]\n",
    "\n",
    "try:\n",
    "    logger.info(\"Asking the model. Please wait...\")\n",
    "    display(Markdown(f\"### Prompt:\\n\\n{prompt}\"))\n",
    "    response = model.generate_content(contents, stream=False)    \n",
    "\n",
    "    display(Markdown(f\"### Response:\\n\"))\n",
    "    display(Markdown(f\"{response.text}\"))\n",
    "except ResourceExhausted as e:\n",
    "    logger.warning(f\"Resource exhausted: {e}\")\n",
    "except Exception as e: # Handle other exceptions separately\n",
    "    logger.exception(f\"An unexpected error occurred: {e}\")   \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Results\n",
    "\n",
    "Wow, that's pretty amazing! This seems like the perfect solution!\n",
    "\n",
    "#### Will It Be Faster With Just Audio Files?\n",
    "\n",
    "This works, but it is a little slow, and potentially costly. I wonder if we can do the same, but using just our audio files instead?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    # See if already defined\n",
    "    sigrid_audio_uri # type: ignore \n",
    "except NameError:\n",
    "    sigrid_audio_uri = get_gcs_uris(BUCKET_NAME, glob=\"audio_files/Sigrid*.m4a\")[0]\n",
    "    \n",
    "logger.info(f\"{sigrid_audio_uri=}\")\n",
    "\n",
    "audio = Part.from_uri(\n",
    "    uri=sigrid_audio_uri,\n",
    "    mime_type=\"audio/mpeg\",\n",
    ")\n",
    "\n",
    "# But we could also do this...\n",
    "# audio = Part.from_data(data=audio_bytes_io, mime_type=\"audio/mpeg\")\n",
    "\n",
    "prompt = \"\"\"\n",
    "In this audio file, please tell me:\n",
    "- What are the lyrics?\n",
    "- Can you summarise them?\n",
    "- What is the meaning of the lyrics?\n",
    "\"\"\"\n",
    "\n",
    "contents = [prompt, audio] # multimodal input\n",
    "\n",
    "try:\n",
    "    logger.info(\"Asking the model. Please wait...\")\n",
    "    display(Markdown(f\"### Prompt:\\n\\n{prompt}\"))\n",
    "    response = model.generate_content(contents, stream=False)    \n",
    "\n",
    "    display(Markdown(f\"### Response:\\n\"))\n",
    "    display(Markdown(f\"{response.text}\"))\n",
    "except ResourceExhausted as e:\n",
    "    logger.warning(f\"Resource exhausted: {e}\")\n",
    "except Exception as e: # Handle other exceptions separately\n",
    "    logger.exception(f\"An unexpected error occurred: {e}\")   \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Results\n",
    "\n",
    "Amazing! Super-easy to modify the code, and it runs in a fraction of the time!\n",
    "\n",
    "#### Ukranian Translation?\n",
    "\n",
    "One last test... Can it do the same with my Ukrainian song?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    # See if already defined\n",
    "    ukrainian_audio_uri # type: ignore \n",
    "except NameError:\n",
    "    ukrainian_audio_uri = get_gcs_uris(BUCKET_NAME, glob=\"audio_files/alyona*.m4a\")[0]\n",
    "    \n",
    "logger.info(f\"{ukrainian_audio_uri=}\")\n",
    "\n",
    "audio = Part.from_uri(\n",
    "    uri=ukrainian_audio_uri,\n",
    "    mime_type=\"audio/mpeg\",\n",
    ")\n",
    "\n",
    "prompt = \"\"\"\n",
    "In this audio file, please tell me:\n",
    "- What languages are being sung?\n",
    "- What are the lyrics? Please show me in the native language, and translated to English.\n",
    "- What is the meaning of the lyrics?\n",
    "\"\"\"\n",
    "\n",
    "contents = [prompt, audio] # multimodal input\n",
    "\n",
    "try:\n",
    "    logger.info(\"Asking the model. Please wait...\")\n",
    "    display(Markdown(f\"### Prompt:\\n\\n{prompt}\"))\n",
    "    response = model.generate_content(contents, stream=False)    \n",
    "\n",
    "    display(Markdown(f\"### Response:\\n\"))\n",
    "    display(Markdown(f\"{response.text}\"))\n",
    "except ResourceExhausted as e:\n",
    "    logger.warning(f\"Resource exhausted: {e}\")\n",
    "except Exception as e: # Handle other exceptions separately\n",
    "    logger.exception(f\"An unexpected error occurred: {e}\")   \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is amazing! It detects the correct language, transcribes it, and translates it.  And it does a MUCH BETTER JOB than the Video Intelligence API!!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## But Can I Do All This Without a Google Cloud Project?\n",
    "\n",
    "### Yes - Use the Gemini API\n",
    "\n",
    "All you need is an API key!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Install the Gemini API SDK\n",
    "%pip install --upgrade google-generativeai"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Test a Simple Prompt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import google.generativeai as genai # Use the Gemini API\n",
    "\n",
    "genai.configure(api_key=GEMINI_API_KEY)\n",
    "model = genai.GenerativeModel(\"gemini-1.5-flash-002\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = model.generate_content(\"Write a story about a silly black and white cat called Mycroft\")\n",
    "display(Markdown(f\"### A Story About Mycroft\\n\\n{response.text}\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Test with a Video"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "video_file = next(Path(output_locn).glob(f'Sigrid*.mp4'), None)\n",
    "uploaded = genai.upload_file(video_file)\n",
    "logger.info(f\"{uploaded=}\")\n",
    "\n",
    "# Videos need to be processed before you can use them.\n",
    "while uploaded.state.name == \"PROCESSING\":\n",
    "    logger.info(\"Processing video...\")\n",
    "    time.sleep(5)\n",
    "    uploaded = genai.get_file(uploaded.name)\n",
    "\n",
    "response = model.generate_content([uploaded, \"Describe this video clip\"])\n",
    "display(Markdown(f\"### Video Description\\n\\n{response.text}\"))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Try the Ukrainian Song"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "audio_file = next(Path(output_locn).glob(f'alyona*.m4a'), None)\n",
    "uploaded = genai.upload_file(audio_file)\n",
    "logger.info(f\"{uploaded=}\")\n",
    "\n",
    "# Videos need to be processed before you can use them.\n",
    "while uploaded.state.name == \"PROCESSING\":\n",
    "    logger.info(\"Processing audio...\")\n",
    "    time.sleep(5)\n",
    "    uploaded = genai.get_file(uploaded.name)\n",
    "\n",
    "prompt = \"\"\"\n",
    "In this audio file, please tell me:\n",
    "- What languages are being sung?\n",
    "- What are the lyrics? Please show me in the native language, and translated to English.\n",
    "- What is the meaning of the lyrics?\n",
    "\"\"\"\n",
    "\n",
    "response = model.generate_content([uploaded, prompt])\n",
    "display(Markdown(f\"### Audio Description\\n\\n{response.text}\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusions\n",
    "\n",
    "This was amazing!\n",
    "\n",
    "- We can use the multimodal Gemini model to transcribe audio from our video clips, including songs, or from our audio files.\n",
    "- It transcribes with a high degree of accuracy; much better than the Video Intelligence API.\n",
    "- It automatically detects the language of the source material.\n",
    "- It successfully translates.\n",
    "- You can do this from within a Google Cloud project using the Vertex AI API.  But you'll need a Google Cloud project, you'll need to enable the necessary APIs.\n",
    "- Or, you can use the Gemini API. The capability is the same, but you don't need a Google Cloud project. All you need is an API key."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".yt-and-vid-conda-env",
   "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.12.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}