CoCalc -- chat_permission.py

GitHub Repository: anasty17/mirror-leech-telegram-bot
Path: blob/master/bot/modules/chat_permission.py
¹⁶³⁰ views
1
from .. import user_data
2
from ..helper.ext_utils.bot_utils import update_user_ldata, new_task
3
from ..helper.ext_utils.db_handler import database
4
from ..helper.telegram_helper.message_utils import send_message
5

6

7
@new_task
8
async def authorize(_, message):
9
    msg = message.text.split()
10
    thread_id = None
11
    try:
12
        if len(msg) > 1:
13
            if "|" in msg:
14
                chat_id, thread_id = list(map(int, msg[1].split("|")))
15
            else:
16
                chat_id = int(msg[1].strip())
17
        elif (
18
            reply_to := message.reply_to_message
19
        ) and reply_to.id != message.message_thread_id:
20
            chat_id = (
21
                reply_to.from_user.id if reply_to.from_user else reply_to.sender_chat.id
22
            )
23
        else:
24
            if message.topic_message:
25
                thread_id = message.message_thread_id
26
            chat_id = message.chat.id
27
        if chat_id in user_data and user_data[chat_id].get("AUTH"):
28
            if (
29
                thread_id is not None
30
                and thread_id in user_data[chat_id].get("thread_ids", [])
31
                or thread_id is None
32
            ):
33
                msg = "Already Authorized!"
34
            else:
35
                if "thread_ids" in user_data[chat_id]:
36
                    user_data[chat_id]["thread_ids"].append(thread_id)
37
                else:
38
                    user_data[chat_id]["thread_ids"] = [thread_id]
39
                msg = "Authorized"
40
        else:
41
            update_user_ldata(chat_id, "AUTH", True)
42
            if thread_id is not None:
43
                update_user_ldata(chat_id, "thread_ids", [thread_id])
44
            await database.update_user_data(chat_id)
45
            msg = "Authorized"
46
    except Exception as e:
47
        msg = f"Error: {e}"
48
    await send_message(message, msg)
49

50

51
@new_task
52
async def unauthorize(_, message):
53
    msg = message.text.split()
54
    thread_id = None
55
    try:
56
        if len(msg) > 1:
57
            if "|" in msg:
58
                chat_id, thread_id = list(map(int, msg[1].split("|")))
59
            else:
60
                chat_id = int(msg[1].strip())
61
        elif (
62
            reply_to := message.reply_to_message
63
        ) and reply_to.id != message.message_thread_id:
64
            chat_id = (
65
                reply_to.from_user.id if reply_to.from_user else reply_to.sender_chat.id
66
            )
67
        else:
68
            if message.topic_message:
69
                thread_id = message.message_thread_id
70
            chat_id = message.chat.id
71
        if chat_id in user_data and user_data[chat_id].get("AUTH"):
72
            if thread_id is not None and thread_id in user_data[chat_id].get(
73
                "thread_ids", []
74
            ):
75
                user_data[chat_id]["thread_ids"].remove(thread_id)
76
            else:
77
                update_user_ldata(chat_id, "AUTH", False)
78
            await database.update_user_data(chat_id)
79
            msg = "Unauthorized"
80
        else:
81
            msg = "Already Unauthorized! Authorized Chats added from config must be removed from config."
82
    except Exception as e:
83
        msg = f"Error: {e}"
84
    await send_message(message, msg)
85

86

87
@new_task
88
async def add_sudo(_, message):
89
    id_ = ""
90
    msg = message.text.split()
91
    try:
92
        if len(msg) > 1:
93
            id_ = int(msg[1].strip())
94
        elif reply_to := message.reply_to_message:
95
            id_ = reply_to.from_user.id if reply_to.from_user else reply_to.sender_chat.id
96
        if id_:
97
            if id_ in user_data and user_data[id_].get("SUDO"):
98
                msg = "Already Sudo!"
99
            else:
100
                update_user_ldata(id_, "SUDO", True)
101
                await database.update_user_data(id_)
102
                msg = "Promoted as Sudo"
103
        else:
104
            msg = "Give ID or Reply To message of whom you want to Promote."
105
    except Exception as e:
106
        msg = f"Error: {e}"
107
    await send_message(message, msg)
108

109

110
@new_task
111
async def remove_sudo(_, message):
112
    id_ = ""
113
    msg = message.text.split()
114
    try:
115
        if len(msg) > 1:
116
            id_ = int(msg[1].strip())
117
        elif reply_to := message.reply_to_message:
118
            id_ = reply_to.from_user.id if reply_to.from_user else reply_to.sender_chat.id
119
        if id_:
120
            if id_ in user_data and user_data[id_].get("SUDO"):
121
                update_user_ldata(id_, "SUDO", False)
122
                await database.update_user_data(id_)
123
                msg = "Demoted"
124
            else:
125
                msg = "Already Not Sudo! Sudo users added from config must be removed from config."
126
        else:
127
            msg = "Give ID or Reply To message of whom you want to remove from Sudo"
128
    except Exception as e:
129
        msg = f"Error: {e}"
130
    await send_message(message, msg)
131

132
Product

Resources

Company
