logger.setLevel(logging.DEBUG if config.DEBUG else logging.INFO)

handler = logging.StreamHandler()

formatter = uvicorn.logging.ColourizedFormatter(

)

handler.setFormatter(formatter)

logger.addHandler(handler)

from .base import Base, SessionLocal, engine # noqa

from .models import JWT, System, User # noqa

"update_admin",

"remove_admin",

"get_admins",

"create_notification_reminder",

"get_notification_reminder",

"delete_notification_reminder",

"GetDB",

"User",

"System",

"JWT",

"Base",

"Session",

]

from sqlalchemy.orm import declarative_base

from sqlalchemy.orm import sessionmaker

if IS_SQLITE:

engine = create_engine(

SQLALCHEMY_DATABASE_URL,

connect_args={"check_same_thread": False},

)

else:

engine = create_engine(

pool_size=10,

max_overflow=30,

pool_recycle=3600,

)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

from sqlalchemy import and_, delete, update, select

from sqlalchemy.orm import Session

from app.models.admin import AdminCreate, AdminModify, AdminPartialModify

from app.models.proxy import InboundHost as InboundHostModify

from app.models.service import Service as ServiceModify, ServiceCreate

from config import NOTIFY_DAYS_LEFT, NOTIFY_REACHED_USAGE_PERCENT

def add_default_hosts(db: Session, inbounds: List[Inbound]):

db.add_all(hosts)

db.commit()

def assure_node_inbounds(db: Session, inbounds: List[Inbound], node_id: int):

updated_tags = set(i.tag for i in list(inbounds))

inbound_additions, tag_deletions = list(), set()

for tag in current_tags:

if tag not in updated_tags:

tag_deletions.add(tag)

for i in db_inbounds:

db.delete(i)

for inb in inbounds:

if inb.tag in current_tags:

)

db.execute(stmt)

else:

inbound_additions.append(inb)

db.add_all(new_inbounds)

db.flush()

for i in new_inbounds:

db.commit()

if isinstance(statuses, list):

query = query.filter(User.status.in_(statuses))

return query.all()

def get_user_hosts(db: Session, user_id: int):

def get_inbound_hosts(db: Session, inbound_id: int) -> List[InboundHost]:

inbound=inbound,

security=host.security,

alpn=host.alpn,

)

inbound.hosts.append(host)

db.commit()

return db_host

inbound = get_or_create_inbound(db, inbound_tag)

inbound.hosts = [

InboundHost(

alpn=host.alpn,

fingerprint=host.fingerprint,

allowinsecure=host.allowinsecure,

]

db.commit()

db.refresh(inbound)

return db.query(User).filter(User.id == user_id).first()

query = db.query(User)

if usernames:

return query.all()

usages = dict()

for node in db.query(Node).all():

usages[node.id] = UserUsageResponse(

)

for v in db.query(NodeUserUsage).filter(cond):

try:

username=user.username,

# proxies=proxies,

key=user.key,

status=user.status,

data_limit=(user.data_limit or None),

expire=(user.expire or None),

dbuser.status = modify.status

if modify.data_limit is not None:

if dbuser.status not in (UserStatus.expired, UserStatus.disabled):

if not dbuser.data_limit or dbuser.used_traffic < dbuser.data_limit:

if dbuser.status != UserStatus.on_hold:

dbuser.status = UserStatus.active

else:

dbuser.status = UserStatus.limited

if modify.expire is not None:

if dbuser.status in (UserStatus.active, UserStatus.expired):

if not dbuser.expire or dbuser.expire > datetime.utcnow().timestamp():

dbuser.status = UserStatus.active

else:

dbuser.status = UserStatus.expired

dbuser.on_hold_expire_duration = modify.on_hold_expire_duration

if modify.services is not None:

dbuser.edit_at = datetime.utcnow()

db.commit()

for dbuser in query.all():

dbuser.used_traffic = 0

dbuser.status = UserStatus.active

dbuser.usage_logs.clear()

dbuser.node_usages.clear()

dbadmin = Admin(

username=admin.username,

hashed_password=admin.hashed_password,

)

db.add(dbadmin)

db.commit()

return dbadmin

if modified_admin.is_sudo is not None:

dbadmin.is_sudo = modified_admin.is_sudo

dbadmin.hashed_password = modified_admin.hashed_password

dbadmin.password_reset_at = datetime.utcnow()

return dbadmin

query = db.query(Admin)

if username:

if offset:

query = query.offset(offset)

if limit:

dbservice = Service(

name=service.name,

inbounds=db.query(Inbound).filter(Inbound.id.in_(service.inbounds)).all(),

)

db.add(dbservice)

db.commit()

db_service.name = modification.name

if modification.inbounds is not None:

db.commit()

db.refresh(db_service)

return db.query(Node).filter(Node.id == node_id).first()

query = db.query(Node)

if status:

return query.all()

usages = dict()

usages[0] = NodeUsageResponse( # Main Core

)

for node in db.query(Node).all():

usages[node.id] = NodeUsageResponse(

)

cond = and_(NodeUsage.created_at >= start, NodeUsage.created_at <= end)

def create_node(db: Session, node: NodeCreate):

db.add(dbnode)

db.commit()

return dbnode

db_node = db.query(Node).where(Node.id == node_id).first()

db_node.status = status

if message:

def create_notification_reminder(

db.add(reminder)

db.commit()

db.refresh(reminder)

def get_notification_reminder(

) -> Optional[NotificationReminder]:

if reminder is None:

return

if reminder.expires_at and reminder.expires_at < datetime.utcnow():

return reminder

"""Deletes notification reminder filtered by user_id and type if exists"""

stmt = delete(NotificationReminder).where(

NotificationReminder.user_id == user_id,

# this is the Alembic Config object, which provides

# access to the values within the .ini file in use.

config = context.config

# Interpret the config file for Python logging.

# This line sets up loggers basically.

)

with connectable.connect() as connection:

with context.begin_transaction():

context.run_migrations()

import os

from datetime import datetime

from sqlalchemy.orm import relationship

from sqlalchemy.sql.expression import text

from app.db.base import Base

from app.models.node import NodeStatus

class Admin(Base):

name = Column(String(64))

# modifications = Column(String(1024))

users = relationship("User", secondary=users_services, back_populates="services")

class User(Base):

id = Column(Integer, primary_key=True)

username = Column(String(32), unique=True, index=True)

key = Column(String(64), unique=True)

inbounds = relationship(

"Inbound",

secondary="join(users_services, inbounds_services, inbounds_services.c.service_id == users_services.c.service_id)"

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)

data_limit = Column(BigInteger)

data_limit_reset_strategy = Column(

Enum(UserDataLimitResetStrategy),

)

ip_limit = Column(Integer, nullable=False, default=-1)

settings = Column(String)

expire = Column(Integer)

admin_id = Column(Integer, ForeignKey("admins.id"))

admin = relationship("Admin", back_populates="users")

class Inbound(Base):

__tablename__ = "inbounds"

id = Column(Integer, primary_key=True)

protocol = Column(Enum(ProxyTypes))

config = Column(String(), nullable=False)

node_id = Column(Integer, ForeignKey("nodes.id"), index=True)

node = relationship("Node", back_populates="inbounds")

hosts = relationship(

"InboundHost", back_populates="inbound", cascade="all, delete, delete-orphan"

)

Enum(InboundHostALPN),

nullable=False,

default=InboundHostSecurity.none,

)

fingerprint = Column(

Enum(InboundHostFingerprint),

nullable=False,

default=InboundHostSecurity.none,

)

inbound_id = Column(Integer, ForeignKey("inbounds.id"), nullable=False)

class Node(Base):

__tablename__ = "nodes"

id = Column(Integer, primary_key=True)

name = Column(String(256), unique=True)

connection_backend = Column(String())

created_at = Column(DateTime, default=datetime.utcnow)

uplink = Column(BigInteger, default=0)

downlink = Column(BigInteger, default=0)

class NodeUserUsage(Base):

__tablename__ = "node_user_usages"

id = Column(Integer, primary_key=True)

created_at = Column(DateTime, nullable=False) # one hour per record

class NodeUsage(Base):

__tablename__ = "node_usages"

id = Column(Integer, primary_key=True)

created_at = Column(DateTime, nullable=False) # one hour per record

yield db

payload = get_admin_payload(token)

if not payload:

return

if not dbadmin:

return

return admin

try:

int(key, 16)

except ValueError:

return db_user

db_user = crud.get_user(db, username)

raise HTTPException(status_code=403, detail="You're not allowed")

if not db_user:

import config

from app.templates import render_template

from . import __version__, telegram

from .routes import api_router

logger = logging.getLogger(__name__)

dashboard_path = "/dashboard/"

"""stores nodes and provides entities to communicate with the nodes"""

from typing import Dict

from . import operations

nodes: Dict[int, MarzNodeBase] = {}

class MarzNodeBase(ABC):

"""updates a user on the node"""

async def fetch_users_stats(self):

class MarzNodeDB:

def list_users(self):

with GetDB() as db:

users = dict()

for rel in relations:

if not users.get(rel[0]):

users[rel[0]]["inbounds"].append(rel[3].tag)

return list(users.values())

from .base import MarzNodeBase

from .database import MarzNodeDB

from .marznode_pb2_grpc import MarzServiceStub

from ..models.node import NodeStatus

logger = logging.getLogger(__name__)

("grpc.keepalive_permit_without_calls", 1),

("grpc.max_connection_idle_ms", INT_MAX),

("grpc.client_idle_timeout_ms", INT_MAX),

]

class MarzNodeGRPCIO(MarzNodeBase, MarzNodeDB):

self.id = node_id

self._address = address

self._port = port

self._stub = MarzServiceStub(self._channel)

asyncio.create_task(self._monitor_channel())

self._streaming_task = None

logger.debug("got something from queue")

user = user_update["user"]

await stream.write(

async def update_user(self, user, inbounds: set[str] | None = None):

if inbounds is None:

await self._updates_queue.put({"user": user, "inbounds": inbounds})

async def _repopulate_users(self, users_data: list[dict]) -> None:

await self._stub.RepopulateUsers(UsersData(users_data=updates))

async def fetch_users_stats(self):

async def get_logs(self, include_buffer=True):

async for response in self._stub.StreamXrayLogs(

):

yield response.line

from .base import MarzNodeBase

from .database import MarzNodeDB

from .marznode_grpc import MarzServiceStub

from ..models.node import NodeStatus

logger = logging.getLogger(__name__)

def string_to_temp_file(content: str):

file.write(content)

file.flush()

return file

class MarzNodeGRPCLIB(MarzNodeBase, MarzNodeDB):

self.id = node_id

self._address = address

self._port = port

else:

if not self.synced:

await self._sync()

self.set_status(NodeStatus.healthy)

logger.info("Connected to node %i", self.id)

await asyncio.sleep(10)

logger.info("got something from queue")

user = user_update["user"]

await stream.send_message(

except (OSError, ConnectionError, GRPCError, StreamTerminatedError):

logger.info("node %i detached", self.id)

self.synced = False

await self._updates_queue.put({"user": user, "inbounds": inbounds})

async def _repopulate_users(self, users_data: list[dict]) -> None:

await self._stub.RepopulateUsers(UsersData(users_data=updates))

async def fetch_users_stats(self):

if db_node.connection_backend == NodeConnectionBackend.grpcio:

node = MarzNodeGRPCIO(db_node.id, db_node.address, db_node.port)

else:

marznode.nodes[db_node.id] = node

name: str

address: str

port: int = 62050

usage_coefficient: float = Field(ge=0, default=1.0)

model_config = ConfigDict(from_attributes=True)

class NodeCreate(Node):

}

class NodeModify(Node):

connection_backend: Optional[NodeConnectionBackend] = Field(None)

status: Optional[NodeStatus] = Field(None)

usage_coefficient: Optional[float] = Field(None)

}

class NodeResponse(Node):

import json

from enum import Enum

from typing import Optional, Union, List

# from uuid import UUID, uuid4

from pydantic import ConfigDict, BaseModel, Field, validator

class XTLSFlows(Enum):

class ShadowsocksMethods(Enum):

class ProxyTypes(str, Enum):

class ServiceCreate(ServiceBase):

users: List[int] = Field([])

inbounds: List[int] = Field([])

}

class ServiceModify(ServiceBase):

inbounds: Optional[List[int]] = Field(None)

}

class ServiceResponse(Service):

from enum import Enum

from typing import List, Optional, Union, Annotated

from config import XRAY_SUBSCRIPTION_URL_PREFIX

@classmethod

def validate_timeout(cls, v: Optional[int]):

# Check if expire is 0 or None and timeout is not 0 or None

return None

return v

model_config = ConfigDict(from_attributes=True)

class UserCreate(User):

status: UserStatusCreate = Field(None, validate_default=True)

services: List[int] = Field([])

}

# TODO[pydantic]: We couldn't refactor the `validator`, please replace it by `field_validator` manually.

# Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.

# TODO[pydantic]: We couldn't refactor the `validator`, please replace it by `field_validator` manually.

# Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.

@field_validator("status", mode="before")

expire = info.data.get("expire")

if status == UserStatusCreate.on_hold:

if on_hold_expire == 0 or on_hold_expire is None:

if expire:

raise ValueError("User cannot be on hold with specified expire.")

return status

status: Optional[UserStatusModify] = Field(None)

services: List[int] = Field([])

data_limit_reset_strategy: Optional[UserDataLimitResetStrategy] = Field(None)

}

# TODO[pydantic]: We couldn't refactor the `validator`, please replace it by `field_validator` manually.

# Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.

expire = values.get("expire")

if status == UserStatusCreate.on_hold:

if not on_hold_expire:

if expire:

raise ValueError("User cannot be on hold with specified expire.")

return status

@property

def subscription_url(self) -> str:

salt = secrets.token_hex(8)

return f"{url_prefix}/sub/{self.username}/{self.key}"

api_router.include_router(system.router, prefix="/api")

api_router.include_router(user.router, prefix="/api")

if not dbadmin:

return None

@router.get("", response_model=Page[Admin])

query = db.query(DBAdmin)

if username:

return paginate(db, query)

@router.post("", response_model=Admin)

try:

dbadmin = crud.create_admin(db, new_admin)

except sqlalchemy.exc.IntegrityError:

@router.post("/token", response_model=Token)

if authenticate_env_sudo(form_data.username, form_data.password):

return Token(access_token=create_admin_token(form_data.username, is_sudo=True))

if dbadmin := authenticate_admin(db, form_data.username, form_data.password):

raise HTTPException(

status_code=status.HTTP_401_UNAUTHORIZED,

@router.put("/{username}", response_model=Admin)

if not (admin.is_sudo or admin.username == username):

raise HTTPException(status_code=403, detail="You're not allowed")

@router.delete("/{username}")

dbadmin = crud.get_admin(db, username)

if not dbadmin:

raise HTTPException(status_code=404, detail="Admin not found")

from app.dependencies import DBDep, sudo_admin

from app.models.proxy import Inbound, InboundHost, InboundHostResponse

@router.get("", response_model=Page[Inbound])

return paginate(db, query)

def get_hosts(db: DBDep):

return paginate(db.query(DBInboundHost))

from app.db.models import Node

from app.dependencies import DBDep, SudoAdminDep, EndDateDep, StartDateDep, get_admin

logger = logging.getLogger(__name__)

@router.get("", response_model=Page[NodeResponse])

query = db.query(Node)

if name:

@router.post("", response_model=NodeResponse)

try:

db_node = crud.create_node(db, new_node)

except sqlalchemy.exc.IntegrityError:

db.rollback()

certificate = get_tls_certificate(db)

await marznode.operations.add_node(db_node, certificate)

@router.get("/settings", response_model=NodeSettings)

tls = crud.get_tls_certificate(db)

@router.get("/usage", response_model=NodesUsageResponse)

"""

Get nodes usage

"""

@router.get("/{node_id}", response_model=NodeResponse)

db_node = crud.get_node_by_id(db, node_id)

if not db_node:

raise HTTPException(status_code=404, detail="Node not found")

@router.websocket("/{node_id}/logs")

admin = get_admin(db, token)

if not admin or not admin.is_sudo:

@router.put("/{node_id}", response_model=NodeResponse)

db_node = crud.get_node_by_id(db, node_id)

if not db_node:

raise HTTPException(status_code=404, detail="Node not found")

@router.delete("/{node_id}")

db_node = crud.get_node_by_id(db, node_id)

if not db_node:

raise HTTPException(status_code=404, detail="Node not found")

@router.post("/{node_id}/resync")

db_node = crud.get_node_by_id(db, node_id)

if not db_node:

raise HTTPException(status_code=404, detail="Node not found")

@router.get("/{node_id}/xray_config")

if not (node := marznode.nodes.get(node_id)):

raise HTTPException(status_code=404, detail="Node not found")

return json.loads(await node.get_xray_config())

@router.put("/{node_id}/xray_config")

if not (node := marznode.nodes.get(node_id)):

raise HTTPException(status_code=404, detail="Node not found")

xray_config = json.dumps(body)

from app.db import crud

from app.db.models import Service

from app.dependencies import DBDep, sudo_admin

@router.get("", response_model=Page[ServiceResponse])

@router.post("", response_model=ServiceResponse)

"""

Add a new service

return crud.create_service(db, new_service)

except sqlalchemy.exc.IntegrityError:

db.rollback()

@router.get("/{id}", response_model=ServiceResponse)

@router.put("/{id}", response_model=ServiceResponse)

"""

Modify Service

# TODO: Update all affected users in nodes

dbservice = crud.get_service(db, id)

if not dbservice:

old_inbounds = {(i.node_id, i.protocol, i.tag) for i in dbservice.inbounds}

try:

response = crud.update_service(db, dbservice, modification)

@router.delete("/{id}")

dbservice = crud.get_service(db, id)

if not dbservice:

raise HTTPException(status_code=404, detail="Service not found")

crud.remove_service(db, dbservice)

return dict()

SUB_PROFILE_TITLE,

SUB_SUPPORT_URL,

SUB_UPDATE_INTERVAL,

)

def get_subscription_user_info(user: UserResponse) -> dict:

@router.get("/{username}/{key}")

"""

Subscription link, V2ray and Clash supported

"""

user: UserResponse = UserResponse.model_validate(db_user)

if "text/html" in accept_header:

return HTMLResponse(

)

response_headers = {

f"{key}={val}"

for key, val in get_subscription_user_info(user).items()

if val is not None

}

crud.update_user_sub(db, db_user, user_agent)

return Response(content=conf, media_type="text/yaml", headers=response_headers)

conf = generate_subscription(user=user, config_format="clash", as_base64=False)

return Response(content=conf, media_type="text/yaml", headers=response_headers)

else:

conf = generate_subscription(user=user, config_format="v2ray", as_base64=True)

@router.get("/{username}/{key}/usage")

usages = crud.get_user_usages(db, db_user, start_date, end_date)

return {"usages": usages, "username": db_user.username}

@router.get("/{username}/{key}/{client_type}")

def user_subscription_with_client_type(

"""

Subscription link, v2ray, clash, sing-box, outline and clash-meta supported

"""

f"{key}={val}"

for key, val in get_subscription_user_info(user).items()

if val is not None

}

if client_type == "clash-meta":

return Response(content=conf, media_type="text/yaml", headers=response_headers)

elif client_type == "sing-box":

elif client_type == "clash":

conf = generate_subscription(user=user, config_format="clash", as_base64=False)

return Response(content=conf, media_type="text/plain", headers=response_headers)

elif client_type == "outline":

else:

raise HTTPException(status_code=400, detail="Invalid subscription type")

dbadmin: Union[Admin, None] = crud.get_admin(db, admin.username)

total_user = crud.get_users_count(db, admin=dbadmin if not admin.is_sudo else None)

realtime_bandwidth_stats = realtime_bandwidth()

return SystemStats(

from app import marznode

from app.db import crud, User

from app.utils import report

logger = logging.getLogger(__name__)

class UsersSortingOptions(str, Enum):

@router.get("", response_model=Page[UserResponse])

"""

Filters users based on the options

"""

@router.post("", response_model=UserResponse)

"""

Add a new user

# TODO expire should be datetime instead of timestamp

try:

except sqlalchemy.exc.IntegrityError:

db.rollback()

raise HTTPException(status_code=409, detail="User already exists")

user = UserResponse.model_validate(db_user)

await marznode.operations.update_user(user=db_user)

logger.info("New user `%s` added", db_user.username)

return user

@router.post("/reset")

"""

Reset all users data usage,

You will need to restart for this to take effect for now

@router.delete("/expired")

"""

Delete expired users

- **passed_time** must be a timestamp

dbadmin = crud.get_admin(db, admin.username)

current_time = int(datetime.now(timezone.utc).timestamp())

expiration_threshold = current_time - passed_time

expired_users = [

if not expired_users:

raise HTTPException(status_code=404, detail="No expired user found.")

for db_user in expired_users:

crud.remove_user(db, db_user)

logger.info("User `%s` removed", db_user.username)

@router.put("/{username}", response_model=UserResponse)

"""

Modify a user

- set **data_limit** to 0 to make the user unlimited in data, null for no change

"""

old_status = db_user.status

await marznode.operations.remove_user(db_user)

old_inbounds = {(i.node_id, i.protocol, i.tag) for i in db_user.inbounds}

if inbound_change and user.status in {UserStatus.active, UserStatus.on_hold}:

await marznode.operations.update_user(new_user, old_inbounds)

logger.info("User `%s` modified", user.username)

if status_change:

return user

@router.delete("/{username}")

"""

Remove a user

"""

crud.remove_user(db, db_user)

db.flush()

logger.info("User %s deleted", db_user.username)

return {}

@router.post("/{username}/reset", response_model=UserResponse)

"""

Reset user data usage

"""

await marznode.operations.update_user(db_user)

user = UserResponse.model_validate(db_user)

logger.info("User `%s`'s usage was reset", db_user.username)

@router.post("/{username}/revoke_sub", response_model=UserResponse)

"""

Revoke users subscription (Subscription link and proxies)

"""

await marznode.operations.remove_user(db_user)

await marznode.operations.update_user(db_user)

user = UserResponse.model_validate(db_user)

logger.info("User %s subscription revoked", db_user.username)

@router.get("/{username}/usage", response_model=UserUsagesResponse)

"""

Get users usage

"""

@router.put("/{username}/set-owner", response_model=UserResponse)

db_user = crud.get_user(db, username)

if not db_user:

raise HTTPException(status_code=404, detail="User not found")

"delete_expired_reminders",

"send_notifications",

"store_bandwidth",

]

from app.marznode import MarzNodeBase

if not params:

return

with GetDB() as db:

# make user usage row if doesn't exist

existings = [r[0] for r in db.execute(select_stmt).fetchall()]

uids_to_insert = set()

for p in params:

if uid in existings:

continue

uids_to_insert.add(uid)

if uids_to_insert:

stmt = insert(NodeUserUsage).values(

created_at=created_at,

node_id=node_id,

)

# record

db.commit()

if not params:

return

with GetDB() as db:

# make node usage row if doesn't exist

notfound = db.execute(select_stmt).first() is None

if notfound:

db.execute(stmt)

db.commit()

# record

db.execute(stmt, params)

# usage_coefficient = {None: 1} # default usage coefficient for the main api instance

results = await asyncio.gather(

api_params = {node_id: params for node_id, params in list(results)}

users_usage = defaultdict(int)

for node_id, params in api_params.items():

for param in params:

if not users_usage:

return

# record users usage

with GetDB() as db:

)

db.execute(stmt, users_usage, execution_options={"synchronize_session": None})

db.commit()

for node_id, params in api_params.items():

async def reset_user_data_usage():

now = datetime.utcnow()

with GetDB() as db:

last_reset_time = user.last_traffic_reset_time

num_days_to_reset = reset_strategy_to_days[user.data_limit_reset_strategy]

if user.status == UserStatus.active and old_status == UserStatus.limited:

await marznode.operations.update_user(user)

from sqlalchemy.orm import Session

from app import marznode

from app.models.user import ReminderType, UserResponse, UserStatus

from app.utils import report

if TYPE_CHECKING:

from app.db.models import User

logger = logging.getLogger(__name__)

if user.data_limit:

report.data_usage_percent_reached(

if user.expire and ((now - user.created_at).days >= NOTIFY_DAYS_LEFT):

expire_days = calculate_expiration_days(user.expire)

report.expire_days_reached(

async def review_users():

await marznode.operations.remove_user(user)

update_user_status(db, user, status)

for user in get_users(db, status=UserStatus.on_hold):

if user.online_at and base_time <= datetime.timestamp(user.online_at):

status = UserStatus.active

# If the user didn't connect within the timeout period, change status to "Active"

status = UserStatus.active

update_user_status(db, user, status)

start_user_expire(db, user)

logger.info(f"on hold user `{user.username}` has been activated")

try:

logger.debug(f"Sending {len(data)} webhook updates to {config.WEBHOOK_ADDRESS}")

if r.ok:

return True

logger.error(r)

notifications_to_send = list()

try:

continue

if notification.send_at > dt.utcnow().timestamp():

continue

notifications_to_send.append(notification)

except IndexError: # if the queue is empty

continue

notification.tries += 1

notification.send_at = ( # schedule notification for n seconds later

queue.append(notification)

def delete_expired_reminders() -> None:

with GetDB() as db:

db.commit()

def record_realtime_bandwidth() -> None:

io = psutil.net_io_counters()

\ No newline at end of file

if bot:

handler_dir = dirname(__file__) + "/handlers/"

for name in handler_names:

spec.loader.exec_module(importlib.util.module_from_spec(spec))

utils.setup()

await bot.polling()

report_user_deletion,

report_status_change,

report_user_usage_reset,

)

__all__ = [

"report_user_deletion",

"report_status_change",

"report_user_usage_reset",

]

from telebot.asyncio_handler_backends import State, StatesGroup

from app.db import GetDB, crud

from app.models.proxy import ProxyTypes

from app.telegram import bot

from app.telegram.utils.keyboard import BotKeyboard

from app.utils.store import MemoryStorage

from app.utils.system import cpu_usage, memory_usage, readable_size

mem_store = MemoryStorage()

class botStates(StatesGroup):

start = State()

add_time = State()

edit_user_data_limit = State()

edit_user_expire = State()

active_users=users_active,

deactivate_users=total_users - users_active,

up_speed=readable_size(realtime_bandwidth().outgoing_bytes),

)

async def cleanup_messages(chat_id: int) -> None:

messages: list[int] = mem_store.get(f"{chat_id}:messages_to_delete", [])

for message_id in messages:

mem_store.set(f"{chat_id}:messages_to_delete", [])

async def help_command(message: types.Message):

await cleanup_messages(message.chat.id)

# bot.clear_step_handler_by_chat_id(message.chat.id)

await bot.delete_state(message.from_user.id, message.chat.id)

{user_link} Welcome to Marzban Telegram-Bot Admin Panel.

Here you can manage your users and proxies.

To get started, use the buttons below.

""".format(

async def system_command(call: types.CallbackQuery):

return await bot.edit_message_text(

get_system_info(),

call.message.chat.id,

call.message.message_id,

parse_mode="MarkdownV2",

)

async def restart_command(call: types.CallbackQuery):

await bot.edit_message_text(

call.message.chat.id,

call.message.message_id,

)

async def delete_user_command(call: types.CallbackQuery):

await bot.edit_message_text(

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

)

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

)

call.message.message_id,

parse_mode="markdown",

reply_markup=BotKeyboard.confirm_action(

)

call.message.message_id,

parse_mode="markdown",

reply_markup=BotKeyboard.confirm_action(

)

async def edit_all_command(call: types.CallbackQuery):

with GetDB() as db:

total_users = crud.get_users_count(db)

disabled_users = crud.get_users_count(db, UserStatus.disabled)

expired_users = crud.get_users_count(db, UserStatus.expired)

limited_users = crud.get_users_count(db, UserStatus.limited)

👥 *Total Users*: `{total_users}`

✅ *Active Users*: `{active_users}`

❌ *Disabled Users*: `{disabled_users}`

🕰 *Expired Users*: `{expired_users}`

return await bot.edit_message_text(

text,

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

)

async def delete_expired_command(call: types.CallbackQuery):

await bot.edit_message_text(

f"⚠️ Are you sure? This will *DELETE All Expired Users*‼️",

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

async def delete_limited_command(call: types.CallbackQuery):

await bot.edit_message_text(

f"⚠️ Are you sure? This will *DELETE All Limited Users*‼️",

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

async def add_data_command(call: types.CallbackQuery):

msg = await bot.edit_message_text(

f"🔋 Enter Data Limit to increase or decrease (GB):",

call.message.chat.id,

call.message.message_id,

schedule_delete_message(call.message.chat.id, call.message.id)

schedule_delete_message(call.message.chat.id, msg.id)

# return bot.register_next_step_handler(call.message, add_data_step)

if not data_limit:

raise ValueError

except ValueError:

schedule_delete_message(message.chat.id, wait_msg.message_id)

schedule_delete_message(message.chat.id, message.message_id)

msg = await bot.send_message(

message.chat.id,

f"⚠️ Are you sure? this will change Data limit of all users according to <b>"

f"{'+' if data_limit > 0 else '-'}{readable_size(abs(data_limit *1024*1024*1024))}</b>",

parse_mode="html",

await bot.delete_state(message.from_user.id, message.chat.id)

await cleanup_messages(message.chat.id)

schedule_delete_message(message.chat.id, msg.id)

async def add_time_command(call: types.CallbackQuery):

msg = await bot.edit_message_text(

f"📅 Enter Days to increase or decrease expiry:",

call.message.chat.id,

call.message.message_id,

schedule_delete_message(call.message.chat.id, call.message.id)

schedule_delete_message(call.message.chat.id, msg.id)

# return bot.register_next_step_handler(call.message, add_time_step)

if not days:

raise ValueError

except ValueError:

schedule_delete_message(message.chat.id, wait_msg.message_id)

schedule_delete_message(message.chat.id, message.message_id)

msg = await bot.send_message(

message.chat.id,

f"⚠️ Are you sure? this will change Expiry Time of all users according to <b>{days} Days</b>",

parse_mode="html",

await cleanup_messages(message.chat.id)

schedule_delete_message(message.chat.id, msg.id)

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

@bot.callback_query_handler(query_startswith("confirm_inbound"), is_admin=True)

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

@bot.callback_query_handler(query_startswith("edit:"), is_admin=True)

db_user = crud.get_user(db, username)

if not db_user:

return await bot.answer_callback_query(

)

services = crud.get_services(db)

user = UserResponse.model_validate(db_user)

await bot.edit_message_text(

f"📝 Editing user `{username}`",

call.message.chat.id,

username=username,

data_limit=db_user.data_limit,

expire_date=mem_store.get(f"{call.message.chat.id}:expire_date"),

)

async def help_edit_command(call: types.CallbackQuery):

await bot.answer_callback_query(

)

async def cancel_command(call: types.CallbackQuery):

# bot.clear_step_handler_by_chat_id(call.message.chat.id)

await bot.delete_state(call.from_user.id, call.message.chat.id)

call.message.chat.id,

call.message.message_id,

parse_mode="MarkdownV2",

)

async def edit_user_command(call: types.CallbackQuery):

_, username, action = call.data.split(":")

schedule_delete_message(call.message.chat.id, call.message.id)

if action == "data":

msg = await bot.send_message(

call.message.chat.id,

)

mem_store.set(f"{call.message.chat.id}:edit_msg_text", call.message.text)

await bot.delete_state(call.from_user.id, call.message.chat.id)

async with bot.retrieve_data(call.from_user.id, call.message.chat.id) as data:

# call.message, edit_user_data_limit_step, username)

schedule_delete_message(call.message.chat.id, msg.message_id)

elif action == "expire":

msg = await bot.send_message(

call.message.chat.id,

mem_store.set(f"{call.message.chat.id}:edit_msg_text", call.message.text)

# bot.clear_step_handler_by_chat_id(call.message.chat.id)

async with bot.retrieve_data(call.from_user.id, call.message.chat.id) as data:

# call.message, edit_user_expire_step, username=username)

schedule_delete_message(call.message.chat.id, msg.message_id)

async def edit_user_data_limit_step(message: types.Message):

try:

if float(message.text) < 0:

schedule_delete_message(message.chat.id, wait_msg.message_id)

data_limit = float(message.text) * 1024**3

except ValueError:

schedule_delete_message(message.chat.id, wait_msg.message_id)

schedule_delete_message(message.chat.id, message.message_id)

text = mem_store.get(f"{message.chat.id}:edit_msg_text")

mem_store.delete(f"{message.chat.id}:edit_msg_text")

text or f"📝 Editing user <code>{username}</code>",

parse_mode="html",

reply_markup=BotKeyboard.select_protocols(

await cleanup_messages(message.chat.id)

try:

now = datetime.now()

today = datetime(

)

expire_date = today

number = int(re.findall(number_pattern, message.text)[0])

symbol = re.findall(symbol_pattern, message.text)[0].upper()

expire_date = today + relativedelta(months=number)

expire_date = today + relativedelta(days=number)

expire_date = datetime.strptime(message.text, "%Y-%m-%d")

else:

expire_date = None

if expire_date and expire_date < today:

schedule_delete_message(message.chat.id, wait_msg.message_id)

except ValueError:

schedule_delete_message(message.chat.id, wait_msg.message_id)

schedule_delete_message(message.chat.id, message.message_id)

text = mem_store.get(f"{message.chat.id}:edit_msg_text")

mem_store.delete(f"{message.chat.id}:edit_msg_text")

text or f"📝 Editing user <code>{username}</code>",

parse_mode="html",

reply_markup=BotKeyboard.select_protocols(

await cleanup_messages(message.chat.id)

async def users_command(call: types.CallbackQuery):

with GetDB() as db:

total_pages = math.ceil(crud.get_users_count(db) / 10)

text = """👥 Users: (Page {page}/{total_pages})

✅ Active

❌ Disabled

🕰 Expired

await bot.edit_message_text(

text,

call.message.chat.id,

call.message.message_id,

parse_mode="HTML",

)

def get_user_info_text(

text = f"""\

┌─{statuses[status]} <b>Status:</b> <code>{status.title()}</code>

│ └─<b>Username:</b> <code>{username}</code>

│

"""

if note:

text += f'└─🚀 <b><a href="{sub_url}">Subscription</a>:</b> <code>{sub_url}</code>'

return text

async def edit_note_command(call: types.CallbackQuery):

with GetDB() as db:

db_user = crud.get_user(db, username)

if not db_user:

schedule_delete_message(call.message.chat.id, call.message.id)

await cleanup_messages(call.message.chat.id)

msg = await bot.send_message(

call.message.chat.id,

parse_mode="HTML",

schedule_delete_message(call.message.chat.id, msg.id)

await bot.set_state(call.from_user.id, botStates.edit_note, call.message.chat.id)

# bot.register_next_step_handler(msg, edit_note_step)

@bot.message_handler(state=botStates.edit_note)

async def edit_note_step(message: types.Message):

if len(note) > 500:

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

if not username:

await cleanup_messages(message.chat.id)

with GetDB() as db:

db_user = crud.get_user(db, username)

last_note = db_user.note

user = UserResponse.model_validate(db_user)

text = get_user_info_text(

status=user.status,

expire=user.expire,

data_limit=user.data_limit,

usage=user.used_traffic,

if TELEGRAM_LOGGER_CHANNEL_ID:

📝 <b>#Edit_Note #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{username}</code>

<b>Last Note :</b> <code>{last_note}</code>

<b>New Note :</b> <code>{note}</code>

➖➖➖➖➖➖➖➖➖

try:

except:

pass

async def user_command(call: types.CallbackQuery):

await bot.delete_state(call.from_user.id, call.message.chat.id)

# bot.clear_step_handler_by_chat_id(call.message.chat.id)

with GetDB() as db:

db_user = crud.get_user(db, username)

if not db_user:

return await bot.answer_callback_query(

)

user = UserResponse.model_validate(db_user)

text = get_user_info_text(

await bot.edit_message_text(

text,

reply_markup=BotKeyboard.user_menu(

@bot.callback_query_handler(query_startswith("revoke_sub:"), is_admin=True)

call.message.chat.id,

call.message.message_id,

parse_mode="markdown",

@bot.callback_query_handler(query_startswith("links:"), is_admin=True)

with GetDB() as db:

db_user = crud.get_user(db, username)

if not db_user:

user = UserResponse.model_validate(db_user)

call.message.chat.id,

call.message.message_id,

parse_mode="HTML",

)

with GetDB() as db:

db_user = crud.get_user(db, username)

if not db_user:

user = UserResponse.model_validate(db_user)

call.message.chat.id,

photo=f,

caption=f"<code>{link}</code>",

)

with io.BytesIO() as f:

qr = qrcode.QRCode(border=6)

sub_url=user.subscription_url,

data_limit=user.data_limit,

usage=user.used_traffic,

),

parse_mode="HTML",

)

try:

await bot.delete_message(call.message.chat.id, call.message.message_id)

call.message.chat.id,

text,

"HTML",

)

async def random_username(call: types.CallbackQuery):

# bot.clear_step_handler_by_chat_id(call.message.chat.id)

await bot.delete_state(call.from_user.id, call.message.chat.id)

template_id = int(call.data.split(":")[1] or 0)

characters = string.ascii_letters + string.digits

schedule_delete_message(call.message.chat.id, call.message.id)

await cleanup_messages(call.message.chat.id)

schedule_delete_message(call.message.chat.id, msg.id)

async with bot.retrieve_data(call.from_user.id, call.message.chat.id) as data:

data["username"] = username

async def add_user_command(call: types.CallbackQuery):

try:

await bot.delete_message(call.message.chat.id, call.message.message_id)

pass

username_msg = await bot.send_message(

call.message.chat.id,

schedule_delete_message(call.message.chat.id, username_msg.id)

# bot.register_next_step_handler(username_msg, add_user_username_step)

@bot.message_handler(state=botStates.add_user_username)

async def add_user_username_step(message: types.Message):

username = message.text

if not username:

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

with GetDB() as db:

if crud.get_user(db, username):

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

schedule_delete_message(message.chat.id, message.id)

await cleanup_messages(message.chat.id)

schedule_delete_message(message.chat.id, msg.id)

async with bot.retrieve_data(message.from_user.id, message.chat.id) as data:

# bot.register_next_step_handler(msg, add_user_data_limit_step, username=username)

async def add_user_data_limit_step(message: types.Message):

try:

if float(message.text) < 0:

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

data_limit = float(message.text) * 1024 * 1024 * 1024

except ValueError:

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

schedule_delete_message(message.chat.id, message.id)

await cleanup_messages(message.chat.id)

schedule_delete_message(message.chat.id, msg.id)

async with bot.retrieve_data(message.from_user.id, message.chat.id) as data:

data["data_limit"] = data_limit

# bot.register_next_step_handler(msg, add_user_expire_step, username=username, data_limit=data_limit)

@bot.message_handler(state=botStates.add_user_expire)

async def add_user_expire_step(message: types.Message):

try:

now = datetime.now()

today = datetime(

)

expire_date = today

number = int(re.findall(number_pattern, message.text)[0])

symbol = re.findall(symbol_pattern, message.text)[0].upper()

expire_date = today + relativedelta(months=number)

expire_date = today + relativedelta(days=number)

expire_date = datetime.strptime(message.text, "%Y-%m-%d")

else:

expire_date = None

if expire_date and expire_date < today:

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

except ValueError:

schedule_delete_message(message.chat.id, wait_msg.id)

schedule_delete_message(message.chat.id, message.id)

async with bot.retrieve_data(message.from_user.id, message.chat.id) as data:

username, data_limit = data["username"], data["data_limit"]

with GetDB() as db:

services = crud.get_services(db)

await cleanup_messages(message.chat.id)

await bot.send_message(

message.chat.id,

),

)

async def switch_service(call: types.CallbackQuery):

service = int(service)

with GetDB() as db:

else:

selected_services.append(service)

if action in ["edit"]:

return await bot.edit_message_text(

"edit",

username=username,

data_limit=mem_store.get(f"{call.message.chat.id}:data_limit"),

)

await bot.edit_message_text(

call.message.text,

call.message.chat.id,

call.message.message_id,

)

async def confirm_user_command(call: types.CallbackQuery):

chat_id = call.from_user.id

full_name = call.from_user.full_name

now = datetime.now()

today = datetime(

@bot.callback_query_handler(query_startswith("do_delete"), is_admin=True)

async def delete_user(call: types.CallbackQuery):

with GetDB() as db:

db_user = crud.get_user(db, username)

crud.remove_user(db, db_user)

await xray.operations.remove_user(db_user)

await bot.edit_message_text(

call.message.chat.id,

call.message.message_id,

)

if TELEGRAM_LOGGER_CHANNEL_ID:

🗑 <b>#Deleted #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{db_user.username}</code>

<b>Expire Date :</b> <code>\

{datetime.fromtimestamp(db_user.expire).strftime('%H:%M:%S %Y-%m-%d') if db_user.expire else "Never"}</code>

➖➖➖➖➖➖➖➖➖

try:

except:

pass

username = call.data.split(":")[1]

with GetDB() as db:

db_user = crud.get_user(db, username)

await xray.operations.remove_user(db_user)

user = UserResponse.model_validate(db_user)

await bot.edit_message_text(

get_user_info_text(

username=username,

sub_url=user.subscription_url,

data_limit=db_user.data_limit,

usage=db_user.used_traffic,

expire=db_user.expire,

),

call.message.chat.id,

call.message.message_id,

if TELEGRAM_LOGGER_CHANNEL_ID:

❌ <b>#Disabled #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username</b> : <code>{username}</code>

➖➖➖➖➖➖➖➖➖

try:

except:

pass

@bot.callback_query_handler(query_startswith("do_activate"), is_admin=True)

async def activate_user(call: types.CallbackQuery):

username = call.data.split(":")[1]

with GetDB() as db:

db_user = crud.get_user(db, username)

await xray.operations.add_user(db_user)

user = UserResponse.model_validate(db_user)

await bot.edit_message_text(

get_user_info_text(

username=username,

sub_url=user.subscription_url,

data_limit=db_user.data_limit,

usage=db_user.used_traffic,

expire=db_user.expire,

),

call.message.chat.id,

call.message.message_id,

if TELEGRAM_LOGGER_CHANNEL_ID:

✅ <b>#Activated #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username</b> : <code>{username}</code>

➖➖➖➖➖➖➖➖➖

try:

except:

pass

@bot.callback_query_handler(query_startswith("do_reset_usage"), is_admin=True)

async def reset_all_usages(call: types.CallbackQuery):

username = call.data.split(":")[1]

data_limit=user.data_limit,

usage=user.used_traffic,

expire=user.expire,

),

call.message.chat.id,

call.message.message_id,

if TELEGRAM_LOGGER_CHANNEL_ID:

🔁 <b>#Reset_usage #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username</b> : <code>{username}</code>

➖➖➖➖➖➖➖➖➖

try:

except:

pass

@bot.callback_query_handler(query_equals("do_restart"), is_admin=True)

async def restart_xray(call: types.CallbackQuery):

m = await bot.edit_message_text(

await xray.core.restart(xray.config.include_db_users())

for node_id, node in list(xray.nodes.items()):

if await node.is_healthy():

await xray.operations.restart_node(node_id, xray.config.include_db_users())

await bot.edit_message_text(

)

@bot.callback_query_handler(query_equals("edit_user_done"), is_admin=True)

async def edit_user(call: types.CallbackQuery):

try:

except Exception:

pass

return await bot.send_message(

call.message.chat.id,

)

return await bot.answer_callback_query(

)

# k: v for k, v in mem_store.get(f'{call.message.chat.id}:protocols').items() if v}

services: List[int] = mem_store.get(f"{call.message.chat.id}:services")

with GetDB() as db:

db_user = crud.get_user(db, username)

if not db_user:

# proxies = {p.type.value: p.settings for p in db_user.proxies}

modify = UserModify(

username=username,

# status=db_user.status,

data_limit=mem_store.get(f"{call.message.chat.id}:data_limit"),

last_user = UserResponse.model_validate(db_user)

db_user = crud.update_user(db, db_user, modify)

data_limit=user.data_limit,

usage=user.used_traffic,

expire=user.expire,

)

await bot.edit_message_text(

text,

call.message.chat.id,

call.message.message_id,

parse_mode="HTML",

)

if TELEGRAM_LOGGER_CHANNEL_ID:

tag = f'\n➖➖➖➖➖➖➖➖➖ \n<b>By :</b> <a href="tg://user?id={call.from_user.id}">{call.from_user.full_name}</a>'

if last_user.data_limit != user.data_limit:

📶 <b>#Traffic_Change #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{user.username}</code>

<b>Last Traffic Limit :</b> <code>{readable_size(last_user.data_limit) if last_user.data_limit else "Unlimited"}</code>

try:

except:

pass

if last_user.expire != user.expire:

📅 <b>#Expiry_Change #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{user.username}</code>

<b>Last Expire Date :</b> <code>\

{datetime.fromtimestamp(last_user.expire).strftime('%H:%M:%S %Y-%m-%d') if last_user.expire else "Never"}</code>

<b>New Expire Date :</b> <code>\

try:

except:

pass

if list(last_user.inbounds.values())[0] != list(user.inbounds.values())[0]:

⚙️ <b>#Inbounds_Change #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{user.username}</code>

<b>Last Proxies :</b> <code>{", ".join(list(last_user.inbounds.values())[0])}</code>

try:

except:

pass

@bot.callback_query_handler(query_startswith("add_user_done"), is_admin=True)

async def delete_limpired(call: types.CallbackQuery):

try:

except Exception:

pass

return await bot.send_message(

call.message.chat.id,

)

return await bot.answer_callback_query(

)

new_user = UserCreate(

try:

with GetDB() as db:

db_user = crud.create_user(db, new_user)

except sqlalchemy.exc.IntegrityError:

db.rollback()

return await bot.answer_callback_query(

)

await xray.operations.add_user(db_user)

data_limit=user.data_limit,

usage=user.used_traffic,

expire=user.expire,

)

await bot.edit_message_text(

text,

call.message.chat.id,

call.message.message_id,

parse_mode="HTML",

if TELEGRAM_LOGGER_CHANNEL_ID:

🆕 <b>#Created #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{user.username}</code>

{datetime.fromtimestamp(user.expire).strftime('%H:%M:%S %Y-%m-%d') if user.expire else "Never"}</code>

<b>Services :</b> <code>{"" if not services else ", ".join([s.name for s in services])}</code>

➖➖➖➖➖➖➖➖➖

try:

except:

pass

async def delete_limpired(call: types.CallbackQuery):

await bot.edit_message_text(

call.message.chat.id,

call.message.message_id,

with GetDB() as db:

depleted_users = crud.get_users(

deleted = 0

for user in depleted_users:

try:

crud.remove_user(db, user)

await xray.operations.remove_user(user)

\t{datetime.fromtimestamp(user.expire) if user.expire else "never"}\

\t{readable_size(user.used_traffic) if user.used_traffic else 0}\

/{readable_size(user.data_limit) if user.data_limit else "Unlimited"}\

except:

db.rollback()

await bot.edit_message_text(

call.message.chat.id,

call.message.message_id,

parse_mode="HTML",

if TELEGRAM_LOGGER_CHANNEL_ID:

🗑 <b>#Delete #{data[7:].title()} #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Count:</b> <code>{deleted}</code>

➖➖➖➖➖➖➖➖➖

try:

os.remove(file_name)

except:

pass

async def add_data_to_all(call: types.CallbackQuery):

schedule_delete_message(

call.message.chat.id,

data_limit = float(call.data.split(":")[1]) * 1024 * 1024 * 1024

with GetDB() as db:

users = crud.get_users(db)

counter = 0

for user in users:

try:

counter += 1

f.write(

f'{user.username}\

\t{datetime.fromtimestamp(user.expire) if user.expire else "never"}\

\t{readable_size(user.used_traffic) if user.used_traffic else 0}\

/{readable_size(user.data_limit) if user.data_limit else "Unlimited"}\

except:

db.rollback()

await cleanup_messages(chat_id)

await bot.send_message(

chat_id,

f'✅ <b>{counter}/{len(users)} Users</b> Data Limit according to <code>{"+" if data_limit > 0 else "-"}{readable_size(abs(data_limit))}</code>',

if TELEGRAM_LOGGER_CHANNEL_ID:

📶 <b>#Traffic_Change #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>According to:</b> <code>{"+" if data_limit > 0 else "-"}{readable_size(abs(data_limit))}</code>

<b>Count:</b> <code>{counter}</code>

➖➖➖➖➖➖➖➖➖

try:

os.remove(file_name)

except:

pass

async def add_time_to_all(call: types.CallbackQuery):

schedule_delete_message(

call.message.chat.id,

days = int(call.data.split(":")[1])

with GetDB() as db:

users = crud.get_users(db)

counter = 0

for user in users:

try:

user = crud.update_user(

UserModify(

username=user.username,

expire=int(

counter += 1

f.write(

f'{user.username}\

\t{datetime.fromtimestamp(user.expire) if user.expire else "never"}\

\t{readable_size(user.used_traffic) if user.used_traffic else 0}\

/{readable_size(user.data_limit) if user.data_limit else "Unlimited"}\

except:

db.rollback()

await cleanup_messages(chat_id)

await bot.send_message(

chat_id,

if TELEGRAM_LOGGER_CHANNEL_ID:

📅 <b>#Expiry_Change #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>According to:</b> <code>{days} Days</code>

<b>Count:</b> <code>{counter}</code>

➖➖➖➖➖➖➖➖➖

try:

os.remove(file_name)

except:

pass

async def revoke_sub(call: types.CallbackQuery):

username = call.data.split(":")[1]

with GetDB() as db:

db_user = crud.get_user(db, username)

if not db_user:

db_user = crud.revoke_user_sub(db, db_user)

user = UserResponse.model_validate(db_user)

text = get_user_info_text(

expire=user.expire,

data_limit=user.data_limit,

usage=user.used_traffic,

await bot.edit_message_text(

if TELEGRAM_LOGGER_CHANNEL_ID:

🚫 <b>#Revoke_sub #From_Bot</b>

➖➖➖➖➖➖➖➖➖

<b>Username:</b> <code>{username}</code>

➖➖➖➖➖➖➖➖➖

try:

except:

pass

with GetDB() as db:

db_user = crud.get_user(db, message.text)

if not db_user:

user = UserResponse.model_validate(db_user)

text = get_user_info_text(

status=user.status,

expire=user.expire,

data_limit=user.data_limit,

usage=user.used_traffic,

if bot and (TELEGRAM_ADMIN_ID or TELEGRAM_LOGGER_CHANNEL_ID):

try:

if TELEGRAM_LOGGER_CHANNEL_ID:

else:

for admin in TELEGRAM_ADMIN_ID:

except ApiTelegramException as e:

logger.error(e)

🆕 <b>#Created</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{username}</code>

<b>Expire Date :</b> <code>{expire_date}</code>

<b>Services :</b> <code>{services}</code>

➖➖➖➖➖➖➖➖➖

by=escape_html(by),

username=escape_html(username),

data_limit=readable_size(data_limit) if data_limit else "Unlimited",

)

return await report(

text,

)

✏️ <b>#Modified</b>

➖➖➖➖➖➖➖➖➖

<b>Username :</b> <code>{username}</code>

<b>Services :</b> <code>{services}</code>

➖➖➖➖➖➖➖➖➖

<b>By :</b> <b>#{by}</b>\

by=escape_html(by),

username=escape_html(username),

data_limit=readable_size(data_limit) if data_limit else "Unlimited",

)

async def report_user_deletion(username: str, by: str):

🗑 <b>#Deleted</b>

➖➖➖➖➖➖➖➖➖

<b>Username</b> : <code>{username}</code>

➖➖➖➖➖➖➖➖➖

<b>By</b> : <b>#{by}</b>\

)

return await report(text)

async def report_status_change(username: str, status: str):

_status = {

}

{status}

➖➖➖➖➖➖➖➖➖

<b>Username</b> : <code>{username}</code>\

)

return await report(text)

➖➖➖➖➖➖➖➖➖

<b>By</b> : <b>#{by}</b>\

""".format(

)

return await report(text)

➖➖➖➖➖➖➖➖➖

<b>By</b> : <b>#{by}</b>\

""".format(

)

return await report(text)

from app.db import GetDB, crud

from app.models.user import UserResponse

from app.telegram import bot

# from pytz import UTC

from telebot.custom_filters import ChatFilter

from telebot.util import extract_arguments

bot.add_custom_filter(ChatFilter())

def usage_command(message):

username = extract_arguments(message.text)

if not username:

with GetDB() as db:

dbuser = crud.get_user(db, username)

return bot.reply_to(message, "No user found with this username")

user = UserResponse.model_validate(dbuser)

┌─{statuses[user.status]} <b>Status:</b> <code>{user.status.title()}</code>

│ └─<b>Username:</b> <code>{user.username}</code>

│

│ └─<b>Data Used:</b> <code>{readable_size(user.used_traffic) if user.used_traffic else "-"}</code>

│

└─📅 <b>Expiry Date:</b> <code>{datetime.fromtimestamp(user.expire).date() if user.expire else 'Never'}</code>

class IsAdminFilter(AdvancedCustomFilter):

async def check(self, message, text):

"""

def main_menu():

keyboard = types.InlineKeyboardMarkup()

keyboard.add(

keyboard.add(

# types.InlineKeyboardButton(text='➕ Create User From Template', callback_data='template_add_user'))

keyboard.add(

return keyboard

@staticmethod

def edit_all_menu():

keyboard = types.InlineKeyboardMarkup()

keyboard.add(

keyboard.add(

# types.InlineKeyboardButton(text='➕ Add Inbound', callback_data='inbound_add'),

# types.InlineKeyboardButton(text='➖ Remove Inbound', callback_data='inbound_remove'))

return keyboard

# keyboard = types.InlineKeyboardMarkup()

# for inbound in inbounds:

# keyboard.add(types.InlineKeyboardButton(text=inbound, callback_data=f'confirm_{action}:{inbound}'))

# keyboard.add(types.InlineKeyboardButton(text='🔙 Back', callback_data='cancel'))

# return keyboard

@staticmethod

keyboard = types.InlineKeyboardMarkup()

return keyboard

@staticmethod

def user_menu(user_info, with_back: bool = True, page: int = 1, note: bool = False):

keyboard = types.InlineKeyboardMarkup()

keyboard.add(

types.InlineKeyboardButton(

),

types.InlineKeyboardButton(

),

)

if note:

keyboard.add(

types.InlineKeyboardButton(

types.InlineKeyboardButton(

keyboard.add(

types.InlineKeyboardButton(

types.InlineKeyboardButton(

else:

keyboard.add(

types.InlineKeyboardButton(

types.InlineKeyboardButton(

keyboard.add(

types.InlineKeyboardButton(

),

types.InlineKeyboardButton(

)

if with_back:

keyboard.add(

types.InlineKeyboardButton(

)

)

return keyboard

keyboard.add(

types.InlineKeyboardButton(

)

)

keyboard.add(

)

return keyboard

@staticmethod

def subscription_page(sub_url: str):

keyboard = types.InlineKeyboardMarkup()

return keyboard

@staticmethod

def confirm_action(action: str, username: str = None):

keyboard = types.InlineKeyboardMarkup()

keyboard.add(

types.InlineKeyboardButton(

),

)

return keyboard

def inline_cancel_action(callback_data: str = "cancel"):

keyboard = types.InlineKeyboardMarkup()

keyboard.add(

)

return keyboard

keyboard = types.InlineKeyboardMarkup()

if len(users) >= 2:

users = [p for p in users]

else:

users = [users]

for user in users:

row = []

for p in user:

status = {

}

keyboard.row(*row)

# if there is more than one page

if total_pages > 1:

if page > 1:

keyboard.add(

types.InlineKeyboardButton(

)

)

if page < total_pages:

keyboard.add(

types.InlineKeyboardButton(

)

)

return keyboard

@staticmethod

keyboard = types.InlineKeyboardMarkup()

if action == "edit":

keyboard.add(

types.InlineKeyboardButton(

)

)

keyboard.add(

types.InlineKeyboardButton(

text=f"{readable_size(data_limit) if data_limit else 'Unlimited'}",

),

types.InlineKeyboardButton(

)

keyboard.add(

types.InlineKeyboardButton(

)

)

keyboard.add(

types.InlineKeyboardButton(

text=f"{expire_date.strftime('%Y-%m-%d') if expire_date else 'Never'}",

),

types.InlineKeyboardButton(

)

for s in services:

keyboard.add(

keyboard.add(

types.InlineKeyboardButton(

)

)

keyboard.add(

types.InlineKeyboardButton(

)

)

env = jinja2.Environment(loader=jinja2.FileSystemLoader(template_directories))

env.filters.update(CUSTOM_FILTERS)

def render_template(template: str, context: Union[dict, None] = None) -> str:

def datetimeformat(dt):

if isinstance(dt, int):

dt = datetime.fromtimestamp(dt)

return formatted_datetime

"except": exclude_keys,

"only": only_keys,

"datetime": datetimeformat,

}

cert = crypto.X509()

cert.get_subject().CN = "Gozargah"

cert.gmtime_adj_notBefore(0)

cert.set_issuer(cert.get_subject())

cert.set_pubkey(k)

cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode("utf-8")

key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, k).decode("utf-8")

def create_secure_context(

) -> ssl.SSLContext:

ctx = ssl.create_default_context(cafile=trusted)

ctx.load_cert_chain(client_cert, client_key)

ctx.check_hostname = False

ctx.verify_mode = ssl.CERT_NONE

return ctx

def create_admin_token(username: str, is_sudo=False) -> str:

if JWT_ACCESS_TOKEN_EXPIRE_MINUTES > 0:

expire = datetime.utcnow() + timedelta(minutes=JWT_ACCESS_TOKEN_EXPIRE_MINUTES)

data["exp"] = expire

payload = jwt.decode(token, get_secret_key(), algorithms=["HS256"])

username: str = payload.get("sub")

access: str = payload.get("access")

return

try:

except KeyError:

created_at = None

except JWTError:

return

import uuid

from app.models.proxy import ProxyTypes

def gen_uuid(key: str) -> str:

return str(uuid.UUID(bytes=xxhash.xxh128(key.encode()).digest()))

def gen_password(key: str) -> str:

return xxhash.xxh128(key.encode()).hexdigest()

def generate_settings(key: str, protocol: ProxyTypes) -> dict:

if protocol in (ProxyTypes.Trojan, ProxyTypes.Shadowsocks):

elif protocol in (ProxyTypes.VMess, ProxyTypes.VLESS):

return ProxyTypes(protocol).settings_model(id=gen_uuid(key)).dict(no_obj=True)

from app.db.models import UserStatus

from app.models.admin import Admin

from app.models.user import ReminderType, UserResponse

try:

await telegram.report_status_change(username, status)

except Exception:

pass

if status == UserStatus.limited:

elif status == UserStatus.expired:

elif status == UserStatus.disabled:

elif status == UserStatus.active:

async def user_created(user: UserResponse, user_id: int, by: Admin) -> None:

expire_date=user.expire,

data_limit=user.data_limit,

# proxies=[]

)

async def user_updated(user: UserResponse, by: Admin) -> None:

)

except Exception:

pass

async def user_deleted(username: str, by: Admin) -> None:

await telegram.report_user_deletion(username=username, by=by.username)

except Exception:

pass

async def user_data_usage_reset(user: UserResponse, by: Admin) -> None:

)

except Exception:

pass

async def user_subscription_revoked(user: UserResponse, by: Admin) -> None:

)

except Exception:

pass

async def data_usage_percent_reached(

await notify(ReachedDaysLeft(username=user.username, user=user, days_left=days))

await create_notification_reminder(

import yaml

from jdatetime import date as jd

from app.templates import render_template

from app.utils.system import get_public_ip, readable_size

from app.utils.keygen import generate_settings

"connect_to_start": "🔌",

}

class V2rayShareLink(str):

@classmethod

def vmess(

spx="",

ais="",

):

if flow and (net in ("tcp", "kcp") and type != "http"):

payload["flow"] = flow

spx="",

ais="",

):

if flow and (net in ("tcp", "kcp") and type != "http"):

payload["flow"] = flow

sni=sni,

host=host,

path=path,

udp=udp,

alpn=alpn,

ais=ais,

def generate_clash_subscription(

) -> str:

if is_meta is True:

conf = ClashMetaConfiguration()

)

conf = SingBoxConfiguration()

format_variables = setup_format_variables(extra_data)

return base64.b64encode("\n".join(links).encode()).decode()

conf = OutlineConfiguration()

format_variables = setup_format_variables(extra_data)

result.append(f"{seconds}s")

return " ".join(result)

from app.models.user import UserStatus

def setup_format_variables(extra_data: dict) -> dict:

user_status = extra_data.get("status")

expire_timestamp = extra_data.get("expire")

) -> Union[List, str]:

salt = secrets.token_hex(8)

results = []

for inb in inbounds:

settings = generate_settings(ukey, protocol)

format_variables.update({"PROTOCOL": protocol.name})

req_host = random.choice(req_host_list).replace("*", salt)

else:

req_host = ""

host_inbound.update(

{

"port": host.port or inbound["port"],

"host": req_host,

"tls": inbound["tls"] if host_tls is None else host_tls,

"alpn": host.alpn.value or inbound.get("alpn", ""),

"fp": host.fingerprint.value or inbound.get("fp", ""),

}

)

results.append(

get_v2ray_link(

remark=host.remark.format_map(format_variables),

inbound=host_inbound,

settings=settings,

)

rt_bw = RealtimeBandwidth(

def realtime_bandwidth() -> RealtimeBandwidthStat:

return RealtimeBandwidthStat(

def random_password() -> str:

def check_port(port: int) -> bool:

s = socket.socket()

try:

return True

except socket.error:

return False

def get_public_ip():

try:

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

return sock.getsockname()[0]

except (socket.error, IndexError):

pass

finally:

sock.close()

def readable_size(size_bytes):

i = int(math.floor(math.log(size_bytes, 1024)))

p = math.pow(1024, i)

s = round(size_bytes / p, 2)

def list_admins(

offset: Optional[int] = typer.Option(None, *utils.FLAGS["offset"]),

limit: Optional[int] = typer.Option(None, *utils.FLAGS["limit"]),

):

"""Displays a table of admins"""

with GetDB() as db:

utils.print_table(

table=Table("Username", "Is sudo", "Created at"),

rows=[

for admin in admins

)

@app.command(name="delete")

def delete_admin(

username: str = typer.Option(..., *utils.FLAGS["username"], prompt=True),

):

"""

Deletes the specified admin

with GetDB() as db:

admin: Union[Admin, None] = crud.get_admin(db, username=username)

if not admin:

crud.remove_admin(db, admin)

utils.success(f'"{username}" deleted successfully.')

else:

def create_admin(

username: str = typer.Option(..., *utils.FLAGS["username"], prompt=True),

is_sudo: bool = typer.Option(None, *utils.FLAGS["is_sudo"], prompt=True),

):

"""

Creates an admin

"""

with GetDB() as db:

try:

utils.success(f'Admin "{username}" created successfully.')

except IntegrityError:

utils.error(f'Admin "{username}" already exists!')

@app.command(name="update")

"""

Updates the specified admin

def _get_modify_model(admin: Admin):

Console().print(

)

is_sudo: bool = typer.confirm("Is sudo", default=admin.is_sudo)

return AdminPartialModify(

is_sudo=is_sudo,

with GetDB() as db:

admin: Union[Admin, None] = crud.get_admin(db, username=username)

if not admin:

crud.partial_update_admin(db, admin, _get_modify_model(admin))

utils.success(f'Admin "{username}" updated successfully.')

@app.command(name="import-from-env")

"""

Imports the sudo admin from env

)

if not (username and password):

with GetDB() as db:

admin: Union[None, Admin] = None

# If env admin already exists

if current_admin := crud.get_admin(db, username=username):

if not yes_to_all and not typer.confirm(

):

utils.error("Aborted.")

admin = crud.partial_update_admin(

)

# If env admin does not exist yet

else:

db.commit()

utils.success(

f'Admin "{username}" imported successfully.\n'

f"{updated_user_count} users' admin_id set to the {username}'s id.\n"

)

@app.command(name="get-link")

"""

Prints the given user's subscription link.

@app.command(name="get-config")

def get_config(

username: str = typer.Option(..., *utils.FLAGS["username"], prompt=True),

output_file: Optional[str] = typer.Option(

),

as_base64: bool = typer.Option(

):

"""

Generates a subscription config.

)

else:

utils.success(

f' using pager for {username}\'s config in "{config_format}" format.',

)

utils.paginate(conf)

def list_users(

offset: Optional[int] = typer.Option(None, *utils.FLAGS["offset"]),

limit: Optional[int] = typer.Option(None, *utils.FLAGS["limit"]),

status: Optional[crud.UserStatus] = typer.Option(None, *utils.FLAGS["status"]),

):

"""

Displays a table of users

"""

with GetDB() as db:

users: list[User] = crud.get_users(

)

utils.print_table(

table=Table(

),

rows=[

(

readable_size(user.data_limit) if user.data_limit else "Unlimited",

user.data_limit_reset_strategy.value,

utils.readable_datetime(user.expire, include_time=False),

)

for user in users

)

@app.command(name="set-owner")

def set_owner(

username: str = typer.Option(None, *utils.FLAGS["username"], prompt=True),

):

"""

Transfers user's ownership

"""

with GetDB() as db:

user: User = utils.raise_if_falsy(

dbadmin = utils.raise_if_falsy(

# Ask for confirmation if user already has an owner

):

utils.error("Aborted.")

def print_table(

):

for row in rows:

table.add_row(*row)

def readable_datetime(

date_time: Union[datetime, int, None],

include_date: bool = True,

):

def get_datetime_format():

dt_format = ""

load_dotenv()

UVICORN_HOST = config("UVICORN_HOST", default="0.0.0.0")

DEBUG = config("DEBUG", default=False, cast=bool)

DOCS = config("DOCS", default=False, cast=bool)

else config("VITE_BASE_API", default="/api/")

)

TELEGRAM_API_TOKEN = config("TELEGRAM_API_TOKEN", default="")

TELEGRAM_ADMIN_ID = config(

default="",

)

TELEGRAM_PROXY_URL = config("TELEGRAM_PROXY_URL", default="")

TELEGRAM_LOGGER_CHANNEL_ID = config("TELEGRAM_LOGGER_CHANNEL_ID", cast=int, default=0)

TELEGRAM_DEFAULT_VLESS_FLOW = config("TELEGRAM_DEFAULT_VLESS_FLOW", default="")

CUSTOM_TEMPLATES_DIRECTORY = config("CUSTOM_TEMPLATES_DIRECTORY", default=None)

HOME_PAGE_TEMPLATE = config("HOME_PAGE_TEMPLATE", default="home/index.html")

SINGBOX_SUBSCRIPTION_TEMPLATE = config(

)

# USERNAME: PASSWORD

else {}

WEBHOOK_ADDRESS = config("WEBHOOK_ADDRESS", default=None)

# recurrent notifications

# timeout between each retry of sending a notification in seconds

# how many times to try after ok response not recevied after sending a notifications

# sends a notification when the user uses this much of thier data

# sends a notification when there is n days left of their service

NOTIFY_DAYS_LEFT = config("NOTIFY_DAYS_LEFT", default=3, cast=int)

# headers: profile-update-interval, support-url, profile-title

SUB_UPDATE_INTERVAL = config("SUB_UPDATE_INTERVAL", default="12")

SUB_SUPPORT_URL = config("SUB_SUPPORT_URL", default="https://t.me/")

\ No newline at end of file