YouTube_API/youtube_api.py

# -*- coding: utf-8 -*-

# Sample Python code for youtube.playlists.update
# See instructions for running these code samples locally:
# https://developers.google.com/explorer-help/code-samples#python

import argparse
import os
import re

from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow

scopes = ["https://www.googleapis.com/auth/youtube.force-ssl"]

channels = {
    "breakbeat_sesiones": "PLZFfy80_qguYVqbpQlX-aHdiYHTwc6-wW",
    "breakbeat_temas": "PLZFfy80_qguZnpBYfSFlkMdjGIfVcz7e7",
    "wods": "PLZFfy80_qguYCbTDQD0GSK4zUzEjbQMQT",
    "rap_español": "PLZFfy80_qgub44wWWDcAykdQCJNCXo4jp",
}


def get_playlist_length(youtube, playlist_id):
    items = youtube.playlists().list(part="snippet,contentDetails", id=playlist_id)
    playlist_info = items.execute()

    return playlist_info["items"][0]["contentDetails"]["itemCount"]


def list_songs(youtube, channel_id, stdout=False):
    max_results = 50
    request = youtube.playlistItems().list(
        part="snippet,contentDetails", maxResults=max_results, playlistId=channel_id
    )
    response = request.execute()
    print()

    # Process the first page of results
    deleted = 0
    songs_list = []

    for idx, item in enumerate(response["items"], start=1):
        title = item["snippet"]["title"]
        if title == "Deleted video":
            deleted += 1
            continue
        songs_list.append(f"{idx-deleted:02d} - {title}")

    # Continue making requests for additional pages using pageToken
    new_starting = 51
    while "nextPageToken" in response:
        request = youtube.playlistItems().list(
            part="snippet",
            playlistId=channel_id,
            maxResults=max_results,
            pageToken=response["nextPageToken"],
        )
        response = request.execute()

        # Process the results for the additional pages
        for idx, item in enumerate(response["items"], start=new_starting):
            title = item["snippet"]["title"]
            if title == "Deleted video":
                deleted += 1
                continue
            songs_list.append(f"{idx-deleted:02d} - {title}")
        new_starting += 50

    if stdout:
        # Print to standard output
        print("\n".join(songs_list))

    # get_key_by_value = lambda dict, val: next(
    #     (k for k, v in dict.items() if v == val), None
    # )
    def _get_key_by_value(dict, val):
        return next((k for k, v in dict.items() if v == val), None)

    channel_name = _get_key_by_value(channels, channel_id)

    # Save to file if output_file is provided
    with open(f"{channel_name} songs", "w", encoding="utf-8") as file:
        file.write("\n".join(songs_list))

    print("\nList created successfully!")


def add_song(youtube, channel_id, song):
    pos = get_playlist_length(youtube, channel_id)

    request = youtube.playlistItems().insert(
        part="snippet",
        body={
            "snippet": {
                "playlistId": channel_id,
                "position": pos,
                "resourceId": {"kind": "youtube#video", "videoId": song},
            }
        },
    )
    # response = request.execute()
    request.execute()
    print("\nSong added successfully!")


def main():
    parser = argparse.ArgumentParser(description="Manage YouTube playlist songs.")
    parser.add_argument(
        "-l", "--list", metavar="channel_name", help="List songs in the playlist"
    )
    parser.add_argument(
        "-a", "--add", metavar="channel_name", help="Name of the playlist"
    )
    parser.add_argument("song", nargs="?", help="Name of the song")

    args = parser.parse_args()
    channel_name = args.add or args.list
    channel_id = channels.get(channel_name)

    if not args.list and not args.add:
        parser.error("Please provide either -l or -a option.")

    if not channel_id:
        print(f"Error: Channel '{channel_name}' not found.")
        return

    # Disable OAuthlib's HTTPS verification when running locally.
    # *DO NOT* leave this option enabled in production.
    os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"

    # scopes is a global variable
    client_secret = "./client_secrets.json"
    flow = InstalledAppFlow.from_client_secrets_file(client_secret, scopes)
    credentials = flow.run_local_server(port=0)

    api_service_name = "youtube"
    api_version = "v3"
    youtube = build(api_service_name, api_version, credentials=credentials)

    if args.list:
        list_songs(youtube, channel_id, stdout=False)
    elif args.add:
        if not args.song:
            parser.error("Please provide the name of the song with -a.")
        song = args.song
        # this assumes a song is passed as in either of the following ways
        # https://youtu.be/eEnyi9L6KP4 --> split by "/"
        # https://youtube.com/watch?v=Ez-gizOF0Wo --> split by "="
        song = re.split(r"/|=", song)[-1]
        add_song(youtube, channel_id, song)


if __name__ == "__main__":
    main()