diff --git a/app/db/__init__.py b/app/db/__init__.py
index 620f4eb..13ae45d 100644
--- a/app/db/__init__.py
+++ b/app/db/__init__.py
@@ -2,7 +2,6 @@ from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from .base import Base, SessionLocal, engine # noqa
-
from .crud import (
create_admin,
create_notification_reminder, # noqa
@@ -27,9 +26,7 @@ from .crud import (
update_user,
update_user_status,
update_user_sub,
- start_user_expire,
)
-
from .models import JWT, System, User # noqa
@@ -56,7 +53,6 @@ __all__ = [
"remove_user",
"update_user",
"update_user_status",
- "start_user_expire",
"update_user_sub",
"revoke_user_sub",
"set_owner",
diff --git a/app/db/crud.py b/app/db/crud.py
index c55200a..d584f23 100644
--- a/app/db/crud.py
+++ b/app/db/crud.py
@@ -7,7 +7,6 @@ from typing import List, Optional, Tuple, Union
from sqlalchemy import and_, delete, update, select
from sqlalchemy.orm import Session
-from app.config.env import NOTIFY_DAYS_LEFT, NOTIFY_REACHED_USAGE_PERCENT
from app.db.models import (
JWT,
TLS,
@@ -34,14 +33,11 @@ from app.models.service import Service as ServiceModify, ServiceCreate
from app.models.user import (
ReminderType,
UserCreate,
- UserDataLimitResetStrategy,
+ UserDataUsageResetStrategy,
UserModify,
UserStatus,
UserUsageResponse,
-)
-from app.utils.helpers import (
- calculate_expiration_days,
- calculate_usage_percent,
+ UserExpireStrategy,
)
@@ -110,19 +106,16 @@ def assure_node_inbounds(db: Session, inbounds: List[Inbound], node_id: int):
def get_node_users(
db: Session,
- node_id: Optional[int],
- statuses: Optional[List[UserStatus]] = None,
+ node_id: int,
):
query = (
db.query(User.id, User.username, User.key, Inbound)
.distinct()
- .where(User.enabled == True)
.join(Inbound.services)
.join(Service.users)
.filter(Inbound.node_id == node_id)
+ .filter(User.activated == True)
)
- if isinstance(statuses, list):
- query = query.filter(User.status.in_(statuses))
return query.all()
@@ -210,12 +203,12 @@ UsersSortingOptions = Enum(
"username": User.username.asc(),
"used_traffic": User.used_traffic.asc(),
"data_limit": User.data_limit.asc(),
- "expire": User.expire.asc(),
+ "expire": User.expire_date.asc(),
"created_at": User.created_at.asc(),
"-username": User.username.desc(),
"-used_traffic": User.used_traffic.desc(),
"-data_limit": User.data_limit.desc(),
- "-expire": User.expire.desc(),
+ "-expire": User.expire_date.desc(),
"-created_at": User.created_at.desc(),
},
)
@@ -226,11 +219,16 @@ def get_users(
offset: Optional[int] = None,
limit: Optional[int] = None,
usernames: Optional[List[str]] = None,
- status: Optional[Union[UserStatus, list]] = None,
sort: Optional[List[UsersSortingOptions]] = None,
admin: Optional[Admin] = None,
- reset_strategy: Optional[Union[UserDataLimitResetStrategy, list]] = None,
- return_with_count: bool = False,
+ reset_strategy: Optional[Union[UserDataUsageResetStrategy, list]] = None,
+ expire_strategy: (
+ UserExpireStrategy | list[UserExpireStrategy] | None
+ ) = None,
+ is_active: bool | None = None,
+ activated: bool | None = None,
+ expired: bool | None = None,
+ data_limit_reached: bool | None = None,
) -> Union[List[User], Tuple[List[User], int]]:
query = db.query(User)
@@ -240,12 +238,6 @@ def get_users(
else:
query = query.filter(User.username.in_(usernames))
- if status:
- if isinstance(status, list):
- query = query.filter(User.status.in_(status))
- else:
- query = query.filter(User.status == status)
-
if reset_strategy:
if isinstance(reset_strategy, list):
query = query.filter(
@@ -256,24 +248,36 @@ def get_users(
User.data_limit_reset_strategy == reset_strategy
)
+ if expire_strategy:
+ if isinstance(expire_strategy, list):
+ query = query.filter(User.expire_strategy.in_(expire_strategy))
+ else:
+ query = query.filter(User.expire_strategy == expire_strategy)
+
+ if isinstance(is_active, bool):
+ query = query.filter(User.is_active == is_active)
+
+ if isinstance(activated, bool):
+ query = query.filter(User.activated == activated)
+
+ if isinstance(expired, bool):
+ query = query.filter(User.expired == expired)
+
+ if isinstance(data_limit_reached, bool):
+ query = query.filter(User.data_limit_reached == data_limit_reached)
+
if admin:
query = query.filter(User.admin == admin)
- # count it before applying limit and offset
- if return_with_count:
- count = query.count()
-
if sort:
query = query.order_by(*(opt.value for opt in sort))
if offset:
query = query.offset(offset)
+
if limit:
query = query.limit(limit)
- if return_with_count:
- return query.all(), count
-
return query.all()
@@ -307,41 +311,54 @@ def get_user_usages(
def get_users_count(
db: Session,
- status: UserStatus | None = None,
admin: Admin | None = None,
enabled: bool | None = None,
online: bool | None = None,
+ expire_strategy: UserExpireStrategy | None = None,
+ is_active: bool | None = None,
+ expired: bool | None = None,
+ data_limit_reached: bool | None = None,
):
query = db.query(User.id)
if admin:
query = query.filter(User.admin_id == admin.id)
- if status:
- query = query.filter(User.status == status)
+ if is_active:
+ query = query.filter(User.is_active == is_active)
+ if expired:
+ query = query.filter(User.expired == expired)
+ if data_limit_reached:
+ query = query.filter(User.data_limit_reached == data_limit_reached)
if enabled:
query = query.filter(User.enabled == enabled)
- if online:
+ if online is True:
query = query.filter(
User.online_at > (datetime.utcnow() - timedelta(seconds=30))
)
+ elif online is False:
+ query = query.filter(
+ User.online_at < (datetime.utcnow() - timedelta(seconds=30))
+ )
+ if expire_strategy:
+ query = query.filter(User.expire_strategy == expire_strategy)
+
return query.count()
def create_user(db: Session, user: UserCreate, admin: Admin = None):
dbuser = User(
username=user.username,
- # proxies=proxies,
key=user.key,
+ expire_strategy=user.expire_strategy,
+ expire_date=user.expire_date,
+ usage_duration=user.usage_duration,
+ activation_deadline=user.activation_deadline,
services=db.query(Service)
.filter(Service.id.in_(user.service_ids))
.all(), # user.services,
- status=user.status,
data_limit=(user.data_limit or None),
- expire=(user.expire or None),
admin=admin,
data_limit_reset_strategy=user.data_limit_reset_strategy,
note=user.note,
- on_hold_expire_duration=(user.on_hold_expire_duration or None),
- on_hold_timeout=(user.on_hold_timeout or None),
)
db.add(dbuser)
db.commit()
@@ -356,64 +373,26 @@ def remove_user(db: Session, dbuser: User):
def update_user(db: Session, dbuser: User, modify: UserModify):
- if modify.status is not None:
- dbuser.status = modify.status
-
if modify.data_limit is not None:
- # in case there is modification to a user's data limit
- dbuser.data_limit = (
- modify.data_limit or None
- ) # set it to the new limit
- if dbuser.status is not UserStatus.expired:
- # in case the user isn't already disabled/expired
- if (
- not dbuser.data_limit
- or dbuser.used_traffic < dbuser.data_limit
- ):
- # in case the user is unlimited or hasn't reached the limit already
- if dbuser.status != UserStatus.on_hold:
- dbuser.status = UserStatus.active
-
- if not dbuser.data_limit or (
- calculate_usage_percent(
- dbuser.used_traffic, dbuser.data_limit
- )
- < NOTIFY_REACHED_USAGE_PERCENT
- ):
- delete_notification_reminder_by_type(
- db, dbuser.id, ReminderType.data_usage
- )
- else:
- # the user has reached the new data limit
- dbuser.status = UserStatus.limited
-
- if modify.expire is not None:
- dbuser.expire = modify.expire or None
- if dbuser.status in (UserStatus.active, UserStatus.expired):
- if not dbuser.expire or dbuser.expire > datetime.utcnow():
- dbuser.status = UserStatus.active
- if not dbuser.expire or (
- calculate_expiration_days(dbuser.expire) > NOTIFY_DAYS_LEFT
- ):
- delete_notification_reminder_by_type(
- db, dbuser.id, ReminderType.expiration_date
- )
- else:
- dbuser.status = UserStatus.expired
+ dbuser.data_limit = modify.data_limit or None
+
+ if modify.expire_strategy is not None:
+ dbuser.expire_strategy = modify.expire_strategy or None
+
+ if modify.expire_date is not None:
+ dbuser.expire_date = modify.expire_date or None
if modify.note is not None:
dbuser.note = modify.note or None
if modify.data_limit_reset_strategy is not None:
- dbuser.data_limit_reset_strategy = (
- modify.data_limit_reset_strategy.value
- )
+ dbuser.data_limit_reset_strategy = modify.data_limit_reset_strategy
- if modify.on_hold_timeout is not None:
- dbuser.on_hold_timeout = modify.on_hold_timeout
+ if modify.activation_deadline is not None:
+ dbuser.activation_deadline = modify.activation_deadline
- if modify.on_hold_expire_duration is not None:
- dbuser.on_hold_expire_duration = modify.on_hold_expire_duration
+ if modify.usage_duration is not None:
+ dbuser.usage_duration = modify.usage_duration
if modify.service_ids is not None:
dbuser.services = (
@@ -432,8 +411,6 @@ def reset_user_data_usage(db: Session, dbuser: User):
dbuser.used_traffic = 0
- if dbuser.status == UserStatus.limited:
- dbuser.status = UserStatus.active.value
db.add(dbuser)
db.commit()
@@ -463,16 +440,8 @@ def reset_all_users_data_usage(db: Session, admin: Optional[Admin] = None):
if admin:
query = query.filter(User.admin == admin)
- for dbuser in query.all():
- dbuser.used_traffic = 0
- if dbuser.status not in [
- UserStatus.on_hold,
- UserStatus.expired,
- ]:
- dbuser.status = UserStatus.active
- dbuser.usage_logs.clear()
- dbuser.node_usages.clear()
- db.add(dbuser)
+ for db_user in query.all():
+ db_user.used_traffic = 0
db.commit()
@@ -491,16 +460,6 @@ def set_owner(db: Session, dbuser: User, admin: Admin):
return dbuser
-def start_user_expire(db: Session, dbuser: User):
- expire = datetime.utcnow() + timedelta(
- seconds=dbuser.on_hold_expire_duration
- )
- dbuser.expire = expire
- db.commit()
- db.refresh(dbuser)
- return dbuser
-
-
def get_system_usage(db: Session):
return db.query(System).first()
diff --git a/app/db/migrations/versions/a83e4dd22672_refactor_user_expire.py b/app/db/migrations/versions/a83e4dd22672_refactor_user_expire.py
new file mode 100644
index 0000000..0afe928
--- /dev/null
+++ b/app/db/migrations/versions/a83e4dd22672_refactor_user_expire.py
@@ -0,0 +1,141 @@
+"""refactor user expire
+
+Revision ID: a83e4dd22672
+Revises: 9001356f0063
+Create Date: 2024-06-27 01:04:34.974520
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import table, column
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "a83e4dd22672"
+down_revision = "9001356f0063"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ bind = op.get_bind()
+ dialect = bind.dialect.name
+
+ if dialect == "postgresql":
+ op.execute(
+ """
+ DO $$
+ BEGIN
+ IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'userexpirestrategy') THEN
+ CREATE TYPE userexpirestrategy AS ENUM ('NEVER', 'FIXED_DATE', 'START_ON_FIRST_USE');
+ END IF;
+ END $$;
+ """
+ )
+ op.add_column(
+ "users",
+ sa.Column(
+ "activated",
+ sa.Boolean(),
+ nullable=False,
+ server_default=sa.sql.true(),
+ ),
+ )
+ op.add_column(
+ "users",
+ sa.Column(
+ "expire_strategy",
+ sa.Enum(
+ "NEVER",
+ "FIXED_DATE",
+ "START_ON_FIRST_USE",
+ name="userexpirestrategy",
+ ),
+ server_default="NEVER",
+ nullable=False,
+ ),
+ )
+
+ if dialect == "sqlite":
+ for old, new, type_ in [
+ ("on_hold_timeout", "activation_deadline", sa.DateTime),
+ ("expire", "expire_date", sa.DateTime),
+ ("on_hold_expire_duration", "usage_duration", sa.Integer),
+ ]:
+ op.add_column("users", sa.Column(new, type_))
+
+ # Step 2: Copy data from the old column to the new column
+ op.execute(f"UPDATE users SET {new} = {old}")
+
+ with op.batch_alter_table("users") as batch_op:
+ batch_op.drop_column(old)
+ else:
+ op.alter_column(
+ "users", "on_hold_timeout", new_column_name="activation_deadline"
+ )
+ op.alter_column("users", "expire", new_column_name="expire_date")
+ op.alter_column(
+ "users",
+ "on_hold_expire_duration",
+ new_column_name="usage_duration",
+ )
+ # ### end Alembic commands ###
+ users_table = table(
+ "users",
+ column("id", sa.Integer),
+ column("status", sa.String),
+ column("expire_date", sa.DateTime),
+ column("expire_strategy", sa.Enum(name="userexpirestrategy")),
+ )
+
+ connection = op.get_bind()
+ results = connection.execute(
+ sa.select(
+ users_table.c.id, users_table.c.status, users_table.c.expire_date
+ )
+ )
+
+ for user in results:
+ status = user[1]
+ expire_date = user[2]
+
+ if status == "on_hold":
+ expire_strategy = "START_ON_FIRST_USE"
+ elif expire_date:
+ expire_strategy = "FIXED_DATE"
+ else:
+ expire_strategy = "NEVER"
+ connection.execute(
+ users_table.update()
+ .where(users_table.c.id == user[0])
+ .values(expire_strategy=expire_strategy)
+ )
+
+ op.drop_column("users", "status")
+
+
+def downgrade() -> None:
+ # ### commands auto generated by Alembic - please adjust! ###
+ op.add_column(
+ "users",
+ sa.Column(
+ "status",
+ postgresql.ENUM(
+ "active", "limited", "expired", "on_hold", name="userstatus"
+ ),
+ autoincrement=False,
+ nullable=False,
+ server_default="active",
+ ),
+ )
+ op.alter_column(
+ "users", "activation_deadline", new_column_name="on_hold_timeout"
+ )
+ op.alter_column(
+ "users", "usage_duration", new_column_name="on_hold_expire_duration"
+ )
+ op.alter_column("users", "expire_date", new_column_name="expire")
+ op.drop_column("users", "expire_strategy")
+ op.drop_column("users", "activated")
+ # ### end Alembic commands ###
diff --git a/app/db/models.py b/app/db/models.py
index 291c08d..4a92a9a 100644
--- a/app/db/models.py
+++ b/app/db/models.py
@@ -15,7 +15,10 @@ from sqlalchemy import (
Table,
UniqueConstraint,
JSON,
+ and_,
+ func,
)
+from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship
from sqlalchemy.sql.expression import text
@@ -29,8 +32,9 @@ from app.models.proxy import (
)
from app.models.user import (
ReminderType,
- UserDataLimitResetStrategy,
+ UserDataUsageResetStrategy,
UserStatus,
+ UserExpireStrategy,
)
@@ -87,6 +91,7 @@ class User(Base):
id = Column(Integer, primary_key=True)
username = Column(String(32), unique=True, index=True)
key = Column(String(64), unique=True)
+ activated = Column(Boolean, nullable=False, default=True)
enabled = Column(
Boolean,
nullable=False,
@@ -106,10 +111,6 @@ class User(Base):
viewonly=True,
distinct_target_key=True,
)
- # proxies = relationship("Proxy", back_populates="user", cascade="all, delete-orphan")
- status = Column(
- Enum(UserStatus), nullable=False, default=UserStatus.active
- )
used_traffic = Column(BigInteger, default=0)
lifetime_used_traffic = Column(
BigInteger, default=0, server_default="0", nullable=False
@@ -127,13 +128,20 @@ class User(Base):
)
data_limit = Column(BigInteger)
data_limit_reset_strategy = Column(
- Enum(UserDataLimitResetStrategy),
+ Enum(UserDataUsageResetStrategy),
nullable=False,
- default=UserDataLimitResetStrategy.no_reset,
+ default=UserDataUsageResetStrategy.no_reset,
)
ip_limit = Column(Integer, nullable=False, default=-1)
settings = Column(String(1024))
- expire = Column(DateTime)
+ expire_strategy = Column(
+ Enum(UserExpireStrategy),
+ nullable=False,
+ default=UserExpireStrategy.NEVER,
+ )
+ expire_date = Column(DateTime)
+ usage_duration = Column(BigInteger)
+ activation_deadline = Column(DateTime)
admin_id = Column(Integer, ForeignKey("admins.id"))
admin = relationship("Admin", back_populates="users")
sub_updated_at = Column(DateTime)
@@ -141,14 +149,50 @@ class User(Base):
created_at = Column(DateTime, default=datetime.utcnow)
note = Column(String(500))
online_at = Column(DateTime)
- on_hold_expire_duration = Column(BigInteger)
- on_hold_timeout = Column(DateTime)
edit_at = Column(DateTime)
@property
def service_ids(self):
return [service.id for service in self.services]
+ @hybrid_property
+ def expired(self):
+ if self.expire_strategy == "fixed_date":
+ return self.expire_date < datetime.utcnow()
+ return False
+
+ @expired.expression
+ def expired(cls):
+ return and_(
+ cls.expire_strategy == "fixed_date", cls.expire_date < func.now()
+ )
+
+ @hybrid_property
+ def data_limit_reached(self):
+ if self.data_limit is not None:
+ return self.used_traffic >= self.data_limit
+ return False
+
+ @data_limit_reached.expression
+ def data_limit_reached(cls):
+ return and_(
+ cls.data_limit.isnot(None), cls.used_traffic >= cls.data_limit
+ )
+
+ @hybrid_property
+ def is_active(self):
+ return (
+ self.enabled and not self.expired and not self.data_limit_reached
+ )
+
+ @is_active.expression
+ def is_active(cls):
+ return and_(cls.enabled == True, ~cls.expired, ~cls.data_limit_reached)
+
+ @property
+ def status(self):
+ return UserStatus.ACTIVE if self.is_active else UserStatus.INACTIVE
+
class Inbound(Base):
__tablename__ = "inbounds"
diff --git a/app/marzneshin.py b/app/marzneshin.py
index db50858..5552947 100644
--- a/app/marzneshin.py
+++ b/app/marzneshin.py
@@ -1,10 +1,9 @@
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 contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI, Request, status
@@ -82,7 +81,7 @@ app.add_middleware(
scheduler = AsyncIOScheduler(timezone="UTC")
scheduler.add_job(record_user_usages, "interval", coalesce=True, seconds=30)
scheduler.add_job(
- review_users, "interval", seconds=600, coalesce=True, max_instances=1
+ review_users, "interval", seconds=30, coalesce=True, max_instances=1
)
scheduler.add_job(reset_user_data_usage, "interval", coalesce=True, hours=1)
scheduler.add_job(
diff --git a/app/marznode/database.py b/app/marznode/database.py
index ac00079..66fff55 100644
--- a/app/marznode/database.py
+++ b/app/marznode/database.py
@@ -1,14 +1,11 @@
from app.db import crud, GetDB
from app.models.node import NodeStatus
-from app.models.user import UserStatus
class MarzNodeDB:
def list_users(self):
with GetDB() as db:
- relations = crud.get_node_users(
- db, self.id, [UserStatus.active, UserStatus.on_hold]
- )
+ relations = crud.get_node_users(db, self.id)
users = dict()
for rel in relations:
if not users.get(rel[0]):
diff --git a/app/marznode/operations.py b/app/marznode/operations.py
index 1443a64..db6f987 100644
--- a/app/marznode/operations.py
+++ b/app/marznode/operations.py
@@ -6,32 +6,32 @@ from app import marznode
from .grpcio import MarzNodeGRPCIO
from .grpclib import MarzNodeGRPCLIB
from ..models.node import NodeConnectionBackend
-from ..models.user import UserStatus
if TYPE_CHECKING:
from app.db import User as DBUser
-async def update_user(user: "DBUser", old_inbounds: set | None = None):
- """updates a user on all nodes
- even though this isn't efficient it is extremely precise"""
+def update_user(
+ user: "DBUser", old_inbounds: set | None = None, remove: bool = False
+):
+ """updates a user on all related nodes"""
if old_inbounds is None:
old_inbounds = set()
node_inbounds = defaultdict(list)
- if user.status in (UserStatus.on_hold, UserStatus.active):
+ if remove:
for inb in user.inbounds:
- node_inbounds[inb.node_id].append(inb.tag)
+ node_inbounds[inb.node_id]
else:
for inb in user.inbounds:
- node_inbounds[inb.node_id]
+ node_inbounds[inb.node_id].append(inb.tag)
for inb in old_inbounds:
node_inbounds[inb[0]]
for node_id, tags in node_inbounds.items():
if marznode.nodes.get(node_id):
- asyncio.create_task(
+ asyncio.ensure_future(
marznode.nodes[node_id].update_user(user=user, inbounds=tags)
)
@@ -41,7 +41,9 @@ async def remove_user(user: "DBUser"):
for node_id in node_ids:
if marznode.nodes.get(node_id):
- await marznode.nodes[node_id].update_user(user=user, inbounds=[])
+ asyncio.ensure_future(
+ marznode.nodes[node_id].update_user(user=user, inbounds=[])
+ )
async def remove_node(node_id: int):
diff --git a/app/models/user.py b/app/models/user.py
index 6655bd2..2c84436 100644
--- a/app/models/user.py
+++ b/app/models/user.py
@@ -1,10 +1,9 @@
import secrets
from datetime import datetime
from enum import Enum
-from typing import List, Union, Annotated, Literal
+from typing import Annotated
from pydantic import (
- field_validator,
ConfigDict,
BaseModel,
Field,
@@ -24,23 +23,11 @@ class ReminderType(str, Enum):
class UserStatus(str, Enum):
- active = "active"
- on_hold = "on_hold"
- limited = "limited"
- expired = "expired"
+ ACTIVE = "active"
+ INACTIVE = "inactive"
-class UserStatusModify(str, Enum):
- active = "active"
- on_hold = "on_hold"
-
-
-class UserStatusCreate(str, Enum):
- active = "active"
- on_hold = "on_hold"
-
-
-class UserDataLimitResetStrategy(str, Enum):
+class UserDataUsageResetStrategy(str, Enum):
no_reset = "no_reset"
day = "day"
week = "week"
@@ -48,129 +35,97 @@ class UserDataLimitResetStrategy(str, Enum):
year = "year"
+class UserExpireStrategy(str, Enum):
+ NEVER = "never"
+ FIXED_DATE = "fixed_date"
+ START_ON_FIRST_USE = "start_on_first_use"
+
+
class User(BaseModel):
id: int | None = None
username: Annotated[
str, StringConstraints(to_lower=True, pattern=USERNAME_REGEXP)
]
- expire: Union[datetime, None, Literal[0]] = Field(None)
+ expire_strategy: UserExpireStrategy
+ expire_date: datetime | None = Field(None)
+ usage_duration: int | None = Field(None)
+ activation_deadline: datetime | None = Field(None)
key: str = Field(default_factory=lambda: secrets.token_hex(16))
data_limit: int | None = Field(
ge=0, default=None, description="data_limit can be 0 or greater"
)
- enabled: bool = Field(default=True)
- data_limit_reset_strategy: UserDataLimitResetStrategy = (
- UserDataLimitResetStrategy.no_reset
+ data_limit_reset_strategy: UserDataUsageResetStrategy = (
+ UserDataUsageResetStrategy.no_reset
)
note: Annotated[str, Field(max_length=500)] | None = None
sub_updated_at: datetime | None = Field(None)
sub_last_user_agent: str | None = Field(None)
online_at: datetime | None = Field(None)
- on_hold_expire_duration: int | None = Field(None)
- on_hold_timeout: datetime | None = Field(None)
model_config = ConfigDict(from_attributes=True)
class UserCreate(User):
- status: UserStatusCreate = Field(UserStatusCreate.active)
service_ids: list[int] = Field([])
model_config = ConfigDict(
json_schema_extra={
"example": {
"username": "user1234",
"services": [1, 2, 3],
- "expire": "2023-11-03T20:30:00",
+ "expire_strategy": "start_on_first_use",
+ "usage_duration": 86400 * 14,
+ "activation_deadline": "2024-11-03T20:30:00",
"data_limit": 0,
"data_limit_reset_strategy": "no_reset",
- "status": "active",
"note": "",
- "on_hold_timeout": "2023-11-03T20:30:00",
- "on_hold_expire_duration": 0,
}
}
)
@model_validator(mode="after")
def validate_expiry(self):
- if self.status == UserStatusCreate.on_hold:
- if not self.on_hold_expire_duration:
+ if self.expire_strategy == UserExpireStrategy.START_ON_FIRST_USE:
+ if not self.usage_duration:
raise ValueError(
- "User cannot be on hold without a valid on_hold_expire_duration."
+ "User expire_strategy can not be start_on_first_use without a valid usage_duration."
)
- if self.expire:
- raise ValueError(
- "User cannot be on hold with specified expire."
- )
- elif self.status == UserStatusCreate.active and (
- self.on_hold_expire_duration or self.on_hold_timeout
+ elif (
+ self.expire_strategy == UserExpireStrategy.FIXED_DATE
+ and not self.expire_date
):
raise ValueError(
- "on hold parameters set when user status isn't on_hold."
+ "User expire_strategy can not be fixed_date without a valid expire date."
)
return self
- @field_validator("expire")
- @classmethod
- def validate_expire(cls, v: Union[datetime, None, Literal[0]]):
- if isinstance(v, datetime) and (v.tzinfo is not None):
- raise ValueError(
- "Expire date should be offset naive, and preferably in utc timezone."
- )
- if isinstance(v, datetime) and v < datetime.utcnow():
- raise ValueError(
- "Expire date should be in the future, not in the past."
- )
- return v
-
-class UserModify(User):
- status: UserStatusModify | None = Field(None)
+class UserModify(UserCreate):
service_ids: list[int] | None = Field(None)
- data_limit_reset_strategy: UserDataLimitResetStrategy | None = Field(None)
+ data_limit_reset_strategy: UserDataUsageResetStrategy | None = Field(None)
model_config = ConfigDict(
json_schema_extra={
"example": {
"username": "mammad1234",
- "services": [1, 2, 3],
- "expire": 0,
+ "service_ids": [1, 2, 3],
+ "expire_strategy": "fixed_date",
+ "expire_date": "2023-11-03T20:30:00",
"data_limit": 0,
"data_limit_reset_strategy": "no_reset",
- "status": "active",
"note": "",
- "on_hold_timeout": "2023-11-03T20:30:00",
- "on_hold_expire_duration": 0,
}
}
)
- @model_validator(mode="after")
- def validate_expiry(self):
- if self.on_hold_expire_duration and self.expire:
- raise ValueError(
- "can't set both expire and on hold expire at the same time."
- )
- return self
-
- @field_validator("expire")
- @classmethod
- def validate_expire(cls, v: Union[datetime, None, Literal[0]]):
- if isinstance(v, datetime) and v.tzinfo is not None:
- raise ValueError(
- "Expire date should be offset naive, and preferably in utc timezone."
- )
- if isinstance(v, datetime) and v < datetime.utcnow():
- raise ValueError(
- "Expire date should be in the future, not in the past."
- )
- return v
-
class UserResponse(User):
id: int
- status: UserStatus
+ activated: bool
+ is_active: bool
+ expired: bool
+ data_limit_reached: bool
+ enabled: bool
used_traffic: int
- lifetime_used_traffic: int = 0
+ lifetime_used_traffic: int
created_at: datetime
service_ids: list[int]
@@ -185,11 +140,11 @@ class UserResponse(User):
class UserUsageResponse(BaseModel):
- node_id: Union[int, None] = None
+ node_id: int | None = None
node_name: str
used_traffic: int
class UserUsagesResponse(BaseModel):
username: str
- usages: List[UserUsageResponse]
+ usages: list[UserUsageResponse]
diff --git a/app/routes/service.py b/app/routes/service.py
index 6873ae1..9a384bf 100644
--- a/app/routes/service.py
+++ b/app/routes/service.py
@@ -110,9 +110,10 @@ async def modify_service(
)
else:
for user in response.users:
- await marznode.operations.update_user(
- user, old_inbounds=old_inbounds
- )
+ if user.activated:
+ marznode.operations.update_user(
+ user, old_inbounds=old_inbounds
+ )
return response
diff --git a/app/routes/system.py b/app/routes/system.py
index aee3cf3..81d2cb1 100644
--- a/app/routes/system.py
+++ b/app/routes/system.py
@@ -7,7 +7,7 @@ from app.dependencies import DBDep, AdminDep, SudoAdminDep
from app.models.node import NodeStatus
from app.models.settings import SubscriptionSettings, TelegramSettings
from app.models.system import UsersStats, NodesStats, AdminsStats
-from app.models.user import UserStatus
+from app.models.user import UserExpireStrategy
router = APIRouter(tags=["System"], prefix="/system")
@@ -67,24 +67,22 @@ def get_users_stats(db: DBDep, admin: AdminDep):
db, admin=admin if not admin.is_sudo else None
),
active=crud.get_users_count(
- db,
- admin=admin if not admin.is_sudo else None,
- status=UserStatus.active,
+ db, admin=admin if not admin.is_sudo else None, is_active=True
),
on_hold=crud.get_users_count(
db,
admin=admin if not admin.is_sudo else None,
- status=UserStatus.on_hold,
+ expire_strategy=UserExpireStrategy.START_ON_FIRST_USE,
),
expired=crud.get_users_count(
db,
admin=admin if not admin.is_sudo else None,
- status=UserStatus.expired,
+ expired=True,
),
limited=crud.get_users_count(
db,
admin=admin if not admin.is_sudo else None,
- status=UserStatus.limited,
+ data_limit_reached=True,
),
online=crud.get_users_count(
db, admin=admin if not admin.is_sudo else None, online=True
diff --git a/app/routes/user.py b/app/routes/user.py
index 33c5825..dd31d12 100644
--- a/app/routes/user.py
+++ b/app/routes/user.py
@@ -1,6 +1,6 @@
import asyncio
import logging
-from datetime import datetime, timezone
+from datetime import datetime, timedelta
from enum import Enum
import sqlalchemy
@@ -25,7 +25,6 @@ from app.models.user import (
UserCreate,
UserModify,
UserResponse,
- UserStatus,
UserUsagesResponse,
)
from app.utils import report
@@ -35,11 +34,21 @@ router = APIRouter(prefix="/users", tags=["User"])
class UsersSortingOptions(str, Enum):
- username = "username"
- used_traffic = "used_traffic"
- data_limit = "data_limit"
- expire = "expire"
- created_at = "created_at"
+ USERNAME = "username"
+ USED_TRAFFIC = "used_traffic"
+ DATA_LIMIT = "data_limit"
+ EXPIRE_DATE = "expire_date"
+ CREATED_AT = "created_at"
+
+
+user_filters = [
+ "username",
+ "is_active",
+ "activated",
+ "expired",
+ "data_limit_reached",
+ "enabled",
+]
@router.get("", response_model=Page[UserResponse])
@@ -47,9 +56,13 @@ def get_users(
db: DBDep,
admin: AdminDep,
username: list[str] = Query(None),
- status: list[UserStatus] = Query(None),
order_by: UsersSortingOptions = Query(None),
descending: bool = Query(False),
+ is_active: bool | None = Query(None),
+ activated: bool | None = Query(None),
+ expired: bool | None = Query(None),
+ data_limit_reached: bool | None = Query(None),
+ enabled: bool | None = Query(None),
):
"""
Filters users based on the options
@@ -60,14 +73,18 @@ def get_users(
admin = dbadmin if not admin.is_sudo else None
- if username:
- if len(username) > 1:
- query = query.filter(User.username.in_(username))
- else:
- query = query.filter(User.username.ilike(f"%{username[0]}%"))
-
- if status:
- query = query.filter(User.status.in_(status))
+ for name in user_filters:
+ value = locals().get(name)
+ if value is not None:
+ if name == "username":
+ if len(username) > 1:
+ query = query.filter(User.username.in_(username))
+ else:
+ query = query.filter(
+ User.username.ilike(f"%{username[0]}%")
+ )
+ else:
+ query = query.filter(getattr(User, name) == value)
if admin:
query = query.filter(User.admin == admin)
@@ -87,7 +104,7 @@ async def add_user(new_user: UserCreate, db: DBDep, admin: AdminDep):
Add a new user
- **username** must have 3 to 32 characters and is allowed to contain a-z, 0-9, and underscores in between
- - **expire** must be a UTC timestamp
+ - **expire_date** must be a datetime
- **data_limit** must be in Bytes, e.g. 1073741824B = 1GB
- **services** list of service ids
"""
@@ -101,8 +118,8 @@ async def add_user(new_user: UserCreate, db: DBDep, admin: AdminDep):
raise HTTPException(status_code=409, detail="User already exists")
user = UserResponse.model_validate(db_user)
- await marznode.operations.update_user(user=db_user)
- asyncio.create_task(
+ marznode.operations.update_user(user=db_user)
+ asyncio.ensure_future(
report.user_created(user=user, user_id=db_user.id, by=admin)
)
logger.info("New user `%s` added", db_user.username)
@@ -121,7 +138,7 @@ async def reset_users_data_usage(db: DBDep, admin: SudoAdminDep):
@router.delete("/expired")
-def delete_expired(passed_time: int, db: DBDep, admin: AdminDep):
+async def delete_expired(passed_time: int, db: DBDep, admin: AdminDep):
"""
Delete expired users
- **passed_time** must be a timestamp
@@ -132,12 +149,12 @@ def delete_expired(passed_time: int, db: DBDep, admin: AdminDep):
db_users = crud.get_users(
db=db,
- status=[UserStatus.expired, UserStatus.limited],
+ expired=True,
admin=dbadmin if not admin.is_sudo else None,
)
- current_time = int(datetime.now(timezone.utc).timestamp())
- expiration_threshold = current_time - passed_time
+ current_time = datetime.utcnow()
+ expiration_threshold = current_time - timedelta(seconds=passed_time)
expired_users = [
user
for user in db_users
@@ -147,9 +164,11 @@ def delete_expired(passed_time: int, db: DBDep, admin: AdminDep):
raise HTTPException(status_code=404, detail="No expired user found.")
for db_user in expired_users:
+ if db_user.activated and not db_user.is_active:
+ marznode.operations.update_user(db_user)
crud.remove_user(db, db_user)
- asyncio.create_task(
+ asyncio.ensure_future(
report.user_deleted(username=db_user.username, by=admin)
)
@@ -173,46 +192,52 @@ async def modify_user(
"""
Modify a user
- - set **expire** to 0 to make the user unlimited in time, null for no change
- set **data_limit** to 0 to make the user unlimited in data, null for no change
"""
- old_status = db_user.status
- status_change = bool(
- modifications.status and modifications.status != db_user.status
- )
+ active_before = db_user.is_active
old_inbounds = {(i.node_id, i.protocol, i.tag) for i in db_user.inbounds}
new_user = crud.update_user(db, db_user, modifications)
+ active_after = new_user.is_active
new_inbounds = {(i.node_id, i.protocol, i.tag) for i in new_user.inbounds}
- user = UserResponse.model_validate(db_user)
inbound_change = old_inbounds != new_inbounds
+
if (
- inbound_change
- and user.enabled
- and user.status in {UserStatus.active, UserStatus.on_hold}
- ):
- await marznode.operations.update_user(new_user, old_inbounds)
+ inbound_change and new_user.is_active
+ ) or active_before != active_after:
+ marznode.operations.update_user(
+ new_user, old_inbounds, remove=~db_user.is_active
+ )
+ db_user.activated = db_user.is_active
+ db.commit()
- asyncio.create_task(report.user_updated(user=user, by=admin))
+ asyncio.ensure_future(
+ report.user_updated(
+ user=UserResponse.model_validate(db_user), by=admin
+ )
+ )
- logger.info("User `%s` modified", user.username)
+ logger.info("User `%s` modified", db_user.username)
- if status_change:
- asyncio.create_task(
+ if active_before != active_after:
+ asyncio.ensure_future(
report.status_change(
- username=user.username, status=user.status, user=user, by=admin
+ username=db_user.username,
+ activation=db_user.activated,
+ user=UserResponse.model_validate(db_user),
+ by=admin,
)
)
logger.info(
- "User `%s` status changed from `%s` to `%s`",
- user.username,
- old_status,
- user.status,
+ "User `%s` activation changed from `%s` to `%s`",
+ db_user.username,
+ active_before,
+ active_after,
)
- return user
+ return db_user
@router.delete("/{username}")
@@ -220,12 +245,12 @@ async def remove_user(db_user: UserDep, db: DBDep, admin: AdminDep):
"""
Remove a user
"""
- await marznode.operations.remove_user(db_user)
+ marznode.operations.update_user(db_user, remove=True)
crud.remove_user(db, db_user)
db.flush()
- asyncio.create_task(
+ asyncio.ensure_future(
report.user_deleted(username=db_user.username, by=admin)
)
logger.info("User %s deleted", db_user.username)
@@ -254,17 +279,16 @@ async def reset_user_data_usage(db_user: UserDep, db: DBDep, admin: AdminDep):
"""
Reset user data usage
"""
- previous_status = db_user.status
+ was_active = db_user.status
db_user = crud.reset_user_data_usage(db, db_user)
- if (
- db_user.status == UserStatus.active
- and previous_status == UserStatus.limited
- ):
- await marznode.operations.update_user(db_user)
+ if db_user.is_active and not was_active:
+ marznode.operations.update_user(db_user)
+ db_user.activated = True
+ db.commit()
user = UserResponse.model_validate(db_user)
- asyncio.create_task(report.user_data_usage_reset(user=user, by=admin))
+ asyncio.ensure_future(report.user_data_usage_reset(user=user, by=admin))
logger.info("User `%s`'s usage was reset", db_user.username)
@@ -279,9 +303,10 @@ async def enable_user(db_user: UserDep, db: DBDep, admin: AdminDep):
if db_user.enabled:
raise HTTPException(409, "User is already enabled")
db_user.enabled = True
+ db_user.activated = True
db.commit()
- await marznode.operations.update_user(db_user)
+ marznode.operations.update_user(db_user)
user = UserResponse.model_validate(db_user)
@@ -298,9 +323,10 @@ async def disable_user(db_user: UserDep, db: DBDep, admin: AdminDep):
if not db_user.enabled:
raise HTTPException(409, "User is not enabled")
db_user.enabled = False
+ db_user.activated = False
db.commit()
- await marznode.operations.remove_user(db_user)
+ marznode.operations.update_user(db_user, remove=True)
user = UserResponse.model_validate(db_user)
@@ -318,11 +344,13 @@ async def revoke_user_subscription(
"""
db_user = crud.revoke_user_sub(db, db_user)
- if db_user.status in [UserStatus.active, UserStatus.on_hold]:
- await marznode.operations.remove_user(db_user)
- await marznode.operations.update_user(db_user)
+ 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.create_task(report.user_subscription_revoked(user=user, by=admin))
+ asyncio.ensure_future(
+ report.user_subscription_revoked(user=user, by=admin)
+ )
logger.info("User %s subscription revoked", db_user.username)
diff --git a/app/tasks/reset_user_data_usage.py b/app/tasks/reset_user_data_usage.py
index 3e009e5..57a2fdb 100644
--- a/app/tasks/reset_user_data_usage.py
+++ b/app/tasks/reset_user_data_usage.py
@@ -3,15 +3,15 @@ from datetime import datetime
from app import marznode
from app.db import crud, GetDB, get_users
-from app.models.user import UserDataLimitResetStrategy, UserStatus
+from app.models.user import UserDataUsageResetStrategy
logger = logging.getLogger(__name__)
reset_strategy_to_days = {
- UserDataLimitResetStrategy.day.value: 1,
- UserDataLimitResetStrategy.week.value: 7,
- UserDataLimitResetStrategy.month.value: 30,
- UserDataLimitResetStrategy.year.value: 365,
+ UserDataUsageResetStrategy.day.value: 1,
+ UserDataUsageResetStrategy.week.value: 7,
+ UserDataUsageResetStrategy.month.value: 30,
+ UserDataUsageResetStrategy.year.value: 365,
}
@@ -21,10 +21,10 @@ async def reset_user_data_usage():
for user in get_users(
db,
reset_strategy=[
- UserDataLimitResetStrategy.day.value,
- UserDataLimitResetStrategy.week.value,
- UserDataLimitResetStrategy.month.value,
- UserDataLimitResetStrategy.year.value,
+ UserDataUsageResetStrategy.day.value,
+ UserDataUsageResetStrategy.week.value,
+ UserDataUsageResetStrategy.month.value,
+ UserDataUsageResetStrategy.year.value,
],
):
last_reset_time = user.traffic_reset_at or user.created_at
@@ -35,14 +35,12 @@ async def reset_user_data_usage():
if not (now - last_reset_time).days >= num_days_to_reset:
continue
- old_status = user.status
+ was_active = user.is_active
crud.reset_user_data_usage(db, user)
# make user active if limited on usage reset
- if (
- user.enabled
- and user.status == UserStatus.active
- and old_status == UserStatus.limited
- ):
- await marznode.operations.update_user(user)
-
- logger.info(f'User data usage reset for User "{user.username}"')
+ if user.is_active and not was_active:
+ marznode.operations.update_user(user)
+ user.activated = True
+ db.commit()
+
+ logger.info("User data usage reset for User `%s`", user.username)
diff --git a/app/tasks/review_users.py b/app/tasks/review_users.py
index 643e0de..97bc044 100644
--- a/app/tasks/review_users.py
+++ b/app/tasks/review_users.py
@@ -1,91 +1,37 @@
import asyncio
import logging
-from datetime import datetime
+from datetime import datetime, timedelta
from typing import TYPE_CHECKING
-from sqlalchemy.orm import Session
-
from app import marznode
-from app.config.env import (
- NOTIFY_DAYS_LEFT,
- NOTIFY_REACHED_USAGE_PERCENT,
- WEBHOOK_ADDRESS,
-)
from app.db import (
GetDB,
- get_notification_reminder,
get_users,
- start_user_expire,
- update_user_status,
)
-from app.models.user import ReminderType, UserResponse, UserStatus
-from app.utils import report
-from app.utils.helpers import (
- calculate_expiration_days,
- calculate_usage_percent,
+from app.models.user import (
+ UserResponse,
+ UserExpireStrategy,
)
+from app.utils import report
if TYPE_CHECKING:
- from app.db.models import User
+ pass
logger = logging.getLogger(__name__)
-def add_notification_reminders(
- db: Session, user: "User", now: datetime = datetime.utcnow()
-) -> None:
- if user.data_limit:
- usage_percent = calculate_usage_percent(
- user.used_traffic, user.data_limit
- )
- if (usage_percent >= NOTIFY_REACHED_USAGE_PERCENT) and (
- not get_notification_reminder(db, user.id, ReminderType.data_usage)
- ):
- report.data_usage_percent_reached(
- db,
- usage_percent,
- UserResponse.model_validate(user),
- user.id,
- user.expire,
- )
-
- if user.expire and ((now - user.created_at).days >= NOTIFY_DAYS_LEFT):
- expire_days = calculate_expiration_days(user.expire)
- if (expire_days <= NOTIFY_DAYS_LEFT) and (
- not get_notification_reminder(
- db, user.id, ReminderType.expiration_date
- )
- ):
- report.expire_days_reached(
- db,
- expire_days,
- UserResponse.model_validate(user),
- user.id,
- user.expire,
- )
-
-
async def review_users():
now = datetime.utcnow()
with GetDB() as db:
- for user in get_users(db, status=UserStatus.active):
-
- limited = user.data_limit and user.used_traffic >= user.data_limit
- expired = user.expire and user.expire <= now
- if limited:
- status = UserStatus.limited
- elif expired:
- status = UserStatus.expired
- else:
- if WEBHOOK_ADDRESS:
- add_notification_reminders(db, user, now)
- continue
-
- await marznode.operations.remove_user(user)
- update_user_status(db, user, status)
-
- asyncio.create_task(
+ for user in get_users(db, activated=True, is_active=False):
+ """looking for expired/to be limited users who are still active"""
+
+ marznode.operations.update_user(user, remove=True)
+ user.activated = False
+ db.commit()
+ db.refresh(user)
+ asyncio.ensure_future(
report.status_change(
user.username,
user.status,
@@ -93,33 +39,42 @@ async def review_users():
)
)
- logger.info(f'User "{user.username}" status changed to {status}')
-
- for user in get_users(db, status=UserStatus.on_hold):
-
- if user.edit_at:
- base_time = user.edit_at
- else:
- base_time = user.created_at
-
- # Check if the user is online After or at 'base_time'
- if user.online_at and base_time <= user.online_at:
- status = UserStatus.active
-
- elif user.on_hold_timeout and (user.on_hold_timeout <= now):
- # If the user didn't connect within the timeout period, change status to "Active"
- status = UserStatus.active
+ logger.info(
+ "User `%s` activation state changed to `%s`",
+ user.username,
+ str(user.activated),
+ )
- else:
+ for user in get_users(
+ db,
+ expire_strategy=UserExpireStrategy.START_ON_FIRST_USE,
+ is_active=True,
+ ):
+ """looking for to be activated, on hold users"""
+ base_time = user.edit_at or user.created_at
+
+ # Check if the user is online After or at 'base_time' or...
+ # If the user didn't connect until activation_deadline; change status to "Active"
+ if not (
+ (user.online_at and base_time <= user.online_at)
+ or (
+ user.activation_deadline
+ and (user.activation_deadline <= now)
+ )
+ ):
continue
- update_user_status(db, user, status)
- start_user_expire(db, user)
- asyncio.create_task(
+ user.expire_date = datetime.utcnow() + timedelta(
+ seconds=user.usage_duration
+ )
+ 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(f"on hold user `{user.username}` has been activated")
+ logger.info("on hold user `%s` has been activated", user.username)
diff --git a/app/templates/subscription/index.html b/app/templates/subscription/index.html
index 539f26e..a16597b 100644
--- a/app/templates/subscription/index.html
+++ b/app/templates/subscription/index.html
@@ -35,17 +35,7 @@
color: white;
}
- .limited {
- background-color: #F44336;
- color: white;
- }
-
- .expired {
- background-color: #FF9800;
- color: white;
- }
-
- .disabled {
+ .inactive {
background-color: #9E9E9E;
color: white;
}
@@ -73,79 +63,81 @@
- User Information
- Username: {{ user.username }}
- Status: {{
- user.status.value }}
- Data Limit: {% if not user.data_limit %}∞{% else %}{{ user.data_limit | bytesformat }}{% endif %}
- Data Used: {{ user.used_traffic | bytesformat }}{% if user.data_limit_reset_strategy != 'no_reset' %} (resets
- every {{ user.data_limit_reset_strategy.value }}){% endif %}
- Expiration Date:
- {% if not user.expire %}
- ∞
- {% else %}
- {% set current_datetime = now() %}
- {% set remaining_days = ((user.expire - current_datetime).days) %}
- {{ user.expire | datetime }} ({{ remaining_days | int }} days
- remaining)
- {% endif %}
-
- {% if user.status == 'active' or user.status == 'on_hold' %}
- Links:
-
-