Iniziamo importando alcune librerie e una chiave API YouTube segreta. Se non disponi di una chiave API, puoi crearne una di seguito questa guida.
import requests
import json
import polars as pl
from my_sk import my_keyfrom youtube_transcript_api import YouTubeTranscriptApi
Successivamente, definiremo le variabili che ci aiutano a estrarre i dati video dall'API di YouTube. Qui, specifico l'ID del mio canale YouTube e l'URL dell'API, inizializza pagina_tokene creare un elenco per la memorizzazione dei dati video.
# define channel ID
channel_id = 'UCa9gErQ9AE5jT2DZLjXBIdA'# define url for API
url = 'https://www.googleapis.com/youtube/v3/search'
# initialize page token
page_token = None
# intialize list to store video data
video_record_list = ()
Il prossimo pezzo di codice potrebbe essere spaventoso, quindi spiegherò prima cosa sta succedendo. Eseguiremo richieste GET all'API di ricerca di YouTube. È proprio come cercare video su YouTube, ma invece di utilizzare l'interfaccia utente, eseguiremo le ricerche in modo programmatico.
Poiché i risultati di ricerca sono limitati a 50 per pagina, dobbiamo eseguire ricerche ricorsivamente per restituire ogni video che corrisponde ai criteri di ricerca. Ecco come appare nel codice Python.
# extract video data across multiple search result pageswhile page_token != 0:
# define parameters for API call
params = {'key': my_key, 'channelId': channel_id,
'part': ("snippet","id"), 'order': "date",
'maxResults':50, 'pageToken': page_token}
# make get request
response = requests.get(url, params=params)
# append video data from page results to list
video_record_list += getVideoRecords(response)
try:
# grab next page token
page_token = json.loads(response.text)('nextPageToken')
except:
# if no next page token kill while loop
page_token = 0
getVideoRecords() è una funzione definita dall'utente che estrae le informazioni rilevanti da una risposta API.
# extract video data from single search result pagedef getVideoRecords(response: requests.models.Response) -> list:
"""
Function to extract YouTube video data from GET request response
"""
# initialize list to store video data from page results
video_record_list = ()
for raw_item in json.loads(response.text)('items'):
# only execute for youtube videos
if raw_item('id')('kind') != "youtube#video":
continue
# extract relevant data
video_record = {}
video_record('video_id') = raw_item('id')('videoId')
video_record('datetime') = raw_item('snippet')('publishedAt')
video_record('title') = raw_item('snippet')('title')
# append record to list
video_record_list.append(video_record)
return video_record_list
Ora che abbiamo informazioni su tutti i miei video YouTube, estraiamo i sottotitoli generati automaticamente. Per facilitare l'accesso agli ID video, memorizzerò i dati video in un dataframe Polars.
# store data in polars dataframe
df = pl.DataFrame(video_record_list)
print(df.head())
Per estrarre le didascalie dei video, userò il file youtube_transcript_api Libreria Python. Eseguirò il loop di ciascun ID video nel dataframe ed estrarrò la trascrizione associata.
# intialize list to store video captions
transcript_text_list = ()# loop through each row of dataframe
for i in range(len(df)):
# try to extract captions
try:
# get transcript
transcript = YouTubeTranscriptApi.get_transcript(df('video_id')(i))
# extract text transcript
transcript_text = extract_text(transcript)
# if not captions available set as n/a
except:
transcript_text = "n/a"
# append transcript text to list
transcript_text_list.append(transcript_text)
Ancora una volta, utilizzo una funzione definita dall'utente chiamata estrai_testo() per estrarre le informazioni necessarie dall'API.
def extract_text(transcript: list) -> str:
"""
Function to extract text from transcript dictionary
"""text_list = (transcript(i)('text') for i in range(len(transcript)))
return ' '.join(text_list)
Quindi possiamo aggiungere le trascrizioni di ciascun video al dataframe.
# add transcripts to dataframe
df = df.with_columns(pl.Series(name="transcript", values=transcript_text_list))
print(df.head())
Con i dati estratti, possiamo trasformarli in modo che siano pronti per il caso d'uso downstream. Ciò richiede alcune analisi esplorative dei dati (EDA).
Consegna dei duplicati
Un buon punto di partenza per EDA è esaminare il numero di righe ed elementi univoci in ciascuna colonna. In questo caso, ci aspettavamo che ogni riga fosse identificata in modo univoco dal video_id. Inoltre, ogni colonna non dovrebbe contenere elementi ripetuti, ad eccezione dei video per i quali non era disponibile alcuna trascrizione, che abbiamo impostato come “n / a”.
Ecco del codice per sondare tali informazioni. Possiamo vedere dall'output che i dati corrispondono alle nostre aspettative.
# shape + unique values
print("shape:", df.shape)
print("n unique rows:", df.n_unique())
for j in range(df.shape(1)):
print("n unique elements (" + df.columns(j) + "):", df(:,j).n_unique())### output
# shape: (84, 4)
# n unique rows: 84
# n unique elements (video_id): 84
# n unique elements (datetime): 84
# n unique elements (title): 84
# n unique elements (transcript): 82
Controlla i dtype
Successivamente, possiamo esaminare i tipi di dati di ciascuna colonna. Nell'immagine sopra, abbiamo visto che tutte le colonne sono stringhe.
Anche se questo è appropriato per il video_id, titoloE trascrizionequesta non è una buona scelta per il appuntamento colonna. Possiamo modificare questo tipo nel modo seguente.
# change datetime to Datetime dtype
df = df.with_columns(pl.col('datetime').cast(pl.Datetime))
print(df.head())
Gestione dei caratteri speciali
Poiché stiamo lavorando con dati di testo, è importante prestare attenzione alle stringhe di caratteri speciali. Ciò richiede un po' di scrematura manuale del testo, ma dopo pochi minuti ho trovato 2 casi particolari: '→' E & → &
Nel codice seguente, ho sostituito queste stringhe con i caratteri appropriati e ho cambiato “sha” A “Shaw”.
# list all special strings and their replacements
special_strings = (''', '&', 'sha ')
special_string_replacements = ("'", "&", "Shaw ")# replace each special string appearing in title and transcript columns
for i in range(len(special_strings)):
df = df.with_columns(df('title').str.replace(special_strings(i),
special_string_replacements(i)).alias('title'))
df = df.with_columns(df('transcript').str.replace(special_strings(i),
special_string_replacements(i)).alias('transcript'))
Poiché il set di dati qui è molto piccolo (84 righe e 4 colonne, ~900.000 caratteri), possiamo archiviare i dati direttamente nella directory del progetto. Questo può essere fatto in una riga di codice utilizzando il comando write_parquet() metodo in Polari. La dimensione finale del file è 341 KB.
# write data to file
df.write_parquet('data/video-transcripts.parquet')
Qui, abbiamo discusso le basi della creazione di pipeline di dati nel contesto di Full Stack Data Science e abbiamo illustrato un esempio concreto utilizzando dati del mondo reale.
Nel prossimo articolo di questa serie, continueremo ad analizzare lo stack tecnologico della scienza dei dati e discuteremo di come possiamo utilizzare questa pipeline di dati per sviluppare un sistema di ricerca semantica per i miei video di YouTube.
Altro in questa serie 👇
Fonte: towardsdatascience.com
