-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathTelegramBot.py
More file actions
86 lines (60 loc) · 2.5 KB
/
Copy pathTelegramBot.py
File metadata and controls
86 lines (60 loc) · 2.5 KB
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
from telegram.ext import ApplicationBuilder, CommandHandler, CallbackQueryHandler
from functools import wraps
class TelegramBot():
'''
A class that creates a bot for telegram, with the ability to add handlers and then launch
'''
__slots__ = ["token", "online", "app"]
handlers = []
job_queue = []
def __init__(self, token: str) -> None:
self.token = token
self.online = False
self.app = ApplicationBuilder().token(self.token).build()
def on(self) -> bool:
if self.online:
print("Bot is already online")
return False
else:
self.app.add_handlers(TelegramBot.handlers)
for job in TelegramBot.job_queue:
if job["repeating"]:
self.app.job_queue.run_repeating(callback=job["func"], interval=job["interval"], first=job["first"])
else:
self.app.job_queue.run_once(callback=job["func"], when=job["first"])
self.app.run_polling()
return True
def off(self) -> bool:
if self.online:
self.app.shutdown()
return True
else:
return False
@classmethod
def AddCommandHandler(cls, command: str, filters=None):
def decorator(func):
async def wrapper(update, context, *args, **kwargs):
return await func(update, context, *args, **kwargs)
handler = CommandHandler(command=command, callback=wrapper, filters=filters)
cls.handlers.append(handler)
return handler
return decorator
@classmethod
def AddCallbackQueryHandler(cls, pattern=None):
def decorator(func):
@wraps(func)
async def wrapper(update, context, *args, **kwargs):
return await func(update, context, *args, **kwargs)
handler = CallbackQueryHandler(callback=wrapper, pattern=pattern)
cls.handlers.append(handler)
return handler
return decorator
@classmethod
def AddJobQuery(cls, repeating=False, first=None, interval=86400.0):
def decorator(func):
@wraps(func)
async def wrapper(context, *args, **kwargs):
return await func(context, *args, **kwargs)
cls.job_queue.append({"repeating": repeating, "func": wrapper, "interval": interval, "first": first})
return wrapper
return decorator