get_transcript_df.py

from youtube_transcript_api import YouTubeTranscriptApi
import pandas as pd
from datetime import timedelta
from urllib.parse import urlparse, parse_qs

TRIGGER_LENGTH = 40000  # in milliseconds


def create_subtitle_df(video_urls):
  base_url_format = "https://www.youtube.com/watch?v={video_id}"
  video_ids = [extract_video_id(video_url) for video_url in video_urls]
  transcripts = [YouTubeTranscriptApi.get_transcript(video_id, languages=['en', 'en-US', 'en-GB']) for video_id in video_ids]

  merged_text_list_all = []
  merged_time_list_all = []
  base_url_list_all = []
  for transcript, video_id in zip(transcripts, video_ids):
    merged_text_list, merged_time_list = merge_text_by_time(transcript)
    base_url_list = [base_url_format.format(video_id = video_id)]*len(merged_time_list)
    merged_text_list_all = merged_text_list_all + merged_text_list
    merged_time_list_all = merged_time_list_all + merged_time_list
    base_url_list_all = base_url_list_all + base_url_list

  return create_split_video_df(merged_text_list_all, merged_time_list_all, base_url_list_all)


def create_split_video_df(merged_text_list, merged_time_list, base_url_list):
  query_params_format = "&t={start}s"
  url_list = [base_url + query_params_format.format(start=merged_time) for base_url, merged_time in zip(base_url_list, merged_time_list)]
  split_video_df = pd.DataFrame({'text': merged_text_list, 'start': url_list})
  return split_video_df


def merge_text_by_time(transcript, trigger_length = TRIGGER_LENGTH):

  merged_text_list = []
  merged_time_list = []
  current_time = timedelta(seconds = 0)  # Initialize to 0
  current_text = ""
  split_time = timedelta(seconds = 0)  # Initialize to 0

  for subtitle in transcript:
    start_time = timedelta(seconds = subtitle['start'])
    duration_time = timedelta(seconds = subtitle['duration'])
    end_time = start_time + duration_time

    # If the current line starts after the next trigger time, add the current text to the list and start a new one.
    if current_time > split_time + timedelta(milliseconds=trigger_length):
      merged_text_list.append(current_text)
      merged_time_list.append(current_time.total_seconds())
      current_text = ""
      split_time = current_time

    # Append the text for the current line to the current text.
    current_text += " " + subtitle['text']
    current_time = start_time

  # Add the last bit of text to the list.
  if current_text:
    merged_text_list.append(current_text)
    merged_time_list.append(current_time.total_seconds())

  return merged_text_list, merged_time_list

from urllib.parse import urlparse, parse_qs

def extract_video_id(youtube_url):
    # Parse the URL into components
    parsed_url = urlparse(youtube_url)
    # Extract the query part of the URL and parse it into a dictionary
    query_params = parse_qs(parsed_url.query)
    # Get the video ID from the dictionary
    video_id = query_params.get('v')
    return video_id[0] if video_id else None