Telegram Bot. Download youtube video and send

index.py

import os
import asyncio
import youtube_dl
import logging
from urllib.parse import urlparse

from telegram import __version__ as TG_VER

try:
    from telegram import __version_info__
except ImportError:
    __version_info__ = (0, 0, 0, 0, 0)

if __version_info__ < (20, 0, 0, "alpha", 1):
    raise RuntimeError(
        "Eski versiyonlarda çalışmaz. Lütfen en son sürümü yükleyin."
    )
from telegram import ForceReply, Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters

from dotenv import load_dotenv

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)

load_dotenv()

NGROK_URL = os.getenv('NGROK_URL')
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')
BOT_NAME = os.getenv('BOT_NAME')


def is_youtube_url(url):
    parsed_url = urlparse(url)
    return parsed_url.hostname in ['www.youtube.com', 'youtube.com', 'youtu.be']


# download youtube video as mp3
async def download_youtube_video(url):
    ydl_opts = {
        'format': 'bestaudio/best',
        'outtmpl': '%(title)s.%(ext)s',
        'postprocessors': [{
            'key': 'FFmpegExtractAudio',
            'preferredcodec': 'mp3',
            'preferredquality': '192',
        }],
    }

    with youtube_dl.YoutubeDL(ydl_opts) as ydl:
        result = await asyncio.get_event_loop().run_in_executor(None, ydl.extract_info, url)
        # get downloaded video file name

        return f"{result['title']}.mp3"


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Send a message when the command /start is issued."""
    user = update.effective_user
    await update.message.reply_html(
        rf"Hi {user.mention_html()}!",
        reply_markup=ForceReply(selective=True),
    )


async def download(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    url = update.message.text
    if is_youtube_url(url):
        await update.message.reply_text('MP3 olarak indiriliyor...')
        title = await download_youtube_video(url)
        await update.message.reply_text(f'{title} indirildi.')
        await update.message.reply_audio(open(title, 'rb'))
    else:
        await update.message.reply_text('Youtube adresi giriniz.')


async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Help!")


async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(update.message.text)


def main() -> None:
    """Start the bot."""
    # Create the Application and pass it your bot's token.
    application = Application.builder().token(TELEGRAM_TOKEN).build()

    # on different commands - answer in Telegram
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("youtube", download))

    # on non command i.e message - echo the message on Telegram
    application.add_handler(MessageHandler(
        filters.TEXT & ~filters.COMMAND, download))

    # Run the bot until the user presses Ctrl-C
    application.run_polling()


if __name__ == "__main__":
    main()