from datetime import datetime from typing import Dict, Optional from aiogram import html from app.models.admin import Admin from app.models.notification import Notification, UserNotification from app.models.user import UserResponse from app.utils.system import readable_size def create_text(notif: Notification) -> str: data = prepare_data(notif) A = UserNotification.Action texts: Dict[UserNotification.Action, str] = { A.user_created: "šŸ†• #Created\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nTraffic Limit : {data_limit}\nExpire Date : {expire_date}\nServices : {services}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBelongs To : {owner_username}\nBy : #{by}", A.user_updated: "āœļø #Modified\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nTraffic Limit : {data_limit}\nExpire Date : {expire_date}\nServices : {services}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBelongs To : {owner_username}\nBy : #{by}", A.user_activated: "āœ… #Activated\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nBelongs To : {owner_username}", A.user_deactivated: "āŒ #Deactivated\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nBelongs To : {owner_username}", A.user_deleted: "šŸ—‘ #Deleted\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBelongs To : {owner_username}\nBy : #{by}", A.user_enabled: "ā˜‘ļø #Enabled\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBelongs To : {owner_username}\nBy : #{by}", A.user_disabled: "šŸ›‘ #Disabled\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBelongs To : {owner_username}\nBy : #{by}", A.data_usage_reset: "šŸ” #Reset\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBy : #{by}", A.subscription_revoked: "šŸ” #Revoked\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBy : #{by}", A.reached_usage_percent: "āš ļø#DataLimitWarning\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nUsed Percent : {usage_percent}\nRemaining Traffic : {remaining_traffic}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBelongs To : {owner_username}", A.reached_days_left: "āš ļø#ExpirationWarning\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nRemaining Days : {remaining_days}\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nBelongs To : {owner_username}", } if notif.action == A.user_deactivated and not notif.by: if notif.user.expired: texts[A.user_deactivated] = ( "šŸ•” #Expired\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nBelongs To : {owner_username}" ) elif notif.user.data_limit_reached: texts[A.user_deactivated] = ( "šŸŖ« #Limited\nāž–āž–āž–āž–āž–āž–āž–āž–āž–\nUsername : {username}\nBelongs To : {owner_username}" ) text = texts.get(notif.action) formatted_message = text.format_map(data) return formatted_message def prepare_data(notif: Notification) -> dict: user = UserResponse.model_validate(notif.user) by: Optional[Admin] = ( Admin.model_validate(notif.by) if hasattr(notif, "by") and notif.by else None ) data = { "username": html.quote(user.username), "data_limit": ( readable_size(user.data_limit) if user.data_limit else "Unlimited" ), "remaining_traffic": ( readable_size(max(user.data_limit - user.used_traffic, 0)) if user.data_limit else "Unlimited" ), "usage_percent": ( f"{round(min((user.used_traffic / user.data_limit) * 100, 100),2)}%" if isinstance(user.data_limit, int) and user.data_limit > 0 else "0%" ), "expire_date": ( user.expire_date.strftime("%H:%M:%S %Y-%m-%d") if user.expire_date else "Never" ), "remaining_days": ( (user.expire_date - datetime.now()).days if user.expire_date else "Never" ), "services": user.service_ids if user.service_ids else "", "owner_username": user.owner_username, "by": html.quote(by.username) if by else None, } return data