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)