colab-enterprise/Campaign-Assets-Video-Create-Shorts.ipynb

{ "cells": [ { "cell_type": "markdown", "metadata": { "id": "HMsUvoF4BP7Y" }, "source": [ "### Overview" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "This demo analyzes our text-to-video and creates \"YouTube\" Shorts from the longer content. We provide Gemini with ABCD best practices, relevant research, and the video itself. Gemini then identifies the most exciting segment suitable for a Short, generating a new voice over script for that portion. Finally, text-to-speech technology produces the audio, completing the Short.\n", "\n", "Process Flow:\n", "\n", "1. Download the full marketing video without audio\n", "2. Download ABCD best practices PDF\n", "3. Download a research paper about \"shorts\"\n", "4. Prompt Gemini to watch the video, read the best practices, and ask for the timestamps of a YouTube short along with a voice over.\n", "5. Extract the video based upon Gemini's outputted timestamps\n", "6. Use text-to-speech to generate the voice over.\n", "7. Merge the video and the voice over.\n", "\n", "Notes:\n", "\n", "1. The notebook does merge the videos, but the quality is much better when you merge with a high quality video tool. The text-to-speech does not always align perfectly with the video.\n", "2. You can also adjust the speed of the text-to-speech to control how fast the words are spoken.\n", "\n", "Cost:\n", "* Low: Gemini, BigQuery\n", "* Medium: Remember to stop your Colab Enterprise Notebook Runtime\n", "\n", "Author: \n", "* Adam Paternostro" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Architecture Diagram\n", "from IPython.display import Image\n", "Image(url='https://storage.googleapis.com/data-analytics-golden-demo/chocolate-ai/v1/Artifacts/Campaign-Assets-Video-Create-Shorts-Architecture.png', width=1200)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Video Walkthrough" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "[![Video](https://storage.googleapis.com/data-analytics-golden-demo/chocolate-ai/v1/Videos/adam-paternostro-video.png)](https://storage.googleapis.com/data-analytics-golden-demo/chocolate-ai/v1/Videos/Campaign-Assets-Video-Create-Shorts.mp4)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from IPython.display import HTML\n", "\n", "HTML(\"\"\"\n", "<video width=\"800\" height=\"600\" controls>\n", " <source src=\"https://storage.googleapis.com/data-analytics-golden-demo/chocolate-ai/v1/Videos/Campaign-Assets-Video-Create-Shorts.mp4\" type=\"video/mp4\">\n", " Your browser does not support the video tag.\n", "</video>\n", "\"\"\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### License" ] }, { "cell_type": "markdown", "metadata": { "id": "jQgQkbOvj55d" }, "source": [ "```\n", "# Copyright 2024 Google LLC\n", "#\n", "# Licensed under the Apache License, Version 2.0 (the \"License\");\n", "# you may not use this file except in compliance with the License.\n", "# You may obtain a copy of the License at\n", "#\n", "# https://www.apache.org/licenses/LICENSE-2.0\n", "#\n", "# Unless required by applicable law or agreed to in writing, software\n", "# distributed under the License is distributed on an \"AS IS\" BASIS,\n", "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n", "# See the License for the specific language governing permissions and\n", "# limitations under the License.\n", "```\n", "\n", "Author: Adam Paternostro" ] }, { "cell_type": "markdown", "metadata": { "id": "uVQQDQqsO8eZ" }, "source": [ "### Initialize" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "-YuApakGO-eh" }, "outputs": [], "source": [ "from PIL import Image\n", "from IPython.display import HTML\n", "from IPython.display import Audio\n", "from functools import reduce\n", "import IPython.display\n", "import google.auth\n", "import requests\n", "import json\n", "import uuid\n", "import base64\n", "import os\n", "import cv2\n", "import random\n", "import time\n", "import datetime\n", "import base64\n", "import random\n", "import datetime\n", "\n", "import logging\n", "from tenacity import retry, wait_exponential, stop_after_attempt, before_sleep_log, retry_if_exception" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "EwstXVf_O-x6" }, "outputs": [], "source": [ "# Set these (run this cell to verify the output)\n", "\n", "bigquery_location = \"${bigquery_location}\"\n", "region = \"${region}\"\n", "location = \"${location}\"\n", "storage_account = \"${chocolate_ai_bucket}\"\n", "dataset_name = \"${bigquery_chocolate_ai_dataset}\"\n", "public_storage_storage_account = \"data-analytics-golden-demo\"\n", "\n", "# Get the current date and time\n", "now = datetime.datetime.now()\n", "\n", "# Format the date and time as desired\n", "formatted_date = now.strftime(\"%Y-%m-%d-%H-%M\")\n", "\n", "# Get some values using gcloud\n", "project_id = !(gcloud config get-value project)\n", "user = !(gcloud auth list --filter=status:ACTIVE --format=\"value(account)\")\n", "\n", "if len(project_id) != 1:\n", " raise RuntimeError(f\"project_id is not set: {project_id}\")\n", "project_id = project_id[0]\n", "\n", "if len(user) != 1:\n", " raise RuntimeError(f\"user is not set: {user}\")\n", "user = user[0]\n", "\n", "print(f\"project_id = {project_id}\")\n", "print(f\"user = {user}\")" ] }, { "cell_type": "markdown", "metadata": { "id": "m65vp54BUFRi" }, "source": [ "### Pip installs" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "5MaWM6H5i6rX" }, "outputs": [], "source": [ "# PIP Installs\n", "import sys\n", "\n", "# https://pypi.org/project/moviepy/\n", "!{sys.executable} -m pip install moviepy" ] }, { "cell_type": "markdown", "metadata": { "id": "sZ6m_wGrK0YG" }, "source": [ "### Helper Methods" ] }, { "cell_type": "markdown", "metadata": { "id": "JbOjdSP1kN9T" }, "source": [ "#### restAPIHelper\n", "Calls the Google Cloud REST API using the current users credentials." ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "40wlwnY4kM11" }, "outputs": [], "source": [ "def restAPIHelper(url: str, http_verb: str, request_body: str) -> str:\n", " \"\"\"Calls the Google Cloud REST API passing in the current users credentials\"\"\"\n", "\n", " import requests\n", " import google.auth\n", " import json\n", "\n", " # Get an access token based upon the current user\n", " creds, project = google.auth.default()\n", " auth_req = google.auth.transport.requests.Request()\n", " creds.refresh(auth_req)\n", " access_token=creds.token\n", "\n", " headers = {\n", " \"Content-Type\" : \"application/json\",\n", " \"Authorization\" : \"Bearer \" + access_token\n", " }\n", "\n", " if http_verb == \"GET\":\n", " response = requests.get(url, headers=headers)\n", " elif http_verb == \"POST\":\n", " response = requests.post(url, json=request_body, headers=headers)\n", " elif http_verb == \"PUT\":\n", " response = requests.put(url, json=request_body, headers=headers)\n", " elif http_verb == \"PATCH\":\n", " response = requests.patch(url, json=request_body, headers=headers)\n", " elif http_verb == \"DELETE\":\n", " response = requests.delete(url, headers=headers)\n", " else:\n", " raise RuntimeError(f\"Unknown HTTP verb: {http_verb}\")\n", "\n", " if response.status_code == 200:\n", " return json.loads(response.content)\n", " #image_data = json.loads(response.content)[\"predictions\"][0][\"bytesBase64Encoded\"]\n", " else:\n", " error = f\"Error restAPIHelper -> ' Status: '{response.status_code}' Text: '{response.text}'\"\n", " raise RuntimeError(error)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "#### RetryCondition (for retrying LLM calls)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def RetryCondition(error):\n", " error_string = str(error)\n", " print(error_string)\n", "\n", " retry_errors = [\n", " \"RESOURCE_EXHAUSTED\",\n", " \"No content in candidate\",\n", " # Add more error messages here as needed\n", " ]\n", "\n", " for retry_error in retry_errors:\n", " if retry_error in error_string:\n", " print(\"Retrying...\")\n", " return True\n", "\n", " return False" ] }, { "cell_type": "markdown", "metadata": { "id": "-xPgu3MncKSG" }, "source": [ "#### Gemini LLM (Pro 1.0 , Pro 1.5)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "IJl_rUxbcMdk" }, "outputs": [], "source": [ "@retry(wait=wait_exponential(multiplier=1, min=1, max=60), stop=stop_after_attempt(10), retry=retry_if_exception(RetryCondition), before_sleep=before_sleep_log(logging.getLogger(), logging.INFO))\n", "def GeminiLLM(prompt, model = \"gemini-2.0-flash\", response_schema = None,\n", " temperature = 1, topP = 1, topK = 32):\n", "\n", " # https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/inference#supported_models\n", " # model = \"gemini-2.0-flash\"\n", "\n", " llm_response = None\n", " if temperature < 0:\n", " temperature = 0\n", "\n", " creds, project = google.auth.default()\n", " auth_req = google.auth.transport.requests.Request() # required to acess access token\n", " creds.refresh(auth_req)\n", " access_token=creds.token\n", "\n", " headers = {\n", " \"Content-Type\" : \"application/json\",\n", " \"Authorization\" : \"Bearer \" + access_token\n", " }\n", "\n", " # https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/inference\n", " url = f\"https://{location}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{location}/publishers/google/models/{model}:generateContent\"\n", "\n", " generation_config = {\n", " \"temperature\": temperature,\n", " \"topP\": topP,\n", " \"maxOutputTokens\": 8192,\n", " \"candidateCount\": 1,\n", " \"responseMimeType\": \"application/json\",\n", " }\n", "\n", " # Add inthe response schema for when it is provided\n", " if response_schema is not None:\n", " generation_config[\"responseSchema\"] = response_schema\n", "\n", " if model == \"gemini-2.0-flash\":\n", " generation_config[\"topK\"] = topK\n", "\n", " payload = {\n", " \"contents\": {\n", " \"role\": \"user\",\n", " \"parts\": {\n", " \"text\": prompt\n", " },\n", " },\n", " \"generation_config\": {\n", " **generation_config\n", " },\n", " \"safety_settings\": {\n", " \"category\": \"HARM_CATEGORY_SEXUALLY_EXPLICIT\",\n", " \"threshold\": \"BLOCK_LOW_AND_ABOVE\"\n", " }\n", " }\n", "\n", " response = requests.post(url, json=payload, headers=headers)\n", "\n", " if response.status_code == 200:\n", " try:\n", " json_response = json.loads(response.content)\n", " except Exception as error:\n", " raise RuntimeError(f\"An error occurred parsing the JSON: {error}\")\n", "\n", " if \"candidates\" in json_response:\n", " candidates = json_response[\"candidates\"]\n", " if len(candidates) > 0:\n", " candidate = candidates[0]\n", " if \"content\" in candidate:\n", " content = candidate[\"content\"]\n", " if \"parts\" in content:\n", " parts = content[\"parts\"]\n", " if len(parts):\n", " part = parts[0]\n", " if \"text\" in part:\n", " text = part[\"text\"]\n", " llm_response = text\n", " else:\n", " raise RuntimeError(\"No text in part: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No parts in content: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No parts in content: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No content in candidate: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No candidates: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No candidates: {response.content}\")\n", "\n", " # Remove some typically response characters (if asking for a JSON reply)\n", " llm_response = llm_response.replace(\"```json\",\"\")\n", " llm_response = llm_response.replace(\"```\",\"\")\n", " llm_response = llm_response.replace(\"\\n\",\"\")\n", "\n", " return llm_response\n", "\n", " else:\n", " raise RuntimeError(f\"Error with prompt:'{prompt}' Status:'{response.status_code}' Text:'{response.text}'\")" ] }, { "cell_type": "markdown", "metadata": { "id": "4brvnJOIlXOT" }, "source": [ "#### Gemini LLM - Multimodal" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "UvhIiqqkcPur" }, "outputs": [], "source": [ "@retry(wait=wait_exponential(multiplier=1, min=1, max=60), stop=stop_after_attempt(10), retry=retry_if_exception(RetryCondition), before_sleep=before_sleep_log(logging.getLogger(), logging.INFO))\n", "def GeminiLLM_Multimodal(multimodal_prompt_list, model = \"gemini-2.0-flash\", response_schema = None,\n", " temperature = 1, topP = 1, topK = 32):\n", "\n", " # https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/inference#supported_models\n", " # model = \"gemini-2.0-flash\"\n", "\n", " llm_response = None\n", " if temperature < 0:\n", " temperature = 0\n", "\n", " creds, project = google.auth.default()\n", " auth_req = google.auth.transport.requests.Request() # required to acess access token\n", " creds.refresh(auth_req)\n", " access_token=creds.token\n", "\n", " headers = {\n", " \"Content-Type\" : \"application/json\",\n", " \"Authorization\" : \"Bearer \" + access_token\n", " }\n", "\n", " # https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/inference\n", " url = f\"https://{location}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{location}/publishers/google/models/{model}:generateContent\"\n", "\n", " generation_config = {\n", " \"temperature\": temperature,\n", " \"topP\": topP,\n", " \"maxOutputTokens\": 8192,\n", " \"candidateCount\": 1,\n", " \"responseMimeType\": \"application/json\",\n", " }\n", "\n", " # Add inthe response schema for when it is provided\n", " if response_schema is not None:\n", " generation_config[\"responseSchema\"] = response_schema\n", "\n", " if model == \"gemini-2.0-flash\":\n", " generation_config[\"topK\"] = topK\n", "\n", " payload = {\n", " \"contents\": {\n", " \"role\": \"user\",\n", " \"parts\": multimodal_prompt_list\n", " },\n", " \"generation_config\": {\n", " **generation_config\n", " },\n", " \"safety_settings\": {\n", " \"category\": \"HARM_CATEGORY_SEXUALLY_EXPLICIT\",\n", " \"threshold\": \"BLOCK_LOW_AND_ABOVE\"\n", " }\n", " }\n", "\n", " response = requests.post(url, json=payload, headers=headers)\n", "\n", " if response.status_code == 200:\n", " try:\n", " json_response = json.loads(response.content)\n", " except Exception as error:\n", " raise RuntimeError(f\"An error occurred parsing the JSON: {error}\")\n", "\n", " if \"candidates\" in json_response:\n", " candidates = json_response[\"candidates\"]\n", " if len(candidates) > 0:\n", " candidate = candidates[0]\n", " if \"content\" in candidate:\n", " content = candidate[\"content\"]\n", " if \"parts\" in content:\n", " parts = content[\"parts\"]\n", " if len(parts):\n", " part = parts[0]\n", " if \"text\" in part:\n", " text = part[\"text\"]\n", " llm_response = text\n", " else:\n", " raise RuntimeError(\"No text in part: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No parts in content: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No parts in content: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No content in candidate: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No candidates: {response.content}\")\n", " else:\n", " raise RuntimeError(\"No candidates: {response.content}\")\n", "\n", " # Remove some typically response characters (if asking for a JSON reply)\n", " llm_response = llm_response.replace(\"```json\",\"\")\n", " llm_response = llm_response.replace(\"```\",\"\")\n", " llm_response = llm_response.replace(\"\\n\",\"\")\n", "\n", " return llm_response\n", "\n", " else:\n", " raise RuntimeError(f\"Error with prompt:'{prompt}' Status:'{response.status_code}' Text:'{response.text}'\")" ] }, { "cell_type": "markdown", "metadata": { "id": "eqIt4BcJlhnk" }, "source": [ "#### Download GCS file" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "YGNfJcHZkqyS" }, "outputs": [], "source": [ "def download_from_gcs(destination_file_name, filename, gcs_storage_bucket, gcs_storage_path):\n", " # prompt: Write python code to download a blob from a gcs bucket. do not use the requests method\n", "\n", " from google.cloud import storage\n", "\n", " # The ID of your GCS object\n", " object_name = gcs_storage_path + filename\n", "\n", " storage_client = storage.Client()\n", "\n", " bucket = storage_client.bucket(gcs_storage_bucket)\n", "\n", " # Construct a client side representation of a blob.\n", " # Note `Bucket.blob` differs from `Bucket.get_blob` as it doesn't retrieve\n", " # any content from Google Cloud Storage. As we don't need additional data,\n", " # using `Bucket.blob` is preferred here.\n", " blob = bucket.blob(object_name)\n", " blob.download_to_filename(destination_file_name)\n", "\n", " print(\n", " \"Downloaded storage object {} from bucket {} to local file {}.\".format(\n", " object_name, gcs_storage_bucket, destination_file_name\n", " )\n", " )" ] }, { "cell_type": "markdown", "metadata": { "id": "IZibXtjkllRV" }, "source": [ "#### Copy file to GCS" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "Hjyf9J8Bk8h0" }, "outputs": [], "source": [ "# This was generated by GenAI\n", "\n", "def copy_file_to_gcs(local_file_path, bucket_name, destination_blob_name):\n", " \"\"\"Copies a file from a local drive to a GCS bucket.\n", "\n", " Args:\n", " local_file_path: The full path to the local file.\n", " bucket_name: The name of the GCS bucket to upload to.\n", " destination_blob_name: The desired name of the uploaded file in the bucket.\n", "\n", " Returns:\n", " None\n", " \"\"\"\n", "\n", " import os\n", " from google.cloud import storage\n", "\n", " # Ensure the file exists locally\n", " if not os.path.exists(local_file_path):\n", " raise FileNotFoundError(f\"Local file '{local_file_path}' not found.\")\n", "\n", " # Create a storage client\n", " storage_client = storage.Client()\n", "\n", " # Get a reference to the bucket\n", " bucket = storage_client.bucket(bucket_name)\n", "\n", " # Create a blob object with the desired destination path\n", " blob = bucket.blob(destination_blob_name)\n", "\n", " # Upload the file from the local filesystem\n", " content_type = \"\"\n", " if local_file_path.endswith(\".html\"):\n", " content_type = \"text/html; charset=utf-8\"\n", "\n", " if local_file_path.endswith(\".json\"):\n", " content_type = \"application/json; charset=utf-8\"\n", "\n", " if content_type == \"\":\n", " blob.upload_from_filename(local_file_path)\n", " else:\n", " blob.upload_from_filename(local_file_path, content_type = content_type)\n", "\n", " print(f\"File '{local_file_path}' uploaded to GCS bucket '{bucket_name}' as '{destination_blob_name}. Content-Type: {content_type}'.\")" ] }, { "cell_type": "markdown", "metadata": { "id": "O-RP_lUAs_8v" }, "source": [ "#### Text-to-Speech" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "8jTIYEkVtBLW" }, "outputs": [], "source": [ "def TextToSpeechLanguageList(language_code):\n", " creds, project = google.auth.default()\n", " auth_req = google.auth.transport.requests.Request()\n", " creds.refresh(auth_req)\n", " access_token=creds.token\n", "\n", " headers = {\n", " \"Content-Type\" : \"application/json\",\n", " \"Authorization\" : \"Bearer \" + access_token,\n", " \"x-goog-user-project\" : project\n", " }\n", "\n", " # https://cloud.google.com/text-to-speech/docs/reference/rest/v1/voices/list\n", " url = f\"https://texttospeech.googleapis.com/v1/voices?languageCode={language_code}\"\n", "\n", " response = requests.get(url, headers=headers)\n", "\n", " if response.status_code == 200:\n", " return response.text\n", " else:\n", " error = f\"Error with language_code:'{language_code}' Status:'{response.status_code}' Text:'{response.text}'\"\n", " raise RuntimeError(error)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "fudhvp8mtBxw" }, "outputs": [], "source": [ "def TextToSpeech(local_filename, text, language_code, language_code_name, ssml_gender, speaking_rate = 1):\n", " creds, project = google.auth.default()\n", " auth_req = google.auth.transport.requests.Request()\n", " creds.refresh(auth_req)\n", " access_token=creds.token\n", "\n", " headers = {\n", " \"Content-Type\" : \"application/json\",\n", " \"Authorization\" : \"Bearer \" + access_token,\n", " \"x-goog-user-project\" : project\n", " }\n", "\n", " # https://cloud.google.com/text-to-speech/docs/reference/rest/v1/text/synthesize\n", " url = f\"https://texttospeech.googleapis.com/v1/text:synthesize\"\n", "\n", " payload = {\n", " \"input\": {\n", " \"text\": text\n", " },\n", " \"voice\": {\n", " \"languageCode\": language_code,\n", " \"name\": language_code_name,\n", " \"ssmlGender\": ssml_gender # FEMALE | MALE\n", " },\n", " \"audioConfig\": {\n", " \"audioEncoding\": \"MP3\",\n", " \"speakingRate\": speaking_rate,\n", " }\n", " }\n", "\n", " response = requests.post(url, json=payload, headers=headers)\n", "\n", " if response.status_code == 200:\n", " audio_data = json.loads(response.content)[\"audioContent\"]\n", " audio_data = base64.b64decode(audio_data)\n", " with open(local_filename, \"wb\") as f:\n", " f.write(audio_data)\n", " print(f\"Audio generated OK.\")\n", " return local_filename\n", " else:\n", " error = f\"Error with text:'{text}' Status:'{response.status_code}' Text:'{response.text}'\"\n", " raise RuntimeError(error)" ] }, { "cell_type": "markdown", "metadata": { "id": "2Vmk1TVJuiRd" }, "source": [ "#### MergeVideoAndAudio and ExtractClip" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "C9JDeCp4ujPT" }, "outputs": [], "source": [ "from moviepy.editor import VideoFileClip, AudioFileClip\n", "\n", "def MergeVideoAndAudio(video_filename, audio_filename, output_filename):\n", " # Load the video and audio files\n", " video = VideoFileClip(video_filename)\n", " audio = AudioFileClip(audio_filename)\n", "\n", " # Combine the video and audio\n", " final_clip = video.set_audio(audio)\n", "\n", " # Save the combined video\n", " final_clip.write_videofile(output_filename)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "tBABNiHMujht" }, "outputs": [], "source": [ "from moviepy.editor import *\n", "\n", "def ExtractClip(input_video_path, start_time, end_time):\n", " # Load the video clip\n", " clip = VideoFileClip(input_video_path)\n", "\n", " # Extract the desired segment\n", " extracted_clip = clip.subclip(start_time, end_time)\n", "\n", " # Save the extracted clip as a new video file\n", " extracted_clip.write_videofile(youtube_extracted_clip_video_filename)" ] }, { "cell_type": "markdown", "metadata": { "id": "rfhk5Jl6FO3Z" }, "source": [ "#### Helper Functions" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "e9GDbcDvFQVE" }, "outputs": [], "source": [ "# prompt: python to delete a file even if it does not exist\n", "\n", "def delete_file(filename):\n", " try:\n", " os.remove(filename)\n", " print(f\"File '{filename}' deleted successfully.\")\n", " except FileNotFoundError:\n", " print(f\"File '{filename}' not found.\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "a9a_8SEKFTbe" }, "outputs": [], "source": [ "def PrettyPrintJson(json_string):\n", " json_object = json.loads(json_string)\n", " json_formatted_str = json.dumps(json_object, indent=2)\n", " #print(json_formatted_str)\n", " return json_formatted_str" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "XWE1mzs_VRG9" }, "outputs": [], "source": [ "# prompt: python code to download a pdf from the internet\n", "\n", "import requests\n", "\n", "def download_http_file(url, filename):\n", " \"\"\"Downloads a PDF file from a given URL.\n", "\n", " Args:\n", " url: The URL of the PDF file to download.\n", " filename: The name to save the downloaded PDF file as.\n", " \"\"\"\n", " try:\n", " response = requests.get(url)\n", " response.raise_for_status() # Raise an exception for bad status codes\n", "\n", " with open(filename, 'wb') as f:\n", " f.write(response.content)\n", "\n", " print(f\"PDF downloaded successfully to {filename}\")\n", "\n", " except requests.exceptions.RequestException as e:\n", " print(f\"An error occurred while downloading the PDF: {e}\")\n", "\n", "# Example usage:\n" ] }, { "cell_type": "markdown", "metadata": { "id": "EYRHDPdVKBzd" }, "source": [ "### Download Existing Ad and Best Pratices for Creating YouTube Shorts" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "Nz2lUhuh5iMY" }, "outputs": [], "source": [ "# Download our sample Chocolate A.I. Ads and upload to our project storage account (this way they will be local and we can access via Gemini)\n", "\n", "download_from_gcs(\"story-01-full-video-no-audio.mp4\", \"full-video-no-audio.mp4\", public_storage_storage_account, \"chocolate-ai/v1/Campaign-Assets-Text-to-Video-01/story-01/\")\n", "download_from_gcs(\"story-02-full-video-no-audio.mp4\", \"full-video-no-audio.mp4\", public_storage_storage_account, \"chocolate-ai/v1/Campaign-Assets-Text-to-Video-01/story-02/\")\n", "download_from_gcs(\"story-03-full-video-no-audio.mp4\", \"full-video-no-audio.mp4\", public_storage_storage_account, \"chocolate-ai/v1/Campaign-Assets-Text-to-Video-01/story-03/\")\n", "\n", "copy_file_to_gcs(\"story-01-full-video-no-audio.mp4\", storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/story-01-full-video-no-audio.mp4\")\n", "copy_file_to_gcs(\"story-02-full-video-no-audio.mp4\", storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/story-02-full-video-no-audio.mp4\")\n", "copy_file_to_gcs(\"story-03-full-video-no-audio.mp4\", storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/story-03-full-video-no-audio.mp4\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "pQwtjvJzVK-d" }, "outputs": [], "source": [ "# https://blog.google/products/ads-commerce/youtube-shorts-ads-select-lineups-abcds/\n", "\n", "abcd_filename = \"Marketing-Shorts-ABCD-One-Sheeter.pdf\"\n", "download_from_gcs(abcd_filename, abcd_filename, public_storage_storage_account, \"chocolate-ai/v1/Artifacts/\")\n", "copy_file_to_gcs(abcd_filename, storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/{abcd_filename}\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "uHLEQTfMY3RX" }, "outputs": [], "source": [ "# https://arxiv.org/html/2402.18208v1\n", "# Shorts on the Rise: Assessing the Effects of YouTube: Shorts on Long-Form Video Content\n", "arxiv_url = \"https://arxiv.org/pdf/2402.18208v1\"\n", "arxiv_filename = \"Shorts-on-the-Rise-Assessing-the-Effects-of-YouTube.pdf\"\n", "download_http_file(arxiv_url, arxiv_filename)\n", "copy_file_to_gcs(arxiv_filename, storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/{arxiv_filename}\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "d_WuyQdoP6HR" }, "outputs": [], "source": [ "print(f\"View the GCS bucket: https://console.cloud.google.com/storage/browser/{storage_account}/chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}\")" ] }, { "cell_type": "markdown", "metadata": { "id": "4PCGPbpIVb1l" }, "source": [ "### YouTube Short - Gemini Prompt" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "uzxW2WR7lygS" }, "outputs": [], "source": [ "# Write me the json in OpenAPI 3.0 schema object for the below object.\n", "# Make all fields required.\n", "# {\n", "# \"youtube_short_description\" : \"text\",\n", "# \"youtube_short_begin_timestamp\" : \"text\",\n", "# \"youtube_short_end_timestamp\" : \"text\",\n", "# \"youtube_short_explanation\" : \"text\",\n", "# \"youtube_short_voice_over\" : \"text\",\n", "# \"abcd_best_practices_in_short\" : [\"text\"]\n", "# }\n", "response_schema = {\n", " \"type\": \"object\",\n", " \"required\": [\n", " \"youtube_short_description\",\n", " \"youtube_short_begin_timestamp\",\n", " \"youtube_short_end_timestamp\",\n", " \"youtube_short_explanation\",\n", " \"youtube_short_voice_over\",\n", " \"abcd_best_practices_in_short\"\n", " ],\n", " \"properties\": {\n", " \"youtube_short_description\": {\n", " \"type\": \"string\"\n", " },\n", " \"youtube_short_begin_timestamp\": {\n", " \"type\": \"string\"\n", " },\n", " \"youtube_short_end_timestamp\": {\n", " \"type\": \"string\"\n", " },\n", " \"youtube_short_explanation\": {\n", " \"type\": \"string\"\n", " },\n", " \"youtube_short_voice_over\": {\n", " \"type\": \"string\"\n", " },\n", " \"abcd_best_practices_in_short\": {\n", " \"type\": \"array\",\n", " \"items\": {\n", " \"type\": \"string\"\n", " }\n", " }\n", " }\n", "}\n", "\n", "prompt = \"\"\"I need you to watch the video and find the most interesting exciting part for a YouTube short.\n", "I attached the file {abcd_filename} which contains the best practices for creating a YouTube short.\n", "Incorporate the best practices into the video.\n", "This is for a company called \"Chocolate A.I.\" based in Paris, France.\n", "Select the best 10 seconds of based upon the {abcd_filename}.\n", "Write a 10 second voice over for the segment.\n", "- The voice over shoud be a short, engaging, and memorable.\n", "- The voice over should have an introduction and conclusion.\n", "- The voice over should be 200 to 300 characters.\n", "- The voice over should be 25 to 30 words.\n", "\n", "- IMPORTANT:\n", " - If you end a sentence with \"Chocolate A.I.\" in the \"youtube_short_voice_over\" field, make sure you place a second period like \"Chocolate A.I..\".\n", " - A second period is needed for the voice-to-text to work properly.\n", " - Example: \"In the heart of Paris, a masterpiece takes shape. Chocolate A.I. Where artistry meets indulgence.\" should become \"In the heart of Paris, a masterpiece takes shape. Chocolate A.I.. Where artistry meets indulgence.\"\n", " - Double check that the voice over is about 30 words (or 3 sentences), if not try again.\n", " \n", "Places the ABCD best practices that will be contained in the short in the \"abcd_best_practices_in_short\" field.\n", "\"\"\"\n", "\n", "# You can you any of these 3 files.\n", "marketing_story = \"01\" # You can use 01, 02 or 03. Try out all three.\n", "\n", "marketing_video_name_to_review = f\"story-{marketing_story}-full-video-no-audio.mp4\"\n", "\n", "multimodal_prompt_list = [\n", " { \"text\": prompt },\n", " { \"fileData\": { \"mimeType\": \"video/mp4\", \"fileUri\": f\"gs://{storage_account}/chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/{marketing_video_name_to_review}\" } },\n", " { \"fileData\": { \"mimeType\": \"application/pdf\", \"fileUri\": f\"gs://{storage_account}/chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/{abcd_filename}\" } },\n", " { \"fileData\": { \"mimeType\": \"application/pdf\", \"fileUri\": f\"gs://{storage_account}/chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/Shorts-on-the-Rise-Assessing-the-Effects-of-YouTube.pdf\" } }\n", " ]\n", "\n", "youtube_short_response = GeminiLLM_Multimodal(multimodal_prompt_list, response_schema=response_schema)\n", "\n", "youtube_short_dict = json.loads(youtube_short_response)\n", "\n", "print(PrettyPrintJson(youtube_short_response))" ] }, { "cell_type": "markdown", "metadata": { "id": "ZP_ydCRuVl5c" }, "source": [ "### YouTube Short - Extract the Video Short" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "gmGzxNEum7jC" }, "outputs": [], "source": [ "youtube_extracted_clip_video_filename = f\"youtube-short-extract-clip-story-{marketing_story}.mp4\"\n", "youtube_extracted_clip_audio_filename = f\"youtube-short-extract-clip-story-{marketing_story}.mp3\"\n", "youtube_short_filename = f\"youtube-short-extract-clip-story-{marketing_story}-final.mp4\"\n", "\n", "ExtractClip(marketing_video_name_to_review, youtube_short_dict['youtube_short_begin_timestamp'], youtube_short_dict['youtube_short_end_timestamp'])" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "RD4OmWrjnERR" }, "outputs": [], "source": [ "# prompt: python to play a mp4 in a jupyter notebook\n", "video_mp4 = open(youtube_extracted_clip_video_filename, 'rb').read()\n", "video_url = \"data:video/mp4;base64,\" + base64.b64encode(video_mp4).decode()" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "mGqtZ7fUnJxK" }, "outputs": [], "source": [ "# 16:9 aspect ratio\n", "HTML(f\"\"\"\n", "YouTube Short (no audio)\n", "<video width=600 height=337 controls>\n", " <source src=\"{video_url}\" type=\"video/mp4\">\n", "</video>\n", "\"\"\")" ] }, { "cell_type": "markdown", "metadata": { "id": "u2GZgiuIVq_M" }, "source": [ "### YouTube Short - Generate the Audio (Voice Over)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "_vQwcTMTsxrV" }, "outputs": [], "source": [ "# Print out a list of language code, select one you want\n", "response_text = TextToSpeechLanguageList(\"en-gb\")\n", "response_json = json.loads(response_text)\n", "print (response_json)\n", "# print(PrettyPrintJson(response_text))\n", "\n", "language_code = \"en-gb\"\n", "language_code_name = \"en-GB-Standard-B\"\n", "ssml_gender = \"MALE\"" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "5yzk0dXfs4-W" }, "outputs": [], "source": [ "# Generate the text-to-speech for each segment\n", "# You can change the speed of the voice [last parameter in TextToSpeech] to either shorted or lengthen the audio lenght.\n", "# Ideally we would generate the voiceover with an exact number of works for 10 seconds\n", "\n", "# Output of video files\n", "voiceover_prompt = youtube_short_dict['youtube_short_voice_over']\n", "print(f\"Generating: {voiceover_prompt}\")\n", "\n", "# Text-to-Speech\n", "TextToSpeech(youtube_extracted_clip_audio_filename, voiceover_prompt, language_code, language_code_name, ssml_gender, .95)\n", "display(Audio(youtube_extracted_clip_audio_filename, autoplay=True,rate=16000))\n", "print()" ] }, { "cell_type": "markdown", "metadata": { "id": "j1NUfC5BVxRb" }, "source": [ "### YouTube Short - Merge Video and Audio for Final Video" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "cB55QHAjttA4" }, "outputs": [], "source": [ "# This does a decent job at merging the video and audio, there are alot more settings you can configure, this is just a basic merge\n", "\n", "MergeVideoAndAudio(youtube_extracted_clip_video_filename,youtube_extracted_clip_audio_filename,youtube_short_filename)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "sykGlyoMuAlc" }, "outputs": [], "source": [ "# prompt: python to play a mp4 in a jupyter notebook\n", "video_mp4 = open(youtube_short_filename, 'rb').read()\n", "video_url = \"data:video/mp4;base64,\" + base64.b64encode(video_mp4).decode()" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "cv1OAmpWuGGV" }, "outputs": [], "source": [ "# 16:9 aspect ratio\n", "HTML(f\"\"\"\n", "YouTube Short (with Audio)\n", "<video width=600 height=337 controls>\n", " <source src=\"{video_url}\" type=\"video/mp4\">\n", "</video>\n", "\"\"\")" ] }, { "cell_type": "markdown", "metadata": { "id": "mzu3RTyvC_0J" }, "source": [ "### YouTube Short - Upload Artifacts to GCS" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "IaWU5rWRDBvj" }, "outputs": [], "source": [ "copy_file_to_gcs(youtube_extracted_clip_video_filename, storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/{youtube_extracted_clip_video_filename}\")\n", "copy_file_to_gcs(youtube_extracted_clip_audio_filename, storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/{youtube_extracted_clip_audio_filename}\")\n", "copy_file_to_gcs(youtube_short_filename, storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/{youtube_short_filename}\")\n", "\n", "# Save the prompt so we know how we got this data\n", "with open(\"gemini_youtube_short_prompt.txt\", \"w\") as f:\n", " f.write(prompt)\n", "copy_file_to_gcs(\"gemini_youtube_short_prompt.txt\", storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/gemini_youtube_short_prompt-{marketing_story}.txt\")\n", "delete_file(\"gemini_youtube_short_prompt.txt\")\n", "\n", "# Save the output of the prompt\n", "with open(\"gemini_youtube_short_prompt_results.txt\", \"w\") as f:\n", " f.write(PrettyPrintJson(json.dumps(youtube_short_dict)))\n", "copy_file_to_gcs(\"gemini_youtube_short_prompt_results.txt\", storage_account, f\"chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}/gemini_youtube_short_prompt_results-{marketing_story}.txt\")\n", "delete_file(\"gemini_youtube_short_prompt_results.txt\")\n", "\n", "# To view the bucket\n", "print()\n", "print(\"Click here to view the bucket\")\n", "print(f\"https://console.cloud.google.com/storage/browser/{storage_account}/chocolate-ai/Campaign-Assets-Video-Create-Shorts/youtube-short-{formatted_date}\")" ] }, { "cell_type": "markdown", "metadata": { "id": "42IxhtRRrvR-" }, "source": [ "### Clean Up" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "id": "6lF2Z7skFbvf" }, "outputs": [], "source": [ "# Placeholder" ] }, { "cell_type": "markdown", "metadata": { "id": "ASQ2BPisXDA0" }, "source": [ "### Reference Links\n" ] }, { "cell_type": "markdown", "metadata": { "id": "UPlXU4Tv5iMZ" }, "source": [ "- [Google.com](https://www.google.com)" ] } ], "metadata": { "colab": { "collapsed_sections": [ "uVQQDQqsO8eZ", "m65vp54BUFRi", "sZ6m_wGrK0YG", "JbOjdSP1kN9T", "-xPgu3MncKSG", "4brvnJOIlXOT", "IZibXtjkllRV", "O-RP_lUAs_8v", "2Vmk1TVJuiRd", "rfhk5Jl6FO3Z", "ZP_ydCRuVl5c", "u2GZgiuIVq_M", "j1NUfC5BVxRb", "42IxhtRRrvR-", "ASQ2BPisXDA0" ], "name": "Campaign-Assets-Video-Create-Shorts", "private_outputs": true, "provenance": [] }, "kernelspec": { "display_name": "Python 3", "name": "python3" }, "language_info": { "name": "python", "version": "3.12.5" } }, "nbformat": 4, "nbformat_minor": 0 }

colab-enterprise/Campaign-Assets-Video-Create-Shorts.ipynb (1,312 lines of code) (raw):