diff --git a/app/db/__init__.py b/app/db/__init__.py
@@ -4,13 +4,10 @@ from sqlalchemy.orm import Session
from .base import Base, SessionLocal, engine # noqa
from .crud import (
create_admin,
- create_notification_reminder, # noqa
create_user,
- delete_notification_reminder,
get_admin,
get_admins,
get_jwt_secret_key,
- get_notification_reminder,
ensure_node_inbounds,
get_system_usage,
get_tls_certificate,
@@ -64,9 +61,6 @@ __all__ = [
"update_admin",
"remove_admin",
"get_admins",
- "create_notification_reminder",
- "get_notification_reminder",
- "delete_notification_reminder",
"GetDB",
"User",
"System",
diff --git a/app/db/crud.py b/app/db/crud.py
@@ -15,7 +15,6 @@ from app.db.models import (
Admin,
Node,
NodeUserUsage,
- NotificationReminder,
InboundHost,
Service,
Inbound,
@@ -33,7 +32,6 @@ from app.models.proxy import InboundHost as InboundHostModify
from app.models.service import Service as ServiceModify, ServiceCreate
from app.models.system import TrafficUsageSeries
from app.models.user import (
- ReminderType,
UserCreate,
UserDataUsageResetStrategy,
UserModify,
@@ -424,7 +422,6 @@ def get_user_usages(
start: datetime,
end: datetime,
) -> UserUsageSeriesResponse:
-
usages = defaultdict(dict)
cond = and_(
@@ -920,59 +917,3 @@ def update_node_status(
db_node.message = message
db_node.last_status_change = datetime.utcnow()
db.commit()
-
-
-def create_notification_reminder(
- db: Session,
- reminder_type: ReminderType,
- expires_at: datetime,
- user_id: int,
-) -> NotificationReminder:
- reminder = NotificationReminder(
- type=reminder_type, expires_at=expires_at, user_id=user_id
- )
- db.add(reminder)
- db.commit()
- db.refresh(reminder)
- return reminder
-
-
-def get_notification_reminder(
- db: Session,
- user_id: int,
- reminder_type: ReminderType,
-) -> Optional[NotificationReminder]:
- reminder = (
- db.query(NotificationReminder)
- .filter(NotificationReminder.user_id == user_id)
- .filter(NotificationReminder.type == reminder_type)
- .first()
- )
- if reminder is None:
- return
- if reminder.expires_at and reminder.expires_at < datetime.utcnow():
- db.delete(reminder)
- db.commit()
- return
- return reminder
-
-
-def delete_notification_reminder_by_type(
- db: Session, user_id: int, reminder_type: ReminderType
-) -> None:
- """Deletes notification reminder filtered by user_id and type if exists"""
- stmt = delete(NotificationReminder).where(
- NotificationReminder.user_id == user_id,
- NotificationReminder.type == reminder_type,
- )
- db.execute(stmt)
- db.commit()
- return
-
-
-def delete_notification_reminder(
- db: Session, dbreminder: NotificationReminder
-) -> None:
- db.delete(dbreminder)
- db.commit()
- return
diff --git a/app/db/migrations/versions/20241127_79271458a973_drop_notification_reminders_table.py b/app/db/migrations/versions/20241127_79271458a973_drop_notification_reminders_table.py
@@ -0,0 +1,45 @@
+"""drop notification reminders table
+
+Revision ID: 79271458a973
+Revises: be0032100c07
+Create Date: 2024-11-27 04:43:24.448736
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "79271458a973"
+down_revision = "be0032100c07"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ # ### commands auto generated by Alembic - please adjust! ###
+ op.drop_table("notification_reminders")
+ # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+ # ### commands auto generated by Alembic - please adjust! ###
+ op.create_table(
+ "notification_reminders",
+ sa.Column("id", sa.Integer(), nullable=False),
+ sa.Column("user_id", sa.Integer(), nullable=True),
+ sa.Column(
+ "type",
+ sa.Enum("expiration_date", "data_usage", name="remindertype"),
+ nullable=False,
+ ),
+ sa.Column("expires_at", sa.DateTime(), nullable=True),
+ sa.Column("created_at", sa.DateTime(), nullable=True),
+ sa.ForeignKeyConstraint(
+ ["user_id"],
+ ["users.id"],
+ ),
+ sa.PrimaryKeyConstraint("id"),
+ )
+ # ### end Alembic commands ###
diff --git a/app/db/models.py b/app/db/models.py
@@ -34,7 +34,6 @@ from app.models.proxy import (
ProxyTypes,
)
from app.models.user import (
- ReminderType,
UserDataUsageResetStrategy,
UserStatus,
UserExpireStrategy,
@@ -182,11 +181,6 @@ class User(Base):
back_populates="user",
cascade="all,delete,delete-orphan",
)
- notification_reminders = relationship(
- "NotificationReminder",
- back_populates="user",
- cascade="all,delete,delete-orphan",
- )
data_limit = Column(BigInteger)
data_limit_reset_strategy = Column(
Enum(UserDataUsageResetStrategy),
@@ -455,17 +449,6 @@ class NodeUsage(Base):
downlink = Column(BigInteger, default=0)
-class NotificationReminder(Base):
- __tablename__ = "notification_reminders"
-
- id = Column(Integer, primary_key=True)
- user_id = Column(Integer, ForeignKey("users.id"))
- user = relationship("User", back_populates="notification_reminders")
- type = Column(Enum(ReminderType), nullable=False)
- expires_at = Column(DateTime)
- created_at = Column(DateTime, default=datetime.utcnow)
-
-
class Settings(Base):
__tablename__ = "settings"
diff --git a/app/marzneshin.py b/app/marzneshin.py
@@ -1,8 +1,5 @@
-import asyncio
import logging
from contextlib import asynccontextmanager
-from datetime import datetime as dt
-from datetime import timedelta as td
from typing import AsyncGenerator
from apscheduler.schedulers.asyncio import AsyncIOScheduler
@@ -25,19 +22,18 @@ from app.config.env import (
UVICORN_SSL_KEYFILE,
UVICORN_UDS,
DASHBOARD_PATH,
- WEBHOOK_ADDRESS,
)
from app.templates import render_template
-from . import __version__, telegram
+from . import __version__
from .routes import api_router
from .tasks import (
- delete_expired_reminders,
nodes_startup,
record_user_usages,
reset_user_data_usage,
review_users,
- send_notifications,
+ expire_days_reached,
)
+from .webhooks import webhooks_router
logger = logging.getLogger(__name__)
@@ -47,8 +43,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
await nodes_startup()
yield
scheduler.shutdown()
- logger.info("Sending pending notifications before shutdown...")
- await send_notifications()
app = FastAPI(
@@ -60,6 +54,8 @@ app = FastAPI(
redoc_url="/redoc" if DOCS else None,
)
+app.webhooks.include_router(webhooks_router)
+
app.include_router(api_router)
add_pagination(app)
@@ -82,19 +78,11 @@ scheduler.add_job(record_user_usages, "interval", coalesce=True, seconds=30)
scheduler.add_job(
review_users, "interval", seconds=30, coalesce=True, max_instances=1
)
+scheduler.add_job(
+ expire_days_reached, "interval", seconds=30, coalesce=True, max_instances=1
+)
scheduler.add_job(reset_user_data_usage, "interval", coalesce=True, hours=1)
-if WEBHOOK_ADDRESS:
- scheduler.add_job(
- send_notifications, "interval", seconds=30, replace_existing=True
- )
- scheduler.add_job(
- delete_expired_reminders,
- "interval",
- hours=2,
- start_date=dt.utcnow() + td(minutes=1),
- )
-
@app.exception_handler(RequestValidationError)
def validation_exception_handler(
@@ -127,7 +115,6 @@ async def main():
name="locales",
)
scheduler.start()
- asyncio.create_task(telegram.start_bot())
cfg = Config(
app=app,
host=UVICORN_HOST,
diff --git a/app/marznode/base.py b/app/marznode/base.py
@@ -2,7 +2,6 @@ from abc import ABC
class MarzNodeBase(ABC):
-
async def stop(self):
"""stops all operations"""
diff --git a/app/models/notification.py b/app/models/notification.py
@@ -0,0 +1,90 @@
+from datetime import datetime as dt
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel
+
+from app.models.admin import Admin
+from app.models.user import UserResponse
+
+
+class Notification(BaseModel):
+ created_at: float = dt.utcnow().timestamp()
+
+
+class AdminNotif(Notification):
+ pass
+
+
+class UserNotification(Notification):
+ user: UserResponse
+
+ class Action(str, Enum):
+ user_created = "user_created"
+ user_updated = "user_updated"
+ user_activated = "user_activated"
+ user_deactivated = "user_deactivated"
+ user_deleted = "user_deleted"
+ user_enabled = "user_enabled"
+ user_disabled = "user_disabled"
+ data_usage_reset = "data_usage_reset"
+ subscription_revoked = "subscription_revoked"
+ reached_usage_percent = "reached_usage_percent"
+ reached_days_left = "reached_days_left"
+
+
+class UserCreated(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.user_created
+ by: Admin
+
+
+class UserUpdated(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.user_updated
+ by: Admin
+
+
+class UserActivated(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.user_activated
+ by: Optional[Admin] = None
+
+
+class UserDeactivated(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.user_deactivated
+ by: Optional[Admin] = None
+
+
+class UserDeleted(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.user_deleted
+ by: Admin
+
+
+class UserEnabled(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.user_enabled
+ by: Optional[Admin] = None
+
+
+class UserDisabled(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.user_disabled
+ by: Optional[Admin] = None
+
+
+class UserDataUsageReset(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.data_usage_reset
+ by: Optional[Admin] = None
+
+
+class UserSubscriptionRevoked(UserNotification):
+ action: UserNotification.Action = (
+ UserNotification.Action.subscription_revoked
+ )
+ by: Admin
+
+
+class ReachedUsagePercent(UserNotification):
+ action: UserNotification.Action = (
+ UserNotification.Action.reached_usage_percent
+ )
+
+
+class ReachedDaysLeft(UserNotification):
+ action: UserNotification.Action = UserNotification.Action.reached_days_left
diff --git a/app/models/user.py b/app/models/user.py
@@ -14,11 +14,6 @@ from pydantic import (
USERNAME_REGEXP = r"^\w{3,32}$"
-class ReminderType(str, Enum):
- expiration_date = "expiration_date"
- data_usage = "data_usage"
-
-
class UserStatus(str, Enum):
ACTIVE = "active"
INACTIVE = "inactive"
diff --git a/app/notification/__init__.py b/app/notification/__init__.py
@@ -0,0 +1,6 @@
+from .factory import get_notification_strategy
+from .services import get_notification_manager
+from .notifiers import notify
+
+
+__all__ = ["get_notification_manager", "get_notification_strategy", "notify"]
diff --git a/app/notification/factory.py b/app/notification/factory.py
@@ -0,0 +1,83 @@
+from abc import ABC, abstractmethod
+from enum import Enum
+from functools import lru_cache
+from typing import Any, Dict, Type, Optional, Union
+
+from app.models.admin import Admin
+from app.models.notification import (
+ Notification,
+ AdminNotif,
+ UserNotification,
+ UserCreated,
+ UserUpdated,
+ UserActivated,
+ UserDeactivated,
+ UserDeleted,
+ UserEnabled,
+ UserDisabled,
+ UserDataUsageReset,
+ UserSubscriptionRevoked,
+ ReachedUsagePercent,
+ ReachedDaysLeft,
+)
+from app.models.user import UserResponse
+
+
+class NotificationFactory(ABC):
+ @abstractmethod
+ def create_notification(self, action: Enum, **kwargs: Any) -> Notification:
+ raise NotImplementedError()
+
+
+class AdminNotificationFactory(NotificationFactory):
+ def create_notification(self, action: Enum, **kwargs: Any) -> Notification:
+ pass
+
+
+class UserNotificationFactory(NotificationFactory):
+ Action = UserNotification.Action
+
+ notification_classes: Dict[
+ UserNotification.Action, Type[UserNotification]
+ ] = {
+ Action.user_created: UserCreated,
+ Action.user_updated: UserUpdated,
+ Action.user_activated: UserActivated,
+ Action.user_deactivated: UserDeactivated,
+ Action.user_deleted: UserDeleted,
+ Action.user_enabled: UserEnabled,
+ Action.user_disabled: UserDisabled,
+ Action.data_usage_reset: UserDataUsageReset,
+ Action.subscription_revoked: UserSubscriptionRevoked,
+ Action.reached_usage_percent: ReachedUsagePercent,
+ Action.reached_days_left: ReachedDaysLeft,
+ }
+
+ def create_notification(
+ self,
+ action: UserNotification.Action,
+ user: UserResponse,
+ by: Optional[Admin] = None,
+ **kwargs: Any
+ ) -> UserNotification:
+ notification_class = self.notification_classes.get(action)
+ return notification_class(user=user, by=by, **kwargs)
+
+
+class NotificationStrategy:
+ def __init__(self):
+ self.user_factory = UserNotificationFactory()
+ self.admin_factory = AdminNotificationFactory()
+
+ def create_notification(
+ self, action: Enum, **kwargs: Any
+ ) -> Union[UserNotification, AdminNotif]:
+ if isinstance(action, UserNotification.Action):
+ return self.user_factory.create_notification(action, **kwargs)
+ elif isinstance(action, AdminNotif.Action):
+ return self.admin_factory.create_notification(action, **kwargs)
+
+
+@lru_cache(maxsize=1)
+def get_notification_strategy() -> NotificationStrategy:
+ return NotificationStrategy()
diff --git a/app/notification/helper.py b/app/notification/helper.py
@@ -0,0 +1,83 @@
+from datetime import datetime
+from typing import Dict, Optional
+
+from aiogram import html
+
+from app.models.admin import Admin
+from app.models.notification import Notification, UserNotification
+from app.models.user import UserResponse
+from app.utils.system import readable_size
+
+
+def create_text(notif: Notification) -> str:
+ data = prepare_data(notif)
+ A = UserNotification.Action
+
+ texts: Dict[UserNotification.Action, str] = {
+ A.user_created: "🆕 <b>#Created</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username :</b> <code>{username}</code>\n<b>Traffic Limit :</b> <code>{data_limit}</code>\n<b>Expire Date :</b> <code>{expire_date}</code>\n<b>Services :</b> <code>{services}</code>\n➖➖➖➖➖➖➖➖➖\n<b>Belongs To :</b> <code>{owner_username}</code>\n<b>By :</b> <b>#{by}</b>",
+ A.user_updated: "✏️ <b>#Modified</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username :</b> <code>{username}</code>\n<b>Traffic Limit :</b> <code>{data_limit}</code>\n<b>Expire Date :</b> <code>{expire_date}</code>\n<b>Services :</b> <code>{services}</code>\n➖➖➖➖➖➖➖➖➖\n<b>Belongs To :</b> <code>{owner_username}</code>\n<b>By :</b> <b>#{by}</b>",
+ A.user_activated: "✅ <b>#Activated</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n<b>Belongs To :</b> <code>{owner_username}</code>",
+ A.user_deactivated: "❌ <b>#Deactivated</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n<b>Belongs To :</b> <code>{owner_username}</code>",
+ A.user_deleted: "🗑 <b>#Deleted</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n➖➖➖➖➖➖➖➖➖\n<b>Belongs To :</b> <code>{owner_username}</code>\n<b>By :</b> <b>#{by}</b>",
+ A.user_enabled: "☑️ <b>#Enabled</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n➖➖➖➖➖➖➖➖➖\n<b>Belongs To :</b> <code>{owner_username}</code>\n<b>By :</b> <b>#{by}</b>",
+ A.user_disabled: "🛑 <b>#Disabled</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n➖➖➖➖➖➖➖➖➖\n<b>Belongs To :</b> <code>{owner_username}</code>\n<b>By :</b> <b>#{by}</b>",
+ A.data_usage_reset: "🔁 <b>#Reset</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n➖➖➖➖➖➖➖➖➖\n<b>By</b> : <b>#{by}</b>",
+ A.subscription_revoked: "🔁 <b>#Revoked</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n➖➖➖➖➖➖➖➖➖\n<b>By</b> : <b>#{by}</b>",
+ A.reached_usage_percent: "⚠️<b>#DataLimitWarning</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n<b>Used Percent</b> : <code>{usage_percent}</code>\n<b>Remaining Traffic</b> : <code>{remaining_traffic}</code>\n➖➖➖➖➖➖➖➖➖\n<b>Belongs To :</b> <code>{owner_username}</code>",
+ A.reached_days_left: "⚠️<b>#ExpirationWarning</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n<b>Remaining Days</b> : <code>{remaining_days}</code>\n➖➖➖➖➖➖➖➖➖\n<b>Belongs To :</b> <code>{owner_username}</code>",
+ }
+
+ if notif.action == A.user_deactivated and not notif.by:
+ if notif.user.expired:
+ texts[A.user_deactivated] = (
+ "🕔 <b>#Expired</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n<b>Belongs To :</b> <code>{owner_username}</code>"
+ )
+ elif notif.user.data_limit_reached:
+ texts[A.user_deactivated] = (
+ "🪫 <b>#Limited</b>\n➖➖➖➖➖➖➖➖➖\n<b>Username</b> : <code>{username}</code>\n<b>Belongs To :</b> <code>{owner_username}</code>"
+ )
+
+ text = texts.get(notif.action)
+ formatted_message = text.format_map(data)
+
+ return formatted_message
+
+
+def prepare_data(notif: Notification) -> dict:
+ user = UserResponse.model_validate(notif.user)
+ by: Optional[Admin] = (
+ Admin.model_validate(notif.by)
+ if hasattr(notif, "by") and notif.by
+ else None
+ )
+ data = {
+ "username": html.quote(user.username),
+ "data_limit": (
+ readable_size(user.data_limit) if user.data_limit else "Unlimited"
+ ),
+ "remaining_traffic": (
+ readable_size(max(user.data_limit - user.used_traffic, 0))
+ if user.data_limit
+ else "Unlimited"
+ ),
+ "usage_percent": (
+ f"{round(min((user.used_traffic / user.data_limit) * 100, 100),2)}%"
+ if isinstance(user.data_limit, int) and user.data_limit > 0
+ else "0%"
+ ),
+ "expire_date": (
+ user.expire_date.strftime("%H:%M:%S %Y-%m-%d")
+ if user.expire_date
+ else "Never"
+ ),
+ "remaining_days": (
+ (user.expire_date - datetime.now()).days
+ if user.expire_date
+ else "Never"
+ ),
+ "services": user.service_ids if user.service_ids else "",
+ "owner_username": user.owner_username,
+ "by": html.quote(by.username) if by else None,
+ }
+
+ return data
diff --git a/app/notification/notifiers.py b/app/notification/notifiers.py
@@ -0,0 +1,17 @@
+from enum import Enum
+from app.notification import (
+ get_notification_strategy,
+ get_notification_manager,
+)
+
+
+async def notify(action: Enum, **kwargs) -> None:
+ try:
+ manager = get_notification_manager()
+ except ValueError:
+ return
+
+ strategy = get_notification_strategy()
+ notification = strategy.create_notification(action=action, **kwargs)
+
+ await manager.send_notification(notification)
diff --git a/app/notification/services.py b/app/notification/services.py
@@ -0,0 +1,48 @@
+from functools import lru_cache
+from typing import List
+from abc import ABC, abstractmethod
+
+from app.config.env import TELEGRAM_API_TOKEN, WEBHOOK_ADDRESS
+from app.models.notification import Notification
+from app.notification import telegram, webhook
+
+
+class BaseNotificationService(ABC):
+ @abstractmethod
+ async def send_notification(self, message: Notification):
+ raise NotImplementedError()
+
+
+class WebhookNotificationService(BaseNotificationService):
+ async def send_notification(self, message: Notification) -> None:
+ await webhook.send_notification(message)
+
+
+class TelegramNotificationService(BaseNotificationService):
+ async def send_notification(self, message: Notification) -> None:
+ await telegram.send_notification(message)
+
+
+class NotificationManager:
+ def __init__(self, strategies: List[BaseNotificationService]) -> None:
+ self.strategies = strategies
+
+ async def send_notification(self, message: Notification) -> None:
+ for strategy in self.strategies:
+ await strategy.send_notification(message)
+
+
+@lru_cache(maxsize=1)
+def get_notification_manager() -> NotificationManager:
+ services = []
+
+ if TELEGRAM_API_TOKEN:
+ services.append(TelegramNotificationService())
+
+ if WEBHOOK_ADDRESS:
+ services.append(WebhookNotificationService())
+
+ if not services:
+ raise ValueError("No notification services provided.")
+
+ return NotificationManager(services)
diff --git a/app/notification/telegram.py b/app/notification/telegram.py
@@ -0,0 +1,68 @@
+import logging
+
+from aiogram import Bot
+from aiogram.client.default import DefaultBotProperties
+from aiogram.client.session.aiohttp import AiohttpSession
+from aiogram.enums import ParseMode
+from aiogram.exceptions import TelegramAPIError
+
+from app.config import TELEGRAM_PROXY_URL
+from app.config.env import (
+ TELEGRAM_API_TOKEN,
+ TELEGRAM_ADMIN_ID,
+ TELEGRAM_LOGGER_CHANNEL_ID,
+)
+from app.models.notification import Notification
+from app.notification.helper import create_text
+
+logger = logging.getLogger(__name__)
+
+
+class BotManager:
+ _instance = None
+
+ @classmethod
+ async def get_instance(cls):
+ if cls._instance is None and TELEGRAM_API_TOKEN:
+ if TELEGRAM_PROXY_URL:
+ session = AiohttpSession(proxy=TELEGRAM_PROXY_URL)
+ else:
+ session = None
+
+ cls._instance = Bot(
+ token=TELEGRAM_API_TOKEN,
+ default=DefaultBotProperties(parse_mode=ParseMode.HTML),
+ session=session,
+ )
+ try:
+ await cls._instance.get_me()
+ except:
+ logger.error("Telegram API token is not valid.")
+ return cls._instance
+
+
+async def send_message(
+ message: str,
+ parse_mode=ParseMode.HTML,
+):
+ if not (bot := await BotManager.get_instance()):
+ return
+
+ for recipient_id in (TELEGRAM_ADMIN_ID or []) + [
+ TELEGRAM_LOGGER_CHANNEL_ID
+ ]:
+ if not recipient_id:
+ continue
+ try:
+ await bot.send_message(
+ recipient_id,
+ message,
+ parse_mode=parse_mode,
+ )
+ except TelegramAPIError as e:
+ logger.error(e)
+
+
+async def send_notification(notif: Notification):
+ text = create_text(notif)
+ await send_message(text)
diff --git a/app/notification/webhook.py b/app/notification/webhook.py
@@ -0,0 +1,43 @@
+import logging
+import aiohttp
+from fastapi.encoders import jsonable_encoder
+
+from app.models.notification import Notification
+from app.config.env import WEBHOOK_ADDRESS, WEBHOOK_SECRET
+
+logger = logging.getLogger(__name__)
+
+headers = {"x-webhook-secret": WEBHOOK_SECRET} if WEBHOOK_SECRET else None
+
+
+async def send_notification(notif: Notification) -> bool:
+ """Send the notification to the webhook address provided by WEBHOOK_ADDRESS
+
+ Args:
+ notif (Notification): A single notification object to be sent.
+
+ Returns:
+ bool: returns True if an ok response received
+ """
+
+ result = await send_req(
+ w_address=WEBHOOK_ADDRESS, data=jsonable_encoder(notif)
+ )
+ return result
+
+
+async def send_req(w_address: str, data):
+ try:
+ logger.debug(f"Sending {len(data)} webhook updates to {w_address}")
+ async with aiohttp.ClientSession() as session:
+ async with session.post(
+ w_address, json=data, headers=headers
+ ) as r:
+ if r.status == 200:
+ return True
+ logger.error(
+ f"Failed to send data to {w_address}: {r.status} {await r.text()}"
+ )
+ except Exception as err:
+ logger.error(err)
+ return False
diff --git a/app/routes/user.py b/app/routes/user.py
@@ -21,6 +21,7 @@ from app.dependencies import (
EndDateDep,
ModifyUsersAccess,
)
+from app.models.notification import UserNotification
from app.models.service import ServiceResponse
from app.models.user import (
UserCreate,
@@ -28,7 +29,7 @@ from app.models.user import (
UserResponse,
UserUsageSeriesResponse,
)
-from app.utils import report
+from app.notification import notify
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/users", tags=["User"])
@@ -136,9 +137,13 @@ async def add_user(new_user: UserCreate, db: DBDep, admin: AdminDep):
user = UserResponse.model_validate(db_user)
marznode.operations.update_user(user=db_user)
+
asyncio.ensure_future(
- report.user_created(user=user, user_id=db_user.id, by=admin)
+ notify(
+ action=UserNotification.Action.user_created, user=user, by=admin
+ )
)
+
logger.info("New user `%s` added", db_user.username)
return user
@@ -189,10 +194,6 @@ async def delete_expired(
for db_user in expired_users:
crud.remove_user(db, db_user)
- asyncio.ensure_future(
- report.user_deleted(username=db_user.username, by=admin)
- )
-
logger.info("User `%s` removed", db_user.username)
return {}
@@ -247,18 +248,25 @@ async def modify_user(
db.commit()
asyncio.ensure_future(
- report.user_updated(
- user=UserResponse.model_validate(db_user), by=admin
+ notify(
+ action=UserNotification.Action.user_updated,
+ user=UserResponse.model_validate(db_user),
+ by=admin,
)
)
logger.info("User `%s` modified", db_user.username)
if active_before != active_after:
+ action = (
+ UserNotification.Action.user_activated
+ if active_after
+ else UserNotification.Action.user_deactivated
+ )
+
asyncio.ensure_future(
- report.status_change(
- username=db_user.username,
- activation=db_user.activated,
+ notify(
+ action=action,
user=UserResponse.model_validate(db_user),
by=admin,
)
@@ -286,12 +294,19 @@ async def remove_user(
"""
marznode.operations.update_user(db_user, remove=True)
+ deleted_username = db_user.username
crud.remove_user(db, db_user)
- db.flush()
+
+ db_user.username = deleted_username
+ user = UserResponse.model_validate(db_user)
+ db.expunge(db_user)
asyncio.ensure_future(
- report.user_deleted(username=db_user.username, by=admin)
+ notify(
+ action=UserNotification.Action.user_deleted, user=user, by=admin
+ )
)
+
logger.info("User %s deleted", db_user.username)
return {}
@@ -333,7 +348,14 @@ async def reset_user_data_usage(
db.commit()
user = UserResponse.model_validate(db_user)
- asyncio.ensure_future(report.user_data_usage_reset(user=user, by=admin))
+
+ asyncio.ensure_future(
+ notify(
+ action=UserNotification.Action.data_usage_reset,
+ user=user,
+ by=admin,
+ )
+ )
logger.info("User `%s`'s usage was reset", db_user.username)
@@ -363,6 +385,12 @@ async def enable_user(
user = UserResponse.model_validate(db_user)
+ asyncio.ensure_future(
+ notify(
+ action=UserNotification.Action.user_enabled, user=user, by=admin
+ )
+ )
+
logger.info("User `%s` has been enabled", db_user.username)
return user
@@ -388,6 +416,12 @@ async def disable_user(
user = UserResponse.model_validate(db_user)
+ asyncio.ensure_future(
+ notify(
+ action=UserNotification.Action.user_disabled, user=user, by=admin
+ )
+ )
+
logger.info("User `%s` has been disabled", db_user.username)
return user
@@ -408,9 +442,15 @@ async def revoke_user_subscription(
if db_user.is_active:
marznode.operations.update_user(db_user, remove=True)
marznode.operations.update_user(db_user)
+
user = UserResponse.model_validate(db_user)
+
asyncio.ensure_future(
- report.user_subscription_revoked(user=user, by=admin)
+ notify(
+ action=UserNotification.Action.subscription_revoked,
+ user=user,
+ by=admin,
+ )
)
logger.info("User %s subscription revoked", db_user.username)
diff --git a/app/tasks/__init__.py b/app/tasks/__init__.py
@@ -2,13 +2,12 @@ from .nodes import nodes_startup
from .record_usages import record_user_usages
from .reset_user_data_usage import reset_user_data_usage
from .review_users import review_users
-from .send_notifications import delete_expired_reminders, send_notifications
+from .expire_days_reached import expire_days_reached
__all__ = [
"nodes_startup",
"record_user_usages",
"reset_user_data_usage",
"review_users",
- "delete_expired_reminders",
- "send_notifications",
+ "expire_days_reached",
]
diff --git a/app/tasks/data_usage_percent_reached.py b/app/tasks/data_usage_percent_reached.py
@@ -0,0 +1,58 @@
+import asyncio
+
+from sqlalchemy import case
+from sqlalchemy.orm import Session
+
+from app.config import NOTIFY_REACHED_USAGE_PERCENT
+from app.db.models import User
+from app.models.notification import UserNotification
+from app.models.user import UserResponse
+from app.notification.notifiers import notify
+
+
+async def data_usage_percent_reached(db: Session, users_usage: list) -> None:
+ """
+ Monitors data usage of active users and sends a notification if usage exceeds NOTIFY_REACHED_USAGE_PERCENT.
+
+ This function performs the following steps:
+ - Receives a list of user data usage, where each user has an ID and the amount of data they have used from all nodes.
+ - If the new total data usage exceeds NOTIFY_REACHED_USAGE_PERCENT of the user's data limit, it sends a notification.
+
+ This function is called when user data usage updates occur.
+
+ """
+
+ users_usage_dict = {user["id"]: user["value"] for user in users_usage}
+
+ case_stmt = case(
+ *[
+ (User.id == user_id, User.used_traffic + increase_amount)
+ for user_id, increase_amount in users_usage_dict.items()
+ ],
+ else_=User.used_traffic,
+ )
+
+ exceeding_users = (
+ db.query(User)
+ .filter(
+ User.id.in_(users_usage_dict.keys()),
+ User.data_limit.isnot(None),
+ User.data_limit > 0,
+ (User.used_traffic / User.data_limit) * 100
+ < NOTIFY_REACHED_USAGE_PERCENT,
+ (case_stmt / User.data_limit) * 100 > NOTIFY_REACHED_USAGE_PERCENT,
+ )
+ .all()
+ )
+
+ for user in exceeding_users:
+ added_traffic = users_usage_dict[user.id]
+ user.used_traffic += added_traffic
+ asyncio.ensure_future(
+ notify(
+ action=UserNotification.Action.reached_usage_percent,
+ user=UserResponse.model_validate(user),
+ )
+ )
+
+ db.expunge_all()
diff --git a/app/tasks/expire_days_reached.py b/app/tasks/expire_days_reached.py
@@ -0,0 +1,43 @@
+import asyncio
+from datetime import datetime, timedelta
+
+from app.config import NOTIFY_DAYS_LEFT
+from app.db import GetDB
+from app.db.models import User
+from app.models.notification import UserNotification
+from app.models.user import UserResponse
+from app.notification.notifiers import notify
+
+
+async def expire_days_reached():
+ """
+ Identifies users whose expiration days have reached a predefined notification period and sends notifications.
+
+ This function checks each active user with an expiration date set and determines if they are within a
+ specific notification window (`NOTIFY_DAYS_LEFT` days before expiration). If the condition is met,
+ it sends a notification.
+
+ How it works:
+ - The current UTC time is obtained.
+ - For each user in the database who is activated and currently active, check if their expiration date is set.
+ - If the user's expiration date is within the `NOTIFY_DAYS_LEFT` threshold (plus a 30-second margin),
+ log the event and trigger a notification.
+ """
+
+ now = datetime.utcnow()
+ cleft = timedelta(days=NOTIFY_DAYS_LEFT) + now
+ cright = timedelta(days=NOTIFY_DAYS_LEFT, seconds=30) + now
+ with GetDB() as db:
+ users = db.query(User).filter(
+ User.is_active == True,
+ User.activated == True,
+ User.expire_date.isnot(None),
+ User.expire_date.between(cleft, cright),
+ )
+ for user in users:
+ asyncio.ensure_future(
+ notify(
+ action=UserNotification.Action.reached_days_left,
+ user=UserResponse.model_validate(user),
+ )
+ )
diff --git a/app/tasks/record_usages.py b/app/tasks/record_usages.py
@@ -8,6 +8,7 @@ from app import marznode
from app.db import GetDB
from app.db.models import NodeUsage, NodeUserUsage, User
from app.marznode import MarzNodeBase
+from app.tasks.data_usage_percent_reached import data_usage_percent_reached
def record_user_usage_logs(
@@ -76,7 +77,6 @@ def record_node_stats(node_id: int, usage: int):
)
with GetDB() as db:
-
# make node usage row if doesn't exist
select_stmt = select(NodeUsage.node_id).where(
and_(
@@ -156,6 +156,8 @@ async def record_user_usages():
# record users usage
with GetDB() as db:
+ await data_usage_percent_reached(db, users_usage)
+
stmt = update(User).values(
used_traffic=User.used_traffic + bindparam("value"),
lifetime_used_traffic=User.lifetime_used_traffic
diff --git a/app/tasks/review_users.py b/app/tasks/review_users.py
@@ -8,11 +8,12 @@ from app.db import (
GetDB,
get_users,
)
+from app.models.notification import UserNotification
from app.models.user import (
UserResponse,
UserExpireStrategy,
)
-from app.utils import report
+from app.notification import notify
if TYPE_CHECKING:
pass
@@ -31,11 +32,11 @@ async def review_users():
user.activated = False
db.commit()
db.refresh(user)
+
asyncio.ensure_future(
- report.status_change(
- user.username,
- user.status,
- UserResponse.model_validate(user),
+ notify(
+ action=UserNotification.Action.user_deactivated,
+ user=UserResponse.model_validate(user),
)
)
@@ -70,11 +71,4 @@ async def review_users():
user.expire_strategy = UserExpireStrategy.FIXED_DATE
db.commit()
db.refresh(user)
- asyncio.ensure_future(
- report.status_change(
- user.username,
- user.status,
- UserResponse.model_validate(user),
- )
- )
logger.info("on hold user `%s` has been activated", user.username)
diff --git a/app/tasks/send_notifications.py b/app/tasks/send_notifications.py
@@ -1,94 +0,0 @@
-import logging
-from datetime import datetime as dt
-from datetime import timedelta as td
-from typing import Any, Dict, List
-
-import aiohttp
-from fastapi.encoders import jsonable_encoder
-from requests import Session
-
-from app import config
-from app.db import GetDB
-from app.db.models import NotificationReminder
-from app.utils.notification import queue
-
-logger = logging.getLogger(__name__)
-
-session = Session()
-
-headers = (
- {"x-webhook-secret": config.WEBHOOK_SECRET}
- if config.WEBHOOK_SECRET
- else None
-)
-
-
-async def send(data: List[Dict[Any, Any]]) -> bool:
- """Send the notification to the webhook address provided by config.WEBHOOK_ADDRESS
-
- Args:
- data (List[Dict[Any, Any]]): list of json encoded notifications
-
- Returns:
- bool: returns True if an ok response received
- """
- try:
- logger.debug(
- f"Sending {len(data)} webhook updates to {config.WEBHOOK_ADDRESS}"
- )
-
- r = await aiohttp.request(
- "post", config.WEBHOOK_ADDRESS, json=data, headers=headers
- )
- if r.ok:
- return True
- logger.error(r)
- except Exception as err:
- logger.error(err)
- return False
-
-
-async def send_notifications():
- if not queue:
- return
-
- notifications_to_send = list()
- try:
- while notification := queue.popleft():
- if notification.tries > config.NUMBER_OF_RECURRENT_NOTIFICATIONS:
- continue
- if notification.send_at > dt.utcnow().timestamp():
- queue.append(
- notification
- ) # add it to the queue again for the next check
- continue
- notifications_to_send.append(notification)
- except IndexError: # if the queue is empty
- pass
-
- if not notifications_to_send:
- return
- if not await send(
- [jsonable_encoder(notif) for notif in notifications_to_send]
- ):
- for notification in notifications_to_send:
- if (
- notification.tries + 1
- ) > config.NUMBER_OF_RECURRENT_NOTIFICATIONS:
- continue
- notification.tries += 1
- notification.send_at = (
- ( # schedule notification for n seconds later
- dt.utcnow()
- + td(seconds=config.RECURRENT_NOTIFICATIONS_TIMEOUT)
- ).timestamp()
- )
- queue.append(notification)
-
-
-def delete_expired_reminders() -> None:
- with GetDB() as db:
- db.query(NotificationReminder).filter(
- NotificationReminder.expires_at < dt.utcnow()
- ).delete()
- db.commit()
diff --git a/app/telegram/__init__.py b/app/telegram/__init__.py
@@ -1,35 +0,0 @@
-from telebot import asyncio_helper
-from telebot.async_telebot import AsyncTeleBot
-
-from app.config.env import TELEGRAM_API_TOKEN, TELEGRAM_PROXY_URL
-from app.telegram.report import (
- report,
- report_new_user,
- report_user_modification,
- report_user_deletion,
- report_status_change,
- report_user_usage_reset,
- report_user_subscription_revoked,
-)
-
-bot = None
-if TELEGRAM_API_TOKEN:
- asyncio_helper.proxy = TELEGRAM_PROXY_URL
- bot = AsyncTeleBot(TELEGRAM_API_TOKEN)
-
-
-async def start_bot():
- if bot:
- await bot.polling()
-
-
-__all__ = [
- "bot",
- "report",
- "report_new_user",
- "report_user_modification",
- "report_user_deletion",
- "report_status_change",
- "report_user_usage_reset",
- "report_user_subscription_revoked",
-]
diff --git a/app/telegram/report.py b/app/telegram/report.py
@@ -1,150 +0,0 @@
-import logging
-from datetime import datetime
-
-from telebot.apihelper import ApiTelegramException
-from telebot.formatting import escape_html
-
-from app import telegram
-from app.config.env import TELEGRAM_ADMIN_ID, TELEGRAM_LOGGER_CHANNEL_ID
-from app.utils.system import readable_size
-
-logger = logging.getLogger(__name__)
-
-
-async def report(message: str, parse_mode="html", keyboard=None):
- if telegram.bot and (TELEGRAM_ADMIN_ID or TELEGRAM_LOGGER_CHANNEL_ID):
- try:
- if TELEGRAM_LOGGER_CHANNEL_ID:
- await telegram.bot.send_message(
- TELEGRAM_LOGGER_CHANNEL_ID, message, parse_mode=parse_mode
- )
- else:
- for admin in TELEGRAM_ADMIN_ID:
- await telegram.bot.send_message(
- admin,
- message,
- parse_mode=parse_mode,
- reply_markup=keyboard,
- )
- except ApiTelegramException as e:
- logger.error(e)
-
-
-async def report_new_user(
- user_id: int,
- username: str,
- by: str,
- expire_date: datetime,
- data_limit: int,
- services: list,
-):
- text = """\
-🆕 <b>#Created</b>
-➖➖➖➖➖➖➖➖➖
-<b>Username :</b> <code>{username}</code>
-<b>Traffic Limit :</b> <code>{data_limit}</code>
-<b>Expire Date :</b> <code>{expire_date}</code>
-<b>Services :</b> <code>{services}</code>
-➖➖➖➖➖➖➖➖➖
-<b>By :</b> <b>#{by}</b>""".format(
- by=escape_html(by),
- username=escape_html(username),
- data_limit=readable_size(data_limit) if data_limit else "Unlimited",
- expire_date=(
- expire_date.strftime("%H:%M:%S %Y-%m-%d")
- if expire_date
- else "Never"
- ),
- services="" if not services else ", ".join([s.name for s in services]),
- )
-
- return await report(text)
-
-
-async def report_user_modification(
- username: str,
- expire_date: datetime,
- data_limit: int,
- services: list,
- by: str,
-):
- text = """\
-✏️ <b>#Modified</b>
-➖➖➖➖➖➖➖➖➖
-<b>Username :</b> <code>{username}</code>
-<b>Traffic Limit :</b> <code>{data_limit}</code>
-<b>Expire Date :</b> <code>{expire_date}</code>
-<b>Services :</b> <code>{services}</code>
-➖➖➖➖➖➖➖➖➖
-<b>By :</b> <b>#{by}</b>\
- """.format(
- by=escape_html(by),
- username=escape_html(username),
- data_limit=readable_size(data_limit) if data_limit else "Unlimited",
- expire_date=(
- expire_date.strftime("%H:%M:%S %Y-%m-%d")
- if expire_date
- else "Never"
- ),
- services=", ".join([s.name for s in services]),
- )
-
- return await report(text)
-
-
-async def report_user_deletion(username: str, by: str):
- text = """\
-🗑 <b>#Deleted</b>
-➖➖➖➖➖➖➖➖➖
-<b>Username</b> : <code>{username}</code>
-➖➖➖➖➖➖➖➖➖
-<b>By</b> : <b>#{by}</b>\
- """.format(
- by=escape_html(by), username=escape_html(username)
- )
- return await report(text)
-
-
-async def report_status_change(username: str, status: str):
- _status = {
- "active": "✅ <b>#Activated</b>",
- "on_hold": "🟡 <b>#On_Hold</b>",
- "limited": "🪫 <b>#Limited</b>",
- "expired": "🕔 <b>#Expired</b>",
- }
- text = """\
-{status}
-➖➖➖➖➖➖➖➖➖
-<b>Username</b> : <code>{username}</code>\
- """.format(
- username=escape_html(username), status=_status[status]
- )
- return await report(text)
-
-
-async def report_user_usage_reset(username: str, by: str):
- text = """
-🔁 <b>#Reset</b>
-➖➖➖➖➖➖➖➖➖
-<b>Username</b> : <code>{username}</code>
-➖➖➖➖➖➖➖➖➖
-<b>By</b> : <b>#{by}</b>\
- """.format(
- by=escape_html(by), username=escape_html(username)
- )
-
- return await report(text)
-
-
-async def report_user_subscription_revoked(username: str, by: str):
- text = """
-🔁 <b>#Revoked</b>
-➖➖➖➖➖➖➖➖➖
-<b>Username</b> : <code>{username}</code>
-➖➖➖➖➖➖➖➖➖
-<b>By</b> : <b>#{by}</b>\
- """.format(
- by=escape_html(by), username=escape_html(username)
- )
-
- return await report(text)
diff --git a/app/utils/helpers.py b/app/utils/helpers.py
@@ -1,9 +0,0 @@
-from datetime import datetime
-
-
-def calculate_usage_percent(used_traffic: int, data_limit: int) -> float:
- return (used_traffic * 100) / data_limit
-
-
-def calculate_expiration_days(expire: datetime) -> int:
- return (expire - datetime.utcnow()).days
diff --git a/app/utils/notification.py b/app/utils/notification.py
@@ -1,104 +0,0 @@
-from collections import deque
-from datetime import datetime as dt
-from enum import Enum
-
-from pydantic import BaseModel
-
-from app.config.env import WEBHOOK_ADDRESS
-from app.models.admin import Admin
-from app.models.user import UserResponse
-
-queue = deque()
-
-
-class Notification(BaseModel):
- class Type(str, Enum):
- user_created = "user_created"
- user_updated = "user_updated"
- user_deleted = "user_deleted"
- user_limited = "user_limited"
- user_expired = "user_expired"
- user_enabled = "user_enabled"
- user_disabled = "user_disabled"
- data_usage_reset = "data_usage_reset"
- subscription_revoked = "subscription_revoked"
-
- reached_usage_percent = "reached_usage_percent"
- reached_days_left = "reached_days_left"
-
- enqueued_at: float = dt.utcnow().timestamp()
- send_at: float = dt.utcnow().timestamp()
- tries: int = 0
-
-
-class UserNotification(Notification):
- username: str
-
-
-class ReachedUsagePercent(UserNotification):
- action: Notification.Type = Notification.Type.reached_usage_percent
- user: UserResponse
- used_percent: float
-
-
-class ReachedDaysLeft(UserNotification):
- action: Notification.Type = Notification.Type.reached_days_left
- user: UserResponse
- days_left: int
-
-
-class UserCreated(UserNotification):
- action: Notification.Type = Notification.Type.user_created
- by: Admin
- user: UserResponse
-
-
-class UserUpdated(UserNotification):
- action: Notification.Type = Notification.Type.user_updated
- by: Admin
- user: UserResponse
-
-
-class UserDeleted(UserNotification):
- action: Notification.Type = Notification.Type.user_deleted
- by: Admin
-
-
-class UserLimited(UserNotification):
- action: Notification.Type = Notification.Type.user_limited
- user: UserResponse
-
-
-class UserExpired(UserNotification):
- action: Notification.Type = Notification.Type.user_expired
- user: UserResponse
-
-
-class UserEnabled(UserNotification):
- action: Notification.Type = Notification.Type.user_enabled
- by: Admin = None
- user: UserResponse
-
-
-class UserDisabled(UserNotification):
- action: Notification.Type = Notification.Type.user_disabled
- by: Admin
- user: UserResponse
- reason: str = None
-
-
-class UserDataUsageReset(UserNotification):
- action: Notification.Type = Notification.Type.data_usage_reset
- by: Admin
- user: UserResponse
-
-
-class UserSubscriptionRevoked(UserNotification):
- action: Notification.Type = Notification.Type.subscription_revoked
- by: Admin
- user: UserResponse
-
-
-async def notify(message: Notification) -> None:
- if WEBHOOK_ADDRESS:
- queue.append(message)
diff --git a/app/utils/report.py b/app/utils/report.py
@@ -1,173 +0,0 @@
-from datetime import datetime as dt
-from typing import Optional
-
-from app import telegram
-from app.db import Session, create_notification_reminder
-from app.models.admin import Admin
-from app.models.user import ReminderType, UserResponse
-from app.utils.notification import (
- Notification,
- ReachedDaysLeft,
- ReachedUsagePercent,
- UserCreated,
- UserDataUsageReset,
- UserDeleted,
- UserEnabled,
- UserLimited,
- UserSubscriptionRevoked,
- UserUpdated,
- notify,
-)
-
-
-async def status_change(
- username: str,
- activation: bool,
- user: UserResponse,
- by: Optional[Admin] = None,
-) -> None:
- try:
- await telegram.report_status_change(username, status)
- except Exception:
- pass
- if activation is False:
- await notify(
- UserLimited(
- username=username,
- action=Notification.Type.user_disabled,
- user=user,
- )
- )
- elif activation is True:
- await notify(
- UserEnabled(
- username=username,
- action=Notification.Type.user_enabled,
- user=user,
- by=by,
- )
- )
-
-
-async def user_created(user: UserResponse, user_id: int, by: Admin) -> None:
- await telegram.report_new_user(
- user_id=user_id,
- username=user.username,
- by=by.username,
- expire_date=user.expire_date,
- data_limit=user.data_limit,
- # proxies=[]
- services=user.services,
- )
-
- await notify(
- UserCreated(
- username=user.username,
- action=Notification.Type.user_created,
- by=by,
- user=user,
- )
- )
-
-
-async def user_updated(user: UserResponse, by: Admin) -> None:
- try:
- await telegram.report_user_modification(
- username=user.username,
- expire_date=user.expire_date,
- data_limit=user.data_limit,
- services=user.services,
- by=by.username,
- )
- except Exception:
- pass
- await notify(
- UserUpdated(
- username=user.username,
- action=Notification.Type.user_updated,
- by=by,
- user=user,
- )
- )
-
-
-async def user_deleted(username: str, by: Admin) -> None:
- try:
- await telegram.report_user_deletion(username=username, by=by.username)
- except Exception:
- pass
- await notify(
- UserDeleted(
- username=username, action=Notification.Type.user_deleted, by=by
- )
- )
-
-
-async def user_data_usage_reset(user: UserResponse, by: Admin) -> None:
- try:
- await telegram.report_user_usage_reset(
- username=user.username,
- by=by.username,
- )
- except Exception:
- pass
- await notify(
- UserDataUsageReset(
- username=user.username,
- action=Notification.Type.data_usage_reset,
- by=by,
- user=user,
- )
- )
-
-
-async def user_subscription_revoked(user: UserResponse, by: Admin) -> None:
- try:
- await telegram.report_user_subscription_revoked(
- username=user.username,
- by=by.username,
- )
- except Exception:
- pass
- await notify(
- UserSubscriptionRevoked(
- username=user.username,
- action=Notification.Type.subscription_revoked,
- by=by,
- user=user,
- )
- )
-
-
-async def data_usage_percent_reached(
- db: Session,
- percent: float,
- user: UserResponse,
- user_id: int,
- expire: Optional[int] = None,
-) -> None:
- await notify(
- ReachedUsagePercent(
- username=user.username, user=user, used_percent=percent
- )
- )
- await create_notification_reminder(
- db,
- ReminderType.data_usage,
- expires_at=dt.utcfromtimestamp(expire) if expire else None,
- user_id=user_id,
- )
-
-
-async def expire_days_reached(
- db: Session, days: int, user: UserResponse, user_id: int, expire: int
-) -> None:
- await notify(
- ReachedDaysLeft(username=user.username, user=user, days_left=days)
- )
- await create_notification_reminder(
- db,
- ReminderType.expiration_date,
- expires_at=dt.utcfromtimestamp(expire),
- user_id=user_id,
- )
diff --git a/app/utils/share.py b/app/utils/share.py
@@ -232,7 +232,6 @@ def generate_user_configs(
user_id: int,
format_variables: dict,
) -> Union[List, str]:
-
salt = secrets.token_hex(8)
configs = []
diff --git a/app/webhooks.py b/app/webhooks.py
@@ -0,0 +1,78 @@
+from fastapi import APIRouter
+from pydantic import BaseModel
+
+from app.models.notification import (
+ UserCreated,
+ UserDeleted,
+ UserActivated,
+ UserDeactivated,
+ UserEnabled,
+ UserDisabled,
+ UserDataUsageReset,
+ UserSubscriptionRevoked,
+ ReachedUsagePercent,
+ UserUpdated,
+)
+
+webhooks_router = APIRouter()
+
+
+class Subscription(BaseModel):
+ username: str
+ monthly_fee: float
+ start_date: int
+
+
+@webhooks_router.post("user-created")
+def user_created(body: UserCreated):
+ pass
+
+
+@webhooks_router.post("user-updated")
+def user_updated(body: UserUpdated):
+ pass
+
+
+@webhooks_router.post("user-activated")
+def user_activated(body: UserActivated):
+ pass
+
+
+@webhooks_router.post("user-deactivated")
+def user_deactivated(body: UserDeactivated):
+ pass
+
+
+@webhooks_router.post("user-deleted")
+def user_deleted(body: UserDeleted):
+ pass
+
+
+@webhooks_router.post("user-enabled")
+def user_enabled(body: UserEnabled):
+ pass
+
+
+@webhooks_router.post("user-disabled")
+def user_disabled(body: UserDisabled):
+ pass
+
+
+@webhooks_router.post("user-data-usage-reset")
+def user_data_usage_reset(body: UserDataUsageReset):
+ pass
+
+
+@webhooks_router.post("user-subscription-revoked")
+def user_subscription_revoked(body: UserSubscriptionRevoked):
+ pass
+
+
+@webhooks_router.post("user-reached-usage")
+def user_reached_usage_percent(body: ReachedUsagePercent):
+ pass
+
+
+@webhooks_router.post("user-reached-days")
+def user_reached_days(ReachedDaysLeft):
+ pass
diff --git a/requirements.txt b/requirements.txt
@@ -11,7 +11,6 @@ protobuf==4.25.2
pydantic==2.7.1
PyMySQL==1.1.1
pyOpenSSL==24.2.1
-pyTelegramBotAPI==4.15.2
python-decouple==3.8
python-dotenv==1.0.0
requests==2.32.2
@@ -31,3 +30,4 @@ uvicorn==0.30.4
rich==13.7.1
python-multipart==0.0.18
httptools==0.6.4
+aiogram==3.12.0