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