Automates YouTube transcript timestamps for DataTalksClub GitHub issues.
import os
import json

import openai
from dotenv import load_dotenv

from src.github_utility import GitHubUtility
from src.youtube_utility import YouTubeUtility
from src.processing_pipeline import ProcessingPipeline
from src.transcript_processor import TranscriptProcessor


def load_env_vars(var_names):
    """
    Load environment variables into a dictionary.

    :param var_names: List of environment variable names to load
    :return: Dictionary of environment variables
    :raises EnvironmentError: If any of the environment variables are not found
    """
    env_vars = {}
    for var in var_names:
        value = os.getenv(var)
        if value is None:
            raise EnvironmentError(f"{var} not found in environment variables")
        env_vars[var] = value
    return env_vars


def load_config(filename):
    """Load configuration from a JSON file.

    :param filename: Path to the JSON configuration file
    :return: Loaded JSON data
    """
    with open(filename, 'r') as f:
        return json.load(f)


def process_repositories(config, env_vars):
    """Process each repository based on the configuration and environment variables.

    :param config: Loaded JSON configuration
    :param env_vars: Loaded environment variables
    """
    for repo_config in config['repositories']:
        organization = repo_config['organization']
        repo_name = repo_config['repo_name']
        playlist_urls = repo_config['playlist_urls']

        print(f"Processing repository: {organization}/{repo_name}")

        github_util = GitHubUtility(
            env_vars['GITHUB_ACCESS_TOKEN'], f"{organization}/{repo_name}"
        )

        for page in range(1, 3):
            print(f"Processing page {page} for issues...")

            video_titles = github_util.get_video_titles_from_issues(page)
            print(f"Video titles: {video_titles}")

            for playlist_url in playlist_urls:
                print(f"Processing playlist: {playlist_url}")

                youtube_util = YouTubeUtility(env_vars['YOUTUBE_API_KEY'], playlist_url)
                openai.api_key = env_vars['OPENAI_API_KEY']
                transcript_processor = TranscriptProcessor(openai, repo_name=repo_name)

                pipeline = ProcessingPipeline(
                    github_util, youtube_util, transcript_processor
                )
                pipeline.process_video(
                    f"{organization}/{repo_name}", playlist_url, video_titles
                )


if __name__ == "__main__":
    """Main function to execute the processing pipeline."""

    print("Loading environment variables...")
    load_dotenv()
    env_vars_needed = ["YOUTUBE_API_KEY", "GITHUB_ACCESS_TOKEN", "OPENAI_API_KEY"]
    env_vars = load_env_vars(env_vars_needed)

    print("Loading JSON configuration...")
    config = load_config('config.json')

    print("Starting to process repositories...")
    process_repositories(config, env_vars)

main.py
dimzachar/DataTalksClub-Timestamp-Issues
import requests
from bs4 import BeautifulSoup
from github import Github


class GitHubUtility:
    """
    Utility class to interact with GitHub for specific repository operations.
    """

    def __init__(self, access_token, repo_name):
        """
        Initialize the GitHub utility.

        :param access_token: GitHub API access token
        :param repo_name: Name of the GitHub repository
        """
        self.g = Github(access_token)
        print(f"Attempting to access repo: {repo_name}")
        self.repo = self.g.get_repo(repo_name)
        self.processed_issues = set()
        print("Successfully accessed repo")

    def get_video_titles_from_issues(self, page=1):
        """
        TODO: Use Github API instead of webscrape (??)
        Fetch video titles from GitHub issues.

        :param page: Page number for GitHub issues pagination
        :return: List of video titles
        """
        print(f"Fetching video titles from issues, page {page}")
        issues_url = f"https://github.com/{self.repo.full_name}/issues?page={page}&q=is%3Aissue+is%3Aopen+sort%3Aupdated-desc"

        try:
            response = requests.get(issues_url)
            response.raise_for_status()
        except requests.RequestException as e:
            print(f"An error occurred while fetching issues: {e}")
            return []

        soup = BeautifulSoup(response.content, "html.parser")
        issue_titles = [
            el.text.strip() for el in soup.find_all("a", class_="Link--primary")
        ]
        video_titles = [
            title.replace("Timecodes for ", "").replace('"', '')
            for title in issue_titles
            if title.startswith("Timecodes for ")
        ]
        print(f"Fetched {len(video_titles)} video titles")
        return video_titles

    def is_already_processed(self, issue_title):
        """
        Check if the issue has already been processed.

        :param issue_title: Title of the issue to check
        :return: True if already processed, False otherwise
        """
        if issue_title in self.processed_issues:
            return True

        issues = self.repo.get_issues(state="open")
        for issue in issues:
            if issue.title == issue_title:
                for comment in issue.get_comments():
                    if comment.user.login == self.g.get_user().login:
                        self.processed_issues.add(issue_title)
                        return True
        return False

    def add_issue_comment_with_confirmation(
        self, issue_titles, comment_body, confirm_prompt=True
    ):
        """
        TODO: Modularize Code (??)
        Add a comment to a GitHub issue with optional confirmation.

        :param issue_titles: List of issue titles to add comments to
        :param comment_body: Text to include in the comment
        :param confirm_prompt: Whether to prompt user for confirmation before adding comment
        """
        print("Entering add_issue_comment_with_confirmation function")
        issues = self.repo.get_issues(state="open")
        for issue in issues:
            print(f"Checking issue: {issue.title}")
            if issue.title.strip() in issue_titles:
                print(f"\nAdding comment to issue '{issue.title}':\n")
                print(comment_body)
                if confirm_prompt:
                    proceed = input("\nProceed with commit? [y/N]: ")
                else:
                    proceed = 'y'

                if proceed.lower() == "y":
                    issue.create_comment(comment_body)
                    print("Comment added.")
                else:
                    print("Skipped.")
                break

src/github_utility.py
dimzachar/DataTalksClub-Timestamp-Issues
from youtube_transcript_api import (
    NoTranscriptFound,
    TranscriptsDisabled,
    YouTubeTranscriptApi,
)


class ProcessingPipeline:
    """
    Class to manage the processing pipeline for adding timecodes to GitHub issues based on video transcripts.
    """

    def __init__(self, github_util, youtube_util, transcript_processor):
        """
        Initialize the ProcessingPipeline.

        :param github_util: GitHubUtility instance for GitHub operations
        :param youtube_util: YouTubeUtility instance for YouTube operations
        :param transcript_processor: TranscriptProcessor instance for transcript operations
        """
        self.github_util = github_util
        self.youtube_util = youtube_util
        self.transcript_processor = transcript_processor

    def match_titles_and_urls(self, issues_titles, video_info):
        """
        Match video titles to their corresponding URLs.

        :param issues_titles: List of issue titles
        :param video_info: List of video information (title, url)
        :return: Dictionary with matched video titles and their video IDs
        """
        matched_videos = {}
        for title in issues_titles:
            for video_title, url in video_info:
                if title.lower() == video_title.lower():
                    video_id = url.split("watch?v=")[1]
                    matched_videos[title] = video_id
                    break
        return matched_videos

    def process_video(self, repo_name, playlist_id, issues_titles):
        """
        Process videos to add timecodes to GitHub issues.

        :param repo_name: GitHub repository name
        :param playlist_id: YouTube playlist ID
        :param issues_titles: List of GitHub issue titles to process
        """
        # Get the video info from YouTube
        video_info = self.youtube_util.get_video_urls_and_titles()

        # Match titles and URLs
        matched_videos = self.match_titles_and_urls(issues_titles, video_info)

        for title, video_id in matched_videos.items():
            print(title)

            new_title = f'Timecodes for "{title}"'
            if self.github_util.is_already_processed(new_title):
                print(f"Issue '{new_title}' has already been processed. Skipping...")
                continue

            print(new_title)
            video_duration = self.youtube_util.get_video_duration(video_id)
            print(f"Video duration: {video_duration} seconds")

            chunk_size = 150 if video_duration <= 600 else 400

            error_messages = {
                TranscriptsDisabled: "Transcripts are disabled for the video.",
                NoTranscriptFound: "No transcript was found in the requested language for video.",
            }

            try:
                transcript = YouTubeTranscriptApi.get_transcript(video_id)
                comment_body = self.transcript_processor.process_transcript(
                    transcript, chunk_size
                )
                self.github_util.add_issue_comment_with_confirmation(
                    new_title, comment_body
                )
            except (TranscriptsDisabled, NoTranscriptFound) as e:
                print(
                    f"Encountered an issue with the video `{video_id}`: {error_messages[type(e)]}"
                )
                self.github_util.add_issue_comment_with_confirmation(
                    new_title, error_messages[type(e)]
                )
            except Exception as e:
                print(
                    f"An unexpected error occurred while processing the video `{video_id}`: {e}"
                )

src/processing_pipeline.py
dimzachar/DataTalksClub-Timestamp-Issues
import os
import re
from datetime import timedelta


class TranscriptProcessor:
    """
    Class to process transcripts and generate timestamps.
    """

    def __init__(self, openai, repo_name):
        """
        Initialize the TranscriptProcessor.

        :param openai: OpenAI API client
        :param repo_name: Repository name to set as the output directory
        """
        self.openai = openai
        self.output_directory = repo_name
        os.makedirs(self.output_directory, exist_ok=True)

    def clean_text(self, text):
        """
        Clean text by removing timestamps.

        :param text: Text to clean
        :return: Cleaned text
        """
        return re.sub(r'(?<!^)(\d{1,2}:\d{2}:\d{2})|(\d{1,2}:\d{2})', '', text).strip()

    def process_chunk(self, current_text_chunk, chunk_start_time):
        """
        Process a chunk of transcript text.

        :param current_text_chunk: The current text chunk to process
        :param chunk_start_time: The start time of the chunk
        :return: Processed timestamp
        """
        prompt = f"Summarize the following in 6 words or less: '{current_text_chunk}'. Strict 6-word limit."
        messages = [
            {"role": "system", "content": "You are a YouTube video creator."},
            {"role": "user", "content": prompt},
        ]

        # Query OpenAI API for text summarization
        response = self.openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=50,
            n=1,
            temperature=0,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0,
        )

        description = self.clean_text(response.choices[0].message['content']).rstrip(
            '.'
        )
        timestamp_string = str(timedelta(seconds=int(chunk_start_time)))
        polished_timestamp = (
            f"{timestamp_string} - {description}".replace("\n", " ")
            .replace('"', '')
            .replace(" - -", " -")
            .rstrip('.')
        )

        return polished_timestamp

    def process_transcript(self, whole_transcript, chunk_size):
        """
        Process an entire transcript.

        :param whole_transcript: The transcript to process
        :param chunk_size: Size of each chunk for processing
        :return: Processed transcript as a comment body
        """
        print("Sending data to OpenAI for processing...")

        current_text_chunk = ""
        comment_body = []
        chunk_start_time = 0

        print("Processing transcript...")
        for current_line in whole_transcript:
            if chunk_start_time == 0:
                chunk_start_time = current_line['start']

            current_text_chunk += " " + current_line['text']

            if len(' '.join(current_text_chunk).split(" ")) > chunk_size:
                polished_timestamp = self.process_chunk(
                    ' '.join(current_text_chunk), chunk_start_time
                )
                comment_body.append(polished_timestamp)

                chunk_start_time = 0
                current_text_chunk = ""

        # Save to output directory
        with open(
            f"{self.output_directory}/timestamps.txt", "a", encoding="utf-8"
        ) as file:
            file.write('\n'.join(comment_body) + '\n')

        return '\n'.join(comment_body)

src/transcript_processor.py
dimzachar/DataTalksClub-Timestamp-Issues
import isodate
from googleapiclient.discovery import build


class YouTubeUtility:
    """
    Class to manage YouTube operations.
    """

    def __init__(self, api_key, playlist_id):
        """
        Initialize the YouTubeUtility class.

        :param api_key: YouTube API key
        :param playlist_id: YouTube playlist ID
        """
        self.youtube = build('youtube', 'v3', developerKey=api_key)
        self.playlist_id = playlist_id

    def update_playlist(self, new_playlist_id):
        """
        Update the playlist ID.

        :param new_playlist_id: New YouTube playlist ID
        """
        self.playlist_id = new_playlist_id

    def get_video_urls_and_titles(self):
        """
        Retrieve video URLs and titles from a playlist.

        :return: List of tuples containing video titles and URLs
        """
        video_info = []
        try:
            request = self.youtube.playlistItems().list(
                part="snippet", maxResults=500, playlistId=self.playlist_id
            )

            while request:
                response = request.execute()
                for item in response.get('items', []):
                    snippet = item.get('snippet', {})
                    video_id = snippet.get('resourceId', {}).get('videoId', '')
                    video_title = snippet.get('title', '')
                    video_url = f"https://www.youtube.com/watch?v={video_id}"
                    video_info.append((video_title, video_url))

                request = self.youtube.playlistItems().list_next(request, response)
        except Exception as e:
            print(f"An error occurred: {e}")
            return []
        return video_info

    def get_video_duration(self, video_id):
        """
        Retrieve the duration of a video.

        :param video_id: YouTube video ID
        :return: Video duration in seconds
        """
        try:
            request = self.youtube.videos().list(part="contentDetails", id=video_id)
            response = request.execute()
            video_duration = (
                response.get('items', [{}])[0]
                .get('contentDetails', {})
                .get('duration', '')
            )
            return isodate.parse_duration(video_duration).total_seconds()
        except Exception as e:
            print(f"An error occurred: {e}")
            return 0

src/youtube_utility.py
dimzachar/DataTalksClub-Timestamp-Issues