Pending changes exported from your codespace

.vscode/launch.json

@@ -0,0 +1,17 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python Debugger: Current File",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "__main__.py",
+            "cwd": "${workspaceFolder}/hiddifypanel_bot",
+            "console": "integratedTerminal",
+            "justMyCode": false
+        }
+    ]
+}

Makefile

@@ -120,3 +120,9 @@ init:             ## Initialize the project based on an application template.
 # __author__ = 'rochacbruno'
 # __repo__ = https://github.com/rochacbruno/python-project-template
 # __sponsor__ = https://github.com/sponsors/rochacbruno/
+
+
+
+generate_hiddify_api:
+	npm install @openapitools/openapi-generator-cli -g 
+	openapi-generator-cli generate -i localhost:9000/api/openapi.json -g python -o hiddifypanel_bot/hiddifyapi_gen --skip-validate-spec

hiddifypanel_bot/__init__.py

@@ -0,0 +1,4 @@
+from .basebot import bot
+from .hiddifyapi import HiddifyApi
+from . import utils
+from . import modules

hiddifypanel_bot/basebot.py

@@ -0,0 +1,31 @@
+#!/usr/bin/python
+
+import telebot
+import logging
+from .hiddifyapi import HiddifyApi
+from telebot import TeleBot,ExceptionHandler
+from telebot import handler_backends
+
+logger = telebot.logger
+telebot.logger.setLevel(logging.DEBUG)  # Outputs debug messages to console.
+
+
+class MyExceptionHandler(ExceptionHandler):
+    async def handle(self, exception):
+        logger.error(exception)
+
+bot:TeleBot = TeleBot('7149818063:AAHfIaKePlc-31SVN0wseELncsT2yioF7c0', exception_handler=MyExceptionHandler(),use_class_middlewares=True)
+
+class Middleware(handler_backends.BaseMiddleware):
+    def __init__(self):
+        self.update_types = ['message'] 
+        pass
+    def pre_process(self, message, data):
+        data['lang']=message.from_user.language_code
+        # data['lang']="en"
+        data['hapi']=HiddifyApi('https://94.242.53.78.sslip.io/OL2Rn5QuO7hF5gDo235/','a49c81ac-e3af-4b5c-af53-98a34121ab96')
+    def post_process(self, message, data, exception):
+        pass
+
+
+bot.setup_middleware(Middleware()) 

hiddifypanel_bot/cli.py

@@ -7,9 +7,13 @@
 - Start a web application
 - Import things from your .base module
 """
+import asyncio
 from . import utils
-
+from . import basebot
 def main():  # pragma: no cover
     utils.setup_translation()
-    import i18n
-    print(i18n.t("start",locale='fa'))
+    # import i18n
+    # print(i18n.t("start",locale='fa'))
+    # asyncio.run(basebot.bot.polling(restart_on_change=True))
+    asyncio.run(basebot.bot.polling())
+

hiddifypanel_bot/hiddifyapi.py

@@ -0,0 +1,215 @@
+# api.py
+
+import uuid
+from datetime import datetime
+import qrcode
+from io import BytesIO
+from qrcode.image.styledpil import StyledPilImage
+from qrcode.image.styles.colormasks import RadialGradiantColorMask
+from qrcode.image.styles.moduledrawers.pil import CircleModuleDrawer
+import aiohttp
+import asyncio
+
+
+class HiddifyApi:
+    def __init__(self, api_url: str, api_key: str):
+        self.base_url = api_url
+        self.headers = {'HIDDIFY-API-KEY': api_key}
+
+
+    async def get_system_status(self) -> dict:
+        """Get the system status."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(f"{self.base_url}/api/v2/admin/server_status/", headers=self.headers) as response:
+                    data = await response.json()
+                    stats = data.get("stats", {})
+                    usage_history = data.get("usage_history", {})
+                    stats["usage_history"] = usage_history
+                    return stats
+            except aiohttp.ClientError as e:
+                print(f"Error in get_system_status: {e}")
+                return {}
+
+    async def make_post_request(self, endpoint: str, json_data: dict) -> dict:
+        """Make a POST request."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(endpoint, json=json_data, headers=self.headers) as response:
+                    return response
+            except aiohttp.ClientError as e:
+                print(f"Error in making POST request: {e}")
+                return False
+
+    async def make_patch_request(self, endpoint: str, json_data: dict) -> dict:
+        """Make a PATCH request."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.patch(endpoint, json=json_data, headers=self.headers) as response:
+                    return response
+            except aiohttp.ClientError as e:
+                print(f"Error in making PATCH request: {e}")
+                return False
+
+    async def get_admin_list(self) -> list:
+        """Get the list of admin users."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(f"{self.base_url}/api/v2/admin/admin_user/", headers=self.headers) as response:
+                    return await response.json()
+            except aiohttp.ClientError as e:
+                print(f"Error in get_admin_list: {e}")
+                return []
+
+    async def delete_admin_user(self, uuid: str) -> bool:
+        """Delete an admin user."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                endpoint = f"{self.base_url}/api/v2/admin/admin_user/{uuid}/"
+                async with session.delete(endpoint, headers=self.headers) as response:
+                    return response.status == 200
+            except aiohttp.ClientError as e:
+                print(f"Error in delete_admin_user: {e}")
+                return False
+
+    async def add_service(self, comment: str, name: str, day: int, traffic: float, telegram_id: int=None,uuid: str=None) -> bool:
+        """Add a new service."""
+        data = {
+            "comment": comment,
+            "current_usage_GB": 0,
+            "mode": "no_reset",
+            "name": name,
+            "package_days": day,
+            "telegram_id": None,
+            "usage_limit_GB": traffic,
+            "uuid": uuid,
+        }
+        endpoint = f"{self.base_url}/api/v2/admin/user/"
+        return await self.make_post_request(endpoint, data)
+
+    async def get_user_list(self) -> list:
+        """Get the list of users."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(f"{self.base_url}/api/v2/admin/user/", headers=self.headers) as response:
+                    return await response.json()
+            except aiohttp.ClientError as e:
+                print(f"Error in get_user_list: {e}")
+                return []
+
+    async def get_user_list_name(self, query_name: str) -> list:
+        """Get the list of users and filter by name containing the query."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(f"{self.base_url}/api/v2/admin/user/", headers=self.headers) as response:
+                    user_list = await response.json()
+                    filtered_users = [user for user in user_list if query_name.lower() in user.get('name', '').lower()]
+                    return filtered_users
+            except aiohttp.ClientError as e:
+                print(f"Error in get_user_list_name: {e}")
+                return []
+
+    async def tele_id(self, uuid: str, telegram_id: int) -> bool:
+        """Add Telegram ID."""
+        data = {
+            "telegram_id": telegram_id
+        }
+        endpoint = f"{self.base_url}/api/v2/admin/user/{uuid}/"
+        return await self.make_patch_request(endpoint, data)
+
+    async def reset_user_last_reset_time(self, uuid: str) -> bool:
+        """Reset the user's last reset time."""
+        user_data = await self.find_service(uuid)
+        if not user_data:
+            print("User not found.")
+            return False
+        user_data['last_reset_time'] = datetime.now().strftime('%Y-%m-%d')
+        user_data['start_date'] = None
+        user_data['current_usage_GB'] = 0
+        endpoint = f"{self.base_url}/api/v2/admin/user/{uuid}/"
+        return await self.make_patch_request(endpoint, user_data)
+
+    async def update_package_days(self, uuid: str) -> bool:
+        """Update the package days for a user."""
+        user_data = await self.find_service(uuid)
+        if not user_data:
+            print("User not found.")
+            return False
+        user_data['last_reset_time'] = datetime.now().strftime('%Y-%m-%d')
+        user_data['start_date'] = None
+        endpoint = f"{self.base_url}/api/v2/admin/user/{uuid}/"
+        return await self.make_patch_request(endpoint, user_data)
+
+    async def update_traffic(self, uuid: str) -> bool:
+        """Reset the traffic limit for a user to 0."""
+        user_data = await self.find_service(uuid)
+        if not user_data:
+            print("User not found.")
+            return False
+        user_data['current_usage_GB'] = 0
+        endpoint = f"{self.base_url}/api/v2/admin/user/{uuid}/"
+        return await self.make_patch_request(endpoint, user_data)
+
+    async def delete_user(self, uuid: str) -> bool:
+        """Delete a user."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                endpoint = f"{self.base_url}/api/v2/admin/user/{uuid}/"
+                async with session.delete(endpoint, headers=self.headers) as response:
+                    return response.status == 200
+            except aiohttp.ClientError as e:
+                print(f"Error in delete_user: {e}")
+                return False
+
+    async def find_service(self, uuid: str) -> dict:
+        """Find a service by UUID."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(f"{self.base_url}/api/v2/admin/user/{uuid}/", headers=self.headers) as response:
+                    if response.status == 200:
+                        return await response.json()
+                    else:
+                        print(f"User with UUID {uuid} not found.")
+                        return {}
+            except aiohttp.ClientError as e:
+                print(f"Error in find_service: {e}")
+                return {}
+
+    async def backup_file(self) -> bytes:
+        """Backup the file."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(f"{self.base_url}/admin/backup/backupfile/", headers=self.headers) as response:
+                    if response.status == 200:
+                        return await response.read()
+                    else:
+                        print(f"Failed to retrieve backup file. Status code: {response.status}")
+                        return None
+            except aiohttp.ClientError as e:
+                print(f"Error in backup_file: {e}")
+                return None
+
+    async def get_app_information(self, uuid: str) -> dict:
+        """Get information about available apps for a given UUID."""
+        async with aiohttp.ClientSession() as session:
+            try:
+                url = f"{self.base_url}/{uuid}/api/v2/user/apps/"
+                querystring = {"platform": "all"}
+                headers = {"Accept": "application/json"}
+                async with session.get(url, headers=headers, params=querystring) as response:
+                    if response.status == 200:
+                        return await response.json()
+                    else:
+                        print("Failed to fetch app information. Status code:", response.status)
+                        return {}
+            except aiohttp.ClientError as e:
+                print(f"Error in get_app_information: {e}")
+                return {}
+
+    @staticmethod
+    def generate_qr_code(data: str) -> BytesIO:
+        """Generate a QR code for the given data."""
+        qr = qrcode.QRCode(version=1, box_size=10, border=2)
+        qr.add_data(data)
+        qr.make(fit=True)
+        qr_img = qr.make_image

hiddifypanel_bot/modules/__init__.py

@@ -0,0 +1,2 @@
+from . import welcome
+from . import add_user

hiddifypanel_bot/modules/add_user.py

@@ -0,0 +1,65 @@
+from hiddifypanel_bot import bot,HiddifyApi
+from hiddifypanel_bot.utils import tghelper
+
+import telebot
+from telebot.types import ReplyParameters,ForceReply,InlineKeyboardButton,InlineKeyboardMarkup
+
+import i18n
+
+@bot.message_handler(commands="add")
+def add_user_name(message,lang,hapi):
+    resp=i18n.t("addname",lang)
+    bot.reply_to(message, resp,reply_markup=ForceReply(False,resp))
+    bot.register_next_step_handler(message, add_user_package_days,lang,hapi)
+
+def add_user_package_days(message,lang,hapi):
+    name = message.text
+    resp=i18n.t("adddays",lang)
+    bot.reply_to(message, resp,reply_markup=ForceReply(False,resp))
+    bot.register_next_step_handler(message, validate_package_days, lang,hapi,name)
+
+def validate_package_days(message,  lang,hapi,name:str):
+    if message.text.isnumeric():
+        package_days = int(message.text)
+        resp=i18n.t("addgb",lang)
+        bot.reply_to(message, resp,reply_markup=ForceReply(False,resp))
+
+
+        bot.register_next_step_handler(message, validate_usage_limit,lang,hapi, name, package_days)
+    else:
+        bot.reply_to(message, i18n.t("invaliddays",message.from_user.language_code))
+        bot.register_next_step_handler(message, validate_package_days,lang,hapi, name)
+
+def validate_usage_limit(message,lang,hapi, name:str, package_days:int):
+    try:
+        usage_limit = float(message.text)
+        add_user_complete(message, lang,hapi,name, package_days, usage_limit)
+    except ValueError:
+        resp=i18n.t("invalidgb",lang)
+        bot.reply_to(message, resp,reply_markup=ForceReply(False,resp))
+        bot.register_next_step_handler(message, validate_usage_limit,lang,hapi, name, package_days)
+
+def add_user_complete(message, lang,hapi:HiddifyApi,name:str, package_days:int, usage_limit:float):
+
+    user_data = hapi.add_service("", name, package_days, usage_limit)
+    if 'msg' not in user_data:
+        uuid=user_data.get('uuid', 'N/A')
+        sublink_data = f"test/{uuid}"
+        qr_code = hapi.generate_qr_code(sublink_data)
+
+        user_info = (
+            f"User UUID: {uuid}\n"
+            f"Name: {user_data.get('name', 'N/A')}\n"
+            f"Usage Limit: {user_data.get('usage_limit_GB', 'N/A')} GB\n"
+            f"Package Days: {user_data.get('package_days', 'N/A')} Days"
+        )
+        inline_keyboard = InlineKeyboardMarkup()
+        inline_keyboard.add(InlineKeyboardButton(text="Open Sublink", web_app=sublink_data))
+
+        bot.send_photo(message.chat.id, qr_code, caption=user_info, reply_markup=inline_keyboard)
+        # else:
+        #     bot.reply_to(message, "Failed to retrieve user data.")
+    else:
+        bot.reply_to(message, "Failed to add user.")
+
+