-
Notifications
You must be signed in to change notification settings - Fork 152
/
Copy pathbot.py
208 lines (184 loc) · 6.28 KB
/
bot.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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
import logging
import time
import humanreadable as hr
from telethon import Button
from telethon.sync import TelegramClient, events
from telethon.tl.custom.message import Message
from telethon.types import UpdateNewMessage
from config import (ADMINS, API_HASH, API_ID, BOT_TOKEN, BOT_USERNAME,
FORCE_LINK)
from redis_db import db
from send_media import VideoSender
from tools import generate_shortenedUrl, is_user_on_chat, remove_all_videos
log = logging.getLogger(__name__)
bot = TelegramClient("bot", API_ID, API_HASH)
@bot.on(
events.NewMessage(
pattern="/start$",
incoming=True,
outgoing=False,
func=lambda x: x.is_private,
)
)
async def start(m: Message):
reply_text = """
Hello there! I'm your friendly video downloader bot specially designed to fetch videos from Terabox. Share the Terabox link with me, and I'll swiftly get started on downloading it for you.
Let's make your video experience even better!
"""
await m.reply(
reply_text,
link_preview=False,
parse_mode="markdown",
buttons=[
[
Button.url(
"Website Source Code", url="https://github.com/r0ld3x/terabox-app"
),
Button.url(
"Bot Source Code",
url="https://github.com/r0ld3x/terabox-downloader-bot",
),
],
[
Button.url("Channel ", url="https://t.me/RoldexVerse"),
Button.url("Group ", url="https://t.me/RoldexVerseChats"),
],
],
)
@bot.on(
events.NewMessage(
pattern="/gen$",
incoming=True,
outgoing=False,
func=lambda x: x.is_private,
)
)
async def generate_token(m: Message):
is_user_active = db.get(f"active_{m.sender_id}")
if is_user_active:
ttl = db.ttl(f"active_{m.sender_id}")
t = hr.Time(str(ttl), default_unit=hr.Time.Unit.SECOND)
return await m.reply(
f"""You are already active.
Your session will expire in {t.to_humanreadable()}."""
)
shortenedUrl = generate_shortenedUrl(m.sender_id)
if not shortenedUrl:
return await m.reply("Something went wrong. Please try again.")
# if_token_avl = db.get(f"token_{m.sender_id}")
# if not if_token_avl:
# else:
# uid, shortenedUrl = if_token_avl.split("|")
text = f"""
Hey {m.sender.first_name or m.sender.username}!
It seems like your Ads token has expired. Please refresh your token and try again.
Token Timeout: 1 hour
What is a token?
This is an Ads token. After viewing 1 ad, you can utilize the bot for the next 1 hour.
Keep the interactions going smoothly! 😊
"""
await m.reply(
text,
link_preview=False,
parse_mode="markdown",
buttons=[Button.url("Click here To Refresh Token", url=shortenedUrl)],
)
@bot.on(
events.NewMessage(
pattern=r"/start (?!token_)([0-9a-f]{8}-[0-9a-f]{4}-[0-5][0-9a-f]{3}-[089ab][0-9a-f]{3}-[0-9a-f]{12})",
incoming=True,
outgoing=False,
func=lambda x: x.is_private,
)
)
async def start_ntoken(m: Message):
if m.sender_id not in ADMINS:
if_token_avl = db.get(f"active_{m.sender_id}")
if not if_token_avl:
return await m.reply(
"Your account is deactivated. send /gen to get activate it again."
)
text = m.pattern_match.group(1)
fileid = db.get_key(str(text))
if fileid:
return await VideoSender.forward_file(
file_id=fileid, message=m, client=bot, uid=text.strip()
)
else:
return await m.reply("""your requested file is not available.""")
@bot.on(
events.NewMessage(
pattern=r"/start token_([0-9a-f]{8}-[0-9a-f]{4}-[0-5][0-9a-f]{3}-[089ab][0-9a-f]{3}-[0-9a-f]{12})",
incoming=True,
outgoing=False,
func=lambda x: x.is_private,
)
)
async def start_token(m: Message):
uuid = m.pattern_match.group(1).strip()
check_if = await is_user_on_chat(bot, FORCE_LINK, m.peer_id)
if not check_if:
return await m.reply(
"You haven't joined @RoldexVerse or @RoldexVerseChats yet. Please join the channel and then send me the link again.\nThank you!",
buttons=[
[
Button.url("RoldexVerse", url="https://t.me/RoldexVerse"),
Button.url("RoldexVerseChats",
url="https://t.me/RoldexVerseChats"),
],
[
Button.url(
"ReCheck ♻️",
url=f"https://{BOT_USERNAME}.t.me?start={uuid}",
),
],
],
)
is_user_active = db.get(f"active_{m.sender_id}")
if is_user_active:
ttl = db.ttl(f"active_{m.sender_id}")
t = hr.Time(str(ttl), default_unit=hr.Time.Unit.SECOND)
return await m.reply(
f"""You are already active.
Your session will expire in {t.to_humanreadable()}."""
)
if_token_avl = db.get(f"token_{uuid}")
if not if_token_avl:
return await generate_token(m)
sender_id, shortenedUrl = if_token_avl.split("|")
if m.sender_id != int(sender_id):
return await m.reply(
"Your token is invalid. Please try again.\n Hit /gen to get a new token."
)
set_user_active = db.set(f"active_{m.sender_id}", time.time(), ex=3600)
db.delete(f"token_{uuid}")
if set_user_active:
return await m.reply("Your account is active. It will expire after 1 hour.")
@bot.on(
events.NewMessage(
pattern="/remove (.*)",
incoming=True,
outgoing=False,
from_users=ADMINS,
)
)
async def remove(m: UpdateNewMessage):
user_id = m.pattern_match.group(1)
if db.get(f"check_{user_id}"):
db.delete(f"check_{user_id}")
await m.reply(f"Removed {user_id} from the list.")
else:
await m.reply(f"{user_id} is not in the list.")
@bot.on(
events.NewMessage(
pattern="/removeall",
incoming=True,
outgoing=False,
from_users=ADMINS,
)
)
async def removeall(m: UpdateNewMessage):
remove_all_videos()
return await m.reply("Removed all videos from the list.")
bot.start(bot_token=BOT_TOKEN)
bot.run_until_disconnected()