import datetime
import logging
from datetime import datetime
from telebot.apihelper import ApiTelegramException
from telebot.formatting import escape_html
from app.telegram import bot
from app.telegram.utils.keyboard import BotKeyboard
from app.utils.system import readable_size
from config import TELEGRAM_ADMIN_ID, TELEGRAM_LOGGER_CHANNEL_ID
logger = logging.getLogger(__name__)
async def report(message: str, parse_mode="html", keyboard=None):
if bot and (TELEGRAM_ADMIN_ID or TELEGRAM_LOGGER_CHANNEL_ID):
try:
if TELEGRAM_LOGGER_CHANNEL_ID:
await bot.send_message(TELEGRAM_LOGGER_CHANNEL_ID, message, parse_mode=parse_mode)
else:
for admin in TELEGRAM_ADMIN_ID:
await bot.send_message(admin, message, parse_mode=parse_mode, reply_markup=keyboard)
except ApiTelegramException as e:
logger.error(e)
async def report_new_user(user_id: int, username: str, by: str, expire_date: int, data_limit: int, services: list):
text = '''\
đ #Created
âââââââââ
Username : {username}
Traffic Limit : {data_limit}
Expire Date : {expire_date}
Services : {services}
âââââââââ
By : #{by}'''.format(
by=escape_html(by),
username=escape_html(username),
data_limit=readable_size(data_limit) if data_limit else "Unlimited",
expire_date=datetime.fromtimestamp(expire_date).strftime("%H:%M:%S %Y-%m-%d") if expire_date else "Never",
services="" if not services else ", ".join([s.name for s in services])
)
return await report(
text,
keyboard=BotKeyboard.user_menu({
'username': username,
'id': user_id,
'status': 'active'
}, with_back=False)
)
async def report_user_modification(username: str, expire_date: int, data_limit: int, services: list, by: str):
text = '''\
âī¸ #Modified
âââââââââ
Username : {username}
Traffic Limit : {data_limit}
Expire Date : {expire_date}
Services : {services}
âââââââââ
By : #{by}\
'''.format(
by=escape_html(by),
username=escape_html(username),
data_limit=readable_size(data_limit) if data_limit else "Unlimited",
expire_date=datetime.fromtimestamp(expire_date).strftime("%H:%M:%S %Y-%m-%d") if expire_date else "Never",
services=", ".join([s.name for s in services])
)
return await report(text, keyboard=BotKeyboard.user_menu({
'username': username,
'status': 'active'
}, with_back=False))
async def report_user_deletion(username: str, by: str):
text = '''\
đ #Deleted
âââââââââ
Username : {username}
âââââââââ
By : #{by}\
'''.format(
by=escape_html(by),
username=escape_html(username)
)
return await report(text)
async def report_status_change(username: str, status: str):
_status = {
'active': 'â
#Activated',
'disabled': 'â #Disabled',
'limited': 'đĒĢ #Limited',
'expired': 'đ #Expired'
}
text = '''\
{status}
âââââââââ
Username : {username}\
'''.format(
username=escape_html(username),
status=_status[status]
)
return await report(text)
async def report_user_usage_reset(username: str, by: str):
text = """
đ #Reset
âââââââââ
Username : {username}
âââââââââ
By : #{by}\
""".format(
by=escape_html(by),
username=escape_html(username)
)
return await report(text)
async def report_user_subscription_revoked(username: str, by: str):
text = """
đ #Revoked
âââââââââ
Username : {username}
âââââââââ
By : #{by}\
""".format(
by=escape_html(by),
username=escape_html(username)
)
return await report(text)