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: datetime, 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=( 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: datetime, 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=( 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", "on_hold": "🟡 #On_Hold", "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)