1from datetime import datetime as dt
  2from typing import Optional
  3
  4from app import telegram
  5from app.db import Session, create_notification_reminder
  6from app.db.models import UserStatus
  7from app.models.admin import Admin
  8from app.models.user import ReminderType, UserResponse
  9from app.utils.notification import (
 10    Notification,
 11    ReachedDaysLeft,
 12    ReachedUsagePercent,
 13    UserCreated,
 14    UserDataUsageReset,
 15    UserDeleted,
 16    UserEnabled,
 17    UserExpired,
 18    UserLimited,
 19    UserSubscriptionRevoked,
 20    UserUpdated,
 21    notify,
 22)
 23
 24
 25async def status_change(
 26    username: str, status: UserStatus, user: UserResponse, by: Optional[Admin] = None
 27) -> None:
 28    try:
 29        await telegram.report_status_change(username, status)
 30    except Exception:
 31        pass
 32    if status == UserStatus.limited:
 33        await notify(
 34            UserLimited(
 35                username=username, action=Notification.Type.user_limited, user=user
 36            )
 37        )
 38    elif status == UserStatus.expired:
 39        await notify(
 40            UserExpired(
 41                username=username, action=Notification.Type.user_expired, user=user
 42            )
 43        )
 44    elif status == UserStatus.active:
 45        await notify(
 46            UserEnabled(
 47                username=username,
 48                action=Notification.Type.user_enabled,
 49                user=user,
 50                by=by,
 51            )
 52        )
 53
 54
 55async def user_created(user: UserResponse, user_id: int, by: Admin) -> None:
 56    await telegram.report_new_user(
 57        user_id=user_id,
 58        username=user.username,
 59        by=by.username,
 60        expire_date=user.expire,
 61        data_limit=user.data_limit,
 62        # proxies=[]
 63        services=user.services,
 64    )
 65
 66    await notify(
 67        UserCreated(
 68            username=user.username,
 69            action=Notification.Type.user_created,
 70            by=by,
 71            user=user,
 72        )
 73    )
 74
 75
 76async def user_updated(user: UserResponse, by: Admin) -> None:
 77    try:
 78        await telegram.report_user_modification(
 79            username=user.username,
 80            expire_date=user.expire,
 81            data_limit=user.data_limit,
 82            services=user.services,
 83            by=by.username,
 84        )
 85    except Exception:
 86        pass
 87    await notify(
 88        UserUpdated(
 89            username=user.username,
 90            action=Notification.Type.user_updated,
 91            by=by,
 92            user=user,
 93        )
 94    )
 95
 96
 97async def user_deleted(username: str, by: Admin) -> None:
 98    try:
 99        await telegram.report_user_deletion(username=username, by=by.username)
100    except Exception:
101        pass
102    await notify(
103        UserDeleted(username=username, action=Notification.Type.user_deleted, by=by)
104    )
105
106
107async def user_data_usage_reset(user: UserResponse, by: Admin) -> None:
108    try:
109        await telegram.report_user_usage_reset(
110            username=user.username,
111            by=by.username,
112        )
113    except Exception:
114        pass
115    await notify(
116        UserDataUsageReset(
117            username=user.username,
118            action=Notification.Type.data_usage_reset,
119            by=by,
120            user=user,
121        )
122    )
123
124
125async def user_subscription_revoked(user: UserResponse, by: Admin) -> None:
126    try:
127        await telegram.report_user_subscription_revoked(
128            username=user.username,
129            by=by.username,
130        )
131    except Exception:
132        pass
133    await notify(
134        UserSubscriptionRevoked(
135            username=user.username,
136            action=Notification.Type.subscription_revoked,
137            by=by,
138            user=user,
139        )
140    )
141
142
143async def data_usage_percent_reached(
144    db: Session,
145    percent: float,
146    user: UserResponse,
147    user_id: int,
148    expire: Optional[int] = None,
149) -> None:
150    await notify(
151        ReachedUsagePercent(username=user.username, user=user, used_percent=percent)
152    )
153    await create_notification_reminder(
154        db,
155        ReminderType.data_usage,
156        expires_at=dt.utcfromtimestamp(expire) if expire else None,
157        user_id=user_id,
158    )
159
160
161async def expire_days_reached(
162    db: Session, days: int, user: UserResponse, user_id: int, expire: int
163) -> None:
164    await notify(ReachedDaysLeft(username=user.username, user=user, days_left=days))
165    await create_notification_reminder(
166        db,
167        ReminderType.expiration_date,
168        expires_at=dt.utcfromtimestamp(expire),
169        user_id=user_id,
170    )