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