1import asyncio
 2import logging
 3from datetime import datetime
 4from typing import TYPE_CHECKING
 5
 6from sqlalchemy.orm import Session
 7
 8from app import marznode
 9from app.db import (GetDB, get_notification_reminder, get_users,
10                    start_user_expire, update_user_status)
11from app.models.user import ReminderType, UserResponse, UserStatus
12from app.utils import report
13from app.utils.helpers import (calculate_expiration_days,
14                               calculate_usage_percent)
15from config import (NOTIFY_DAYS_LEFT, NOTIFY_REACHED_USAGE_PERCENT,
16                    WEBHOOK_ADDRESS)
17
18if TYPE_CHECKING:
19    from app.db.models import User
20
21
22logger = logging.getLogger(__name__)
23
24
25def add_notification_reminders(db: Session, user: "User", now: datetime = datetime.utcnow()) -> None:
26    if user.data_limit:
27        usage_percent = calculate_usage_percent(
28            user.used_traffic, user.data_limit)
29        if (usage_percent >= NOTIFY_REACHED_USAGE_PERCENT) and (not get_notification_reminder(db, user.id, ReminderType.data_usage)):
30            report.data_usage_percent_reached(
31                db, usage_percent, UserResponse.model_validate(user),
32                user.id, user.expire)
33
34    if user.expire and ((now - user.created_at).days >= NOTIFY_DAYS_LEFT):
35        expire_days = calculate_expiration_days(user.expire)
36        if (expire_days <= NOTIFY_DAYS_LEFT) and (not get_notification_reminder(db, user.id, ReminderType.expiration_date)):
37            report.expire_days_reached(
38                db, expire_days, UserResponse.model_validate(user),
39                user.id, user.expire)
40
41
42async def review_users():
43    now = datetime.utcnow()
44    with GetDB() as db:
45        for user in get_users(db, status=UserStatus.active):
46
47            limited = user.data_limit and user.used_traffic >= user.data_limit
48            expired = user.expire and user.expire <= now.timestamp()
49            if limited:
50                status = UserStatus.limited
51            elif expired:
52                status = UserStatus.expired
53            else:
54                if WEBHOOK_ADDRESS:
55                    add_notification_reminders(db, user, now)
56                continue
57
58            await marznode.operations.remove_user(user)
59            update_user_status(db, user, status)
60
61            asyncio.create_task(report.status_change(
62                user.username,
63                user.status,
64                UserResponse.model_validate(user)))
65
66            logger.info(f"User \"{user.username}\" status changed to {status}")
67
68        for user in get_users(db, status=UserStatus.on_hold):
69
70            if user.edit_at:
71                base_time = datetime.timestamp(user.edit_at)
72            else:
73                base_time = datetime.timestamp(user.created_at)
74
75            # Check if the user is online After or at 'base_time'
76            if user.online_at and base_time <= datetime.timestamp(user.online_at):
77                status = UserStatus.active
78
79            elif user.on_hold_timeout and (datetime.timestamp(user.on_hold_timeout) <= (now.timestamp())):
80                # If the user didn't connect within the timeout period, change status to "Active"
81                status = UserStatus.active
82
83            else:
84                continue
85
86            update_user_status(db, user, status)
87            start_user_expire(db, user)
88            asyncio.create_task(report.status_change(
89                user.username,
90                user.status,
91                UserResponse.model_validate(user)))
92            logger.info(f"on hold user `{user.username}` has been activated")
93
94