add integrated telegram bot

to report changes in users
and some early commands

.env.example

@@ -15,4 +15,7 @@ XRAY_ASSETS_PATH = "/usr/local/share/xray"
 # Bridge@192.168.1.10
 # "
 
-# XRAY_SUBSCRIPTION_URL_PREFIX = "http://example.com"
+# XRAY_SUBSCRIPTION_URL_PREFIX = "http://example.com"
+
+# TELEGRAM_API_TOKEN = 123456789:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
+# TELEGRAM_ADMIN_ID = 987654321

app/__init__.py

@@ -1,3 +1,4 @@
+from datetime import timezone
 import logging
 
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -14,7 +15,7 @@
     redoc_url='/redoc' if DOCS else None
 )
 app.openapi = custom_openapi(app)
-scheduler = BackgroundScheduler({'apscheduler.job_defaults.max_instances': 5})
+scheduler = BackgroundScheduler({'apscheduler.job_defaults.max_instances': 5}, timezone='UTC')
 logger = logging.getLogger('uvicorn.error')
 app.add_middleware(
     CORSMiddleware,
@@ -25,7 +26,7 @@
 )
 
 
-from app import dashboard, jobs, views  # noqa
+from app import dashboard, jobs, views, telegram  # noqa
 
 
 @app.on_event("startup")

app/jobs/review_users.py

@@ -35,6 +35,10 @@ def review():
 
             update_user_status(db, user, status)
 
+            try:
+                telegram.report_status_change(user.username, status)
+            except Exception:
+                pass
             logger.info(f"User \"{user.username}\" status changed to {status}")
 
 

app/models/admin.py

@@ -33,7 +33,6 @@ def get_current(cls,
         )
 
         payload = get_admin_payload(token)
-        print(payload)
         if not payload:
             raise exc
 

app/telegram/__init__.py

@@ -0,0 +1,46 @@
+import glob
+import importlib.util
+from os.path import basename, dirname, join
+from threading import Thread
+from config import TELEGRAM_API_TOKEN, TELEGRAM_PROXY_URL
+from app import app
+from telebot import TeleBot, apihelper
+
+
+bot = None
+if TELEGRAM_API_TOKEN:
+    apihelper.proxy = {'http': TELEGRAM_PROXY_URL, 'https': TELEGRAM_PROXY_URL}
+    bot = TeleBot(TELEGRAM_API_TOKEN)
+
+
+@app.on_event("startup")
+def start_bot():
+    if bot:
+        handler = glob.glob(join(dirname(__file__), "*.py"))
+        for file in handler:
+            name = basename(file).replace('.py', '')
+            if name.startswith('_'):
+                continue
+            spec = importlib.util.spec_from_file_location(name, file)
+            spec.loader.exec_module(importlib.util.module_from_spec(spec))
+
+        thread = Thread(target=bot.infinity_polling, daemon=True)
+        thread.start()
+
+
+from .report import (  # noqa
+    report,
+    report_new_user,
+    report_user_modification,
+    report_user_deletion,
+    report_status_change
+)
+
+__all__ = [
+    "bot",
+    "report",
+    "report_new_user",
+    "report_user_modification",
+    "report_user_deletion",
+    "report_status_change"
+]

app/telegram/admin.py

@@ -0,0 +1,84 @@
+from app import xray
+import math
+from app.db import GetDB, crud
+from app.models.user import UserStatus
+from app.telegram import bot
+from app.utils.system import cpu_usage, memory_usage
+from config import TELEGRAM_ADMIN_ID
+from telebot.custom_filters import ChatFilter
+
+bot.add_custom_filter(ChatFilter())
+
+
+commands_text = """🚀 Marzban's bot commands:
+
+/system
+_\- Get system info_
+
+/restart
+_\- Restart Xray core_
+"""
+
+system_text = """⚙️ System statistics
+
+*CPU Cores*: `{cpu_cores}`
+*CPU Usage*: `{cpu_percent}%`
+
+*Total Memory*: `{total_memory}`
+*In Use Memory*: `{used_memory}`
+*Free Memory*: `{free_memory}`
+
+*Total Bandwidth Usage*: `{total_bandwidth}`
+*Upload Bandwidth Usage*: `{up_bandwidth}`
+*Download Bandwidth Usage*: `{down_bandwidth}`
+
+*Total Users*: `{total_users}`
+*Active Users*: `{active_users}`
+*Deactive Users*: `{deactive_users}`
+"""
+
+
+def readable_size(size_bytes):
+    if size_bytes == 0:
+        return "0B"
+    size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
+    i = int(math.floor(math.log(size_bytes, 1024)))
+    p = math.pow(1024, i)
+    s = round(size_bytes / p, 2)
+    return f'{s} {size_name[i]}'
+
+
+@bot.message_handler(commands=['start', 'help'], chat_id=[TELEGRAM_ADMIN_ID])
+def help_command(message):
+    return bot.reply_to(message, commands_text, parse_mode="MarkdownV2")
+
+
+@bot.message_handler(commands=['system'], chat_id=[TELEGRAM_ADMIN_ID])
+def system_command(message):
+    mem = memory_usage()
+    cpu = cpu_usage()
+    with GetDB() as db:
+        bandwidth = crud.get_system_usage(db)
+        total_users = crud.get_users_count(db)
+        users_active = crud.get_users_count(db, UserStatus.active)
+
+    text = system_text.format(cpu_cores=cpu.cores,
+                              cpu_percent=cpu.percent,
+                              total_memory=readable_size(mem.total),
+                              used_memory=readable_size(mem.used),
+                              free_memory=readable_size(mem.free),
+                              total_bandwidth=readable_size(bandwidth.uplink + bandwidth.downlink),
+                              up_bandwidth=readable_size(bandwidth.uplink),
+                              down_bandwidth=readable_size(bandwidth.downlink),
+                              total_users=total_users,
+                              active_users=users_active,
+                              deactive_users=total_users - users_active)
+
+    return bot.reply_to(message, text, parse_mode='MarkdownV2')
+
+
+@bot.message_handler(commands=['restart'], chat_id=[TELEGRAM_ADMIN_ID])
+def restart_command(message):
+    m = bot.reply_to(message, '🔄 Restarting...')
+    xray.core.restart()
+    bot.edit_message_text('✅ Xray core restarted.', m.chat.id, m.message_id)

app/telegram/report.py

@@ -0,0 +1,29 @@
+from app import logger
+from app.telegram import bot
+from telebot.apihelper import ApiTelegramException
+from config import TELEGRAM_ADMIN_ID
+from telebot.formatting import escape_markdown
+
+
+def report(message: str, parse_mode="MarkdownV2"):
+    if bot and TELEGRAM_ADMIN_ID:
+        try:
+            bot.send_message(TELEGRAM_ADMIN_ID, message, parse_mode=parse_mode)
+        except ApiTelegramException as e:
+            logger.error(e)
+
+
+def report_new_user(username: str, by: str):
+    return report(f"New user *{escape_markdown(username)}* added by *{escape_markdown(by)}*")
+
+
+def report_user_modification(username: str, by: str):
+    return report(f"User *{escape_markdown(username)}* modified by *{escape_markdown(by)}*")
+
+
+def report_user_deletion(username: str, by: str):
+    return report(f"User *{escape_markdown(username)}* deleted by *{escape_markdown(by)}*")
+
+
+def report_status_change(username: str, status: str):
+    return report(f"User *{escape_markdown(username)}*'s status has changed to _{status}_")

app/utils/system.py

@@ -12,9 +12,19 @@ class MemoryStat():
     free: int
 
 
+@dataclass
+class CPUStat():
+    cores: int
+    percent: int
+
+
+def cpu_usage() -> CPUStat:
+    return CPUStat(cores=psutil.cpu_count(), percent=psutil.cpu_percent())
+
+
 def memory_usage() -> MemoryStat:
     mem = psutil.virtual_memory()
-    return MemoryStat(total=mem.total, used=mem.used, free=mem.free)
+    return MemoryStat(total=mem.total, used=mem.used, free=mem.available)
 
 
 def random_password() -> str:

app/views/user.py

@@ -2,17 +2,18 @@
 from typing import List
 
 import sqlalchemy
-from app import app, logger, xray
+from app import app, logger, telegram, xray
 from app.db import Session, crud, get_db
 from app.models.admin import Admin
 from app.models.proxy import ProxyTypes
 from app.models.user import UserCreate, UserModify, UserResponse, UserStatus
 from app.xray import INBOUNDS
-from fastapi import Depends, HTTPException
+from fastapi import BackgroundTasks, Depends, HTTPException
 
 
 @app.post("/api/user", tags=['User'], response_model=UserResponse)
 def add_user(new_user: UserCreate,
+             bg: BackgroundTasks,
              db: Session = Depends(get_db),
              admin: Admin = Depends(Admin.get_current)):
     """
@@ -47,11 +48,16 @@ def add_user(new_user: UserCreate,
             except xray.exc.EmailExistsError:
                 pass
 
+    bg.add_task(
+        telegram.report_new_user,
+        username=dbuser.username,
+        by=admin.username
+    )
     logger.info(f"New user \"{dbuser.username}\" added")
     return dbuser
 
 
-@app.get("/api/user/{username}", tags=['User'], response_model=UserResponse)
+@ app.get("/api/user/{username}", tags=['User'], response_model=UserResponse)
 def get_user(username: str,
              db: Session = Depends(get_db),
              admin: Admin = Depends(Admin.get_current)):
@@ -68,9 +74,10 @@ def get_user(username: str,
     return dbuser
 
 
-@app.put("/api/user/{username}", tags=['User'], response_model=UserResponse)
+@ app.put("/api/user/{username}", tags=['User'], response_model=UserResponse)
 def modify_user(username: str,
                 modified_user: UserModify,
+                bg: BackgroundTasks,
                 db: Session = Depends(get_db),
                 admin: Admin = Depends(Admin.get_current)):
     """
@@ -97,15 +104,31 @@ def modify_user(username: str,
 
     if modified_user.expire is not None and dbuser.status != UserStatus.limited:
         if not dbuser.expire or dbuser.expire > datetime.utcnow().timestamp():
-            dbuser = crud.update_user_status(db, dbuser, UserStatus.active)
+            status = UserStatus.active
         else:
-            dbuser = crud.update_user_status(db, dbuser, UserStatus.expired)
+            status = UserStatus.expired
+        dbuser = crud.update_user_status(db, dbuser, status)
+
+        bg.add_task(
+            telegram.report_status_change,
+            username=dbuser.username,
+            status=status
+        )
+        logger.info(f"User \"{dbuser.username}\" status changed to {status}")
 
     if modified_user.data_limit is not None and dbuser.status != UserStatus.expired:
         if not dbuser.data_limit or dbuser.used_traffic < dbuser.data_limit:
-            dbuser = crud.update_user_status(db, dbuser, UserStatus.active)
+            status = UserStatus.active
         else:
-            dbuser = crud.update_user_status(db, dbuser, UserStatus.limited)
+            status = UserStatus.limited
+        dbuser = crud.update_user_status(db, dbuser, status)
+
+        bg.add_task(
+            telegram.report_status_change,
+            username=dbuser.username,
+            status=status
+        )
+        logger.info(f"User \"{dbuser.username}\" status changed to {status}")
 
     user = UserResponse.from_orm(dbuser)
 
@@ -119,12 +142,18 @@ def modify_user(username: str,
                 account = user.get_account(proxy_type)
                 xray.api.add_inbound_user(tag=inbound['tag'], user=account)
 
+    bg.add_task(
+        telegram.report_user_modification,
+        username=dbuser.username,
+        by=admin.username
+    )
     logger.info(f"User \"{user.username}\" modified")
     return user
 
 
-@app.delete("/api/user/{username}", tags=['User'])
+@ app.delete("/api/user/{username}", tags=['User'])
 def remove_user(username: str,
+                bg: BackgroundTasks,
                 db: Session = Depends(get_db),
                 admin: Admin = Depends(Admin.get_current)):
     """
@@ -145,12 +174,16 @@ def remove_user(username: str,
                 xray.api.remove_inbound_user(tag=inbound['tag'], email=username)
             except xray.exc.EmailNotFoundError:
                 pass
-
+    bg.add_task(
+        telegram.report_user_deletion,
+        username=dbuser.username,
+        by=admin.username
+    )
     logger.info(f"User \"{username}\" deleted")
     return {}
 
 
-@app.get("/api/users", tags=['User'], response_model=List[UserResponse])
+@ app.get("/api/users", tags=['User'], response_model=List[UserResponse])
 def get_users(offset: int = None,
               limit: int = None,
               username: str = None,

config.py

@@ -40,6 +40,10 @@
 XRAY_SUBSCRIPTION_URL_PREFIX = config("XRAY_SUBSCRIPTION_URL_PREFIX", default="").strip("/")
 
 
+TELEGRAM_API_TOKEN = config("TELEGRAM_API_TOKEN", default=None)
+TELEGRAM_ADMIN_ID = config("TELEGRAM_ADMIN_ID", cast=int, default=0)
+TELEGRAM_PROXY_URL = config("TELEGRAM_PROXY_URL", default=None)
+
 JWT_ACCESS_TOKEN_EXPIRE_MINUTES = config("JWT_ACCESS_TOKEN_EXPIRE_MINUTES", cast=int, default=1440)