À propos de cet atelier de programmation
1. Introduction
Dernière mise à jour:12/03/2025
Clause de non-responsabilité
Voici un exemple de code qui analyse des vidéos à l'aide de l'API YouTube Data et de Gemini. L'utilisateur est responsable de son utilisation. Ce code utilisé dans des environnements réels doit être soigneusement examiné. L'auteur décline toute responsabilité en cas de problème lié à l'utilisation de ce code. De plus, en raison de la nature de l'intelligence artificielle, il est toujours possible que les résultats diffèrent des faits réels. Par conséquent, vous ne devez pas faire confiance aveuglément aux résultats et vous devez les examiner attentivement.
Objectif de ce projet
L'objectif principal est d'identifier les vidéos et les créateurs YouTube adaptés à la promotion de votre marque en analysant le contenu et le sentiment des vidéos.
Présentation
Le projet utilise l'API YouTube Data pour extraire des informations sur les vidéos et l'API Vertex AI GCP avec le modèle Gemini pour analyser le contenu vidéo. Il s'exécute sur Google Colab.
Vous pouvez coller les codes qui seront publiés à l'avenir dans Colab et les exécuter un par un.
Points abordés
- Utiliser l'API YouTube Data pour extraire des informations sur les vidéos
- Utiliser l'API Vertex AI GCP avec le modèle Gemini pour analyser du contenu vidéo
- Utiliser Google Colab pour exécuter le code
- Créer une feuille de calcul à partir des données analysées
Prérequis
Pour mettre en œuvre cette solution, vous avez besoin des éléments suivants:
- Un projet Google Cloud Platform.
- Activez les API YouTube Data v3, Vertex AI, Generative Language, Google Drive et Google Sheets dans le projet.
- Créez une clé API dans l'onglet "Identifiants" avec une autorisation pour l'API YouTube Data v3.
Cette solution utilise l'API YouTube Data et l'API Vertex AI GCP.
2. Code et explication
La première chose à faire est d'importer les bibliothèques que nous souhaitons utiliser. Connectez-vous ensuite avec votre compte Google et autorisez l'accès à Google Drive.
# library
# colab
import ipywidgets as widgets
from IPython.display import display
from google.colab import auth
# cloud
from google import genai
from google.genai.types import Part, GenerateContentConfig
# function, util
import requests, os, re, time
from pandas import DataFrame
from datetime import datetime, timedelta
auth.authenticate_user()
[Action Required]
La clé API et l'ID de projet de GCP sont les valeurs qui doivent généralement être modifiées. Les cellules ci-dessous sont destinées aux valeurs de paramètres GCP.
# GCP Setting
LANGUAGE_MODEL = 'gemini-1.5-pro' # @param {type:"string"}
API_KEY = 'Please write your API_KEY' # @param {type:"string"}
PROJECT_ID = 'Please write your GCP_ID' # @param {type:"string"}
LOCATION = 'us-central1' # @param {type:"string"}
[Action Required]
Veuillez modifier les valeurs des variables tout en vérifiant les codes sous "Entrée".
À l'aide de la marque"Google" comme exemple, cet article vous explique comment rechercher sur YouTube des vidéos sur un sujet spécifique (par exemple, "IA de Google") tout en excluant les vidéos de la chaîne de la marque.
Variables d'entrée pour l'analyse vidéo YouTube
- BRAND_NAME (obligatoire) : nom de la marque à analyser (par exemple, (Google, par exemple).
- MY_COMPANY_INFO (Obligatoire) : brève description et contexte de la marque.
- SEARCH_QUERY (Obligatoire) : terme de recherche pour les vidéos YouTube (par exemple, Google AI).
- VIEWER_COUNTRY::code pays du spectateur (code pays à deux lettres: ISO 3166-1 alpha-2) (par exemple, Corée).
- GENERATION_LANGUAGE (obligatoire) : langue des résultats de Gemini (par exemple, Coréen).
- EXCEPT_CHANNEL_IDS::ID de chaîne à exclure, séparés par une virgule.
Vous pouvez trouver l'ID de la chaîne sur YouTube.
- VIDEO_TOPIC::ID de sujet YouTube à affiner.
Vous trouverez la valeur du sujet de la vidéo dans Search: list | YouTube Data API | Google for Developers (Recherche : liste | API YouTube Data | Google pour les développeurs).
- DATE_INPUT (Obligatoire) : date de début de la vidéo publiée (AAAA-MM-JJ).
# Input
BRAND_NAME = "Google" # @param {type:"string"}
MY_COMPANY_INFO = "Google is a multinational technology company specializing in internet-related services and products." # @param {type:"string"}
SEARCH_QUERY = 'Google AI' # @param {type:"string"}
VIEWER_COUNTRY = 'KR' # @param {type:"string"}
GENERATION_LANGUAGE = 'Korean' # @param {type:"string"}
EXCEPT_CHANNEL_IDS = 'UCK8sQmJBp8GCxrOtXWBpyEA, UCdc_SRhKUlH3grljQXA0skw' # @param {type:"string"}
VIDEO_TOPIC = '/m/07c1v' # @param {type: "string"}
DATE_INPUT = '2025-01-01' # @param {type:"date"}
# Auth Scope
SCOPE = [
'https://www.googleapis.com/auth/youtube.readonly',
'https://www.googleapis.com/auth/spreadsheets',
'https://www.googleapis.com/auth/drive',
'https://www.googleapis.com/auth/cloud-platform'
]
# validation check
if not SEARCH_QUERY or not DATE_INPUT:
raise ValueError("Search query and date input are required.")
EXCEPT_CHANNEL_IDS = [id.strip() for id in EXCEPT_CHANNEL_IDS.split(',')]
Le texte fourni liste les fonctions clés liées à l'interaction avec l'API YouTube Data.
# YouTube API function
def get_youtube_videos(q, viewer_country_code, topic_str, start_period):
page_token_number = 1
next_page_token = ''
merged_array = []
published_after_date = f"{start_period}T00:00:00Z"
while page_token_number < 9 and len(merged_array) <= 75:
result = search_youtube(q, topic_str, published_after_date, viewer_country_code, '', next_page_token, 50)
merged_array = list(set(merged_array + result['items']))
next_page_token = result['nextPageToken']
page_token_number += 1
return merged_array
def search_youtube(query, topic_id, published_after, region_code, relevance_language, next_page_token, max_results=50):
if not query:
return None
q = query
url = f'https://www.googleapis.com/youtube/v3/search?key={API_KEY}&part=snippet&q={q}&publishedAfter={published_after}®ionCode={region_code}&type=video&topicId={topic_id}&maxResults={max_results}&pageToken={next_page_token}&gl={region_code.lower()}'
response = requests.get(url)
data = response.json()
results = data.get('items', [])
next_page_token = data.get('nextPageToken', '')
return_results = [item['id']['videoId'] for item in results]
print(url)
return {
"nextPageToken": next_page_token,
"items": return_results
}
def get_date_string(days_ago):
date = datetime.now() + timedelta(days=days_ago)
return date.strftime('%Y-%m-%dT00:00:00Z')
def get_video_details(video_id):
url = f'https://www.googleapis.com/youtube/v3/videos?part=snippet,contentDetails&id={video_id}&key={API_KEY}'
response = requests.get(url)
data = response.json()
if data.get('items'):
video = data['items'][0]
snippet = video['snippet']
content_details = video['contentDetails']
title = snippet.get('title', 'no title')
description = snippet.get('description', 'no description')
duration_iso = content_details.get('duration', None)
channel_id = snippet.get('channelId', 'no channel id')
channel_title = snippet.get('channelTitle', 'no channel title')
return {'title': title, 'description': description, 'duration': duration_to_seconds(duration_iso), 'channel_id': channel_id, 'channel_title': channel_title}
else:
return None
def duration_to_seconds(duration_str):
match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?', duration_str)
if not match:
return None
hours, minutes, seconds = match.groups()
total_seconds = 0
if hours:
total_seconds += int(hours) * 3600
if minutes:
total_seconds += int(minutes) * 60
if seconds:
total_seconds += int(seconds)
return total_seconds
Le texte fournit un modèle d'invite pouvant être ajusté si nécessaire, ainsi que des fonctions clés pour interagir avec l'API Vertex AI GCP.
# GCP Vertex AI API
client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)
model = client.models
def request_gemini(prompt, video_link):
video_extraction_json_generation_config = GenerateContentConfig(
temperature=0.0,
max_output_tokens=2048,
)
contents = [
Part.from_uri(
file_uri=video_link,
mime_type="video/mp4",
),
prompt
]
response = model.generate_content(
model=LANGUAGE_MODEL,
contents=contents,
config=video_extraction_json_generation_config
)
try:
return response.text
except:
return response.GenerateContentResponse
def create_prompt(yt_title, yt_description, yt_link):
return f"""### Task: You are a highly specialized marketer and YouTube expert working for the brand or company, {BRAND_NAME}.
Your boss is wondering which a video to use to promote their company's advertisements and which a YouTuber to promote their advertisements with in the future. You are the expert who can give your boss the most suitable suggestions.
Analyze the video according to the criteria below and solve your boss's worries.
### Criteria: Now you review the video.
If you evaluate it using the following criteria, you will be able to receive a better evaluation.
1. Whether the video mentions brand, {BRAND_NAME}.
2. Whether the video views {BRAND_NAME} positively or negatively.
3. Whether the video would be suitable for marketing purposes.
### Context and Contents:
Your Company Information:
- Company Description: {MY_COMPANY_INFO}
- Brand: {BRAND_NAME}
Analysis subject:
- YouTube title: {yt_title}
- YouTube description: {yt_description}
- YouTube link: {yt_link}
### Answer Format:
brand_relevance_score: (Integer between 0 and 100 - If this video is more relative about the {BRAND_NAME}, it will score higher)
brand_positive_score: (Integer between 0 and 100 - If this video is positive about the {BRAND_NAME}, it will score higher)
brand_negative_score: (Integer between 0 and 100 - If this video is negative about the {BRAND_NAME}, it will score higher)
video_content_summary: (Summarize the content of the video like overview)
video_brand_summary: (Summarize the content about your brand, {BRAND_NAME})
opinion: (Why this video is suitable for promoting your company or product)
### Examples:
brand_relevance_score: 100
brand_positive_score: 80
brand_negative_score: 0
video_content_summary: YouTubers introduce various electronic products in their videos.
video_brand_summary: The brand products mentioned in the video have their advantages well explained by the YouTuber.
opinion: Consumers are more likely to think positively about the advantages of the product.
### Caution:
DO NOT fabricate information.
DO NOT imagine things.
DO NOT Markdown format.
DO Analyze each video based on the criteria mentioned above.
DO Analyze after watching the whole video.
DO write the results for summary as {GENERATION_LANGUAGE}."""
def parse_response(response: str):
brand_relevance_score_pattern = r"brand_relevance_score:\s*(\d{1,3})"
brand_positive_score_pattern = r"brand_positive_score:\s*(\d{1,3})"
brand_negative_score_pattern = r"brand_negative_score:\s*(\d{1,3})"
video_content_summary_pattern = r"video_content_summary:\s*(.*)"
video_brand_summary_pattern = r"video_brand_summary:\s*(.*)"
opinion_pattern = r"opinion:\s*(.*)"
brand_relevance_score_match = re.search( brand_relevance_score_pattern, response )
brand_relevance_score = ( int(brand_relevance_score_match.group(1)) if brand_relevance_score_match else 0 )
brand_positive_score_match = re.search( brand_positive_score_pattern, response )
brand_positive_score = ( int(brand_positive_score_match.group(1)) if brand_positive_score_match else 0 )
brand_negative_score_match = re.search( brand_negative_score_pattern, response )
brand_negative_score = ( int(brand_negative_score_match.group(1)) if brand_negative_score_match else 0 )
video_content_score_match = re.search( video_content_summary_pattern, response )
video_content_summary = ( video_content_score_match.group(1) if video_content_score_match else '' )
video_brand_summary_match = re.search( video_brand_summary_pattern, response )
video_brand_summary = ( video_brand_summary_match.group(1) if video_brand_summary_match else '' )
opinion_match = re.search( opinion_pattern, response )
opinion = ( opinion_match.group(1) if opinion_match else '' )
return ( brand_relevance_score, brand_positive_score, brand_negative_score, video_content_summary, video_brand_summary, opinion)
def request_gemini_with_retry(prompt, youtube_link='', max_retries=1):
retries = 0
while retries <= max_retries:
try:
response = request_gemini(prompt, youtube_link)
( brand_relevance_score,
brand_positive_score,
brand_negative_score,
video_content_summary,
video_brand_summary,
opinion) = parse_response(response)
if ( validate_score(brand_relevance_score) and
validate_score(brand_positive_score) and
validate_score(brand_negative_score) and
validate_summary(video_content_summary) and
validate_summary(video_brand_summary) ):
return ( brand_relevance_score,
brand_positive_score,
brand_negative_score,
video_content_summary,
video_brand_summary,
opinion
)
else:
retries += 1
ValueError(
"The value may be incorrect, there may be a range issue, a parsing"
" issue, or a response issue with Gemini: score -"
f" {brand_relevance_score}, {brand_positive_score},"
f" {brand_negative_score} , summary - {video_content_summary},"
f" {video_brand_summary}" )
except Exception as e:
print(f"Request failed: {e}")
retries += 1
if retries <= max_retries:
print(f"retry ({retries}/{max_retries})...")
else:
print("Maximum number of retries exceeded")
return 0, 0, 0, "", "", ""
def validate_score(score):
return score >= 0 and score <= 100
def validate_summary(summary):
return len(summary) > 0
Ce bloc de code est responsable de trois fonctions principales: créer un cadre de données, exécuter une analyse Gemini et mettre à jour le cadre de données par la suite.
def df_youtube_videos():
youtube_video_list = get_youtube_videos(SEARCH_QUERY, VIEWER_COUNTRY, VIDEO_TOPIC, DATE_INPUT)
youtube_video_link_list = []
youtube_video_title_list = []
youtube_video_description_list = []
youtube_video_channel_title_list = []
youtube_video_duration_list = []
for video_id in youtube_video_list:
video_details = get_video_details(video_id)
# https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/video-understanding
if video_details['duration'] < 50*60 and not video_details['channel_id'] in EXCEPT_CHANNEL_IDS:
youtube_video_link_list.append(f'https://www.youtube.com/watch?v={video_id}')
if video_details:
youtube_video_title_list.append(video_details['title'])
youtube_video_description_list.append(video_details['description'])
youtube_video_channel_title_list.append(video_details['channel_title'])
duration_new_format = f"{video_details['duration'] // 3600:02d}:{(video_details['duration'] % 3600) // 60:02d}:{video_details['duration'] % 60:02d}" # HH:MM:SS
youtube_video_duration_list.append(duration_new_format)
else:
youtube_video_title_list.append('')
youtube_video_description_list.append('')
youtube_video_channel_title_list.append(video_details['channel_title'])
youtube_video_duration_list.append('')
df = DataFrame({
'video_id': youtube_video_link_list,
'title': youtube_video_title_list,
'description': youtube_video_description_list,
'channel_title': youtube_video_channel_title_list,
'length': youtube_video_duration_list
})
return df
def run_gemini(df):
for index, row in df.iterrows():
video_title = row['title']
video_description = row['description']
video_link = row['video_id']
prompt = create_prompt(video_title, video_description, video_link)
( brand_relevance_score,
brand_positive_score,
brand_negative_score,
video_content_summary,
video_brand_summary,
opinion) = request_gemini_with_retry(prompt, video_link)
df.at[index, 'gemini_brand_relevance_score'] = brand_relevance_score
df.at[index, 'gemini_brand_positive_score'] = brand_positive_score
df.at[index, 'gemini_brand_negative_score'] = brand_negative_score
df.at[index, 'gemini_video_content_summary'] = video_content_summary
df.at[index, 'gemini_video_brand_summary'] = video_brand_summary
df.at[index, 'gemini_opinion'] = opinion
# https://cloud.google.com/vertex-ai/generative-ai/docs/quotas
time.sleep(1)
print(f"Processing: {index}/{len(df)}")
print(f"video_title: {video_title}")
return df
Il s'agit d'un bloc de code qui exécute tout le code écrit jusqu'à présent. Il extrait les données de YouTube, les analyse à l'aide de Gemini, puis crée un DataFrame.
# main
df = df_youtube_videos()
run_gemini(df)
df['gemini_brand_positive_score'] = df[ 'gemini_brand_positive_score' ].astype('int64')
df['gemini_brand_relevance_score'] = df[ 'gemini_brand_relevance_score' ].astype('int64')
df['gemini_brand_negative_score'] = df[ 'gemini_brand_negative_score' ].astype('int64')
df = df.sort_values( 'gemini_brand_positive_score', ascending=False )
df
La dernière étape consiste à créer une feuille de calcul à partir du frame data. Pour vérifier votre progression, utilisez l'URL de sortie.
import gspread
from google.auth import default
today_date = datetime.now().strftime('%Y-%m-%d')
my_spreadsheet_title = f"Partner's Video Finder, {BRAND_NAME}, {SEARCH_QUERY}, {VIEWER_COUNTRY} ({DATE_INPUT}~{today_date})"
creds, _ = default()
gc = gspread.authorize(creds)
sh = gc.create(my_spreadsheet_title)
worksheet = gc.open(my_spreadsheet_title).sheet1
cell_list = df.values.tolist()
worksheet.update([df.columns.values.tolist()] + cell_list)
print("URL: ", sh.url)
3. Référence
J'ai utilisé les ressources suivantes pour écrire le code. Si vous devez modifier le code ou si vous souhaitez en savoir plus sur son utilisation, veuillez cliquer sur le lien ci-dessous.