import asyncio import os import logging import re import json import random from datetime import datetime from zoneinfo import ZoneInfo from urllib.parse import quote import aiohttp from pyrogram import Client, filters, idle from pyrogram.handlers import MessageHandler, CallbackQueryHandler, InlineQueryHandler from pyrogram.enums import ChatType, ChatAction from pyrogram.types import ( Message, InlineKeyboardMarkup, InlineKeyboardButton, InlineQueryResultArticle, InputTextMessageContent ) from pyrogram.raw import functions from pyrogram.errors import SessionPasswordNeeded, ChatSendInlineForbidden from aiohttp import web import pyrogram.utils import asyncio # ==================== تنظیمات ==================== API_ID = 111111111 API_HASH = "111111111" BOT_TOKEN = "8980730102:AAGgklxmn1j6j56lL0ibrpxSbSCcYEKSxZA" OWNER_ID = 7965858650 BOT_USERNAME = "self_dnw_bot" DATA_FILE = "bot_data.json" WEBHOOK_PORT = 8080 WEBHOOK_PATH = "/webhook" WEBHOOK_URL = "https://your-domain.com/webhook" # تغییر دهید TEHRAN_TIMEZONE = ZoneInfo("Asia/Tehran") logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(levelname)s - %(message)s') # ==================== پچ Pyrogram ==================== 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() # ==================== دیتا ==================== ENEMY_REPLIES = [ "کیرم تو رحم اجاره ای و خونی مالی مادرت", "دو میلیون شبی پول ویلا بدم تا مادرتو تو گوشه کناراش بگام", "احمق مادر کونی من کس مادرت گذاشتم تو بازم داری کسشر میگی", "هی بیناموس کیرم بره تو کس ننت واس بابات نشآخ", "کیرم تو کس سیاه مادرت خارکصده", "حروم زاده باک کص ننت با ابکیرم پر میکنم", "منبع اب ایرانو با اب کص مادرت تامین میکنم", "خارکسته میخای مادرتو بگام بعد بیای ادعای شرف کنی", ] FONT_STYLES = { 'stylized': {'0':'𝟎','1':'𝟏','2':'𝟐','3':'𝟑','4':'𝟒','5':'𝟓','6':'𝟔','7':'𝟕','8':'𝟖','9':'𝟗',':':':'}, 'cursive': {'0':'𝟎','1':'𝟏','2':'𝟐','3':'𝟑','4':'𝟒','5':'𝟓','6':'𝟔','7':'𝟕','8':'𝟖','9':'𝟗',':':':'}, 'doublestruck': {'0':'𝟘','1':'𝟙','2':'𝟚','3':'𝟛','4':'𝟜','5':'𝟝','6':'𝟞','7':'𝟟','8':'𝟠','9':'𝟡',':':':'}, 'monospace': {'0':'𝟶','1':'𝟷','2':'𝟸','3':'𝟹','4':'𝟺','5':'𝟻','6':'𝟼','7':'𝟽','8':'𝟾','9':'𝟿',':':':'}, 'normal': {'0':'0','1':'1','2':'2','3':'3','4':'4','5':'5','6':'6','7':'7','8':'8','9':'9',':':':'}, } FONT_KEYS = ["stylized", "cursive", "doublestruck", "monospace", "normal"] ALL_CLOCK_CHARS = "".join(set(char for font in FONT_STYLES.values() for char in font.values())) CLOCK_CHARS_REGEX_CLASS = f"[{re.escape(ALL_CLOCK_CHARS)}]" # ==================== کلاس مدیریت دیتا ==================== class DataManager: def __init__(self, file_path): self.file_path = file_path self.data = self.load_data() 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: return self.get_default_data() return self.get_default_data() def get_default_data(self): return {"users": {}, "sessions": {}} def save_data(self): with open(self.file_path, 'w', encoding='utf-8') as f: json.dump(self.data, f, ensure_ascii=False, indent=2) def get_user(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": "", "settings": { "font": "stylized", "clock": True, "bold": False, "secretary": False, "secretary_msg": "", "auto_seen": False, "pv_lock": False, "anti_login": False, "typing": False, "playing": False, "global_enemy": False, "copy_mode": False, }, "enemies": [], "muted": [], "reactions": {}, "replied_users": [] } self.save_data() return self.data["users"][user_id_str] def update_user(self, user_id, updates): user = self.get_user(user_id) for key, value in updates.items(): if key == "settings": for sk, sv in value.items(): user["settings"][sk] = sv else: user[key] = value self.save_data() return user def get_setting(self, user_id, key): return self.get_user(user_id)["settings"].get(key) def set_setting(self, user_id, key, value): self.update_user(user_id, {"settings": {key: value}}) data_manager = DataManager(DATA_FILE) # ==================== متغیرهای وضعیت ==================== active_bots = {} user_states = {} login_states = {} # ==================== توابع کمکی ==================== def stylize_time(time_str: str, style: str) -> str: font_map = FONT_STYLES.get(style, FONT_STYLES["stylized"]) return ''.join(font_map.get(char, char) for char in time_str) def generate_panel_markup(user_id): settings = data_manager.get_user(user_id)["settings"] s_clock = "✔" if settings.get("clock", True) else "✖" s_bold = "✔" if settings.get("bold", False) else "✖" s_sec = "✔" if settings.get("secretary", False) else "✖" s_seen = "✔" if settings.get("auto_seen", False) else "✖" s_pv = "🔒" if settings.get("pv_lock", False) else "🔓" s_anti = "✔" if settings.get("anti_login", False) else "✖" s_type = "✔" if settings.get("typing", False) else "✖" s_game = "✔" if settings.get("playing", False) else "✖" s_enemy = "✔" if settings.get("global_enemy", False) else "✖" preview = stylize_time("12:34", settings.get("font", "stylized")) return InlineKeyboardMarkup([ [InlineKeyboardButton(f"ساعت {s_clock}", callback_data=f"toggle_clock_{user_id}"), InlineKeyboardButton(f"بولد {s_bold}", callback_data=f"toggle_bold_{user_id}")], [InlineKeyboardButton(f"تغییر فونت: {preview}", callback_data=f"cycle_font_{user_id}")], [InlineKeyboardButton(f"منشی {s_sec}", callback_data=f"toggle_sec_{user_id}"), InlineKeyboardButton(f"سین {s_seen}", callback_data=f"toggle_seen_{user_id}")], [InlineKeyboardButton(f"پیوی {s_pv}", callback_data=f"toggle_pv_{user_id}"), InlineKeyboardButton(f"انتی لوگین {s_anti}", callback_data=f"toggle_anti_{user_id}")], [InlineKeyboardButton(f"تایپ {s_type}", callback_data=f"toggle_type_{user_id}"), InlineKeyboardButton(f"دشمن همگانی {s_enemy}", callback_data=f"toggle_g_enemy_{user_id}")], [InlineKeyboardButton(f"بازی {s_game}", callback_data=f"toggle_game_{user_id}")], [InlineKeyboardButton("بستن پنل ✖", callback_data=f"close_panel_{user_id}")] ]) async def update_profile_clock(client, user_id): """به‌روزرسانی ساعت در اسم""" while user_id in active_bots: try: if data_manager.get_setting(user_id, "clock") and not data_manager.get_setting(user_id, "copy_mode"): me = await client.get_me() current_name = me.first_name base_name = re.sub(r'(?:\s*' + CLOCK_CHARS_REGEX_CLASS + r'+)+$', '', current_name).strip() current_time = datetime.now(TEHRAN_TIMEZONE).strftime("%H:%M") font = data_manager.get_setting(user_id, "font") stylized_time = stylize_time(current_time, font) new_name = f"{base_name} {stylized_time}" if new_name != current_name: await client.update_profile(first_name=new_name) await asyncio.sleep(60) except Exception as e: logging.error(f"Clock update error: {e}") await asyncio.sleep(60) async def start_bot_instance(session_string, phone, user_id): """راه‌اندازی یک نمونه ربات برای کاربر""" 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: {e}") return if user_id in active_bots: for task in active_bots[user_id][1]: task.cancel() # ========== هندلرها ========== @client.on_message(filters.private & ~filters.me) async def private_handler(client, message): user_id = client.me.id # بررسی قفل پیوی if data_manager.get_setting(user_id, "pv_lock"): await message.delete() return # سین (دیده شدن) if data_manager.get_setting(user_id, "auto_seen"): await client.read_chat_history(message.chat.id) # واکنش خودکار reactions = data_manager.get_user(user_id)["reactions"] if str(message.from_user.id) in reactions: try: await client.send_reaction(message.chat.id, message.id, reactions[str(message.from_user.id)]) except: pass # دشمن enemies = data_manager.get_user(user_id)["enemies"] is_enemy = any(e["user_id"] == message.from_user.id and e["chat_id"] == message.chat.id for e in enemies) if is_enemy or data_manager.get_setting(user_id, "global_enemy"): await message.reply_text(random.choice(ENEMY_REPLIES)) return # منشی if data_manager.get_setting(user_id, "secretary"): replied = data_manager.get_user(user_id)["replied_users"] if message.from_user.id not in replied: msg = data_manager.get_setting(user_id, "secretary_msg") or "سلام! در حال حاضر آفلاین هستم." await message.reply_text(msg) replied.append(message.from_user.id) data_manager.update_user(user_id, {"replied_users": replied}) # سکوت muted = data_manager.get_user(user_id)["muted"] if any(m["user_id"] == message.from_user.id and m["chat_id"] == message.chat.id for m in muted): await message.delete() @client.on_message(filters.me & filters.text & ~filters.reply) async def outgoing_handler(client, message): user_id = client.me.id text = message.text if data_manager.get_setting(user_id, "bold"): if not text.startswith(('`', '**', '__', '~~')): try: await message.edit_text(f"**{text}**") except: pass @client.on_message(filters.me & filters.regex(r"^(پنل|panel)$")) async def panel_handler(client, message): user_id = client.me.id await message.edit_text("⚡️ **پنل مدیریت**", reply_markup=generate_panel_markup(user_id)) @client.on_message(filters.me & filters.regex(r"^راهنما$")) async def help_handler(client, message): help_text = """ **🛠 دستورات** ━━━━━━━━━━━━━━━━━━━━ **✦ مدیریت:** » `حذف [تعداد]` » `ذخیره` (ریپلای) » `تکرار [تعداد]` (ریپلای) **✦ دفاعی:** » `دشمن روشن/خاموش` » `بلاک روشن/خاموش` » `سکوت روشن/خاموش` » `ریاکشن [شکلک]` **✦ سرگرمی:** » `تاس` » `بولینگ` **✦ پنل:** `پنل` """ await message.edit_text(help_text) @client.on_message(filters.me & filters.reply) async def reply_handler(client, message): user_id = client.me.id text = message.text replied = message.reply_to_message if not replied or not replied.from_user: return target_id = replied.from_user.id # دشمن if text == "دشمن روشن": enemies = data_manager.get_user(user_id)["enemies"] enemies.append({"user_id": target_id, "chat_id": message.chat.id}) data_manager.update_user(user_id, {"enemies": enemies}) await message.edit_text("⚔️ دشمن اضافه شد.") return if text == "دشمن خاموش": enemies = data_manager.get_user(user_id)["enemies"] enemies = [e for e in enemies if not (e["user_id"] == target_id and e["chat_id"] == message.chat.id)] data_manager.update_user(user_id, {"enemies": enemies}) await message.edit_text("🏳️ دشمن حذف شد.") return # بلاک if text == "بلاک روشن": await client.block_user(target_id) await message.edit_text("🚫 کاربر بلاک شد.") return if text == "بلاک خاموش": await client.unblock_user(target_id) await message.edit_text("⭕️ کاربر آنبلاک شد.") return # سکوت if text == "سکوت روشن": muted = data_manager.get_user(user_id)["muted"] muted.append({"user_id": target_id, "chat_id": message.chat.id}) data_manager.update_user(user_id, {"muted": muted}) await message.edit_text("🔇 کاربر ساکت شد.") return if text == "سکوت خاموش": muted = data_manager.get_user(user_id)["muted"] muted = [m for m in muted if not (m["user_id"] == target_id and m["chat_id"] == message.chat.id)] data_manager.update_user(user_id, {"muted": muted}) await message.edit_text("🔊 کاربر از سکوت خارج شد.") return # ریاکشن if text.startswith("ریاکشن "): emoji = text.split()[1] reactions = data_manager.get_user(user_id)["reactions"] reactions[str(target_id)] = emoji data_manager.update_user(user_id, {"reactions": reactions}) await message.edit_text(f"👍 واکنش {emoji} تنظیم شد.") return if text == "ریاکشن خاموش": reactions = data_manager.get_user(user_id)["reactions"] reactions.pop(str(target_id), None) data_manager.update_user(user_id, {"reactions": reactions}) await message.edit_text("❌ واکنش حذف شد.") return # ذخیره if text == "ذخیره": await replied.forward("me") await message.edit_text("💾 ذخیره شد.") return # تکرار if text.startswith("تکرار "): try: count = int(text.split()[1]) for _ in range(min(count, 10)): await replied.copy(message.chat.id) await asyncio.sleep(0.1) await message.delete() except: pass return # حذف if text.startswith("حذف "): try: count = int(text.split()[1]) msg_ids = [] async for m in client.get_chat_history(message.chat.id, limit=count + 1): if m.from_user and m.from_user.is_self: msg_ids.append(m.id) if msg_ids: await client.delete_messages(message.chat.id, msg_ids) await message.delete() except: pass return # کپی پروفایل if text == "کپی روشن": data_manager.set_setting(user_id, "copy_mode", True) data_manager.set_setting(user_id, "clock", False) await message.edit_text("👤 هویت جعل شد.") return if text == "کپی خاموش": data_manager.set_setting(user_id, "copy_mode", False) await message.edit_text("👤 هویت بازگردانده شد.") return @client.on_callback_query() async def callback_handler(client, callback): user_id = int(callback.data.split("_")[-1]) if callback.from_user.id != user_id: await callback.answer("⛔️ دسترسی غیرمجاز!", show_alert=True) return action = "_".join(callback.data.split("_")[:-1]) if action == "toggle_clock": val = not data_manager.get_setting(user_id, "clock") data_manager.set_setting(user_id, "clock", val) elif action == "toggle_bold": val = not data_manager.get_setting(user_id, "bold") data_manager.set_setting(user_id, "bold", val) elif action == "cycle_font": current = data_manager.get_setting(user_id, "font") idx = (FONT_KEYS.index(current) + 1) % len(FONT_KEYS) data_manager.set_setting(user_id, "font", FONT_KEYS[idx]) data_manager.set_setting(user_id, "clock", True) elif action == "toggle_sec": val = not data_manager.get_setting(user_id, "secretary") data_manager.set_setting(user_id, "secretary", val) elif action == "toggle_seen": val = not data_manager.get_setting(user_id, "auto_seen") data_manager.set_setting(user_id, "auto_seen", val) elif action == "toggle_pv": val = not data_manager.get_setting(user_id, "pv_lock") data_manager.set_setting(user_id, "pv_lock", val) elif action == "toggle_anti": val = not data_manager.get_setting(user_id, "anti_login") data_manager.set_setting(user_id, "anti_login", val) elif action == "toggle_type": val = not data_manager.get_setting(user_id, "typing") data_manager.set_setting(user_id, "typing", val) if val: data_manager.set_setting(user_id, "playing", False) elif action == "toggle_game": val = not data_manager.get_setting(user_id, "playing") data_manager.set_setting(user_id, "playing", val) if val: data_manager.set_setting(user_id, "typing", False) elif action == "toggle_g_enemy": val = not data_manager.get_setting(user_id, "global_enemy") data_manager.set_setting(user_id, "global_enemy", val) elif action == "close_panel": try: await callback.message.delete() except: pass await callback.answer() return await callback.answer() try: await callback.message.edit_text("⚡️ **پنل مدیریت**", reply_markup=generate_panel_markup(user_id)) except: pass # ========== تسک‌های پس‌زمینه ========== tasks = [ asyncio.create_task(update_profile_clock(client, user_id)) ] active_bots[user_id] = (client, tasks) logging.info(f"✅ Bot started for user {user_id}") # ==================== ربات مدیر ==================== manager_bot = Client("manager_bot", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN) @manager_bot.on_message(filters.private & ~filters.bot) async def manager_handler(client, message): user_id = message.from_user.id text = message.text or "" if text == "/start": await message.reply_text("👋 ربات سلف‌بات\nارسال شماره برای اتصال:") return # Login flow if user_id in login_states and login_states[user_id]["step"] == "waiting_code": code = re.sub(r"\D+", "", text) state = login_states[user_id] user_client = state["client"] try: await user_client.sign_in(state["phone"], state["hash"], code) session_string = await user_client.export_session_string() me = await user_client.get_me() data_manager.update_user(me.id, { "phone": state["phone"], "session_string": session_string, "first_name": me.first_name or "", "username": me.username or "" }) await user_client.disconnect() del login_states[user_id] await message.reply_text("✅ اتصال موفق! دستور `پنل` را بزنید.") asyncio.create_task(start_bot_instance(session_string, state["phone"], me.id)) except Exception as e: await message.reply_text(f"❌ خطا: {e}") return # شماره if text.startswith("+") or text.isdigit(): try: user_client = Client(f"login_{user_id}", api_id=API_ID, api_hash=API_HASH, in_memory=True) await user_client.connect() sent = await user_client.send_code(text) login_states[user_id] = { "step": "waiting_code", "phone": text, "client": user_client, "hash": sent.phone_code_hash } await message.reply_text("✅ کد تأیید را وارد کنید:") except Exception as e: await message.reply_text(f"❌ خطا: {e}") @manager_bot.on_inline_query() async def inline_handler(client, query): if query.query == "panel": user_id = query.from_user.id result = InlineQueryResultArticle( title="پنل مدیریت", input_message_content=InputTextMessageContent(f"⚡️ **پنل مدیریت**\n👤 کاربر: {user_id}"), reply_markup=generate_panel_markup(user_id) ) await query.answer([result], cache_time=0) @manager_bot.on_callback_query() async def manager_callback(client, callback): user_id = int(callback.data.split("_")[-1]) if callback.from_user.id != user_id: await callback.answer("⛔️ دسترسی غیرمجاز!", show_alert=True) return action = "_".join(callback.data.split("_")[:-1]) if action == "toggle_clock": val = not data_manager.get_setting(user_id, "clock") data_manager.set_setting(user_id, "clock", val) elif action == "toggle_bold": val = not data_manager.get_setting(user_id, "bold") data_manager.set_setting(user_id, "bold", val) elif action == "cycle_font": current = data_manager.get_setting(user_id, "font") idx = (FONT_KEYS.index(current) + 1) % len(FONT_KEYS) data_manager.set_setting(user_id, "font", FONT_KEYS[idx]) data_manager.set_setting(user_id, "clock", True) elif action == "toggle_sec": val = not data_manager.get_setting(user_id, "secretary") data_manager.set_setting(user_id, "secretary", val) elif action == "toggle_seen": val = not data_manager.get_setting(user_id, "auto_seen") data_manager.set_setting(user_id, "auto_seen", val) elif action == "toggle_pv": val = not data_manager.get_setting(user_id, "pv_lock") data_manager.set_setting(user_id, "pv_lock", val) elif action == "toggle_anti": val = not data_manager.get_setting(user_id, "anti_login") data_manager.set_setting(user_id, "anti_login", val) elif action == "toggle_type": val = not data_manager.get_setting(user_id, "typing") data_manager.set_setting(user_id, "typing", val) if val: data_manager.set_setting(user_id, "playing", False) elif action == "toggle_game": val = not data_manager.get_setting(user_id, "playing") data_manager.set_setting(user_id, "playing", val) if val: data_manager.set_setting(user_id, "typing", False) elif action == "toggle_g_enemy": val = not data_manager.get_setting(user_id, "global_enemy") data_manager.set_setting(user_id, "global_enemy", val) elif action == "close_panel": await callback.answer() try: await callback.message.delete() except: pass return await callback.answer() try: await callback.message.edit_text("⚡️ **پنل مدیریت**", reply_markup=generate_panel_markup(user_id)) except: pass # ==================== Webhook Handler ==================== async def webhook_handler(request): """پردازش Webhook از تلگرام""" data = await request.json() logging.info(f"Webhook received: {data}") # ارسال به ربات مدیر update = data if "message" in update: await manager_bot.process_update(update) elif "callback_query" in update: await manager_bot.process_update(update) elif "inline_query" in update: await manager_bot.process_update(update) return web.Response(status=200) async def set_webhook(): """تنظیم Webhook در تلگرام""" url = f"https://api.telegram.org/bot{BOT_TOKEN}/setWebhook" webhook_url = f"{WEBHOOK_URL}{WEBHOOK_PATH}" async with aiohttp.ClientSession() as session: async with session.post(url, data={"url": webhook_url}) as resp: result = await resp.json() logging.info(f"Webhook set result: {result}") return result async def start_webhook_server(): """راه‌اندازی سرور Webhook""" app = web.Application() app.router.add_post(WEBHOOK_PATH, webhook_handler) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, "0.0.0.0", WEBHOOK_PORT) await site.start() logging.info(f"🌐 Webhook server running on port {WEBHOOK_PORT}") return runner, site # ==================== Main ==================== async def main(): """تابع اصلی""" # راه‌اندازی Webhook runner, site = await start_webhook_server() # تنظیم Webhook در تلگرام await set_webhook() # شروع ربات مدیر await manager_bot.start() logging.info("✅ Manager bot started") # بارگذاری نشست‌های قبلی for phone, session_data in data_manager.data["sessions"].items(): session_string = session_data["string"] user_id = session_data["user_id"] asyncio.create_task(start_bot_instance(session_string, phone, user_id)) try: # نگهداری برنامه await asyncio.Event().wait() except KeyboardInterrupt: logging.info("Shutting down...") # پاکسازی for user_id, (client, tasks) in active_bots.items(): for task in tasks: task.cancel() await client.stop() await manager_bot.stop() await runner.cleanup() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logging.info("👋 Goodbye!")