/
__init__.py
150 lines (135 loc) · 6.3 KB
/
__init__.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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
try:
from argparse import ArgumentParser, Namespace
except (ImportError, ModuleNotFoundError) as e:
print("Modules needed not found: " + str(e))
from out import cPrint, Colors
from upgrader import PiPUpgrader
from database import DatabaseOperationsBase
from utils import Constants, logger
from handlers.StartHandler import StartHandler
from handlers.HelpHandler import HelpHandler
from handlers.DeveloperHandler import DeveloperHandler
from handlers.VideoIDHandler import VideoIDHandler
from handlers.URLHandler import URLHandler
from handlers.TextHandler import TextHandler
from handlers.UnexpectedHandler import UnexpectedHandler
def handler_definer():
# type: () -> list
from telegram import MessageEntity
from telegram.ext import CommandHandler, MessageHandler, Filters
handlers = []
with open(Constants.A_APP_MESSAGES, 'r') as messages_file:
messages: dict = messages_file.read()
start = StartHandler(messages["welcome"])
help_handler = HelpHandler(messages["help"])
dev = DeveloperHandler(messages["dev"])
video = VideoIDHandler(messages["video_id"])
url = URLHandler(messages["url_messages"])
text = TextHandler(messages["text"])
unexpected = UnexpectedHandler(messages["unexpected"])
handlers.append(CommandHandler("start", start.start))
handlers.append(CommandHandler("help", help_handler.help))
handlers.append(CommandHandler("develop", dev.develop))
handlers.append(MessageHandler(Filters.command, video.video_handler))
handlers.append(MessageHandler(Filters.text & (Filters.entity(MessageEntity.URL) |
Filters.entity(MessageEntity.TEXT_LINK)), url.url_handler))
handlers.append(MessageHandler(Filters.text, text.message_handler))
handlers.append(MessageHandler(Filters.all, unexpected.unexpected))
return handlers
def main(arguments: Namespace):
try:
import pickle
import logging
import secrets
import string
from os import path
from telegram.ext import Updater
except (ImportError, ModuleNotFoundError) as import_error:
print("Modules needed not found: " + str(import_error))
exit(-1)
else:
token = arguments.token
youtube_api_key = arguments.youtube
creator_id = arguments.creator
database_user = arguments.db_user
database_password = arguments.db_password
must_show_version = arguments.version
if must_show_version:
cPrint("Version: " + Constants.A_APP_VERSION + "-" + Constants.A_APP_TAG_R +
" (" + Constants.A_APP_TAG + ")", Colors.BOLD)
exit(0)
if not path.exists(Constants.A_APP_DATA_FILE):
if not token:
raise ValueError("You must add token at least the first time you execute this app")
elif not youtube_api_key:
raise ValueError("You must include the YouTube API Key at least the first time you execute this app")
elif not creator_id:
raise ValueError("You must include the creator ID (Telegram) at least the first time you execute "
"this app")
else:
if not database_user:
database_user = "youtube_md_bot"
if not database_password:
alphabet = string.ascii_letters + string.digits
database_password = ''.join(secrets.choice(alphabet) for i in range(32))
with open(Constants.A_APP_DATA_FILE, "wb") as app_data_file:
app_data = {"TOKEN": token,
"YT_API": youtube_api_key,
"CREATOR_ID": creator_id,
"DB_USER": database_user,
"DB_PASSWORD:": database_password}
pickle.dump(app_data, app_data_file, pickle.HIGHEST_PROTOCOL)
main(arguments)
else:
cPrint("Initializing bot...", Colors.GREEN)
cPrint("Looking for packages updates...", Colors.GREEN)
upgrader = PiPUpgrader(Constants.A_APP_REQ_FILE)
upgrader.upgradePackages()
cPrint("Obtaining values...", Colors.GREEN)
with open(Constants.A_APP_DATA_FILE, "rb") as app_data_file:
app_data = pickle.load(app_data_file)
cPrint("Starting database system...", Colors.GREEN)
db_manager = DatabaseOperationsBase(username=database_user, password=database_password)
cPrint("Defining handlers...", Colors.GREEN)
handlers = handler_definer()
updater = Updater(token=app_data["TOKEN"], workers=50)
dispatcher = updater.dispatcher
for handler in handlers:
dispatcher.add_handler(handler)
cPrint("Defining log...", Colors.GREEN)
logging.basicConfig(format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
level=logging.DEBUG)
try:
updater.start_polling(poll_interval=5, timeout=60)
except KeyboardInterrupt:
cPrint("Exiting program... Wait while closing threads and pending petitions...", Colors.FAIL)
updater.idle()
db_manager.finishConnection()
exit(0)
if __name__ == '__main__':
args = ArgumentParser()
args.add_argument("-t",
"--token",
help="Telegram token obtained via BotFather",
type=str)
args.add_argument("-y",
"--youtube",
help="YouTube API Key",
type=str)
args.add_argument("-c",
"--creator",
help="Telegram ID of the creator",
type=int)
args.add_argument("-dbu",
"--db_user",
help="Database user (can be empty)",
type=str)
args.add_argument("-dbp",
"--db_password",
help="Database password (can be empty)",
type=str)
args.add_argument("-v",
"--version",
help="Application version",
action="store_true")
main(args.parse_args())