From 3ba1ccf8e87e4875cf99ce977aefa0ec716ad2cb Mon Sep 17 00:00:00 2001 From: justuser-31 Date: Wed, 4 Feb 2026 14:20:44 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D0=B5=20=D1=82=D0=B5=D0=BB=D0=B5=D0=B3=D1=80=D0=B0?= =?UTF-8?q?=D0=BC=20=D0=B1=D0=BE=D1=82=D0=B0=20=D0=B4=D0=BB=D1=8F=20=D1=80?= =?UTF-8?q?=D0=B5=D0=B3=D0=B8=D1=81=D1=82=D1=80=D0=B0=D1=86=D0=B8=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- call2api.py.py => call2api.py | 4 +- requirements.txt | 1 + telegram_bot/.gitignore | 1 + telegram_bot/call2api.py | 82 ++++++++++++++ telegram_bot/sfs_bot.py | 203 ++++++++++++++++++++++++++++++++++ 5 files changed, 289 insertions(+), 2 deletions(-) rename call2api.py.py => call2api.py (94%) create mode 100644 telegram_bot/.gitignore create mode 100644 telegram_bot/call2api.py create mode 100644 telegram_bot/sfs_bot.py diff --git a/call2api.py.py b/call2api.py similarity index 94% rename from call2api.py.py rename to call2api.py index fa0ec6b..b28b7a1 100644 --- a/call2api.py.py +++ b/call2api.py @@ -78,5 +78,5 @@ def delete_account(api_url: str, api_token: str, username: str) -> dict: # print(set_quota("http://proxy.del.pw:50020", "test", "test", 1000)) -# print(delete_account("http://proxy.del.pw:50020", "test", "test")) -# print(create_account("http://proxy.del.pw:50020", "test", "test", "test", 100)) +# print(delete_account("http://127.0.0.1:8000", "test", "test")) +# print(create_account("http://127.0.0.1:8000", "test", "test", "test", 999)) diff --git a/requirements.txt b/requirements.txt index 6c9fdba..d24632e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ PyYAML requests +pyTelegramBotAPI diff --git a/telegram_bot/.gitignore b/telegram_bot/.gitignore new file mode 100644 index 0000000..a6c57f5 --- /dev/null +++ b/telegram_bot/.gitignore @@ -0,0 +1 @@ +*.json diff --git a/telegram_bot/call2api.py b/telegram_bot/call2api.py new file mode 100644 index 0000000..b28b7a1 --- /dev/null +++ b/telegram_bot/call2api.py @@ -0,0 +1,82 @@ +import requests + + +def create_account( + api_url: str, api_token: str, username: str, password: str, drive_quota: int +) -> dict: + """ + Call the create account API endpoint. + + Args: + api_url: Base URL of the API server (e.g., 'http://localhost:8000') + api_token: Security token for API authentication + username: Username for the new account + password: Password for the new account + drive_quota: Storage quota in MB + + Returns: + Dictionary with 'status_code' and 'message' keys + """ + url = f"{api_url.rstrip('/')}/create_account" + data = { + "token": api_token, + "username": username, + "password": password, + "drive_quota": str(drive_quota), + } + + try: + response = requests.post(url, data=data) + return {"status_code": response.status_code, "message": response.text} + except requests.RequestException as e: + return {"status_code": -1, "message": f"Request failed: {str(e)}"} + + +def set_quota(api_url: str, api_token: str, username: str, drive_quota: int) -> dict: + """ + Call the set quota API endpoint. + + Args: + api_url: Base URL of the API server (e.g., 'http://localhost:8000') + api_token: Security token for API authentication + username: Username whose quota should be updated + drive_quota: New storage quota in MB + + Returns: + Dictionary with 'status_code' and 'message' keys + """ + url = f"{api_url.rstrip('/')}/set_quota" + data = {"token": api_token, "username": username, "drive_quota": str(drive_quota)} + + try: + response = requests.post(url, data=data) + return {"status_code": response.status_code, "message": response.text} + except requests.RequestException as e: + return {"status_code": -1, "message": f"Request failed: {str(e)}"} + + +def delete_account(api_url: str, api_token: str, username: str) -> dict: + """ + Call the delete account API endpoint. + + Args: + api_url: Base URL of the API server (e.g., 'http://localhost:8000') + api_token: Security token for API authentication + username: Username of the account to delete + + Returns: + Dictionary with 'status_code' and 'message' keys + """ + url = f"{api_url.rstrip('/')}/delete_account" + data = {"token": api_token, "username": username} + + try: + response = requests.post(url, data=data) + return {"status_code": response.status_code, "message": response.text} + except requests.RequestException as e: + return {"status_code": -1, "message": f"Request failed: {str(e)}"} + + +# print(set_quota("http://proxy.del.pw:50020", "test", "test", 1000)) +# print(delete_account("http://127.0.0.1:8000", "test", "test")) +# print(create_account("http://127.0.0.1:8000", "test", "test", "test", 999)) diff --git a/telegram_bot/sfs_bot.py b/telegram_bot/sfs_bot.py new file mode 100644 index 0000000..a0ec519 --- /dev/null +++ b/telegram_bot/sfs_bot.py @@ -0,0 +1,203 @@ +import json +import os +import random +import string + +import telebot + +from call2api import create_account, delete_account + +# Load configuration +CONFIG_FILE = "config.json" +if not os.path.exists(CONFIG_FILE): + with open(CONFIG_FILE, "w") as f: + f.write("""{ + "bot_token": "YOUR_BOT_TOKEN_HERE", + "chat_id": 1111, + "api_url": "http://localhost:8000", + "api_token": "YOUR_API_TOKEN_HERE", + "drive_quota": 100 +}""") + raise FileNotFoundError(f"Configuration file {CONFIG_FILE} not found") + +with open(CONFIG_FILE, "r") as f: + config = json.load(f) + +# Bot configuration +BOT_TOKEN = config.get("bot_token") +CHAT_ID = config.get("chat_id", 1111) +API_URL = config.get("api_url") +API_TOKEN = config.get("api_token") +DRIVE_QUOTA = config.get("drive_quota", 100) + +# Initialize bot +bot = telebot.TeleBot(BOT_TOKEN) + + +# Load user registrations +def load_registrations(): + if os.path.exists("registrations.json"): + with open("registrations.json", "r") as f: + return json.load(f) + return {} + + +# Save user registrations +def save_registrations(registrations): + with open("registrations.json", "w") as f: + json.dump(registrations, f, indent=2) + + +# Generate random password +def generate_password(length=20): + characters = string.ascii_letters + string.digits + return "".join(random.choice(characters) for _ in range(length)) + + +# Generate registration token +def generate_registration_token(length=20): + characters = string.ascii_letters + string.digits + return "".join(random.choice(characters) for _ in range(length)) + + +# Store pending registrations +global pending_registrations +pending_registrations = {} + + +@bot.message_handler(commands=["reg_sfs"]) +def register_user(message): + user_id = str(message.from_user.id) + chat_id = message.chat.id + + # Check if this is private messages + if message.chat.type == "private": + complete_registration(message) + return + # Check if command was sent in the correct chat + if chat_id != CHAT_ID: + bot.reply_to(message, "Please use this command in the designated chat.") + return + + # Load current registrations + registrations = load_registrations() + + # Check if user is already registered + if user_id in registrations: + bot.reply_to(message, "You are already registered!") + return + + # Generate registration token and store it + reg_token = generate_registration_token() + global pending_registrations + pending_registrations[user_id] = reg_token + + # Send instruction to user + instruction_msg = ( + f"To complete registration, type in bot's private messages:\n" + f"`/reg_sfs {reg_token} YOUR_USERNAME`" + ) + bot.reply_to(message, instruction_msg, parse_mode="Markdown") + + +def complete_registration(message): + global pending_registrations + user_id = str(message.from_user.id) + chat_type = message.chat.type + + # Registration must happen in private chat + if chat_type != "private": + return + + # Check if user has initiated registration + if user_id not in pending_registrations: + bot.reply_to( + message, "Please start registration in the group chat first using /reg_sfs" + ) + return + + parts = message.text.split() + if len(parts) != 3: + bot.reply_to( + message, + "Invalid format. Use: `/reg_sfs TOKEN USERNAME`", + parse_mode="Markdown", + ) + return + + _, provided_token, username = parts + + # Verify token + if pending_registrations[user_id] != provided_token: + bot.reply_to(message, "Invalid registration token.") + return + + # Generate password + password = generate_password() + + if username == "YOUR_USERNAME": + bot.reply_to( + message, + "You should use own username instead example `YOUR_USERNAME`.", + parse_mode="Markdown", + ) + return + + # Create account via API + result = create_account(API_URL, API_TOKEN, username, password, DRIVE_QUOTA) + + if result["status_code"] == 200: + # Save registration + registrations = load_registrations() + registrations[user_id] = username + registrations[username] = user_id + save_registrations(registrations) + + # Remove pending registration + del pending_registrations[user_id] + + # Send success message + success_msg = ( + f"Successful registration:\nUsername: `{username}`\nPassword: `{password}`" + ) + bot.reply_to(message, success_msg, parse_mode="Markdown") + else: + bot.reply_to( + message, + f"Registration failed: {result['message']}", + parse_mode="HTML", + ) + + +@bot.message_handler(content_types=["left_chat_member"]) +def handle_user_leave(message): + user_id = str(message.left_chat_member.id) + + # Load registrations + registrations = load_registrations() + + # Check if user was registered + if user_id in registrations: + username = registrations[user_id] + + # Delete account via API + result = delete_account(API_URL, API_TOKEN, username) + + # Remove registration + del registrations[user_id] + if username in registrations: + del registrations[username] + save_registrations(registrations) + + # Notify in chat if deletion was successful + if result["status_code"] == 200: + bot.send_message(CHAT_ID, f"Account for {username} has been deleted.") + else: + bot.send_message( + CHAT_ID, f"Failed to delete account for {username}: {result['message']}" + ) + + +if __name__ == "__main__": + print("SFS Bot started...") + bot.polling(none_stop=True)