From 426995e94c48fb00bd8b880b985460c96f76240e Mon Sep 17 00:00:00 2001 From: Zeta-05 Date: Wed, 6 Nov 2024 17:32:15 +0530 Subject: [PATCH] Update welcome_sql.py --- Database/sql/welcome_sql.py | 506 ------------------------------------ 1 file changed, 506 deletions(-) diff --git a/Database/sql/welcome_sql.py b/Database/sql/welcome_sql.py index 1b73f046..8b137891 100644 --- a/Database/sql/welcome_sql.py +++ b/Database/sql/welcome_sql.py @@ -1,507 +1 @@ -""" -MIT License -Copyright (c) 2022 Aʙɪsʜɴᴏɪ - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. -""" - -import random -import threading -from typing import Union - -from sqlalchemy import BigInteger, Boolean, Column, Integer, String, UnicodeText - -from Database.sql import BASE, SESSION -from Mikobot.plugins.helper_funcs.msg_types import Types - -DEFAULT_WELCOME = "ʜᴇʏ {first}, ʜᴏᴡ ᴀʀᴇ ʏᴏᴜ?" -DEFAULT_GOODBYE = "ɴɪᴄᴇ ᴋɴᴏᴡɪɴɢ ʏᴀ!" - -DEFAULT_WELCOME_MESSAGES = [ - "{first} ɪs ʜᴇʀᴇ!", # Discord welcome messages copied - "ʀᴇᴀᴅʏ ᴘʟᴀʏᴇʀ {first}", - "ᴡᴇʟᴄᴏᴍᴇ ʙʀᴏ {first}", -] - -DEFAULT_GOODBYE_MESSAGES = [ - "{first} ᴡɪʟʟ ʙᴇ ᴍɪssᴇᴅ.", - "{first} ᴡʜᴇɴ ʙᴀᴄᴋ ?.", -] - - -class Welcome(BASE): - __tablename__ = "welcome_pref" - chat_id = Column(String(14), primary_key=True) - should_welcome = Column(Boolean, default=True) - should_goodbye = Column(Boolean, default=True) - custom_content = Column(UnicodeText, default=None) - - custom_welcome = Column( - UnicodeText, default=random.choice(DEFAULT_WELCOME_MESSAGES) - ) - welcome_type = Column(Integer, default=Types.TEXT.value) - - custom_leave = Column(UnicodeText, default=random.choice(DEFAULT_GOODBYE_MESSAGES)) - leave_type = Column(Integer, default=Types.TEXT.value) - - clean_welcome = Column(BigInteger) - - def __init__(self, chat_id, should_welcome=True, should_goodbye=True): - self.chat_id = chat_id - self.should_welcome = should_welcome - self.should_goodbye = should_goodbye - - def __repr__(self): - return "<ᴄʜᴀᴛ {} sʜᴏᴜʟᴅ sʜᴏᴜʟᴅ ɴᴇᴡ ᴜsᴇʀs: {}>".format( - self.chat_id, self.should_welcome - ) - - -class WelcomeButtons(BASE): - __tablename__ = "welcome_urls" - id = Column(Integer, primary_key=True, autoincrement=True) - chat_id = Column(String(14), primary_key=True) - name = Column(UnicodeText, nullable=False) - url = Column(UnicodeText, nullable=False) - same_line = Column(Boolean, default=False) - - def __init__(self, chat_id, name, url, same_line=False): - self.chat_id = str(chat_id) - self.name = name - self.url = url - self.same_line = same_line - - -class GoodbyeButtons(BASE): - __tablename__ = "leave_urls" - id = Column(Integer, primary_key=True, autoincrement=True) - chat_id = Column(String(14), primary_key=True) - name = Column(UnicodeText, nullable=False) - url = Column(UnicodeText, nullable=False) - same_line = Column(Boolean, default=False) - - def __init__(self, chat_id, name, url, same_line=False): - self.chat_id = str(chat_id) - self.name = name - self.url = url - self.same_line = same_line - - -class WelcomeMute(BASE): - __tablename__ = "welcome_mutes" - chat_id = Column(String(14), primary_key=True) - welcomemutes = Column(UnicodeText, default=False) - - def __init__(self, chat_id, welcomemutes): - self.chat_id = str(chat_id) # ensure string - self.welcomemutes = welcomemutes - - -class WelcomeMuteUsers(BASE): - __tablename__ = "human_checks" - user_id = Column(BigInteger, primary_key=True) - chat_id = Column(String(14), primary_key=True) - human_check = Column(Boolean) - - def __init__(self, user_id, chat_id, human_check): - self.user_id = user_id # ensure string - self.chat_id = str(chat_id) - self.human_check = human_check - - -class CleanServiceSetting(BASE): - __tablename__ = "clean_service" - chat_id = Column(String(14), primary_key=True) - clean_service = Column(Boolean, default=True) - - def __init__(self, chat_id): - self.chat_id = str(chat_id) - - def __repr__(self): - return "<ᴄʜᴀᴛ ᴜsᴇᴅ ᴄʟᴇᴀɴ sᴇʀᴠɪᴄᴇ ({})>".format(self.chat_id) - - -class RaidMode(BASE): - __tablename__ = "raid_mode" - chat_id = Column(String(14), primary_key=True) - status = Column(Boolean, default=False) - time = Column(Integer, default=21600) - acttime = Column(Integer, default=3600) - # permanent = Column(Boolean, default=False) - - def __init__(self, chat_id, status, time, acttime): - self.chat_id = str(chat_id) - self.status = status - self.time = time - self.acttime = acttime - # self.permanent = permanent - - -Welcome.__table__.create(checkfirst=True) -WelcomeButtons.__table__.create(checkfirst=True) -GoodbyeButtons.__table__.create(checkfirst=True) -WelcomeMute.__table__.create(checkfirst=True) -WelcomeMuteUsers.__table__.create(checkfirst=True) -CleanServiceSetting.__table__.create(checkfirst=True) -RaidMode.__table__.create(checkfirst=True) - -INSERTION_LOCK = threading.RLock() -WELC_BTN_LOCK = threading.RLock() -LEAVE_BTN_LOCK = threading.RLock() -WM_LOCK = threading.RLock() -CS_LOCK = threading.RLock() -RAID_LOCK = threading.RLock() - - -def welcome_mutes(chat_id): - try: - welcomemutes = SESSION.query(WelcomeMute).get(str(chat_id)) - if welcomemutes: - return welcomemutes.welcomemutes - return False - finally: - SESSION.close() - - -def set_welcome_mutes(chat_id, welcomemutes): - with WM_LOCK: - prev = SESSION.query(WelcomeMute).get((str(chat_id))) - if prev: - SESSION.delete(prev) - welcome_m = WelcomeMute(str(chat_id), welcomemutes) - SESSION.add(welcome_m) - SESSION.commit() - - -def set_human_checks(user_id, chat_id): - with INSERTION_LOCK: - human_check = SESSION.query(WelcomeMuteUsers).get((user_id, str(chat_id))) - if not human_check: - human_check = WelcomeMuteUsers(user_id, str(chat_id), True) - - else: - human_check.human_check = True - - SESSION.add(human_check) - SESSION.commit() - - return human_check - - -def get_human_checks(user_id, chat_id): - try: - human_check = SESSION.query(WelcomeMuteUsers).get((user_id, str(chat_id))) - if not human_check: - return None - human_check = human_check.human_check - return human_check - finally: - SESSION.close() - - -def get_welc_mutes_pref(chat_id): - welcomemutes = SESSION.query(WelcomeMute).get(str(chat_id)) - SESSION.close() - - if welcomemutes: - return welcomemutes.welcomemutes - - return False - - -def get_welc_pref(chat_id): - welc = SESSION.query(Welcome).get(str(chat_id)) - SESSION.close() - if welc: - return ( - welc.should_welcome, - welc.custom_welcome, - welc.custom_content, - welc.welcome_type, - ) - - else: - # Welcome by default. - return True, DEFAULT_WELCOME, None, Types.TEXT - - -def get_gdbye_pref(chat_id): - welc = SESSION.query(Welcome).get(str(chat_id)) - SESSION.close() - if welc: - return welc.should_goodbye, welc.custom_leave, welc.leave_type - else: - # Welcome by default. - return True, DEFAULT_GOODBYE, Types.TEXT - - -def set_clean_welcome(chat_id, clean_welcome): - with INSERTION_LOCK: - curr = SESSION.query(Welcome).get(str(chat_id)) - if not curr: - curr = Welcome(str(chat_id)) - - curr.clean_welcome = int(clean_welcome) - - SESSION.add(curr) - SESSION.commit() - - -def get_clean_pref(chat_id): - welc = SESSION.query(Welcome).get(str(chat_id)) - SESSION.close() - - if welc: - return welc.clean_welcome - - return False - - -def set_welc_preference(chat_id, should_welcome): - with INSERTION_LOCK: - curr = SESSION.query(Welcome).get(str(chat_id)) - if not curr: - curr = Welcome(str(chat_id), should_welcome=should_welcome) - else: - curr.should_welcome = should_welcome - - SESSION.add(curr) - SESSION.commit() - - -def set_gdbye_preference(chat_id, should_goodbye): - with INSERTION_LOCK: - curr = SESSION.query(Welcome).get(str(chat_id)) - if not curr: - curr = Welcome(str(chat_id), should_goodbye=should_goodbye) - else: - curr.should_goodbye = should_goodbye - - SESSION.add(curr) - SESSION.commit() - - -def set_custom_welcome( - chat_id, custom_content, custom_welcome, welcome_type, buttons=None -): - if buttons is None: - buttons = [] - - with INSERTION_LOCK: - welcome_settings = SESSION.query(Welcome).get(str(chat_id)) - if not welcome_settings: - welcome_settings = Welcome(str(chat_id), True) - - if custom_welcome or custom_content: - welcome_settings.custom_content = custom_content - welcome_settings.custom_welcome = custom_welcome - welcome_settings.welcome_type = welcome_type.value - - else: - welcome_settings.custom_welcome = DEFAULT_WELCOME - welcome_settings.welcome_type = Types.TEXT.value - - SESSION.add(welcome_settings) - - with WELC_BTN_LOCK: - prev_buttons = ( - SESSION.query(WelcomeButtons) - .filter(WelcomeButtons.chat_id == str(chat_id)) - .all() - ) - for btn in prev_buttons: - SESSION.delete(btn) - - for b_name, url, same_line in buttons: - button = WelcomeButtons(chat_id, b_name, url, same_line) - SESSION.add(button) - - SESSION.commit() - - -def get_custom_welcome(chat_id): - welcome_settings = SESSION.query(Welcome).get(str(chat_id)) - ret = DEFAULT_WELCOME - if welcome_settings and welcome_settings.custom_welcome: - ret = welcome_settings.custom_welcome - - SESSION.close() - return ret - - -def set_custom_gdbye(chat_id, custom_goodbye, goodbye_type, buttons=None): - if buttons is None: - buttons = [] - - with INSERTION_LOCK: - welcome_settings = SESSION.query(Welcome).get(str(chat_id)) - if not welcome_settings: - welcome_settings = Welcome(str(chat_id), True) - - if custom_goodbye: - welcome_settings.custom_leave = custom_goodbye - welcome_settings.leave_type = goodbye_type.value - - else: - welcome_settings.custom_leave = DEFAULT_GOODBYE - welcome_settings.leave_type = Types.TEXT.value - - SESSION.add(welcome_settings) - - with LEAVE_BTN_LOCK: - prev_buttons = ( - SESSION.query(GoodbyeButtons) - .filter(GoodbyeButtons.chat_id == str(chat_id)) - .all() - ) - for btn in prev_buttons: - SESSION.delete(btn) - - for b_name, url, same_line in buttons: - button = GoodbyeButtons(chat_id, b_name, url, same_line) - SESSION.add(button) - - SESSION.commit() - - -def get_custom_gdbye(chat_id): - welcome_settings = SESSION.query(Welcome).get(str(chat_id)) - ret = DEFAULT_GOODBYE - if welcome_settings and welcome_settings.custom_leave: - ret = welcome_settings.custom_leave - - SESSION.close() - return ret - - -def get_welc_buttons(chat_id): - try: - return ( - SESSION.query(WelcomeButtons) - .filter(WelcomeButtons.chat_id == str(chat_id)) - .order_by(WelcomeButtons.id) - .all() - ) - finally: - SESSION.close() - - -def get_gdbye_buttons(chat_id): - try: - return ( - SESSION.query(GoodbyeButtons) - .filter(GoodbyeButtons.chat_id == str(chat_id)) - .order_by(GoodbyeButtons.id) - .all() - ) - finally: - SESSION.close() - - -def clean_service(chat_id: Union[str, int]) -> bool: - try: - chat_setting = SESSION.query(CleanServiceSetting).get(str(chat_id)) - if chat_setting: - return chat_setting.clean_service - return False - finally: - SESSION.close() - - -def set_clean_service(chat_id: Union[int, str], setting: bool): - with CS_LOCK: - chat_setting = SESSION.query(CleanServiceSetting).get(str(chat_id)) - if not chat_setting: - chat_setting = CleanServiceSetting(chat_id) - - chat_setting.clean_service = setting - SESSION.add(chat_setting) - SESSION.commit() - - -def migrate_chat(old_chat_id, new_chat_id): - with INSERTION_LOCK: - chat = SESSION.query(Welcome).get(str(old_chat_id)) - if chat: - chat.chat_id = str(new_chat_id) - - with WELC_BTN_LOCK: - chat_buttons = ( - SESSION.query(WelcomeButtons) - .filter(WelcomeButtons.chat_id == str(old_chat_id)) - .all() - ) - for btn in chat_buttons: - btn.chat_id = str(new_chat_id) - - with LEAVE_BTN_LOCK: - chat_buttons = ( - SESSION.query(GoodbyeButtons) - .filter(GoodbyeButtons.chat_id == str(old_chat_id)) - .all() - ) - for btn in chat_buttons: - btn.chat_id = str(new_chat_id) - - SESSION.commit() - - -def getRaidStatus(chat_id): - try: - if stat := SESSION.query(RaidMode).get(str(chat_id)): - return stat.status, stat.time, stat.acttime - return False, 21600, 3600 # default - finally: - SESSION.close() - - -def setRaidStatus(chat_id, status, time=21600, acttime=3600): - with RAID_LOCK: - if prevObj := SESSION.query(RaidMode).get(str(chat_id)): - SESSION.delete(prevObj) - newObj = RaidMode(str(chat_id), status, time, acttime) - SESSION.add(newObj) - SESSION.commit() - - -def toggleRaidStatus(chat_id): - newObj = True - with RAID_LOCK: - prevObj = SESSION.query(RaidMode).get(str(chat_id)) - if prevObj: - newObj = not prevObj.status - stat = RaidMode( - str(chat_id), newObj, prevObj.time or 21600, prevObj.acttime or 3600 - ) - SESSION.add(stat) - SESSION.commit() - return newObj - - -def _ResetRaidOnRestart(): - with RAID_LOCK: - raid = SESSION.query(RaidMode).all() - for r in raid: - r.status = False - SESSION.commit() - - -# it uses a cron job to turn off so if the bot restarts and there is a pending raid disable job then raid will stay on -_ResetRaidOnRestart()