1import logging
2from datetime import datetime
3
4from telebot.apihelper import ApiTelegramException
5from telebot.formatting import escape_html
6
7from app import telegram
8from app.utils.system import readable_size
9from config import TELEGRAM_ADMIN_ID, TELEGRAM_LOGGER_CHANNEL_ID
10
11logger = logging.getLogger(__name__)
12
13
14async def report(message: str, parse_mode="html", keyboard=None):
15 if telegram.bot and (TELEGRAM_ADMIN_ID or TELEGRAM_LOGGER_CHANNEL_ID):
16 try:
17 if TELEGRAM_LOGGER_CHANNEL_ID:
18 await telegram.bot.send_message(
19 TELEGRAM_LOGGER_CHANNEL_ID, message, parse_mode=parse_mode
20 )
21 else:
22 for admin in TELEGRAM_ADMIN_ID:
23 await telegram.bot.send_message(
24 admin, message, parse_mode=parse_mode, reply_markup=keyboard
25 )
26 except ApiTelegramException as e:
27 logger.error(e)
28
29
30async def report_new_user(
31 user_id: int,
32 username: str,
33 by: str,
34 expire_date: datetime,
35 data_limit: int,
36 services: list,
37):
38 text = """\
39🆕 <b>#Created</b>
40➖➖➖➖➖➖➖➖➖
41<b>Username :</b> <code>{username}</code>
42<b>Traffic Limit :</b> <code>{data_limit}</code>
43<b>Expire Date :</b> <code>{expire_date}</code>
44<b>Services :</b> <code>{services}</code>
45➖➖➖➖➖➖➖➖➖
46<b>By :</b> <b>#{by}</b>""".format(
47 by=escape_html(by),
48 username=escape_html(username),
49 data_limit=readable_size(data_limit) if data_limit else "Unlimited",
50 expire_date=(
51 expire_date.strftime("%H:%M:%S %Y-%m-%d") if expire_date else "Never"
52 ),
53 services="" if not services else ", ".join([s.name for s in services]),
54 )
55
56 return await report(text)
57
58
59async def report_user_modification(
60 username: str, expire_date: datetime, data_limit: int, services: list, by: str
61):
62 text = """\
63✏️ <b>#Modified</b>
64➖➖➖➖➖➖➖➖➖
65<b>Username :</b> <code>{username}</code>
66<b>Traffic Limit :</b> <code>{data_limit}</code>
67<b>Expire Date :</b> <code>{expire_date}</code>
68<b>Services :</b> <code>{services}</code>
69➖➖➖➖➖➖➖➖➖
70<b>By :</b> <b>#{by}</b>\
71 """.format(
72 by=escape_html(by),
73 username=escape_html(username),
74 data_limit=readable_size(data_limit) if data_limit else "Unlimited",
75 expire_date=(
76 expire_date.strftime("%H:%M:%S %Y-%m-%d") if expire_date else "Never"
77 ),
78 services=", ".join([s.name for s in services]),
79 )
80
81 return await report(text)
82
83
84async def report_user_deletion(username: str, by: str):
85 text = """\
86🗑 <b>#Deleted</b>
87➖➖➖➖➖➖➖➖➖
88<b>Username</b> : <code>{username}</code>
89➖➖➖➖➖➖➖➖➖
90<b>By</b> : <b>#{by}</b>\
91 """.format(
92 by=escape_html(by), username=escape_html(username)
93 )
94 return await report(text)
95
96
97async def report_status_change(username: str, status: str):
98 _status = {
99 "active": "✅ <b>#Activated</b>",
100 "on_hold": "🟡 <b>#On_Hold</b>",
101 "limited": "🪫 <b>#Limited</b>",
102 "expired": "🕔 <b>#Expired</b>",
103 }
104 text = """\
105{status}
106➖➖➖➖➖➖➖➖➖
107<b>Username</b> : <code>{username}</code>\
108 """.format(
109 username=escape_html(username), status=_status[status]
110 )
111 return await report(text)
112
113
114async def report_user_usage_reset(username: str, by: str):
115 text = """
116🔁 <b>#Reset</b>
117➖➖➖➖➖➖➖➖➖
118<b>Username</b> : <code>{username}</code>
119➖➖➖➖➖➖➖➖➖
120<b>By</b> : <b>#{by}</b>\
121 """.format(
122 by=escape_html(by), username=escape_html(username)
123 )
124
125 return await report(text)
126
127
128async def report_user_subscription_revoked(username: str, by: str):
129 text = """
130🔁 <b>#Revoked</b>
131➖➖➖➖➖➖➖➖➖
132<b>Username</b> : <code>{username}</code>
133➖➖➖➖➖➖➖➖➖
134<b>By</b> : <b>#{by}</b>\
135 """.format(
136 by=escape_html(by), username=escape_html(username)
137 )
138
139 return await report(text)