|
|
import datetime |
|
|
import time |
|
|
import markdown |
|
|
import smtplib |
|
|
from motor import motor_asyncio |
|
|
from motor.core import AgnosticClient |
|
|
from motor.motor_asyncio import AsyncIOMotorClient |
|
|
|
|
|
from email.mime.multipart import MIMEMultipart |
|
|
from email.mime.text import MIMEText |
|
|
|
|
|
from config import * |
|
|
from akn.utils.logger import LOGS |
|
|
|
|
|
client_tiktok = AsyncIOMotorClient(MONGO_URL) |
|
|
db_tiktok = client_tiktok["tiktokbot"] |
|
|
otp_collection = db_tiktok["otps"] |
|
|
|
|
|
async def send_check_otp_email(receiver_email, text): |
|
|
try: |
|
|
smtp_server = "smtp.gmail.com" |
|
|
port = 587 |
|
|
sender_email = ME_GMAIL |
|
|
password = ME_GMAIL_PASSWORD |
|
|
msg = MIMEMultipart() |
|
|
msg['From'] = 'Akeno AI <[email protected]>' |
|
|
msg['To'] = receiver_email |
|
|
msg['Subject'] = "Verified OTP- Akeno AI API" |
|
|
html = markdown.markdown(text) |
|
|
msg.attach(MIMEText(html, 'html')) |
|
|
server = smtplib.SMTP(smtp_server, port) |
|
|
server.starttls() |
|
|
server.login(sender_email, password) |
|
|
server.sendmail(sender_email, receiver_email, msg.as_string()) |
|
|
server.quit() |
|
|
LOGS.info("Email sent successfully!") |
|
|
except Exception: |
|
|
pass |
|
|
|
|
|
class Database: |
|
|
def __init__(self, uri: str) -> None: |
|
|
self.client: AgnosticClient = motor_asyncio.AsyncIOMotorClient(uri) |
|
|
self.db = self.client["Akeno"] |
|
|
|
|
|
|
|
|
self.db_tiktok = self.client["tiktokbot"] |
|
|
self.magic_bot = self.db_tiktok["magicbot"] |
|
|
self.gemini_bot = self.db_tiktok["geminibot"] |
|
|
self.gpt_4_bot = self.db_tiktok["gpt_4_bot"] |
|
|
self.meta_bot = self.db_tiktok["metabot"] |
|
|
self.youtube_bot = self.db_tiktok["youtubebot"] |
|
|
self.session_bot = self.db_tiktok["sessionbot"] |
|
|
self.captcha_bot = self.db_tiktok["captchabot"] |
|
|
self.alldl_bot = self.db_tiktok["alldlbot"] |
|
|
self.privary_bot = self.db_tiktok["privarybot"] |
|
|
self.maintance_bot = self.db_tiktok["maintancebot"] |
|
|
|
|
|
|
|
|
self.afk = self.db["afk"] |
|
|
self.antiflood = self.db["antiflood"] |
|
|
self.autopost = self.db["autopost"] |
|
|
self.blacklist = self.db["blacklist"] |
|
|
self.echo = self.db["echo"] |
|
|
self.env = self.db["env"] |
|
|
self.filter = self.db["filter"] |
|
|
self.forcesub = self.db["forcesub"] |
|
|
self.gachabots = self.db["gachabots"] |
|
|
self.cohere = self.db["cohere"] |
|
|
self.chatbot = self.db["chatbot"] |
|
|
self.backup_chatbot = self.db["backupchatbot"] |
|
|
self.antiarabic = self.db["antiarabic"] |
|
|
self.gban = self.db["gban"] |
|
|
self.gmute = self.db["gmute"] |
|
|
self.greetings = self.db["greetings"] |
|
|
self.mute = self.db["mute"] |
|
|
self.pmpermit = self.db["pmpermit"] |
|
|
self.session = self.db["session"] |
|
|
self.snips = self.db["snips"] |
|
|
self.stan_users = self.db["stan_users"] |
|
|
self.expired_users = self.db["expired_users"] |
|
|
|
|
|
async def connect(self): |
|
|
try: |
|
|
await self.client.admin.command("ping") |
|
|
LOGS.info(f"Database Connection Established!") |
|
|
except Exception as e: |
|
|
LOGS.info(f"DatabaseErr: {e} ") |
|
|
quit(1) |
|
|
|
|
|
async def _close(self): |
|
|
await self.client.close() |
|
|
|
|
|
def get_datetime(self) -> str: |
|
|
return datetime.datetime.now().strftime("%d/%m/%Y - %H:%M") |
|
|
|
|
|
async def claim_privacy_policy(self, user_id): |
|
|
return await self.privary_bot.update_one( |
|
|
{"user_id": user_id}, |
|
|
{"$set": {"is_claimed": True}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def set_pic_in_allbot(self, id, link, is_pic): |
|
|
return await self.privary_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
{"$set": {"link_pic": link, "is_pic": is_pic}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def get_pic_in_allbot(self, id): |
|
|
user_data = await self.privary_bot.find_one({"bot_id": id}) |
|
|
if user_data: |
|
|
return user_data.get("is_pic"), user_data.get("link_pic") |
|
|
return False, None |
|
|
|
|
|
async def fixed_maintance(self, is_maintance): |
|
|
return await self.maintance_bot.update_one( |
|
|
{"bot_id": 1}, |
|
|
{"$set": {"is_maintance": is_maintance}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def get_maintance(self): |
|
|
user_data = await self.maintance_bot.find_one({"bot_id": 1}) |
|
|
return user_data.get("is_maintance") if user_data else False |
|
|
|
|
|
async def get_privacy_policy(self, user_id): |
|
|
user_data = await self.privary_bot.find_one({"user_id": user_id}) |
|
|
return user_data.get("is_claimed") if user_data else None |
|
|
|
|
|
async def remove_bot_token_magic(self, user_id): |
|
|
remove_bot_clone = { |
|
|
"bot_token": None, |
|
|
"user_id": user_id |
|
|
} |
|
|
return await self.magic_bot.update_one({"user_id": user_id}, {"$unset": remove_bot_clone}) |
|
|
|
|
|
async def remove_expired_date(self, user_id): |
|
|
remove_filter = {"expire_date": None} |
|
|
return await self.expired_users.update_one({"user_id": user_id}, {"$unset": remove_filter}) |
|
|
|
|
|
async def get_expired_date(self, user_id): |
|
|
user = await self.expired_users.find_one({"user_id": user_id}) |
|
|
if user: |
|
|
return user |
|
|
else: |
|
|
return None |
|
|
|
|
|
async def set_expired_date( |
|
|
self, |
|
|
user_id, |
|
|
first_name, |
|
|
username, |
|
|
expire_date, |
|
|
bot_token, |
|
|
client_name, |
|
|
disconnected: bool = False |
|
|
): |
|
|
add_expired = { |
|
|
"expire_date": expire_date, |
|
|
"first_name": first_name, |
|
|
"username": username, |
|
|
"user_id": user_id, |
|
|
"user_client": { |
|
|
"bot_token": bot_token, |
|
|
"client_name": client_name, |
|
|
"disconnected": disconnected |
|
|
} |
|
|
} |
|
|
return await self.expired_users.update_one( |
|
|
{"user_id": user_id}, |
|
|
{"$set": add_expired}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def set_env(self, name: str, value: str) -> None: |
|
|
await self.env.update_one( |
|
|
{"name": name}, {"$set": {"value": value}}, upsert=True |
|
|
) |
|
|
|
|
|
async def get_env(self, name: str) -> str | None: |
|
|
if await self.is_env(name): |
|
|
data = await self.env.find_one({"name": name}) |
|
|
return data["value"] |
|
|
return None |
|
|
|
|
|
async def rm_env(self, name: str) -> None: |
|
|
await self.env.delete_one({"name": name}) |
|
|
|
|
|
async def is_env(self, name: str) -> bool: |
|
|
if await self.env.find_one({"name": name}): |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def get_all_env(self) -> list: |
|
|
return [i async for i in self.env.find({})] |
|
|
|
|
|
async def is_stan(self, client: int, user_id: int) -> bool: |
|
|
if await self.stan_users.find_one({"client": client, "user_id": user_id}): |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def add_stan(self, client: int, user_id: int) -> bool: |
|
|
if await self.is_stan(client, user_id): |
|
|
return False |
|
|
await self.stan_users.insert_one( |
|
|
{"client": client, "user_id": user_id, "date": self.get_datetime()} |
|
|
) |
|
|
return True |
|
|
|
|
|
async def rm_stan(self, client: int, user_id: int) -> bool: |
|
|
if not await self.is_stan(client, user_id): |
|
|
return False |
|
|
await self.stan_users.delete_one({"client": client, "user_id": user_id}) |
|
|
return True |
|
|
|
|
|
async def get_stans(self, client: int) -> list: |
|
|
return [i async for i in self.stan_users.find({"client": client})] |
|
|
|
|
|
async def get_all_stans(self) -> list: |
|
|
return [i async for i in self.stan_users.find({})] |
|
|
|
|
|
async def is_session(self, user_id: int) -> bool: |
|
|
if await self.session.find_one({"user_id": user_id}): |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def update_session( |
|
|
self, |
|
|
user_id: int, |
|
|
api_id: int, |
|
|
api_hash: str, |
|
|
session: str, |
|
|
email: str, |
|
|
phone_number: str, |
|
|
verified_password=None |
|
|
) -> None: |
|
|
await self.session.update_one( |
|
|
{"user_id": user_id}, |
|
|
{"$set": { |
|
|
"session": session, |
|
|
"api_id": api_id, |
|
|
"api_hash": api_hash, |
|
|
"email": email, |
|
|
"phone_number": phone_number, |
|
|
"verified_password": verified_password, |
|
|
"date": self.get_datetime()} |
|
|
}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def rm_session(self, user_id: int) -> None: |
|
|
await self.session.delete_one({"user_id": user_id}) |
|
|
|
|
|
async def get_session(self, user_id: int): |
|
|
if not await self.is_session(user_id): |
|
|
return False |
|
|
data = await self.session.find_one({"user_id": user_id}) |
|
|
return data |
|
|
|
|
|
async def add_bot_token_meta(self, user_id, bot_token): |
|
|
add_bot_clone = { |
|
|
"bot_token": bot_token, |
|
|
"user_id": user_id |
|
|
} |
|
|
return await self.meta_bot.update_one({"user_id": user_id}, {"$set": add_bot_clone}, upsert=True) |
|
|
|
|
|
async def get_all_sessions(self) -> list: |
|
|
return [i async for i in self.session.find({})] |
|
|
|
|
|
async def get_all_meta_bot(self) -> list: |
|
|
return [i async for i in self.meta_bot.find({})] |
|
|
|
|
|
async def get_all_magic_bot(self) -> list: |
|
|
return [i async for i in self.magic_bot.find({})] |
|
|
|
|
|
async def get_all_gemini_bot(self) -> list: |
|
|
return [i async for i in self.gemini_bot.find({})] |
|
|
|
|
|
async def get_all_youtube_bot(self) -> list: |
|
|
return [i async for i in self.youtube_bot.find({})] |
|
|
|
|
|
async def get_all_session_bot(self) -> list: |
|
|
return [i async for i in self.session_bot.find({})] |
|
|
|
|
|
async def get_all_captcha_bot(self) -> list: |
|
|
return [i async for i in self.captcha_bot.find({})] |
|
|
|
|
|
async def get_all_alldl_bot(self) -> list: |
|
|
return [i async for i in self.alldl_bot.find({})] |
|
|
|
|
|
async def get_alldlbot_by_user_id(self, user_id, id): |
|
|
user_data = await self.alldl_bot.find_one({ |
|
|
"user_id": user_id, |
|
|
"bot_id": int(id) |
|
|
}) |
|
|
LOGS.info(user_data) |
|
|
return True if user_data else False |
|
|
|
|
|
async def get_alldlbot_by_no_button(self, id): |
|
|
user_data = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
if not user_data: |
|
|
return False |
|
|
return user_data.get("is_rm_button", False) |
|
|
|
|
|
|
|
|
|
|
|
async def remove_strings_userv(self, user: int) -> bool: |
|
|
if not user: |
|
|
LOGS.error(f"Invalid session User format: {type(user)}") |
|
|
return False |
|
|
try: |
|
|
result = await self.session.delete_one({"user_id": user}) |
|
|
if result.deleted_count == 1: |
|
|
LOGS.info(f"Session User deleted: {user}") |
|
|
return True |
|
|
LOGS.warning(f"Session User not found: {user}...") |
|
|
return False |
|
|
except Exception as e: |
|
|
LOGS.critical(f"Deletion failed for {user}...: {str(e)}") |
|
|
return False |
|
|
|
|
|
async def remove_bot_token_meta(self, bot_token: str) -> bool: |
|
|
if not bot_token or not isinstance(bot_token, str): |
|
|
LOGS.error(f"Invalid token format: {type(bot_token)}") |
|
|
return False |
|
|
try: |
|
|
result = await self.meta_bot.delete_one({"bot_token": bot_token}) |
|
|
if result.deleted_count == 1: |
|
|
LOGS.info(f"Token deleted: {bot_token[:6]}...{bot_token[-3:]}") |
|
|
return True |
|
|
LOGS.warning(f"Token not found: {bot_token[:6]}...") |
|
|
return False |
|
|
except Exception as e: |
|
|
LOGS.critical(f"Deletion failed for {bot_token[:6]}...: {str(e)}") |
|
|
return False |
|
|
|
|
|
async def remove_bot_token_magic(self, bot_token: str) -> bool: |
|
|
if not bot_token or not isinstance(bot_token, str): |
|
|
LOGS.error(f"Invalid token format: {type(bot_token)}") |
|
|
return False |
|
|
try: |
|
|
result = await self.magic_bot.delete_one({"bot_token": bot_token}) |
|
|
if result.deleted_count == 1: |
|
|
LOGS.info(f"Token deleted: {bot_token[:6]}...{bot_token[-3:]}") |
|
|
return True |
|
|
LOGS.warning(f"Token not found: {bot_token[:6]}...") |
|
|
return False |
|
|
except Exception as e: |
|
|
LOGS.critical(f"Deletion failed for {bot_token[:6]}...: {str(e)}") |
|
|
return False |
|
|
|
|
|
async def remove_bot_token_youtube(self, bot_token: str) -> bool: |
|
|
if not bot_token or not isinstance(bot_token, str): |
|
|
LOGS.error(f"Invalid token format: {type(bot_token)}") |
|
|
return False |
|
|
try: |
|
|
result = await self.youtube_bot.delete_one({"bot_token": bot_token}) |
|
|
if result.deleted_count == 1: |
|
|
LOGS.info(f"Token deleted: {bot_token[:6]}...{bot_token[-3:]}") |
|
|
return True |
|
|
LOGS.warning(f"Token not found: {bot_token[:6]}...") |
|
|
return False |
|
|
except Exception as e: |
|
|
LOGS.critical(f"Deletion failed for {bot_token[:6]}...: {str(e)}") |
|
|
return False |
|
|
|
|
|
async def remove_bot_token_sessionbot(self, bot_token: str) -> bool: |
|
|
if not bot_token or not isinstance(bot_token, str): |
|
|
LOGS.error(f"Invalid token format: {type(bot_token)}") |
|
|
return False |
|
|
try: |
|
|
result = await self.session_bot.delete_one({"bot_token": bot_token}) |
|
|
if result.deleted_count == 1: |
|
|
LOGS.info(f"Token deleted: {bot_token[:6]}...{bot_token[-3:]}") |
|
|
return True |
|
|
LOGS.warning(f"Token not found: {bot_token[:6]}...") |
|
|
return False |
|
|
except Exception as e: |
|
|
LOGS.critical(f"Deletion failed for {bot_token[:6]}...: {str(e)}") |
|
|
return False |
|
|
|
|
|
async def remove_bot_token_gemini(self, bot_token: str) -> bool: |
|
|
if not bot_token or not isinstance(bot_token, str): |
|
|
LOGS.error(f"Invalid token format: {type(bot_token)}") |
|
|
return False |
|
|
try: |
|
|
result = await self.gemini_bot.delete_one({"bot_token": bot_token}) |
|
|
if result.deleted_count == 1: |
|
|
LOGS.info(f"Token deleted: {bot_token[:6]}...{bot_token[-3:]}") |
|
|
return True |
|
|
LOGS.warning(f"Token not found: {bot_token[:6]}...") |
|
|
return False |
|
|
except Exception as e: |
|
|
LOGS.critical(f"Deletion failed for {bot_token[:6]}...: {str(e)}") |
|
|
return False |
|
|
|
|
|
async def remove_bot_token_alldlbot(self, bot_token: str) -> bool: |
|
|
if not bot_token or not isinstance(bot_token, str): |
|
|
LOGS.error(f"Invalid token format: {type(bot_token)}") |
|
|
return False |
|
|
try: |
|
|
result = await self.alldl_bot.delete_one({"bot_token": bot_token}) |
|
|
if result.deleted_count == 1: |
|
|
LOGS.info(f"Token deleted: {bot_token[:6]}...{bot_token[-3:]}") |
|
|
return True |
|
|
LOGS.warning(f"Token not found: {bot_token[:6]}...") |
|
|
return False |
|
|
except Exception as e: |
|
|
LOGS.critical(f"Deletion failed for {bot_token[:6]}...: {str(e)}") |
|
|
return False |
|
|
|
|
|
|
|
|
|
|
|
async def update_alldlbot_broadcast(self, id, user_id, action): |
|
|
if action == "add": |
|
|
update_query = {"$addToSet": {"stats_bc": user_id}} |
|
|
elif action == "remove": |
|
|
update_query = {"$pull": {"stats_bc": user_id}} |
|
|
else: |
|
|
raise ValueError("Invalid action. Use 'add' or 'remove'.") |
|
|
return await self.alldl_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
update_query, |
|
|
upsert=True if action == "add" else False |
|
|
) |
|
|
|
|
|
async def update_alldlbot_sudouser(self, id, user, action): |
|
|
if action == "add": |
|
|
update_query = {"$addToSet": {"sudo_user": user}} |
|
|
elif action == "remove": |
|
|
update_query = {"$pull": {"sudo_user": user}} |
|
|
else: |
|
|
raise ValueError("Invalid action. Use 'add' or 'remove'.") |
|
|
return await self.alldl_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
update_query, |
|
|
upsert=True if action == "add" else False |
|
|
) |
|
|
|
|
|
async def get_alldlbot_sudouser(self, id, user_id): |
|
|
user_data = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
if not user_data or "sudo_user" not in user_data: |
|
|
return [] |
|
|
return user_id in user_data["sudo_user"] |
|
|
|
|
|
async def get_alldlbot_allsudouser(self, id: int): |
|
|
y = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
if not y or "sudo_user" not in y: |
|
|
return [] |
|
|
return y["blacklist_chat"] |
|
|
|
|
|
async def update_alldlbot_broadcast_in_group(self, id, chat, action): |
|
|
if action == "add": |
|
|
update_query = {"$addToSet": {"stats_gc": chat}} |
|
|
elif action == "remove": |
|
|
update_query = {"$pull": {"stats_gc": chat}} |
|
|
else: |
|
|
raise ValueError("Invalid action. Use 'add' or 'remove'.") |
|
|
return await self.alldl_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
update_query, |
|
|
upsert=True if action == "add" else False |
|
|
) |
|
|
|
|
|
async def alldlbot_broadcast_stats(self, id): |
|
|
user_data = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
return len(user_data["stats_bc"]) if user_data and "stats_bc" in user_data else 0 |
|
|
|
|
|
async def alldlbot_all_broadcast(self, id): |
|
|
user_data = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
return user_data["stats_bc"] |
|
|
|
|
|
async def alldlbot_all_broadcast_group(self, id): |
|
|
user_data = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
return user_data["stats_gc"] |
|
|
|
|
|
async def alldlbot_blacklist_stats(self, id): |
|
|
user_data = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
return len(user_data["blacklist_user"]) if user_data and "blacklist_user" in user_data else 0 |
|
|
|
|
|
async def get_alldlbot_is_blacklist_user(self, id: int, user_id: int) -> bool: |
|
|
data = await self.get_alldlbot_blacklist_user(id) |
|
|
if not data or "blacklist_user" not in data: |
|
|
return False |
|
|
return user_id in data["blacklist_user"] |
|
|
|
|
|
async def get_alldlbot_blacklist_user(self, id: int): |
|
|
return await self.alldl_bot.find_one({"bot_id": id}) |
|
|
|
|
|
async def get_alldlbot_allblacklist_user(self, id: int): |
|
|
data = await self.get_alldlbot_blacklist_user(id) |
|
|
if not data or "blacklist_user" not in data: |
|
|
return [] |
|
|
return data["blacklist_user"] |
|
|
|
|
|
async def add_alldlbot_blacklist_user(self, id, user_id): |
|
|
return await self.alldl_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
{"$push": {"blacklist_user": user_id}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def add_alldlbot_unblacklist_user(self, id, user_id): |
|
|
return await self.alldl_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
{"$pull": {"blacklist_user": user_id}} |
|
|
) |
|
|
|
|
|
async def add_alldlbot_blacklist_chat(self, id, chat): |
|
|
return await self.alldl_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
{"$push": {"blacklist_chat": chat}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def get_alldlbot_blacklist_chat(self, id: int): |
|
|
chat = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
if not chat or "blacklist_chat" not in chat: |
|
|
return [] |
|
|
return chat["blacklist_chat"] |
|
|
|
|
|
async def get_alldlbot_is_blacklist_chat(self, id: int, chat_id: int) -> bool: |
|
|
data = await self.get_alldlbot_blacklist_chat(id) |
|
|
return chat_id in data |
|
|
|
|
|
async def alldlbot_whitelist_chat(self, id: int, chat_id: int) -> bool: |
|
|
chat = await self.alldl_bot.find_one({"bot_id": id}) |
|
|
if chat and "blacklist_chat" in chat and chat_id in chat["blacklist_chat"]: |
|
|
await self.alldl_bot.update_one( |
|
|
{"bot_id": id}, |
|
|
{"$pull": {"blacklist_chat": chat_id}} |
|
|
) |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def add_alldlbot_by_no_button(self, id, is_rm_button=False): |
|
|
_ups = {"is_rm_button": is_rm_button} |
|
|
return await self.alldl_bot.update_one({"bot_id": id}, {"$set": _ups}, upsert=True) |
|
|
|
|
|
async def is_gbanned(self, user_id: int) -> bool: |
|
|
if await self.gban.find_one({"user_id": user_id}): |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def add_gban(self, user_id: int, reason: str) -> bool: |
|
|
if await self.is_gbanned(user_id): |
|
|
return False |
|
|
await self.gban.insert_one( |
|
|
{"user_id": user_id, "reason": reason, "date": self.get_datetime()} |
|
|
) |
|
|
return True |
|
|
|
|
|
async def rm_gban(self, user_id: int): |
|
|
if not await self.is_gbanned(user_id): |
|
|
return None |
|
|
reason = (await self.gban.find_one({"user_id": user_id}))["reason"] |
|
|
await self.gban.delete_one({"user_id": user_id}) |
|
|
return reason |
|
|
|
|
|
async def get_gban(self) -> list: |
|
|
return [i async for i in self.gban.find({})] |
|
|
|
|
|
async def get_gban_user(self, user_id: int) -> dict | None: |
|
|
if not await self.is_gbanned(user_id): |
|
|
return None |
|
|
return await self.gban.find_one({"user_id": user_id}) |
|
|
|
|
|
async def is_gmuted(self, user_id: int) -> bool: |
|
|
if await self.gmute.find_one({"user_id": user_id}): |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def add_gmute(self, user_id: int, reason: str) -> bool: |
|
|
if await self.is_gmuted(user_id): |
|
|
return False |
|
|
await self.gmute.insert_one( |
|
|
{"user_id": user_id, "reason": reason, "date": self.get_datetime()} |
|
|
) |
|
|
return True |
|
|
|
|
|
async def rm_gmute(self, user_id: int): |
|
|
if not await self.is_gmuted(user_id): |
|
|
return None |
|
|
reason = (await self.gmute.find_one({"user_id": user_id}))["reason"] |
|
|
await self.gmute.delete_one({"user_id": user_id}) |
|
|
return reason |
|
|
|
|
|
async def get_gmute(self) -> list: |
|
|
return [i async for i in self.gmute.find({})] |
|
|
|
|
|
async def add_mute(self, client: int, user_id: int, chat_id: int, reason: str): |
|
|
await self.mute.update_one( |
|
|
{"client": client, "user_id": user_id, "chat_id": chat_id}, |
|
|
{"$set": {"reason": reason, "date": self.get_datetime()}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_mute(self, client: int, user_id: int, chat_id: int) -> str: |
|
|
reason = (await self.get_mute(client, user_id, chat_id))["reason"] |
|
|
await self.mute.delete_one({"client": client, "user_id": user_id, "chat_id": chat_id}) |
|
|
return reason |
|
|
|
|
|
async def is_muted(self, client: int, user_id: int, chat_id: int) -> bool: |
|
|
if await self.get_mute(client, user_id, chat_id): |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def get_mute(self, client: int, user_id: int, chat_id: int): |
|
|
data = await self.mute.find_one({"client": client, "user_id": user_id, "chat_id": chat_id}) |
|
|
return data |
|
|
|
|
|
async def set_afk( |
|
|
self, user_id: int, reason: str, media: int, media_type: str |
|
|
) -> None: |
|
|
await self.afk.update_one( |
|
|
{"user_id": user_id}, |
|
|
{ |
|
|
"$set": { |
|
|
"reason": reason, |
|
|
"time": time.time(), |
|
|
"media": media, |
|
|
"media_type": media_type, |
|
|
} |
|
|
}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def get_afk(self, user_id: int): |
|
|
data = await self.afk.find_one({"user_id": user_id}) |
|
|
return data |
|
|
|
|
|
async def is_afk(self, user_id: int) -> bool: |
|
|
if await self.afk.find_one({"user_id": user_id}): |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def rm_afk(self, user_id: int) -> None: |
|
|
await self.afk.delete_one({"user_id": user_id}) |
|
|
|
|
|
async def set_flood(self, client_chat: tuple[int, int], settings: dict): |
|
|
await self.antiflood.update_one( |
|
|
{"client": client_chat[0], "chat": client_chat[1]}, |
|
|
{"$set": settings}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def get_flood(self, client_chat: tuple[int, int]): |
|
|
data = await self.antiflood.find_one( |
|
|
{"client": client_chat[0], "chat": client_chat[1]} |
|
|
) |
|
|
return data or {} |
|
|
|
|
|
async def is_flood(self, client_chat: tuple[int, int]) -> bool: |
|
|
data = await self.get_flood(client_chat) |
|
|
|
|
|
if not data: |
|
|
return False |
|
|
|
|
|
if data["limit"] == 0: |
|
|
return False |
|
|
|
|
|
return True |
|
|
|
|
|
async def get_all_floods(self) -> list: |
|
|
return [i async for i in self.antiflood.find({})] |
|
|
|
|
|
async def set_autopost(self, client: int, from_channel: int, to_channel: int): |
|
|
await self.autopost.update_one( |
|
|
{"client": client}, |
|
|
{ |
|
|
"$push": { |
|
|
"autopost": { |
|
|
"from_channel": from_channel, |
|
|
"to_channel": to_channel, |
|
|
"date": self.get_datetime(), |
|
|
} |
|
|
} |
|
|
}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def get_autopost(self, client: int, from_channel: int): |
|
|
data = await self.autopost.find_one( |
|
|
{ |
|
|
"client": client, |
|
|
"autopost": {"$elemMatch": {"from_channel": from_channel}}, |
|
|
} |
|
|
) |
|
|
return data |
|
|
|
|
|
async def is_autopost( |
|
|
self, client: int, from_channel: int, to_channel: int = None |
|
|
) -> bool: |
|
|
if to_channel: |
|
|
data = await self.autopost.find_one( |
|
|
{ |
|
|
"client": client, |
|
|
"autopost": { |
|
|
"$elemMatch": { |
|
|
"from_channel": from_channel, |
|
|
"to_channel": to_channel, |
|
|
} |
|
|
}, |
|
|
} |
|
|
) |
|
|
else: |
|
|
data = await self.autopost.find_one( |
|
|
{ |
|
|
"client": client, |
|
|
"autopost": {"$elemMatch": {"from_channel": from_channel}}, |
|
|
} |
|
|
) |
|
|
return True if data else False |
|
|
|
|
|
async def rm_autopost(self, client: int, from_channel: int, to_channel: int): |
|
|
await self.autopost.update_one( |
|
|
{"client": client}, |
|
|
{ |
|
|
"$pull": { |
|
|
"autopost": { |
|
|
"from_channel": from_channel, |
|
|
"to_channel": to_channel, |
|
|
} |
|
|
} |
|
|
}, |
|
|
) |
|
|
|
|
|
async def get_all_autoposts(self, client: int) -> list: |
|
|
return [i async for i in self.autopost.find({"client": client})] |
|
|
|
|
|
async def add_blacklist(self, client: int, chat: int, blacklist: str): |
|
|
await self.blacklist.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$push": {"blacklist": blacklist}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_blacklist(self, client: int, chat: int, blacklist: str): |
|
|
await self.blacklist.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$pull": {"blacklist": blacklist}}, |
|
|
) |
|
|
|
|
|
async def is_blacklist(self, client: int, chat: int, blacklist: str) -> bool: |
|
|
blacklists = await self.get_all_blacklists(client, chat) |
|
|
if blacklist in blacklists: |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def get_all_blacklists(self, client: int, chat: int) -> list: |
|
|
data = await self.blacklist.find_one({"client": client, "chat": chat}) |
|
|
|
|
|
if not data: |
|
|
return [] |
|
|
|
|
|
return data["blacklist"] |
|
|
|
|
|
async def get_blacklist_clients(self) -> list: |
|
|
return [i async for i in self.blacklist.find({})] |
|
|
|
|
|
async def set_echo(self, client: int, chat: int, user: int): |
|
|
await self.echo.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$push": {"echo": user}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_echo(self, client: int, chat: int, user: int): |
|
|
await self.echo.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$pull": {"echo": user}}, |
|
|
) |
|
|
|
|
|
async def is_echo(self, client: int, chat: int, user: int) -> bool: |
|
|
data = await self.get_all_echo(client, chat) |
|
|
if user in data: |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def get_all_echo(self, client: int, chat: int) -> list: |
|
|
data = await self.echo.find_one({"client": client, "chat": chat}) |
|
|
|
|
|
if not data: |
|
|
return [] |
|
|
|
|
|
return data["echo"] |
|
|
|
|
|
async def set_filter(self, client: int, chat: int, keyword: str, msgid: int): |
|
|
await self.filter.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$push": {"filter": {"keyword": keyword, "msgid": msgid}}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_filter(self, client: int, chat: int, keyword: str): |
|
|
await self.filter.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$pull": {"filter": {"keyword": keyword}}}, |
|
|
) |
|
|
|
|
|
async def rm_all_filters(self, client: int, chat: int): |
|
|
await self.filter.delete_one({"client": client, "chat": chat}) |
|
|
|
|
|
async def is_filter(self, client: int, chat: int, keyword: str) -> bool: |
|
|
data = await self.get_filter(client, chat, keyword) |
|
|
return True if data else False |
|
|
|
|
|
async def get_filter(self, client: int, chat: int, keyword: str): |
|
|
data = await self.filter.find_one( |
|
|
{ |
|
|
"client": client, |
|
|
"chat": chat, |
|
|
"filter": {"$elemMatch": {"keyword": keyword}}, |
|
|
} |
|
|
) |
|
|
return data |
|
|
|
|
|
async def get_all_filters(self, client: int, chat: int) -> list: |
|
|
data = await self.filter.find_one({"client": client, "chat": chat}) |
|
|
|
|
|
if not data: |
|
|
return [] |
|
|
|
|
|
return data["filter"] |
|
|
|
|
|
async def set_snip(self, client: int, chat: int, keyword: str, msgid: int): |
|
|
await self.snips.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$push": {"snips": {"keyword": keyword, "msgid": msgid}}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_snip(self, client: int, chat: int, keyword: str): |
|
|
await self.snips.update_one( |
|
|
{"client": client, "chat": chat}, |
|
|
{"$pull": {"snips": {"keyword": keyword}}}, |
|
|
) |
|
|
|
|
|
async def rm_all_snips(self, client: int, chat: int): |
|
|
await self.snips.delete_one({"client": client, "chat": chat}) |
|
|
|
|
|
async def is_snip(self, client: int, chat: int, keyword: str) -> bool: |
|
|
data = await self.get_snip(client, chat, keyword) |
|
|
return True if data else False |
|
|
|
|
|
async def get_snip(self, client: int, chat: int, keyword: str): |
|
|
data = await self.snips.find_one( |
|
|
{ |
|
|
"client": client, |
|
|
"chat": chat, |
|
|
"snips": {"$elemMatch": {"keyword": keyword}}, |
|
|
} |
|
|
) |
|
|
return data |
|
|
|
|
|
async def get_all_snips(self, client: int, chat: int) -> list: |
|
|
data = await self.snips.find_one({"client": client, "chat": chat}) |
|
|
|
|
|
if not data: |
|
|
return [] |
|
|
|
|
|
return data["snips"] |
|
|
|
|
|
async def add_pmpermit(self, client: int, user: int): |
|
|
await self.pmpermit.update_one( |
|
|
{"client": client, "user": user}, |
|
|
{"$set": {"date": self.get_datetime()}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_pmpermit(self, client: int, user: int): |
|
|
await self.pmpermit.delete_one({"client": client, "user": user}) |
|
|
|
|
|
async def is_pmpermit(self, client: int, user: int) -> bool: |
|
|
data = await self.get_pmpermit(client, user) |
|
|
return True if data else False |
|
|
|
|
|
async def get_pmpermit(self, client: int, user: int): |
|
|
data = await self.pmpermit.find_one({"client": client, "user": user}) |
|
|
return data |
|
|
|
|
|
async def get_all_pmpermits(self, client: int) -> list: |
|
|
return [i async for i in self.pmpermit.find({"client": client})] |
|
|
|
|
|
async def set_welcome(self, client: int, chat: int, message: int): |
|
|
await self.greetings.update_one( |
|
|
{"client": client, "chat": chat, "welcome": True}, |
|
|
{"$set": {"message": message}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_welcome(self, client: int, chat: int): |
|
|
await self.greetings.delete_one( |
|
|
{"client": client, "chat": chat, "welcome": True} |
|
|
) |
|
|
|
|
|
async def is_welcome(self, client: int, chat: int) -> bool: |
|
|
data = await self.get_welcome(client, chat) |
|
|
return True if data else False |
|
|
|
|
|
async def get_welcome(self, client: int, chat: int): |
|
|
data = await self.greetings.find_one( |
|
|
{"client": client, "chat": chat, "welcome": True} |
|
|
) |
|
|
return data |
|
|
|
|
|
async def set_goodbye(self, client: int, chat: int, message: int): |
|
|
await self.greetings.update_one( |
|
|
{"client": client, "chat": chat, "welcome": False}, |
|
|
{"$set": {"message": message}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_goodbye(self, client: int, chat: int): |
|
|
await self.greetings.delete_one( |
|
|
{"client": client, "chat": chat, "welcome": False} |
|
|
) |
|
|
|
|
|
async def is_goodbye(self, client: int, chat: int) -> bool: |
|
|
data = await self.get_goodbye(client, chat) |
|
|
return True if data else False |
|
|
|
|
|
async def get_goodbye(self, client: int, chat: int): |
|
|
data = await self.greetings.find_one( |
|
|
{"client": client, "chat": chat, "welcome": False} |
|
|
) |
|
|
return data |
|
|
|
|
|
async def get_all_greetings(self, client: int) -> list: |
|
|
return [i async for i in self.greetings.find({"client": client})] |
|
|
|
|
|
async def add_forcesub_only_bot(self, id: int, must_join: str): |
|
|
await self.forcesub.update_one( |
|
|
{"bot_id": id}, |
|
|
{"$set": {"must_join": must_join}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def get_forcesub_only_bot(self, id: int): |
|
|
data = await self.forcesub.find_one({"bot_id": id}) |
|
|
if data: |
|
|
return data.get("must_join") |
|
|
return None |
|
|
|
|
|
async def add_forcesub(self, chat: int, must_join: int): |
|
|
await self.forcesub.update_one( |
|
|
{"chat": chat}, |
|
|
{"$push": {"must_join": must_join}}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_forcesub(self, chat: int, must_join: int) -> int: |
|
|
await self.forcesub.update_one( |
|
|
{"chat": chat}, |
|
|
{"$pull": {"must_join": must_join}}, |
|
|
) |
|
|
data = await self.forcesub.find_one({"chat": chat}) |
|
|
return len(data["must_join"]) |
|
|
|
|
|
async def rm_all_forcesub(self, in_chat: int): |
|
|
await self.forcesub.delete_one({"chat": in_chat}) |
|
|
|
|
|
async def is_forcesub(self, chat: int, must_join: int) -> bool: |
|
|
data = await self.get_forcesub(chat) |
|
|
if must_join in data["must_join"]: |
|
|
return True |
|
|
return False |
|
|
|
|
|
async def get_forcesub(self, in_chat: int): |
|
|
data = await self.forcesub.find_one({"chat": in_chat}) |
|
|
return data |
|
|
|
|
|
async def get_all_forcesubs(self) -> list: |
|
|
return [i async for i in self.forcesub.find({})] |
|
|
|
|
|
async def add_gachabot( |
|
|
self, client: int, bot: tuple[int, str], catch_command: str, chat_id: int |
|
|
): |
|
|
await self.gachabots.update_one( |
|
|
{"client": client, "bot": bot[0]}, |
|
|
{ |
|
|
"$set": { |
|
|
"username": bot[1], |
|
|
"catch_command": catch_command, |
|
|
"chat_id": chat_id, |
|
|
"date": self.get_datetime(), |
|
|
} |
|
|
}, |
|
|
upsert=True, |
|
|
) |
|
|
|
|
|
async def rm_gachabot(self, client: int, bot: int, chat_id: int = None): |
|
|
if chat_id: |
|
|
await self.gachabots.delete_one( |
|
|
{"client": client, "bot": bot, "chat_id": chat_id} |
|
|
) |
|
|
else: |
|
|
await self.gachabots.delete_one({"client": client, "bot": bot}) |
|
|
|
|
|
async def is_gachabot(self, client: int, bot: int, chat_id: int) -> bool: |
|
|
data = await self.get_gachabot(client, bot, chat_id) |
|
|
return True if data else False |
|
|
|
|
|
async def get_gachabot(self, client: int, bot: int, chat_id: int): |
|
|
data = await self.gachabots.find_one( |
|
|
{"client": client, "bot": bot, "chat_id": chat_id} |
|
|
) |
|
|
|
|
|
return data |
|
|
|
|
|
async def get_all_gachabots(self, client: int) -> list: |
|
|
return [i async for i in self.gachabots.find({"client": client})] |
|
|
|
|
|
async def get_all_gachabots_id(self) -> list: |
|
|
data = await self.gachabots.distinct("bot") |
|
|
return data |
|
|
|
|
|
async def _get_cohere_chat_from_db(self, user_id): |
|
|
user_data = await self.cohere.find_one({"user_id": user_id}) |
|
|
return user_data.get("cohere_chat", []) if user_data else [] |
|
|
|
|
|
async def _update_cohere_chat_in_db(self, user_id, cohere_chat): |
|
|
await self.cohere.update_one( |
|
|
{"user_id": user_id}, |
|
|
{"$set": {"cohere_chat": cohere_chat}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def _clear_history_in_db(self, user_id): |
|
|
unset_clear = {"cohere_chat": None} |
|
|
return await self.cohere.update_one({"user_id": user_id}, {"$unset": unset_clear}) |
|
|
|
|
|
async def clear_database(self, user_id): |
|
|
"""Clear the cohere history for the current user.""" |
|
|
result = await self._clear_history_in_db(user_id) |
|
|
if result.modified_count > 0: |
|
|
return "Chat history cleared successfully." |
|
|
else: |
|
|
return "No chat history found to clear." |
|
|
|
|
|
async def set_chat_setting(self, chat_id, isboolean): |
|
|
await self.antiarabic.update_one( |
|
|
{"chat_id": chat_id}, |
|
|
{"$set": {"arabic": isboolean}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def chat_antiarabic(self, chat_id): |
|
|
user_data = await self.antiarabic.find_one({"chat_id": chat_id}) |
|
|
if user_data: |
|
|
return user_data.get("arabic", False) |
|
|
return False |
|
|
|
|
|
async def add_chatbot(self, chat_id, user_id): |
|
|
await self.chatbot.update_one( |
|
|
{"chat_id": chat_id}, |
|
|
{"$set": {"user_id": user_id}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def add_gpt_plan(self, user_id): |
|
|
await self.gpt_4_bot.update_one( |
|
|
{"user_id": user_id}, |
|
|
{"$set": {"is_payment": True}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def is_gpt_plan(self, user_id): |
|
|
user_data = await self.gpt_4_bot.find_one({"user_id": user_id}) |
|
|
return user_data.get("is_payment") if user_data else None |
|
|
|
|
|
async def get_chatbot(self, chat_id): |
|
|
user_data = await self.chatbot.find_one({"chat_id": chat_id}) |
|
|
return user_data.get("user_id") if user_data else None |
|
|
|
|
|
async def remove_chatbot(self, chat_id): |
|
|
unset_data = {"user_id": None} |
|
|
return await self.chatbot.update_one({"chat_id": chat_id}, {"$unset": unset_data}) |
|
|
|
|
|
async def _update_chatbot_chat_in_db(self, user_id, chatbot_chat): |
|
|
await self.backup_chatbot.update_one( |
|
|
{"user_id": user_id}, |
|
|
{"$set": {"chatbot_chat": chatbot_chat}}, |
|
|
upsert=True |
|
|
) |
|
|
|
|
|
async def _get_chatbot_chat_from_db(self, user_id): |
|
|
user_data = await self.backup_chatbot.find_one({"user_id": user_id}) |
|
|
return user_data.get("chatbot_chat", []) if user_data else [] |
|
|
|
|
|
async def _clear_chatbot_history_in_db(self, user_id): |
|
|
unset_clear = {"chatbot_chat": None} |
|
|
return await self.backup_chatbot.update_one({"user_id": user_id}, {"$unset": unset_clear}) |
|
|
|
|
|
async def _clear_chatbot_database(self, user_id): |
|
|
result = await self._clear_chatbot_history_in_db(user_id) |
|
|
if result.modified_count > 0: |
|
|
return "Chat history cleared successfully." |
|
|
else: |
|
|
return "No chat history found to clear." |
|
|
|
|
|
db = Database(MONGO_URL) |
|
|
|
