使用 Python 实现 Youtube 视频下载 Telegram 机器人

我将通过本文向您展示如何使用 Python 构建一个实用电报机器人(Telegram bot)并将其免费部署到 Heroku 上。

本电报机器人可实现油管(Youtube)视频下载功能——用户在电报对话框发送油管视频地址后，机器人将下载该 Youtube 视频并将下载链接发送给用户。

通过这样的方式，您可以轻松在 Telegram 上运营一个 Youtube 视频下载服务供用户订阅，积累自己的用户群体。

我知道这听起来可能有点太复杂了，但相信我，事实并非如此。下面我们开始吧！

设置

首先，请在您的项目目录中添加以下文件：

bot.py
requirements.txt
Procfile

bot.py 文件，您可能从文件后缀名就能猜到它是一个 Python 文件，没错，我们将在其中编写 Telegram 机器人代码。

requirements.txt 文件以及 Procfile 则是在 Heroku 服务器上部署机器人所需的文件。稍后我将讨论这两个文件各自负责的内容。请注意，Procfile 文件没有扩展名！

给机器人编程

我们建议初学者使用如下所示的 Telegram 机器人原始模板，并在其上构建我们的应用程序。

from telegram import *
from telegram.ext import Updater, CommandHandler, CallbackContext, \
    MessageHandler, Filters

def start(update: Update, context: CallbackContext) -> None:
    update.message.reply_text(text='Welcome to URL downloader!\nPlease provide a valid url')


def textHandler(update: Update, context: CallbackContext) -> None:
    if update.message.parse_entities(types=MessageEntity.URL):
        update.message.reply_text(text='You sent a valid URL!', quote=True)

def main():
    TOKEN = "YOUR BOT TOKEN"
    
    updater = Updater(TOKEN, use_context=True)
    updater.dispatcher.add_handler(CommandHandler('start', start))
    updater.dispatcher.add_handler(MessageHandler(Filters.all & ~Filters.command, textHandler, run_async=True))
    updater.start_polling()
    updater.idle()


if __name__ == '__main__':
    main()

理解上述代码片段的作用非常重要。特别是，我们正在创建两个处理函数：一个命令处理函数( def start() ) 和一个消息处理函数( def textHandler() )。

每次用户在 Telegram 聊天中发送 /start 命令时都会执行 start() 函数，并且每次向机器人发送文本消息时都会执行 textHandler() 函数。在本电报机器人实例中，我们不希望用户发送任何类型的纯文本消息。事实上，我们只想接受 URL 类型的文本。这就是我们在 def textHandler() 中添加以下 if 语句的原因：

if update.message.parse_entities(types=MessageEntity.URL)

现在我们已经了解了电报机器人的基础知识，是时候想办法获取用户发送的 URL 的下载地址了。有两种可能的方法来解决这个问题。老式的方法是创建一个网络抓取工具，该抓取工具会转到该特定 URL 并下载内容。但是，这种方法不是最有效的方法，因为它需要大量 RAM 并且速度很慢。

因此，我们将坚持使用为我们完成这项工作的公共 API的新型方法。更具体地说，有很多免费的 API 可以为我们提供几乎任何视频 URL 的下载链接。请注意，这取决于 API 支持的视频网站类型。

对于本应用，我们将使用 get-video-and-audio-url API。

免费注册此 API 后，您将能够通过密钥连接到它并发送 get/post 请求。现在让我们将 API 集成到我们的 bot.py 文件中。

import requests
from telegram import *
from telegram.ext import Updater, CommandHandler, CallbackContext, \
    MessageHandler, Filters


def get_Download_URL_From_API(url):
    API_URL = "https://getvideo.p.rapidapi.com/"
    querystring = {"url": f"{url}"}

    headers = {
        'x-rapidapi-host': "getvideo.p.rapidapi.com",
        'x-rapidapi-key': "f46d0d682dmsh95ed9f4dc7225e2p146570jsn1d8f86b7de46"  # This is your API key token. Keep it secret!
    }

    response = requests.request("GET", API_URL, headers=headers, params=querystring)
    data = response.json()
    return data['streams'][0]['url']

def start(update: Update, context: CallbackContext) -> None:
    update.message.reply_text(text='Welcome to URL downloader!\nPlease provide a valid url')


def textHandler(update: Update, context: CallbackContext) -> None:
    user_message = str(update.message.text)

    if update.message.parse_entities(types=MessageEntity.URL):
        download_url = get_Download_URL_From_API(user_message)
        update.message.reply_text(text=f'Your download url is: {download_url}')



def main():
    TOKEN = "YOUR BOT TOKEN"
    updater = Updater(TOKEN, use_context=True)
    updater.dispatcher.add_handler(CommandHandler('start', start))
    updater.dispatcher.add_handler(MessageHandler(Filters.all & ~Filters.command, textHandler, run_async=True))
    updater.start_polling()
    updater.idle()


if __name__ == '__main__':
    main()

所以我们刚才所做的就是在我们的文件中添加一个名为 def get_Download_URL_From_API(url) 的新函数。这个函数接受一个 url 参数并将这个 url 传递给 API。然后，API 返回一个 json 响应，其中包含诸如标题、网站、上传时间、持续时间、视频格式以及视频下载 url 等数据。

请记住，json 数据的结构因 API 而异，因此查看 API 的 json 响应并查看您要查找的数据项非常重要。在本电报机器人应用中我们需要下载网址字段，可以通过以下方式访问：

data[‘streams’][0][‘url’]

现在让我们回顾一下代码逻辑，每次用户发送一个类型为 URL 的文本消息时，我们将调用 get_Download_URL_From_API(url) 函数，将用户的 url 消息作为参数传递给该函数。然后该函数将向 API 发送一个 get 请求，并返回视频的下载 url 给用户。是不是很简单，就是这样！

接下来，让我们看看这个机器人是如何运作的。

您可以向代码添加更多功能，例如打印视频标题、时长、播放次数等，使其看起来更专业、更美观。

在 Heroku 上部署机器人

为方便部署，我们需要对 bot.py 文件进行一些小改动。我们先导入 os，然后设置监听 webhook 端口号。

import os
PORT = int(os.environ.get('PORT', 5000))

然后将以下代码行：

updater.start_polling()

替换为：

updater.start_webhook(listen="0.0.0.0",
                          port=int(PORT),
                          url_path=TOKEN)
updater.bot.setWebhook('https://yourherokuappname.herokuapp.com/' + TOKEN)

我们从长轮询(long polling)切换到 webhook 的主要原因在于如果使用长轮询频繁地请求 Telegram 服务器更新机器人状态，将是非常低效的。而触发式 webhook 则能够在需要时向机器人主动推送处理消息。

现在让我们处理之前提到的 Procfile 和 requirements.txt 文件。Procfile 文件定义应用在启动时所需额外执行的命令。我们将以下代码添加到 Procfile 文件：

web: python3 bot.py

简单来说，上面一行告诉 Heroku 服务器执行 bot.py 文件。

同样，requirements.txt 文件告诉 Heroku 服务器需要哪些依赖项才能运行 bot.py 文件。要生成 requirements.txt 文件，只需进入项目目录并在 cmd 或终端中执行以下命令：

pip freeze > requirements.txt

现在我们拥有通过 Telegram 机器人下载 Youtube 视频所需的一切功能，但在将本机器人应用部署到 Heroku 上之前，您需要将以上代码文件部署到 Github 中。Heruko 将直接从您的 Github 项目库中部署应用。

您接下来只需执行以下步骤：

登录您的 Heroku 帐户 → 创建新应用 → 将您的 Github 项目库连接到您的 Heruko 帐户 → 部署您的机器人

完成以上步骤后，我们的电报机器人就大功告成了！快去将您的油管视频下载机器人分享给大家吧！