#!/usr/bin/env python3 import os import json import logging import requests from datetime import datetime, timedelta from telegram import InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Updater, CommandHandler, CallbackQueryHandler, MessageHandler, Filters # ============ تنظیمات ============ TOKEN = "8736951649:AAGGRxU-Hibc_vHRt1ic9X5teW2eShSmeDs" ADMIN_ID = 8833990029 PANEL_CONFIG_FILE = "panel_config.json" ORDERS_FILE = "orders.json" logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ============ فایل‌ها ============ def load_json(file): if os.path.exists(file): with open(file, 'r', encoding='utf-8') as f: return json.load(f) return {} def save_json(file, data): with open(file, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) def load_panel_config(): return load_json(PANEL_CONFIG_FILE) def save_panel_config(data): save_json(PANEL_CONFIG_FILE, data) # ============ ارتباط با پنل Marzban ============ def test_panel_connection(panel_url, username, password): """تست اتصال به پنل""" try: panel_url = panel_url.rstrip('/') # تصحیح لینک پنل if 'marzban-panel-production' in panel_url: # لینک را به فرمت صحیح تغییر دهید if not panel_url.endswith('/api'): panel_url = panel_url.replace('/dashboard', '') url = f"{panel_url}/api/admin/token" data = {"username": username, "password": password} logger.info(f"اتصال به: {url}") response = requests.post(url, json=data, verify=False, timeout=15) if response.status_code == 200: token = response.json().get("access_token") return {"status": "success", "token": token} else: return {"status": "error", "message": f"کد خطا: {response.status_code} - {response.text}"} except requests.exceptions.ConnectionError: return {"status": "error", "message": "اتصال به پنل برقرار نشد! لینک پنل را بررسی کنید."} except requests.exceptions.Timeout: return {"status": "error", "message": "زمان اتصال به پایان رسید! پنل را بررسی کنید."} except Exception as e: return {"status": "error", "message": str(e)} def create_user(panel_url, token, username, data_limit_gb, days=30): """ایجاد کاربر در پنل Marzban""" try: panel_url = panel_url.rstrip('/') if '/dashboard' in panel_url: panel_url = panel_url.replace('/dashboard', '') url = f"{panel_url}/api/user" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } expire = (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S") user_data = { "username": username, "data_limit": data_limit_gb * 1073741824, "expire": expire, "status": "active" } response = requests.post(url, json=user_data, headers=headers, verify=False, timeout=15) if response.status_code in [200, 201]: return response.json() return None except Exception as e: logger.error(f"خطا در ایجاد کاربر: {e}") return None def get_user_sub(panel_url, username, token): """دریافت لینک سابسکریپشن کاربر""" try: panel_url = panel_url.rstrip('/') if '/dashboard' in panel_url: panel_url = panel_url.replace('/dashboard', '') url = f"{panel_url}/api/user/{username}" headers = {"Authorization": f"Bearer {token}"} response = requests.get(url, headers=headers, verify=False, timeout=15) if response.status_code == 200: data = response.json() return data.get("subscription_url", "") return None except Exception as e: logger.error(f"خطا در دریافت لینک ساب: {e}") return None def get_users(panel_url, token): """دریافت لیست کاربران""" try: panel_url = panel_url.rstrip('/') if '/dashboard' in panel_url: panel_url = panel_url.replace('/dashboard', '') url = f"{panel_url}/api/users" headers = {"Authorization": f"Bearer {token}"} response = requests.get(url, headers=headers, verify=False, timeout=15) if response.status_code == 200: return response.json() return [] except Exception as e: logger.error(f"خطا در دریافت لیست کاربران: {e}") return [] # ============ منو ============ def menu(): kb = [ [InlineKeyboardButton("⚙️ تنظیمات پنل", callback_data="panel_settings")], [InlineKeyboardButton("📦 ساخت کانفیگ جدید", callback_data="new_config")], [InlineKeyboardButton("📋 لیست کاربران", callback_data="users_list")], [InlineKeyboardButton("🔗 لینک ساب من", callback_data="my_sub")], [InlineKeyboardButton("❓ راهنما", callback_data="help")] ] return InlineKeyboardMarkup(kb) def start(update, context): uid = update.effective_user.id if uid != ADMIN_ID: update.message.reply_text("❌ شما دسترسی به این ربات ندارید!") return update.message.reply_text( "👋 **به ربات مدیریت Marzban خوش آمدید!**\n\n" "📌 لطفاً روی دکمه '⚙️ تنظیمات پنل' کلیک کنید و اطلاعات پنل را وارد کنید.\n\n" "🌐 لینک پنل: marzban-panel-production-780c.up.railway.app\n" "👤 نام کاربری: x4g\n" "🔑 رمز عبور: Svgahauakabagah", reply_markup=menu(), parse_mode='Markdown' ) # ============ تنظیمات پنل ============ def panel_settings(update, context): q = update.callback_query q.answer() uid = q.from_user.id if uid != ADMIN_ID: q.edit_message_text("❌ شما دسترسی به این بخش ندارید!") return config = load_panel_config() # تنظیم پیش‌فرض if not config.get('panel_url'): config['panel_url'] = "marzban-panel-production-780c.up.railway.app" config['username'] = "x4g" config['password'] = "Svgahauakabagah" save_panel_config(config) panel_url = config.get('panel_url', 'تنظیم نشده') username = config.get('username', 'تنظیم نشده') password = config.get('password', '') password_display = '********' if password else 'تنظیم نشده' text = ( "⚙️ **تنظیمات پنل Marzban**\n\n" f"🌐 **لینک پنل:** `{panel_url}`\n" f"👤 **نام کاربری:** `{username}`\n" f"🔑 **رمز عبور:** `{password_display}`\n\n" "برای تغییر هر کدام، روی دکمه مربوطه کلیک کنید:" ) kb = [ [InlineKeyboardButton("🌐 تنظیم لینک پنل", callback_data="set_panel_url")], [InlineKeyboardButton("👤 تنظیم نام کاربری", callback_data="set_username")], [InlineKeyboardButton("🔑 تنظیم رمز عبور", callback_data="set_password")], [InlineKeyboardButton("🔄 تست اتصال به پنل", callback_data="test_connection")], [InlineKeyboardButton("🔙 بازگشت", callback_data="back")] ] q.edit_message_text(text, reply_markup=InlineKeyboardMarkup(kb), parse_mode='Markdown') def set_panel_url(update, context): q = update.callback_query q.answer() q.edit_message_text( "🌐 **لینک پنل را وارد کنید:**\n\n" "📌 مثال: `marzban-panel-production-780c.up.railway.app`\n" "⚠️ بدون `https://` و بدون `/dashboard`", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("🔙 بازگشت", callback_data="panel_settings")]]), parse_mode='Markdown' ) context.user_data['waiting_panel_url'] = True def set_username(update, context): q = update.callback_query q.answer() q.edit_message_text( "👤 **نام کاربری پنل را وارد کنید:**", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("🔙 بازگشت", callback_data="panel_settings")]]), parse_mode='Markdown' ) context.user_data['waiting_username'] = True def set_password(update, context): q = update.callback_query q.answer() q.edit_message_text( "🔑 **رمز عبور پنل را وارد کنید:**", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("🔙 بازگشت", callback_data="panel_settings")]]), parse_mode='Markdown' ) context.user_data['waiting_password'] = True def test_connection(update, context): q = update.callback_query q.answer() config = load_panel_config() panel_url = config.get('panel_url') username = config.get('username') password = config.get('password') if not panel_url or not username or not password: q.edit_message_text( "❌ **تنظیمات پنل کامل نیست!**\n\n" "لطفاً لینک، نام کاربری و رمز عبور را تنظیم کنید.", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⚙️ تنظیمات پنل", callback_data="panel_settings")]]), parse_mode='Markdown' ) return q.edit_message_text("⏳ در حال تست اتصال به پنل...") result = test_panel_connection(panel_url, username, password) if result["status"] == "success": config['token'] = result["token"] save_panel_config(config) q.edit_message_text( "✅ **اتصال به پنل موفقیت‌آمیز بود!**\n\n" f"🌐 لینک: `{panel_url}`\n" f"👤 کاربر: `{username}`\n\n" "📌 حالا می‌توانید کانفیگ بسازید.", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("📦 ساخت کانفیگ", callback_data="new_config")]]), parse_mode='Markdown' ) else: q.edit_message_text( f"❌ **خطا در اتصال به پنل!**\n\n" f"📌 دلیل: {result.get('message', 'نامشخص')}\n\n" "🔹 لینک پنل را بررسی کنید\n" "🔹 نام کاربری و رمز را بررسی کنید\n" "🔹 مطمئن شوید پنل روشن است\n" "🔹 لینک را به صورت `marzban-panel-production-780c.up.railway.app` وارد کنید", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⚙️ تنظیمات پنل", callback_data="panel_settings")]]), parse_mode='Markdown' ) def handle_panel_settings(update, context): uid = update.effective_user.id if uid != ADMIN_ID: return text = update.message.text.strip() config = load_panel_config() if context.user_data.get('waiting_panel_url'): text = text.rstrip('/') # حذف https:// اگر وجود داشت if text.startswith('https://'): text = text.replace('https://', '') config['panel_url'] = text save_panel_config(config) context.user_data['waiting_panel_url'] = False update.message.reply_text("✅ لینک پنل با موفقیت ذخیره شد!") panel_settings_callback(update, context) elif context.user_data.get('waiting_username'): config['username'] = text save_panel_config(config) context.user_data['waiting_username'] = False update.message.reply_text("✅ نام کاربری با موفقیت ذخیره شد!") panel_settings_callback(update, context) elif context.user_data.get('waiting_password'): config['password'] = text save_panel_config(config) context.user_data['waiting_password'] = False update.message.reply_text("✅ رمز عبور با موفقیت ذخیره شد!") panel_settings_callback(update, context) def panel_settings_callback(update, context): """فراخوانی تنظیمات پنل""" q = update.message config = load_panel_config() panel_url = config.get('panel_url', 'تنظیم نشده') username = config.get('username', 'تنظیم نشده') password = config.get('password', '') password_display = '********' if password else 'تنظیم نشده' text = ( "⚙️ **تنظیمات پنل Marzban**\n\n" f"🌐 **لینک پنل:** `{panel_url}`\n" f"👤 **نام کاربری:** `{username}`\n" f"🔑 **رمز عبور:** `{password_display}`\n\n" "برای تغییر هر کدام، روی دکمه مربوطه کلیک کنید:" ) kb = [ [InlineKeyboardButton("🌐 تنظیم لینک پنل", callback_data="set_panel_url")], [InlineKeyboardButton("👤 تنظیم نام کاربری", callback_data="set_username")], [InlineKeyboardButton("🔑 تنظیم رمز عبور", callback_data="set_password")], [InlineKeyboardButton("🔄 تست اتصال به پنل", callback_data="test_connection")], [InlineKeyboardButton("🔙 بازگشت", callback_data="back")] ] q.reply_text(text, reply_markup=InlineKeyboardMarkup(kb), parse_mode='Markdown') # ============ ساخت کانفیگ ============ def new_config(update, context): q = update.callback_query q.answer() uid = q.from_user.id if uid != ADMIN_ID: q.edit_message_text("❌ شما دسترسی به این بخش ندارید!") return config = load_panel_config() if not config.get("panel_url") or not config.get("username") or not config.get("password"): q.edit_message_text( "⚠️ **تنظیمات پنل کامل نیست!**\n\n" "لطفاً ابتدا پنل را تنظیم کنید.", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⚙️ تنظیمات پنل", callback_data="panel_settings")]]), parse_mode='Markdown' ) return q.edit_message_text( "📦 **ساخت کانفیگ جدید**\n\n" "لطفاً حجم مورد نظر را به **گیگابایت** وارد کنید:\n" "مثال: `10` یا `20` یا `50`\n\n" "🔹 کانفیگ ۳۰ روزه خواهد بود", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("🔙 بازگشت", callback_data="back")]]), parse_mode='Markdown' ) context.user_data['waiting_volume'] = True def handle_volume(update, context): uid = update.effective_user.id if uid != ADMIN_ID: return if not context.user_data.get('waiting_volume'): return try: volume = int(update.message.text.strip()) if volume <= 0: update.message.reply_text("❌ لطفاً یک عدد مثبت وارد کنید.") return config = load_panel_config() panel_url = config['panel_url'] result = test_panel_connection(panel_url, config['username'], config['password']) if result["status"] != "success": update.message.reply_text( f"❌ خطا در اتصال به پنل: {result.get('message')}\n" "لطفاً تنظیمات را بررسی کنید." ) return token = result["token"] username = f"user_{datetime.now().strftime('%Y%m%d%H%M%S')}" user = create_user(panel_url, token, username, volume, 30) if not user: update.message.reply_text("❌ خطا در ایجاد کاربر در پنل!") return sub_link = get_user_sub(panel_url, username, token) if sub_link: text = ( f"✅ **کانفیگ با موفقیت ساخته شد!**\n\n" f"📦 **حجم:** {volume} گیگابایت\n" f"👤 **نام کاربری:** `{username}`\n" f"📅 **مدت:** ۳۰ روز\n" f"🔗 **لینک سابسکریپشن:**\n`{sub_link}`\n\n" "🌍 ۱۲ لوکیشن مختلف\n" "⏱ بلافاصله فعال می‌شود" ) update.message.reply_text(text, parse_mode='Markdown') else: update.message.reply_text( f"✅ **کاربر ساخته شد اما لینک ساب دریافت نشد!**\n\n" f"👤 نام کاربری: `{username}`", parse_mode='Markdown' ) context.user_data['waiting_volume'] = False except ValueError: update.message.reply_text("❌ لطفاً فقط عدد وارد کنید!") # ============ لیست کاربران ============ def users_list(update, context): q = update.callback_query q.answer() uid = q.from_user.id if uid != ADMIN_ID: q.edit_message_text("❌ شما دسترسی به این بخش ندارید!") return config = load_panel_config() if not config.get("panel_url") or not config.get("username") or not config.get("password"): q.edit_message_text("⚠️ تنظیمات پنل کامل نیست!") return q.edit_message_text("⏳ در حال دریافت لیست کاربران...") result = test_panel_connection(config['panel_url'], config['username'], config['password']) if result["status"] != "success": q.edit_message_text(f"❌ خطا: {result.get('message')}") return users = get_users(config['panel_url'], result["token"]) if not users: q.edit_message_text("📭 هیچ کاربری در پنل وجود ندارد!") return text = "📋 **لیست کاربران پنل**\n\n" for user in users[:20]: username = user.get('username', 'نامشخص') status = user.get('status', 'نامشخص') data_limit = user.get('data_limit', 0) / 1073741824 used = user.get('used_traffic', 0) / 1073741824 expire = user.get('expire', 'نامشخص') status_emoji = "🟢" if status == "active" else "🔴" text += f"{status_emoji} **{username}**\n" text += f" 📊 {used:.1f}/{data_limit:.1f} GB\n" text += f" 📅 {expire[:10] if expire else 'نامحدود'}\n\n" kb = [[InlineKeyboardButton("🔙 بازگشت", callback_data="back")]] q.edit_message_text(text, reply_markup=InlineKeyboardMarkup(kb), parse_mode='Markdown') # ============ لینک ساب ============ def my_sub(update, context): q = update.callback_query q.answer() uid = q.from_user.id if uid != ADMIN_ID: q.edit_message_text("❌ این بخش فقط برای ادمین است!") return q.edit_message_text( "🔗 **دریافت لینک سابسکریپشن**\n\n" "لطفاً نام کاربری را وارد کنید:", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("🔙 بازگشت", callback_data="back")]]), parse_mode='Markdown' ) context.user_data['waiting_sub_username'] = True def handle_sub_username(update, context): uid = update.effective_user.id if uid != ADMIN_ID: return if not context.user_data.get('waiting_sub_username'): return username = update.message.text.strip() config = load_panel_config() result = test_panel_connection(config['panel_url'], config['username'], config['password']) if result["status"] != "success": update.message.reply_text(f"❌ خطا: {result.get('message')}") return sub_link = get_user_sub(config['panel_url'], username, result["token"]) if sub_link: update.message.reply_text( f"🔗 **لینک سابسکریپشن کاربر `{username}`:**\n\n`{sub_link}`", parse_mode='Markdown' ) else: update.message.reply_text(f"❌ کاربر `{username}` پیدا نشد!", parse_mode='Markdown') context.user_data['waiting_sub_username'] = False # ============ راهنما ============ def help_cmd(update, context): q = update.callback_query q.answer() text = ( "📌 **راهنمای استفاده**\n\n" "1️⃣ **تنظیم پنل:**\n" " لینک، نام کاربری و رمز پنل را تنظیم کنید.\n\n" "2️⃣ **تست اتصال:**\n" " اتصال به پنل را تست کنید.\n\n" "3️⃣ **ساخت کانفیگ:**\n" " حجم را وارد کنید تا کانفیگ ساخته شود.\n\n" "4️⃣ **لیست کاربران:**\n" " لیست تمام کاربران پنل را نمایش می‌دهد.\n\n" "5️⃣ **لینک ساب:**\n" " لینک سابسکریپشن کاربر مورد نظر را دریافت کنید.\n\n" "🌍 **۱۲ لوکیشن مختلف**\n" "⏱ **بلافاصله فعال می‌شود**" ) kb = [[InlineKeyboardButton("🔙 بازگشت", callback_data="back")]] q.edit_message_text(text, reply_markup=InlineKeyboardMarkup(kb), parse_mode='Markdown') def back(update, context): q = update.callback_query q.answer() q.edit_message_text("🏠 منوی اصلی", reply_markup=menu()) # ============ اجرا ============ def main(): updater = Updater(TOKEN, use_context=True) dp = updater.dispatcher dp.add_handler(CommandHandler("start", start)) dp.add_handler(CallbackQueryHandler(panel_settings, pattern="^panel_settings$")) dp.add_handler(CallbackQueryHandler(set_panel_url, pattern="^set_panel_url$")) dp.add_handler(CallbackQueryHandler(set_username, pattern="^set_username$")) dp.add_handler(CallbackQueryHandler(set_password, pattern="^set_password$")) dp.add_handler(CallbackQueryHandler(test_connection, pattern="^test_connection$")) dp.add_handler(CallbackQueryHandler(new_config, pattern="^new_config$")) dp.add_handler(CallbackQueryHandler(users_list, pattern="^users_list$")) dp.add_handler(CallbackQueryHandler(my_sub, pattern="^my_sub$")) dp.add_handler(CallbackQueryHandler(help_cmd, pattern="^help$")) dp.add_handler(CallbackQueryHandler(back, pattern="^back$")) dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_panel_settings)) dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_volume)) dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_sub_username)) logger.info("🤖 ربات مدیریت Marzban راه‌اندازی شد!") updater.start_polling() updater.idle() if __name__ == '__main__': main()