forked from tgbot-collection/ytdlbot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
ytdl.py
132 lines (102 loc) · 4.5 KB
/
ytdl.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
#!/usr/local/bin/python3
# coding: utf-8
# ytdlbot - new.py
# 8/14/21 14:37
#
__author__ = "Benny <benny.think@gmail.com>"
import logging
import os
import re
import tempfile
import typing
from pyrogram import Client, filters, types
from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup
from tgbot_ping import get_runtime
from downloader import convert_flac, upload_hook, ytdl_download
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(filename)s [%(levelname)s]: %(message)s')
api_id = int(os.getenv("APP_ID", 0))
api_hash = os.getenv("APP_HASH")
token = os.getenv("TOKEN")
app = Client("ytdl", api_id, api_hash, bot_token=token, workers=100)
@app.on_message(filters.command(["start"]))
def start_handler(client: "Client", message: "types.Message"):
chat_id = message.chat.id
logging.info("Welcome to youtube-dl bot!")
client.send_chat_action(chat_id, "typing")
client.send_message(message.chat.id, "Wrapper for youtube-dl.")
@app.on_message(filters.command(["help"]))
def help_handler(client: "Client", message: "types.Message"):
chat_id = message.chat.id
client.send_chat_action(chat_id, "typing")
client.send_message(chat_id, "Stop working? "
"Wait a few seconds, send your link again or report bugs at "
"https://github.com/tgbot-collection/ytdlbot/issues")
@app.on_message(filters.command(["ping"]))
def ping_handler(client: "Client", message: "types.Message"):
chat_id = message.chat.id
client.send_chat_action(chat_id, "typing")
bot_info = get_runtime("botsrunner_ytdl_1", "YouTube-dl")
client.send_message(chat_id, bot_info)
@app.on_message(filters.command(["about"]))
def help_handler(client: "Client", message: "types.Message"):
chat_id = message.chat.id
client.send_chat_action(chat_id, "typing")
client.send_message(chat_id, "YouTube-DL by @BennyThink\n"
"GitHub: https://github.com/tgbot-collection/ytdlbot")
@app.on_message()
def download_handler(client: "Client", message: "types.Message"):
if message.chat.type != "private" and not message.text.lower().startswith("/ytdl"):
logging.warning("%s, it's annoying me...🙄️ ", message.text)
return
chat_id = message.chat.id
url = re.sub(r'/ytdl\s*', '', message.text)
logging.info("start %s", url)
if not re.findall(r"^https?://", url.lower()):
message.reply_text("I think you should send me a link.", quote=True)
return
bot_msg: typing.Union["types.Message", "typing.Any"] = message.reply_text("Processing", quote=True)
client.send_chat_action(chat_id, 'upload_video')
temp_dir = tempfile.TemporaryDirectory()
result = ytdl_download(url, temp_dir.name, bot_msg)
logging.info("Download complete.")
markup = InlineKeyboardMarkup(
[
[ # First row
InlineKeyboardButton( # Generates a callback query when pressed
"audio",
callback_data="audio"
)
]
]
)
if result["status"]:
client.send_chat_action(chat_id, 'upload_document')
video_path = result["filepath"]
bot_msg.edit_text('Download complete. Sending now...')
client.send_video(chat_id, video_path, supports_streaming=True, caption=url,
progress=upload_hook, progress_args=(bot_msg,), reply_markup=markup)
bot_msg.edit_text('Download success!✅')
else:
client.send_chat_action(chat_id, 'typing')
tb = result["error"][0:4000]
bot_msg.edit_text(f"{url} download failed❌:\n```{tb}```")
temp_dir.cleanup()
@app.on_callback_query()
def answer(client: "Client", callback_query: types.CallbackQuery):
callback_query.answer(f"Converting to audio...please wait patiently")
msg = callback_query.message
chat_id = msg.chat.id
mp4_name = msg.video.file_name # 'youtube-dl_test_video_a.mp4'
flac_name = mp4_name.replace("mp4", "m4a")
with tempfile.NamedTemporaryFile() as tmp:
logging.info("downloading to %s", tmp.name)
client.send_chat_action(chat_id, 'record_video_note')
client.download_media(msg, tmp.name)
logging.info("downloading complete %s", tmp.name)
# execute ffmpeg
client.send_chat_action(chat_id, 'record_audio')
flac_tmp = convert_flac(flac_name, tmp)
client.send_chat_action(chat_id, 'upload_audio')
client.send_audio(chat_id, flac_tmp)
if __name__ == '__main__':
app.run()