-
Notifications
You must be signed in to change notification settings - Fork 156
/
mongo.py
112 lines (92 loc) · 3.18 KB
/
mongo.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
#
# Copyright (C) 2021-2022 by TeamYukki@Github, < https://github.com/YukkiChatBot >.
#
# This file is part of < https://github.com/TeamYukki/YukkiChatBot > project,
# and is released under the "GNU v3.0 License Agreement".
# Please see < https://github.com/TeamYukki/YukkiChatBot/blob/master/LICENSE >
#
# All rights reserved.
#
from motor.motor_asyncio import AsyncIOMotorClient as MongoClient
from config import MONGO_DB_URI
db = None
if MONGO_DB_URI != None:
mongo = MongoClient(MONGO_DB_URI)
db = mongo.ChatBot
usersdb = db.users
blockeddb = db.block
modedb = db.mode
modelist = {}
# Served Users
async def is_served_user(user_id: int) -> bool:
user = await usersdb.find_one({"user_id": user_id})
if not user:
return False
return True
async def get_served_users() -> list:
users_list = []
async for user in usersdb.find({"user_id": {"$gt": 0}}):
users_list.append(user)
return users_list
async def add_served_user(user_id: int):
is_served = await is_served_user(user_id)
if is_served:
return
return await usersdb.insert_one({"user_id": user_id})
# Banned Users
async def get_banned_users() -> list:
results = []
async for user in blockeddb.find({"user_id": {"$gt": 0}}):
user_id = user["user_id"]
results.append(user_id)
return results
async def get_banned_count() -> int:
users = blockeddb.find({"user_id": {"$gt": 0}})
users = await users.to_list(length=100000)
return len(users)
async def is_banned_user(user_id: int) -> bool:
user = await blockeddb.find_one({"user_id": user_id})
if not user:
return False
return True
async def add_banned_user(user_id: int):
is_gbanned = await is_banned_user(user_id)
if is_gbanned:
return
return await blockeddb.insert_one({"user_id": user_id})
async def remove_banned_user(user_id: int):
is_gbanned = await is_banned_user(user_id)
if not is_gbanned:
return
return await blockeddb.delete_one({"user_id": user_id})
# Forward Mode
async def is_group() -> bool:
chat_id = 123
mode = modelist.get(chat_id)
if not mode:
user = await modedb.find_one({"chat_id": chat_id})
if not user:
modelist[chat_id] = False
return False
modelist[chat_id] = True
return True
return mode
async def group_on():
chat_id = 123
modelist[chat_id] = True
user = await modedb.find_one({"chat_id": chat_id})
if not user:
return await modedb.insert_one({"chat_id": chat_id})
async def group_off():
chat_id = 123
modelist[chat_id] = False
user = await modedb.find_one({"chat_id": chat_id})
if user:
return await modelist.delete_one({"chat_id": chat_id})
else:
async def is_group() -> bool:
return False
async def is_banned_user(user_id: int) -> bool:
return False
async def add_served_user(user_id: int):
return True