-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathmain.py
More file actions
13557 lines (12248 loc) · 631 KB
/
main.py
File metadata and controls
13557 lines (12248 loc) · 631 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
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
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import json
import base64
import uuid
import aiohttp
import time
import datetime
import discord
import asyncio
import re
import secrets
import string
import os
import functools
import io
import calendar
import pytz
import unicodedata
import random
from urllib.parse import quote
from dateutil.relativedelta import relativedelta
from aiohttp_socks import ProxyConnector, ProxyConnectionError
from itertools import cycle
from dotenv import load_dotenv
from functools import wraps
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase, AsyncIOMotorCollection
from datetime import datetime, timezone, timedelta
from typing import Optional, Dict, Any, List, Literal
from discord.ext import commands, tasks
from discord import app_commands, Embed, ButtonStyle, Color, TextStyle, SelectOption
from discord.ui import View, Button, Modal, TextInput, Select
from uuid import uuid4
load_dotenv()
intents = discord.Intents.all()
bot = commands.Bot(command_prefix="/", intents=intents)
AUTH_URL = "https://discord.com/oauth2/authorize?client_id=1367892670390730812"
FOUNDER_ROLE = 1308322106005782549
BOT_STOPPED = False
HIDE_EMOJI = "<:Hide:1367929687338123497>"
CALENDAR_EMOJI = "<:Calendar:1367929720439832596>"
CHECK_EMOJI = "<:Check:1367939014732283914>"
FAIL_EMOJI = "<:Fail:1367939006163324928>"
DIAMOND_TROPHY_EMOJI = "<:DiamondTrophy:1367929741763543130>"
DUO_MAD_EMOJI = "<:DuoMad:1367929136525348864>"
MAX_EMOJI = "<:Max:1367929152493064223>"
DUOLINGO_TRAINING_EMOJI = "<:DuolingoTraining:1367946724147990558>"
EYES_EMOJI = "<:Eyes:1367929752945561660>"
TRASH_EMOJI = "<:Trash:1367929731495759882>"
GEM_EMOJI = "<:Gem:1367929660373078076>"
GLOBE_DUOLINGO_EMOJI = "<:GlobeDuolingo:1367929764677029970>"
HOME_EMOJI = "<:Home:1367939034311295006>"
XP_EMOJI = "<:XP:1367938999565680690>"
STREAK_EMOJI = "<:Streak:1367929779285790831>"
SUPER_EMOJI = "<:Super:1367929702479560764>"
NERD_EMOJI = "<:Nerd:1367929711447244861>"
QUEST_EMOJI = "<:Quest:1367928949740540016>"
SHOP = "<#1308337662519803904>"
TOKEN = os.getenv('DISCORD_TOKEN')
MONGODB_URI = os.getenv('MONGODB_URI')
YEUMONEY_TOKEN = os.getenv('YEUMONEY_TOKEN')
LINK4M_TOKEN = os.getenv('LINK4M_TOKEN')
LINK2M_TOKEN = os.getenv('LINK2M_TOKEN')
CLKSH_TOKEN = os.getenv('CLKSH_TOKEN')
SHRINKEARN_TOKEN = os.getenv('SHRINKEARN_TOKEN')
SEOTRIEUVIEW_TOKEN = os.getenv('SEOTRIEUVIEW_TOKEN')
VERSION = "3.9.8"
CODENAME = "monoFlow Plus"
TOTAL_SERVERS_CHANNEL_ID = 1329827047505399888
LOG_ID = 1327639092435095573
ERROR_ID = 1308338102158495824
SUCCESS_ID = 1316357045905264660
SERVER = 1308317770017935380
PRE_ROLE = 1308322457584795678
FREE_ROLE = 1343214504707751936
ULTRA_ROLE = 1368635360288178276
MOD_ROLE = 1308376130453245982
USER_COUNT = 1308357361227792457
XP = 1308351056606138428
STREAK_BUFF = 1308351371505958913
GEM = 1308351421413982219
STREAK_SAVER = 1308351483980283975
LEAGUE = 1308351543644389397
LEAGUE_FARMERS = 1315952244541362206
DUO_ACCOUNT = 1308351593808396288
FOLLOW_ACCOUNT = 1352800705144033362
STATUS = 1308351660992757760
GIFT_CHANNEL = 1311590373386358784
QUEST_SAVER = 1321915274634723378
LOOP_COOLDOWN = 86400
CMD_COOLDOWN = 600
LOOPS = 10000
NAME = "www․DuoXPy․site"
FREE_ACCOUNT_LIMIT = 2
PREMIUM_ACCOUNT_LIMIT = 20
USE_PROXY_INDEX = 0
class Task:
def __init__(self, discord_id, duolingo_id, task_type, amount, message_id, position=None):
self.task_id = str(uuid4())
self.discord_id = discord_id
self.duolingo_id = duolingo_id
self.task_type = task_type
self.amount = amount
self.position = position
self.message_id = message_id
self.start_time = datetime.now(pytz.UTC)
self.end_time = None
self.status = "running"
self.progress = 0
self.total = 0
self.estimated_time_left = None
self.gained = 0
def to_dict(self):
return self.__dict__
TASKS = {}
def check_bot_running():
def decorator(func):
@functools.wraps(func)
async def wrapper(interaction: discord.Interaction, *args, **kwargs):
if BOT_STOPPED:
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Smart Maintenance",
description=f"{FAIL_EMOJI} The bot is currently under maintenance by itself to fix some issues. Please try again later.",
color=discord.Color.red()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.response.send_message(embed=embed, ephemeral=True)
return
return await func(interaction, *args, **kwargs)
return wrapper
return decorator
def is_in_guild():
async def predicate(interaction: discord.Interaction):
specific_guild = bot.get_guild(SERVER)
if specific_guild:
member = specific_guild.get_member(interaction.user.id)
if member is None:
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Access Denied",
description=f"{FAIL_EMOJI} This bot is exclusively available to members of the official [server](https://discord.gg/KBzNXRZRdz).",
color=discord.Color.red()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.response.send_message(embed=embed, ephemeral=True)
return False
return True
def decorator(func):
@wraps(func)
async def wrapper(interaction: discord.Interaction, *args, **kwargs):
if await predicate(interaction):
return await func(interaction, *args, **kwargs)
return wrapper
return decorator
def is_owner():
async def predicate(interaction: discord.Interaction):
guild = bot.get_guild(SERVER)
member = guild.get_member(interaction.user.id)
if member is None or not any(role.id == FOUNDER_ROLE for role in member.roles):
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Access Denied",
description=f"{FAIL_EMOJI} This command is only available to users with the specified role.",
color=discord.Color.red()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.response.send_message(embed=embed, ephemeral=True)
return False
return True
def decorator(func):
@wraps(func)
async def wrapper(interaction: discord.Interaction, *args, **kwargs):
if await predicate(interaction):
return await func(interaction, *args, **kwargs)
return wrapper
return decorator
def is_premium():
async def predicate(interaction: discord.Interaction):
if not await is_premium_user(interaction.user):
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Access Denied",
description=f"{FAIL_EMOJI} This command is only available to DuoXPy Premium users.\n{GEM_EMOJI} You can purchase our premium plan at {SHOP}\n{QUEST_EMOJI} Or using `/free` to get a free premium for 1 day.",
color=discord.Color.red()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.response.send_message(embed=embed, ephemeral=True)
return False
return True
def decorator(func):
@wraps(func)
async def wrapper(interaction: discord.Interaction, *args, **kwargs):
if await predicate(interaction):
return await func(interaction, *args, **kwargs)
return wrapper
return decorator
def is_paid_premium():
async def predicate(interaction: discord.Interaction):
if await is_premium_free_user(interaction.user):
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Access Denied",
description=f"{FAIL_EMOJI} This command is only available to DuoXPy Paid Premium users.\n{GEM_EMOJI} You can purchase our premium plan at {SHOP}.",
color=discord.Color.red()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.response.send_message(embed=embed, ephemeral=True)
return False
return True
def decorator(func):
@wraps(func)
async def wrapper(interaction: discord.Interaction, *args, **kwargs):
if await predicate(interaction):
return await func(interaction, *args, **kwargs)
return wrapper
return decorator
def custom_cooldown(rate: int, per: float):
def decorator(func):
command_name = func.__name__
@wraps(func)
async def wrapper(interaction: discord.Interaction, *args, **kwargs):
log_channel = interaction.client.get_channel(LOG_ID)
channel_info = (
f"{CALENDAR_EMOJI} Channel: **{interaction.channel.name}**"
if isinstance(interaction.channel, discord.TextChannel)
else f"{CALENDAR_EMOJI} Channel: **DM**"
)
log_embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Usage Log",
description=(
f"{DUOLINGO_TRAINING_EMOJI} Command: **{command_name}**\n"
f"{NERD_EMOJI} User: {interaction.user.mention}\n"
f"{channel_info}\n"
f"{CHECK_EMOJI} Server: **{interaction.guild.name if interaction.guild else 'DM'}**"
),
color=Color.teal(),
timestamp=datetime.now()
)
log_embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await log_channel.send(embed=log_embed)
if await is_premium_user(interaction.user) and not await is_premium_free_user(interaction.user):
return await func(interaction, *args, **kwargs)
current_time = time.time()
user_id = str(interaction.user.id)
cooldown_data = await db.cooldowns.find_one({
"_id": f"{user_id}:{command_name}"
})
if not cooldown_data:
cooldown_data = {
"_id": f"{user_id}:{command_name}",
"uses_left": rate,
"last_used": current_time,
"rate": rate,
"per": per
}
await db.cooldowns.insert_one(cooldown_data)
else:
if cooldown_data["rate"] != rate or cooldown_data["per"] != per:
cooldown_data["rate"] = rate
cooldown_data["per"] = per
await db.cooldowns.update_one(
{"_id": f"{user_id}:{command_name}"},
{"$set": {"rate": rate, "per": per}}
)
time_since_last = current_time - cooldown_data["last_used"]
if time_since_last >= cooldown_data["per"]:
cooldown_data["uses_left"] = rate
cooldown_data["last_used"] = current_time
await db.cooldowns.update_one(
{"_id": f"{user_id}:{command_name}"},
{"$set": cooldown_data}
)
if cooldown_data["uses_left"] > 0:
cooldown_data["uses_left"] -= 1
await db.cooldowns.update_one(
{"_id": f"{user_id}:{command_name}"},
{"$set": {
"uses_left": cooldown_data["uses_left"],
"last_used": current_time
}}
)
return await func(interaction, *args, **kwargs)
time_until_next = cooldown_data["last_used"] + cooldown_data["per"] - current_time
next_hours = int(time_until_next // 3600)
next_minutes = int((time_until_next % 3600) // 60)
next_seconds = int(time_until_next % 60)
next_time_str = f"{next_hours}h {next_minutes}m {next_seconds}s".strip()
cooldown_period_str = f"{int(per // 3600)}h {int((per % 3600) // 60)}m {int(per % 60)}s".strip()
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Cooldown",
description=(
f"{FAIL_EMOJI} You can use this command in {next_time_str}.\n"
f"{DUOLINGO_TRAINING_EMOJI} Uses left: **{cooldown_data['uses_left']}/{rate}**\n"
f"{CALENDAR_EMOJI} Cooldown period: **{cooldown_period_str}**"
),
color=Color.red()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.response.send_message(embed=embed, ephemeral=True)
return
return wrapper
return decorator
class DatabaseManager:
def __init__(self, uri: str) -> None:
self.client = AsyncIOMotorClient(uri)
self.database: AsyncIOMotorDatabase = self.client["duoxpy"]
self.discord: AsyncIOMotorCollection = self.database["discord"]
self.duolingo: AsyncIOMotorCollection = self.database["duolingo"]
self.keys: AsyncIOMotorCollection = self.database["keys"]
self.proxies: AsyncIOMotorCollection = self.database["proxies"]
self.superlinks: AsyncIOMotorCollection = self.database["superlinks"]
self.loop_usage: AsyncIOMotorCollection = self.database["loop_usage"]
self.cooldowns: AsyncIOMotorCollection = self.database["cooldowns"]
self.follow: AsyncIOMotorCollection = self.database["follow"]
self.report: AsyncIOMotorCollection = self.database["report"]
self.victim: AsyncIOMotorCollection = self.database["victim"]
async def login_dis(self, discord_id: int) -> None:
await self.discord.find_one_and_update(
{"_id": discord_id},
{"$set": {
"selected": None,
"streaksaver": False,
"questsaver": False,
"hide": False,
"autoleague": {
"active": False,
"target": None,
"autoblock": False
},
"premium": {
"active": False,
"type": None,
"duration": None,
"start": None,
"end": None
}
}},
upsert=True
)
async def login(self, discord_id: int, duolingo_id: int, jwt_token: str, duolingo_username: str, timezone: str) -> None:
await self.duolingo.find_one_and_update(
{"_id": duolingo_id},
{"$set": {
"username": duolingo_username,
"discord_id": discord_id,
"jwt_token": jwt_token,
"timezone": timezone,
"paused": False
}},
upsert=True,
)
await self.discord.find_one_and_update(
{"_id": discord_id},
{"$set": {
"selected": duolingo_id
}},
upsert=True
)
async def list_profiles(self, discord_id: int) -> List[Dict[str, Any]]:
profile_list = self.duolingo.find({"discord_id": discord_id})
return [profile async for profile in profile_list]
async def get_selected_profile(self, discord_id: int) -> Optional[Dict[str, Any]]:
discord_account = await self.discord.find_one({"_id": discord_id})
if not discord_account:
return None
duolingo_id = discord_account.get("selected")
if not duolingo_id:
return None
return await self.duolingo.find_one({"_id": duolingo_id})
async def select_profile(self, discord_id: int, duolingo_id: int) -> None:
await self.discord.find_one_and_update(
{"_id": discord_id},
{"$set": {"selected": duolingo_id}},
upsert=True
)
async def load_keys(self) -> Dict[str, Any]:
cursor = self.keys.find({})
keys_dict = {}
async for doc in cursor:
key_id = doc.pop("_id")
keys_dict[key_id] = doc
return keys_dict
async def load_proxies(self) -> Dict[str, bool]:
cursor = self.proxies.find({})
return {doc["_id"]: True async for doc in cursor}
async def save_proxies(self, proxies: Dict[str, bool]) -> None:
await self.proxies.delete_many({})
if proxies:
await self.proxies.insert_many(
[{"_id": proxy} for proxy in proxies.keys()]
)
async def load_superlinks(self) -> Dict[str, bool]:
cursor = self.superlinks.find({})
return {doc["_id"]: True async for doc in cursor}
async def save_superlinks(self, superlinks: Dict[str, bool]) -> None:
await self.superlinks.delete_many({})
if superlinks:
await self.superlinks.insert_many(
[{"_id": link} for link in superlinks.keys()]
)
db = DatabaseManager(MONGODB_URI)
class LinkView(discord.ui.View):
def __init__(self, user_id):
super().__init__(timeout=60)
self.user_id = user_id
self.message = None
self.is_completed = False
self.jwt_guide_shown = False
async def on_timeout(self):
if not self.is_completed and self.message:
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Timeout",
description=f"{FAIL_EMOJI} You took too long to respond. Please try again.",
color=discord.Color.red()
)
if self.message.interaction_metadata:
embed.set_author(
name=self.message.interaction_metadata.user.display_name,
icon_url=self.message.interaction_metadata.user.avatar.url if self.message.interaction_metadata.user.avatar else self.message.interaction_metadata.user.display_avatar.url
)
await self.message.edit(embed=embed, view=None)
async def setup_buttons(self):
discord_account = await db.discord.find_one({"_id": self.user_id})
if not discord_account:
link_discord_button = discord.ui.Button(
label="Link Discord Account",
style=ButtonStyle.secondary,
emoji=QUEST_EMOJI
)
async def link_discord_callback(interaction: discord.Interaction):
await db.login_dis(self.user_id)
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Account Linked",
description=f"{CHECK_EMOJI} Discord account linked successfully! Please use `/account` again to link your Duolingo account.",
color=discord.Color.green()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.response.edit_message(embed=embed, view=None)
self.is_completed = True
link_discord_button.callback = link_discord_callback
self.add_item(link_discord_button)
else:
email_button = discord.ui.Button(
label="Link with Email/Username",
style=ButtonStyle.secondary,
emoji=QUEST_EMOJI
)
email_button.callback = self.email_button_callback
self.add_item(email_button)
jwt_button = discord.ui.Button(
label="Link with JWT Token",
style=ButtonStyle.secondary,
emoji=DUO_MAD_EMOJI
)
jwt_button.callback = self.jwt_button_callback
self.add_item(jwt_button)
async def email_button_callback(self, interaction: discord.Interaction):
await interaction.response.send_modal(EmailLoginModal(self))
async def jwt_button_callback(self, interaction: discord.Interaction):
if not self.jwt_guide_shown:
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: JWT Token Guide",
description=(
"### To Get Your JWT Token:\n"
"- **Desktop**: Go to https://duolingo.com while logged in, open Developer Tools (**Ctrl + Shift + i**), and navigate to the Console tab.\n"
"- Paste the following code in the Console to retrieve your JWT token:\n"
"```js\n"
"document.cookie.match(new RegExp('(^| )jwt_token=([^;]+)'))[0].slice(11)\n"
"```\n"
"- **Mobile**: Use [Web Inspector for Apple](https://apps.apple.com/us/app/web-inspector/id1584825745) or [Kiwi Browser for Android](https://play.google.com/store/apps/details?id=com.kiwibrowser.browser&hl=en&pli=1).\n\n"
"Click **Continue** below to enter your token."
),
color=Color.teal()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
continue_view = discord.ui.View(timeout=60)
continue_button = discord.ui.Button(label="Continue", style=ButtonStyle.secondary, emoji=CHECK_EMOJI)
async def continue_callback(i: discord.Interaction):
await i.response.send_modal(JWTModal(self))
continue_button.callback = continue_callback
continue_view.add_item(continue_button)
await interaction.response.edit_message(embed=embed, view=continue_view)
continue_view.message = await interaction.original_response()
self.jwt_guide_shown = True
else:
await interaction.response.send_modal(JWTModal(self))
class EmailLoginModal(Modal, title="Login Form"):
def __init__(self, link_view: LinkView):
super().__init__()
self.link_view = link_view
email = TextInput(
label="Email/Username",
style=TextStyle.short,
placeholder="Enter your Duolingo email or username",
required=True
)
password = TextInput(
label="Password",
style=TextStyle.short,
placeholder="Enter your password",
required=True
)
async def on_submit(self, interaction: discord.Interaction):
await interaction.response.defer(ephemeral=True)
user = interaction.user
misc = Miscellaneous()
wait_embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Please Wait",
description=f"{CHECK_EMOJI} Logging in to your account...",
color=Color.teal()
)
wait_embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=wait_embed, view=None)
distinct_id = ''.join(secrets.choice(string.ascii_letters + string.digits) for _ in range(32))
try:
async with await get_session(direct=True) as session:
login_data = {
"distinctId": distinct_id,
"identifier": self.email.value,
"password": self.password.value
}
async with session.post(
"https://www.duolingo.com/2017-06-30/login?fields=",
json=login_data,
headers={
"Host": "www.duolingo.com",
"User-Agent": misc.randomize_mobile_user_agent(),
"Accept": "application/json",
"X-Amzn-Trace-Id": "User=0",
"Accept-Encoding": "gzip, deflate, br"
}
) as response:
if response.status != 200:
self.link_view.is_completed = True
self.link_view.stop()
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Link Failed",
description=f"{FAIL_EMOJI} Invalid username or password. Please try again.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed, view=None)
return
jwt_token = response.headers.get('jwt')
if not jwt_token:
self.link_view.is_completed = True
self.link_view.stop()
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Link Failed",
description=f"{FAIL_EMOJI} Failed to get authentication token. Please try again.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed, view=None)
return
await process_jwt_token(interaction, jwt_token, user)
self.link_view.is_completed = True
self.link_view.stop()
except Exception as e:
self.link_view.is_completed = True
self.link_view.stop()
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Error",
description=f"{FAIL_EMOJI} An error occurred during link. Please try again later.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed, view=None)
class JWTModal(Modal, title="Link Account Form"):
def __init__(self, link_view: LinkView):
super().__init__()
self.link_view = link_view
jwt_token_input = TextInput(
label="JWT Token",
style=TextStyle.short,
placeholder="Enter your JWT token",
required=True,
min_length=10,
max_length=2000,
)
async def on_submit(self, interaction: discord.Interaction):
await interaction.response.defer(ephemeral=True)
user = interaction.user
wait_embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Please Wait",
description=f"{CHECK_EMOJI} Processing your JWT token...",
color=Color.teal()
)
wait_embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=wait_embed, view=None)
jwt_token = self.jwt_token_input.value.strip().replace(" ", "").replace("'", "")
await process_jwt_token(interaction, jwt_token, user)
self.link_view.is_completed = True
self.link_view.stop()
async def process_jwt_token(interaction: discord.Interaction, jwt_token: str, user):
user_id = interaction.user.id
if not re.match(r'^[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+$', jwt_token):
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Invalid Token",
description=f"{FAIL_EMOJI} The JWT token is invalid. Please ensure it's correctly formatted and try again.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed, view=None)
return
duo_id = await extract_duolingo_user_id(jwt_token)
if not duo_id:
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Error",
description=f"{FAIL_EMOJI} Failed to extract Duolingo user ID from JWT token. Please ensure your token is valid.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed, view=None)
return
existing_duo_profile = await db.duolingo.find_one({"_id": duo_id})
if existing_duo_profile:
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Account Already Linked",
description=f"{FAIL_EMOJI} This Duolingo account is already linked to another Discord account. Please unlink it first.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed, view=None)
return
async with await get_session(direct=True) as session:
headers = await getheaders(jwt_token, duo_id)
url = f"https://www.duolingo.com/2017-06-30/users/{duo_id}"
async with session.get(url, headers=headers) as response:
if response.status != 200:
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Profile Retrieval Failed",
description=f"{FAIL_EMOJI} Failed to retrieve Duolingo profile. Please check your token or try again later.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed)
return
profile_data = await response.json()
if profile_data:
display_name = profile_data.get('name', '').lower()
normalized_name = unicodedata.normalize('NFKD', display_name)
normalized_name = ''.join(c for c in normalized_name if c.isalnum()).lower()
prohibited_patterns = [
r'.*a.*u.*t.*o.*d.*u.*o.*',
r'.*d.*u.*o.*s.*m.*',
r'.*c.*l.*i.*c.*k.*',
r'.*s.*u.*p.*e.*r.*d.*u.*o.*l.*i.*n.*g.*o.*',
r'.*f.*a.*m.*i.*l.*y.*',
r'.*s.*u.*p.*e.*r.*d.*u.*o.*',
]
is_prohibited = any(re.match(pattern, normalized_name) for pattern in prohibited_patterns)
if is_prohibited:
try:
guild = bot.get_guild(SERVER)
member = guild.get_member(user_id)
if member:
await member.ban(reason=f"Using prohibited display name: {display_name}")
print(f"[LOG] Banned user {user_id} for using prohibited display name: {display_name}")
except Exception as e:
print(f"[LOG] Failed to ban user {user_id}: {e}")
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Prohibited Name",
description=f"{FAIL_EMOJI} This account's display name contains prohibited terms.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed, view=None)
return
duolingo_username = profile_data.get("username", "Unknown")
profile_link = f"https://www.duolingo.com/profile/{duolingo_username}"
timezone = profile_data.get("timezone", "Asia/Saigon")
is_premium = await is_premium_user(interaction.user)
if not is_premium:
payload = {
"name": NAME,
"username": duolingo_username,
"email": profile_data.get("email", ""),
"signal": None
}
patch_url = f"https://www.duolingo.com/2017-06-30/users/{duo_id}"
async with session.patch(patch_url, headers=headers, json=payload) as patch_response:
if patch_response.status != 200:
print(f"[LOG] Failed to update display name for free user")
await db.login(user_id, duo_id, jwt_token, duolingo_username, timezone)
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Linked",
description=f"{CHECK_EMOJI} Successfully linked to profile [**{duolingo_username}**]({profile_link}).",
color=Color.green()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed)
else:
embed = Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Profile Retrieval Failed",
description=f"{FAIL_EMOJI} Failed to retrieve Duolingo profile. Please check your token or try again later.",
color=Color.red()
)
embed.set_author(name=user.display_name, icon_url=user.avatar.url if user.avatar else user.display_avatar.url)
await interaction.edit_original_response(embed=embed)
async def fetch_username_from_id(duo_id, session):
url = f"https://www.duolingo.com/2017-06-30/users/{duo_id}?fields=username"
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
return data.get('username')
return None
async def check_account_limit(user_id, interaction):
profiles = await db.list_profiles(user_id)
account_count = len(profiles)
is_premium = await is_premium_user(interaction.user)
is_free = await is_premium_free_user(interaction.user)
limit = PREMIUM_ACCOUNT_LIMIT if is_premium else FREE_ACCOUNT_LIMIT if is_free else FREE_ACCOUNT_LIMIT
if account_count >= limit:
embed = discord.Embed(
title=f"{MAX_EMOJI} DuoXPy Max: Account Limit Reached",
description=f"{FAIL_EMOJI} You have reached the maximum number of **{limit}** accounts you can link.",
color=discord.Color.red()
)
embed.set_author(
name=interaction.user.display_name,
icon_url=interaction.user.avatar.url if interaction.user.avatar else interaction.user.display_avatar.url
)
await interaction.edit_original_response(embed=embed, view=None)
return False
return True
async def check_proxy_health(proxy):
try:
connector = ProxyConnector.from_url(proxy, force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
async with session.get("https://www.duolingo.com") as response:
return response.status == 200
except Exception as e:
print(f"[LOG] Proxy health check failed for {proxy}: {str(e)}")
return False
async def get_session(slot: Optional[int] = None, direct: bool = False):
global BOT_STOPPED, USE_PROXY_INDEX
if direct:
connector = aiohttp.TCPConnector(force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
return aiohttp.ClientSession(connector=connector, timeout=timeout)
proxies = await db.load_proxies()
proxy_list = list(proxies.keys())
if slot is not None and proxy_list:
for i in range(len(proxy_list)):
adjusted_slot = (slot + i) % len(proxy_list)
proxy = proxy_list[adjusted_slot]
connector = ProxyConnector.from_url(proxy, force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
try:
async with session.get("https://www.duolingo.com") as response:
if response.status == 200:
BOT_STOPPED = False
return session
except:
await session.close()
continue
connector = aiohttp.TCPConnector(force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
try:
async with session.get("https://www.duolingo.com") as response:
if response.status == 200:
BOT_STOPPED = False
return session
await session.close()
except:
await session.close()
BOT_STOPPED = True
connector = aiohttp.TCPConnector(force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
return aiohttp.ClientSession(connector=connector, timeout=timeout)
if not proxy_list:
connector = aiohttp.TCPConnector(force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
try:
async with session.get("https://www.duolingo.com") as response:
if response.status == 200:
BOT_STOPPED = False
return session
await session.close()
except:
await session.close()
BOT_STOPPED = True
connector = aiohttp.TCPConnector(force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
return aiohttp.ClientSession(connector=connector, timeout=timeout)
current_proxy_index = USE_PROXY_INDEX % len(proxy_list)
proxy = proxy_list[current_proxy_index]
connector = ProxyConnector.from_url(proxy, force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
try:
async with session.get("https://www.duolingo.com") as response:
if response.status == 200:
BOT_STOPPED = False
USE_PROXY_INDEX = (USE_PROXY_INDEX + 1) % len(proxy_list)
return session
await session.close()
except:
await session.close()
for i in range(len(proxy_list) - 1):
current_proxy_index = (current_proxy_index + 1) % len(proxy_list)
proxy = proxy_list[current_proxy_index]
connector = ProxyConnector.from_url(proxy, force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
try:
async with session.get("https://www.duolingo.com") as response:
if response.status == 200:
BOT_STOPPED = False
USE_PROXY_INDEX = (current_proxy_index + 1) % len(proxy_list)
return session
await session.close()
except:
await session.close()
continue
connector = aiohttp.TCPConnector(force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
session = aiohttp.ClientSession(connector=connector, timeout=timeout)
try:
async with session.get("https://www.duolingo.com") as response:
if response.status == 200:
BOT_STOPPED = False
return session
await session.close()
except:
await session.close()
BOT_STOPPED = True
connector = aiohttp.TCPConnector(force_close=True)
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
return aiohttp.ClientSession(connector=connector, timeout=timeout)
async def check_duolingo_access():
global BOT_STOPPED
timeout = aiohttp.ClientTimeout(total=600, connect=600, sock_read=600, sock_connect=600)
connector = aiohttp.TCPConnector(force_close=True)
direct_failed = False
try:
async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
async with session.get("https://www.duolingo.com/") as response:
if response.status != 200:
direct_failed = True
except:
direct_failed = True
proxy_list = await db.load_proxies()
all_proxies_failed = True
if proxy_list:
for proxy in proxy_list:
try:
connector = ProxyConnector.from_url(proxy, force_close=True)
async with aiohttp.ClientSession(connector=connector, timeout=timeout, force_close=True) as proxy_session:
async with proxy_session.get("https://www.duolingo.com/") as response:
if response.status == 200:
all_proxies_failed = False
break
except:
continue
if direct_failed and all_proxies_failed:
BOT_STOPPED = True
print("[LOG] All connections failed. Setting BOT_STOPPED to True...")
else:
BOT_STOPPED = False
print("[LOG] At least one connection successful. Setting BOT_STOPPED to False...")
async def extract_code(urlorcode: str) -> str:
if 'family-plan' in urlorcode:
return urlorcode.split('/')[-1]
return urlorcode