convert youtube video to captivate.fm podcast

youtube_to_captivate.py

#!/usr/bin/python

import os
from datetime import datetime, time
from typing import Dict, Union

import requests
from dotenv import load_dotenv

from yt_dlp import YoutubeDL

load_dotenv()

USER_ID = os.getenv("USER_ID")
API_KEY = os.getenv("CAPTIVATE_API_KEY")
SHOWS_ID = os.getenv("SHOWS_ID")


def youtube_to_captivatefm(url: str):
    info = download_youtube_video(url)
    formated_upload_date = format_date(info["upload_date"])
    print("getting user token")
    token = get_token(user_id=USER_ID, api_key=API_KEY)
    print("uploading media", info["file_name"])
    media_id = upload_media(token=token, file_name=info["file_name"])
    print("creating podcast")
    episode_url = create_podcast(
        token=token,
        media_id=media_id,
        date=formated_upload_date,
        title=info["title"],
        shownotes=info["description"],
    )
    print(episode_url)


def get_token(user_id: str, api_key: str) -> Union[str, None]:
    """
    This function gets a token from the captivate.fm API, using the user_id and api_key as authentication.

    :param user_id: The user_id of the account to get a token for
    :type user_id: str
    :param api_key: The api_key for the account
    :type api_key: str
    :return: The token from the API
    :rtype: Union[str, None]
    :raise: Exception if the API request fails.
    """
    url = "https://api.captivate.fm/authenticate/token"

    payload = {"username": user_id, "token": api_key}
    files = []
    headers = {}

    try:
        response = requests.request(
            "POST", url, headers=headers, data=payload, files=files
        )
        response.raise_for_status()
        r = response.json()
        return r["user"]["token"]
    except requests.exceptions.HTTPError as error:
        print(f"An HTTP error occurred: {error}")
        return


def download_youtube_video(url: str) -> Dict[str, str]:
    """
    This function downloads a YouTube video from the given URL, and returns a dictionary containing the video's title, description, file name and upload date.

    :param url: The URL of the YouTube video to download
    :type url: str
    :return: A dictionary containing the video's title, description, file name, and upload date
    :rtype: Dict[str, str]
    """
    with YoutubeDL({
        "format": "bestaudio/best",
        "outtmpl": "%(title)s.%(ext)s",
    }) as ydl:
        video_info = ydl.extract_info(url)
    return {
        "title": video_info["title"],
        "description": video_info.get("description"),
        "file_name": video_info['requested_downloads'][0]['filepath'],
        "upload_date": video_info.get("upload_date"),
    }


def format_date(date_str: str) -> Union[str, None]:
    """
    This function takes in a date string in the format "YYYYMMDD" and returns it in the format "YYYY-MM-DD 12:00:00"


    :param date_str: The date string to be formatted
    :type date_str: str
    :return: The formatted date string
    :rtype: Union[str, None]
    """
    try:
        time_str = time(6, 30, 0)
        date_obj = datetime.strptime(date_str, "%Y%m%d").date()
        dt_obj = datetime.combine(date_obj, time_str)
        formatted_date = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
        return formatted_date

    except ValueError as e:
        print(e)
        return None


def upload_media(token: str, file_name: str) -> Union[str, None]:
    """
    This function uploads a file to captivate.fm using an API, and returns the media_id.

    :param token: The API token to be used for authentication
    :type token: str
    :param file_name: The name of the file to be uploaded
    :type file_name: str
    :return: The media_id of the uploaded file
    :rtype: Union[str, None]
    :raise: Exception if the file upload fails.
    """
    headers = {
        "Authorization": "Bearer " + token,
    }

    files = {
        "file": open(
            file_name,
            "rb",
        ),
    }

    try:
        response = requests.post(
            f"https://api.captivate.fm/shows/{SHOWS_ID}/media",
            headers=headers, files=files,)
        response.raise_for_status()
        r = response.json()
        return r["media"]["id"]
    except requests.exceptions.HTTPError as error:
        print(f"An HTTP error occurred: {error}")
        return None


def create_podcast(
    token: str,
    title: str,
    media_id: str,
    date: str,
    shownotes: str,
    shows_id: str = SHOWS_ID,
    status: str = "draft",
    episode_season: str = "1",
    episode_number: str = "1",
) -> Union[str, None]:
    """
    This function creates a podcast on captivate.fm using the API, by taking in all the parameters and putting them in the payload, and returns the response.

    :param token: The API token to be used for authentication
    :type token: str
    :param title: The title of the episode
    :type title: str
    :param media_id: The media_id of the episode
    :type media_id: str
    :param date: The date of the episode
    :type date: str
    :param shownotes: The shownotes of the episode
    :type shownotes: str
    :param shows_id: The id of the show
    :type shows_id: str
    :param status: The status of the episode
    :type status: str
    :param episode_season: The season of the episode
    :type episode_season: str
    :param episode_number: The number of the episode
    :type episode_number: str

    """
    url = "https://api.captivate.fm/episodes"

    payload = {
        "shows_id": shows_id,
        "title": title,
        "media_id": media_id,
        "date": date,
        "status": status,
        "shownotes": shownotes,
        "episode_season": episode_season,
        "episode_number": episode_number,
    }

    files = []
    headers = {"Authorization": "Bearer " + token}

    try:
        response = requests.request(
            "POST", url, headers=headers, data=payload, files=files
        )
        response.raise_for_status()
        r = response.json()

        return f"https://player.captivate.fm/episode/{r['id']}"
    except requests.exceptions.HTTPError as error:
        print(f"An HTTP error occurred: {error}")
        return None


if __name__ == "__main__":
    youtube_to_captivatefm(input("Enter url of youtube video: "))