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