# script was programmed by: Kodo # For any inquiries: @DevKodo # Data source and operation: @SourceKodo # com.py version: 1.2 (broadcast/open-msg run in background with dedicated session) import asyncio import logging import aiohttp import aiosqlite import sys import os import re FACTORY_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, FACTORY_DIR) from aiogram import Bot, Dispatcher, Router, F from aiogram.types import ( Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton, ChatMemberUpdated, ) from aiogram.filters import CommandStart from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage from cobots.communication.db import ( init_db, get_or_create_user, get_user, get_setting, set_setting, get_all_settings, get_mandatory_channels, save_message, update_dev_message_id, update_group_message_id, save_reply, get_message_by_user_msg_id, get_message_by_dev_msg_id, get_message_by_group_msg_id, get_message_by_reply_dev_msg_id, get_message_by_reply_message_id, get_muted_users, set_user_muted, get_admins, is_admin, add_admin, remove_admin, clear_admins, get_groups, remove_mandatory_channel, remove_group, increment_sent, increment_received, get_statistics, add_mandatory_channel, add_group, get_all_active_users, update_mandatory_channel_link, get_db_path, add_custom_button, get_custom_buttons, get_custom_button, update_custom_button, delete_custom_button, ) from information.database import get_session, is_bot_paid as _is_paid_bot from information.models import Bot as BotModel, FactorySettings from sqlalchemy import select logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s" ) logger = logging.getLogger(__name__) router = Router() dev_router = Router() class UserStates(StatesGroup): waiting_message = State() class DevStates(StatesGroup): waiting_reply = State() waiting_broadcast_msg = State() waiting_broadcast_fwd = State() waiting_add_admin = State() waiting_mand_type = State() waiting_mand_fwd = State() waiting_group_fwd = State() waiting_welcome = State() waiting_close_msg = State() waiting_open_msg = State() waiting_send_btn_label = State() waiting_back_btn_label = State() waiting_cbtn_label = State() waiting_cbtn_content = State() waiting_cbtn_edit_label = State() waiting_cbtn_edit_content = State() waiting_cbtn_emoji = State() waiting_cbtn_name = State() BOT_TOKEN : str = "" BOT_USERNAME : str = "" OWNER_ID : int = 0 FACTORY_TOKEN: str = "" _FACTORY_URL_CACHE: dict = {} # توكن المصنع -> رابطه (يمنع تداخل الحقوق) MAKER_MSG = "🤖 ادري عجبك البوت يلا اصنع واحد مثله مجاني !" BROADCAST_BATCH = 5 BROADCAST_DELAY = 2 async def _build_dev_panel_text(bot_label: str) -> str: from information.database import get_bot_subscription from information.models import Bot as _BotModel from information.database import get_session as _get_session from sqlalchemy import select as _select text = ( f"» هذه هي لوحة التحكم وادارة {bot_label} الخاص بك.\n\n" "» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، " "متابعة الإحصائيات الفورية وإدارة القنوات وكل شيء من مكان واحد." ) factory_url = await get_factory_url() factory_username = factory_url.replace("https://t.me/", "") if factory_url else "" sub_deeplink = f"https://t.me/{factory_username}?start=subscription" if factory_username else "" async with _get_session() as _sess: _res = await _sess.execute( _select(_BotModel).where(_BotModel.username.ilike(BOT_USERNAME), _BotModel.is_active == True) ) _bot_rec = _res.scalar_one_or_none() if _bot_rec and _bot_rec.is_paid: _sub = await get_bot_subscription(_bot_rec.id) _expires = _sub.expires_at.strftime("%d-%m-%Y") if _sub else "—" text += ( f"\n» البوت مدفوع حتى: {_expires}" f"\n» يمكنك ادارة الاشتراك من هنا:" + (f'\n@{factory_username}' if sub_deeplink else "") ) elif _bot_rec and not _bot_rec.is_paid and sub_deeplink: text += ( f"\n» اجعل بوتك مدفوعاً من هنا:" f'\n@{factory_username}' ) text += "\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:" return text async def get_factory_url() -> str: # cache لكل توكن على حدة (يمنع تداخل الحقوق بين البوتات المختلفة) if not FACTORY_TOKEN: return "" cached = _FACTORY_URL_CACHE.get(FACTORY_TOKEN) if cached: return cached try: async with aiohttp.ClientSession() as session: async with session.get( f"https://api.telegram.org/bot{FACTORY_TOKEN}/getMe", timeout=aiohttp.ClientTimeout(total=10) ) as resp: data = await resp.json() if data.get("ok"): username = data["result"]["username"] url = f"https://t.me/{username}" _FACTORY_URL_CACHE[FACTORY_TOKEN] = url return url except Exception: pass return "" async def notify_owner(bot: Bot, text: str, reply_markup=None): try: await bot.send_message(OWNER_ID, text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=reply_markup) except Exception: pass async def is_dev_or_admin(user_id: int) -> bool: if user_id == OWNER_ID: return True return await is_admin(BOT_USERNAME, user_id) def make_btn(label: str, color: str = "default", emoji: str = "", callback_data: str = None, url: str = None) -> InlineKeyboardButton: text = f"{emoji} {label}".strip() if emoji else label kwargs = {"text": text} if url: kwargs["url"] = url elif callback_data: kwargs["callback_data"] = callback_data if color and color != "default": color_map = {"green": "success", "blue": "primary", "red": "danger"} style = color_map.get(color) if style: kwargs["style"] = style return InlineKeyboardButton(**kwargs) async def get_send_btn() -> InlineKeyboardButton: s = await get_all_settings(BOT_USERNAME) return make_btn( label = s.get("send_btn_label", "إرسال"), color = s.get("send_btn_color", "default"), emoji = s.get("send_btn_emoji", ""), callback_data = "user:send" ) async def get_back_btn(cb: str = "user:back") -> InlineKeyboardButton: s = await get_all_settings(BOT_USERNAME) return make_btn( label = s.get("back_btn_label", "رجوع"), color = s.get("back_btn_color", "default"), emoji = s.get("back_btn_emoji", ""), callback_data = cb ) def build_dev_keyboard(user_id: int, user_msg_id: int) -> InlineKeyboardMarkup: return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="كتم", callback_data=f"dev:mute:{user_id}", style="danger")], [InlineKeyboardButton(text="معلومات المستخدم", callback_data=f"dev:info:{user_id}", style="primary")], ]) async def check_subscription(bot: Bot, user_id: int) -> list[dict]: channels = await get_mandatory_channels(BOT_USERNAME) not_subbed = [] for ch in channels: try: member = await asyncio.wait_for( bot.get_chat_member(int(ch["channel_id"]), user_id), timeout=5 ) if member.status in ("left", "kicked", "banned"): not_subbed.append(ch) except Exception: not_subbed.append(ch) return not_subbed async def send_sub_request(bot: Bot, chat_id: int, channel: dict): link = channel.get("invite_link") or ( f"https://t.me/{channel['channel_username']}" if channel.get("channel_username") else None ) if not link: return if channel.get("display_type") == "buttons": await bot.send_message( chat_id, "عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton( text=f"@{channel.get('channel_username') or channel['channel_title']}", url=link )], [InlineKeyboardButton(text="✅ تحقق من الاشتراك", callback_data="check_sub")], ]) ) else: await bot.send_message( chat_id, f"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:\n" f"- {channel['channel_title']}\n\n" "اشترك ثم أرسل /start", parse_mode="HTML", disable_web_page_preview=True) def _entities_to_json(entities): import json as _json if not entities: return None return _json.dumps([ {"type": e.type, "offset": e.offset, "length": e.length, "custom_emoji_id": getattr(e, "custom_emoji_id", None), "url": getattr(e, "url", None)} for e in entities ]) def _entities_from_json(s): import json as _json if not s: return None try: from aiogram.types import MessageEntity as _ME data = _json.loads(s) ents = [] for d in data: kwargs = {"type": d["type"], "offset": d["offset"], "length": d["length"]} if d.get("custom_emoji_id"): kwargs["custom_emoji_id"] = d["custom_emoji_id"] if d.get("url"): kwargs["url"] = d["url"] ents.append(_ME(**kwargs)) return ents except Exception: return None def _has_custom_emoji(entities): if not entities: return False return any(getattr(e, "type", None) == "custom_emoji" for e in entities) async def build_welcome_message(user): settings = await get_all_settings(BOT_USERNAME) welcome = settings.get("welcome_message", "") ent_json = settings.get("welcome_message_entities", "") factory_url = await get_factory_url() maker_rights = await get_factory_maker_rights() name_link = f'{user.first_name}' entities = _entities_from_json(ent_json) if welcome and _has_custom_emoji(entities): ph = "{name}"; idx = welcome.find(ph) text = welcome.replace(ph, user.first_name) if idx >= 0: shift = len(user.first_name) - len(ph) for e in entities: if e.offset > idx: e.offset += shift return text, entities if welcome: text = welcome.replace("{name}", name_link) else: text = ( f"» اهلاً عزيزي: {name_link}\n» بوت التواصل الخاص بي يرحب بك 🤍.\n» ارسل رسالتك الان وسأقوم بإرسالها الى فريق البوت.\n» سيتم الرد عليك في اقرب وقت ممكن." ) if maker_rights and factory_url: fac_un = factory_url.replace("https://t.me/", "").replace("/", "") text += f'\n» 🤖 ادري عجبك البوت يلا اصنع واحد مثلة مجاني!\n» @{fac_un}' return text, None async def get_factory_maker_rights() -> bool: try: if await _is_paid_bot(BOT_USERNAME): return False async with get_session() as session: result = await session.execute(select(FactorySettings).limit(1)) settings = result.scalar_one_or_none() return settings.cobots_maker_rights if settings else False except Exception: return False async def get_receive_targets() -> tuple[bool, list[dict]]: in_group = await get_setting(BOT_USERNAME, "receive_in_group") == "1" groups = await get_groups(BOT_USERNAME) if in_group else [] return in_group, groups async def send_to_dev(bot: Bot, original_msg: Message, user_id: int, user_msg_id: int) -> tuple[int | None, int | None, int | None]: in_group, groups = await get_receive_targets() keyboard = build_dev_keyboard(user_id, user_msg_id) action_text = "» وصلت رسالة جديدة:" dev_msg_id = None group_msg_id = None group_chat_id = None if not in_group: try: fwd = await original_msg.forward(OWNER_ID) dev_msg_id = fwd.message_id await bot.send_message( OWNER_ID, action_text, parse_mode = "HTML", disable_web_page_preview = True, reply_markup = keyboard, reply_to_message_id = fwd.message_id, ) except Exception as e: logger.error(f"خطأ في إرسال للمطور: {e}") else: for group in groups: try: fwd = await original_msg.forward(group["chat_id"]) sent = await bot.send_message( group["chat_id"], action_text, parse_mode = "HTML", disable_web_page_preview = True, reply_markup = keyboard, reply_to_message_id = fwd.message_id, ) group_msg_id = fwd.message_id group_chat_id = group["chat_id"] except Exception as e: logger.error(f"خطأ في إرسال للمجموعة {group['chat_id']}: {e}") return dev_msg_id, group_msg_id, group_chat_id @router.message(CommandStart()) async def cmd_start(message: Message, state: FSMContext, bot: Bot): await state.clear() user = message.from_user is_open = await get_setting(BOT_USERNAME, "is_open") == "1" if not is_open and not await is_dev_or_admin(user.id): closed_msg = await get_setting(BOT_USERNAME, "closed_message") await message.answer(closed_msg or "🔴 البوت مقفول حالياً.") return db_user = await get_or_create_user(BOT_USERNAME, user.id, user.username, user.first_name) if db_user["is_blocked"]: await message.answer("⛔ أنت محظور من استخدام هذا البوت.") return not_subbed = await check_subscription(bot, user.id) if not_subbed: await send_sub_request(bot, message.chat.id, not_subbed[0]) return settings = await get_all_settings(BOT_USERNAME) if settings.get("notify_new_users") == "1": from datetime import datetime, timedelta joined = db_user.get("joined_at", "") if joined: try: jt = datetime.fromisoformat(joined) if (datetime.utcnow() - jt).total_seconds() < 10: name = f"@{user.username}" if user.username else user.first_name await notify_owner( bot, f"👤 مستخدم جديد!\n" f"الاسم: {user.first_name}\n" f"المعرف: {name}\n" f"الـ ID: {user.id}" ) except Exception: pass if await is_dev_or_admin(user.id): panel_text = await _build_dev_panel_text("بوت التواصل") if await get_factory_maker_rights(): panel_text += f"\n{MAKER_MSG}" await message.answer(panel_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_dev_panel()) welcome_text, w_ents = await build_welcome_message(user) if w_ents: await message.answer(welcome_text, entities=w_ents, disable_web_page_preview=True, reply_markup=await _user_custom_kb()) else: await message.answer(welcome_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await _user_custom_kb()) await state.set_state(UserStates.waiting_message) @router.callback_query(F.data == "check_sub") async def cb_check_sub(callback: CallbackQuery, state: FSMContext, bot: Bot): user = callback.from_user not_subbed = await check_subscription(bot, user.id) if not_subbed: await send_sub_request(bot, callback.message.chat.id, not_subbed[0]) await callback.answer() return welcome_text, w_ents = await build_welcome_message(user) if w_ents: await callback.message.answer(welcome_text, entities=w_ents, disable_web_page_preview=True, reply_markup=await _user_custom_kb()) else: await callback.message.answer(welcome_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await _user_custom_kb()) await state.set_state(UserStates.waiting_message) await callback.answer() @router.message((F.text | F.photo | F.video | F.document | F.voice | F.audio | F.sticker | F.animation), ~F.reply_to_message) async def user_message_handler_any(message: Message, state: FSMContext, bot: Bot): if await is_dev_or_admin(message.from_user.id): return current_state = await state.get_state() if current_state != UserStates.waiting_message.state: await state.set_state(UserStates.waiting_message) await user_message_handler(message, state, bot) @router.message(UserStates.waiting_message) async def user_message_handler(message: Message, state: FSMContext, bot: Bot): user = message.from_user db_user = await get_user(BOT_USERNAME, user.id) if not db_user: await cmd_start(message, state, bot) return if db_user["is_muted"]: await message.answer("🔇 أنت مكتوم ولا يمكنك إرسال رسائل حالياً.") return if db_user["is_blocked"]: await message.answer("⛔ أنت محظور من استخدام هذا البوت.") return not_subbed = await check_subscription(bot, user.id) if not_subbed: await send_sub_request(bot, message.chat.id, not_subbed[0]) return data = { "pending_message_id": message.message_id, "pending_user_id": user.id, "pending_text": message.text or "", "pending_caption": message.caption or "", "pending_photo_id": message.photo[-1].file_id if message.photo else None, "pending_video_id": message.video.file_id if message.video else None, "pending_document_id": message.document.file_id if message.document else None, "pending_voice_id": message.voice.file_id if message.voice else None, "pending_audio_id": message.audio.file_id if message.audio else None, "pending_sticker_id": message.sticker.file_id if message.sticker else None, } dev_msg_id, group_msg_id, group_chat_id = await send_to_dev( bot, message, user.id, message.message_id ) await save_message( BOT_USERNAME, user.id, message.message_id, dev_msg_id, group_msg_id, group_chat_id ) await increment_received(BOT_USERNAME) async with aiosqlite.connect(get_db_path(BOT_USERNAME)) as _db: async with _db.execute( "SELECT COUNT(*) FROM messages WHERE user_id = ?", (user.id,) ) as _cur: (_count,) = await _cur.fetchone() if _count <= 1: await message.answer( "» تم ارسال رسالتك، انتضر قليلاً ليتم الرد عليك.", parse_mode="HTML", disable_web_page_preview=True) @router.message(F.reply_to_message, ~F.via_bot) async def user_reply_handler(message: Message, state: FSMContext, bot: Bot): user = message.from_user if await is_dev_or_admin(user.id): return db_user = await get_user(BOT_USERNAME, user.id) if not db_user or db_user.get("is_muted") or db_user.get("is_blocked"): return replied_id = message.reply_to_message.message_id ctx = await get_message_by_reply_message_id(BOT_USERNAME, replied_id) if ctx: in_group, groups = await get_receive_targets() keyboard = build_dev_keyboard(user.id, message.message_id) action = "» وصلت رسالة جديدة:" if not in_group: try: fwd = await message.forward(OWNER_ID) await bot.send_message( OWNER_ID, action, parse_mode="HTML", disable_web_page_preview=True, reply_markup=keyboard, reply_to_message_id=fwd.message_id, ) await save_message(BOT_USERNAME, user.id, message.message_id, fwd.message_id, None, None) await increment_received(BOT_USERNAME) except Exception as e: logger.error(f"خطأ في user_reply: {e}") else: for group in groups: try: fwd = await message.forward(group["chat_id"]) await bot.send_message( group["chat_id"], action, parse_mode="HTML", disable_web_page_preview=True, reply_markup=keyboard, reply_to_message_id=fwd.message_id, ) await save_message(BOT_USERNAME, user.id, message.message_id, None, fwd.message_id, group["chat_id"]) await increment_received(BOT_USERNAME) except Exception as e: logger.error(f"خطأ في user_reply group: {e}") else: await user_message_handler(message, state, bot) @router.callback_query(F.data == "user:back") async def cb_user_back(callback: CallbackQuery, state: FSMContext, bot: Bot): await state.clear() user = callback.from_user welcome_text, w_ents = await build_welcome_message(user) if w_ents: await callback.message.edit_text(welcome_text, entities=w_ents, disable_web_page_preview=True) else: await callback.message.edit_text(welcome_text, parse_mode="HTML", disable_web_page_preview=True) await state.set_state(UserStates.waiting_message) await callback.answer() @router.callback_query(F.data == "user:send") async def cb_user_send(callback: CallbackQuery, state: FSMContext, bot: Bot): data = await state.get_data() pending_msg_id = data.get("pending_message_id") user = callback.from_user if not pending_msg_id: await callback.answer("⚠️ لم يتم العثور على الرسالة!", show_alert=True) return dev_msg_id, group_msg_id, group_chat_id = await send_to_dev_from_data( bot, data, user.id, pending_msg_id ) await save_message( BOT_USERNAME, user.id, pending_msg_id, dev_msg_id, group_msg_id, group_chat_id ) await increment_received(BOT_USERNAME) await callback.message.edit_text( "» تم ارسال رسالتك، انتضر قليلاً ليتم الرد عليك.\n\n", parse_mode="HTML", disable_web_page_preview=True) await state.set_state(UserStates.waiting_message) await callback.answer() async def send_to_dev_from_data(bot: Bot, data: dict, user_id: int, user_msg_id: int) -> tuple[int | None, int | None, int | None]: in_group, groups = await get_receive_targets() keyboard = build_dev_keyboard(user_id, user_msg_id) text = data.get("pending_text", "") caption = data.get("pending_caption", "") photo_id = data.get("pending_photo_id") video_id = data.get("pending_video_id") document_id = data.get("pending_document_id") voice_id = data.get("pending_voice_id") audio_id = data.get("pending_audio_id") sticker_id = data.get("pending_sticker_id") header = f"» رسالة جديدة:\n{text or caption}" if (text or caption) else "» رسالة جديدة:" targets = [OWNER_ID] if not in_group else [g["chat_id"] for g in groups] dev_msg_id = None group_msg_id = None group_chat_id = None action_text = "» وصلت رسالة جديدة:" for target in targets: try: if text: fwd = await bot.send_message(target, text, parse_mode="HTML", disable_web_page_preview=True) elif photo_id: fwd = await bot.send_photo(target, photo_id, caption=caption or "", parse_mode="HTML", disable_web_page_preview=True) elif video_id: fwd = await bot.send_video(target, video_id, caption=caption or "", parse_mode="HTML", disable_web_page_preview=True) elif document_id: fwd = await bot.send_document(target, document_id, caption=caption or "", parse_mode="HTML", disable_web_page_preview=True) elif voice_id: fwd = await bot.send_voice(target, voice_id, caption=caption or "", parse_mode="HTML", disable_web_page_preview=True) elif audio_id: fwd = await bot.send_audio(target, audio_id, caption=caption or "", parse_mode="HTML", disable_web_page_preview=True) elif sticker_id: fwd = await bot.send_sticker(target, sticker_id) else: fwd = await bot.send_message(target, "» رسالة جديدة", parse_mode="HTML", disable_web_page_preview=True) sent = await bot.send_message( target, action_text, parse_mode = "HTML", disable_web_page_preview = True, reply_markup = keyboard, reply_to_message_id = fwd.message_id, ) if target == OWNER_ID: dev_msg_id = fwd.message_id else: group_msg_id = fwd.message_id group_chat_id = target except Exception as e: logger.error(f"خطأ في إرسال للهدف {target}: {e}") return dev_msg_id, group_msg_id, group_chat_id @router.edited_message() async def user_edited_message(message: Message, bot: Bot): logger.info(f"✏️ user_edited_message triggered: user={message.from_user.id} msg_id={message.message_id}") user_id = message.from_user.id if await is_dev_or_admin(user_id): return msg_record = await get_message_by_user_msg_id(BOT_USERNAME, user_id, message.message_id) if not msg_record: return new_text = message.text or message.caption or "" new_caption = message.caption if not message.text else None if msg_record.get("dev_message_id"): try: if message.text: await bot.edit_message_text( chat_id = OWNER_ID, message_id = msg_record["dev_message_id"], text = f"» رسالة جديدة:\n{new_text}", parse_mode = "HTML", disable_web_page_preview=True, reply_markup = build_dev_keyboard(user_id, message.message_id), ) elif message.caption: await bot.edit_message_caption( chat_id = OWNER_ID, message_id = msg_record["dev_message_id"], caption = f"» رسالة جديدة:\n{new_caption}", parse_mode = "HTML", disable_web_page_preview=True, reply_markup = build_dev_keyboard(user_id, message.message_id), ) except Exception as e: logger.error(f"خطأ في تعديل رسالة المطور: {e}") if msg_record.get("group_message_id") and msg_record.get("group_chat_id"): try: if message.text: await bot.edit_message_text( chat_id = msg_record["group_chat_id"], message_id = msg_record["group_message_id"], text = f"» رسالة جديدة:\n{new_text}", parse_mode = "HTML", disable_web_page_preview=True, reply_markup = build_dev_keyboard(user_id, message.message_id), ) elif message.caption: await bot.edit_message_caption( chat_id = msg_record["group_chat_id"], message_id = msg_record["group_message_id"], caption = f"» رسالة جديدة:\n{new_caption}", parse_mode = "HTML", disable_web_page_preview=True, reply_markup = build_dev_keyboard(user_id, message.message_id), ) except Exception as e: logger.error(f"خطأ في تعديل رسالة المجموعة: {e}") async def _is_dev_reply_filter(message: Message) -> bool: if not message.reply_to_message: return False return await is_dev_or_admin(message.from_user.id) @dev_router.message(_is_dev_reply_filter) async def handle_direct_reply(message: Message, state: FSMContext, bot: Bot): replied_msg_id = message.reply_to_message.message_id msg_record = ( await get_message_by_dev_msg_id(BOT_USERNAME, replied_msg_id) or await get_message_by_group_msg_id(BOT_USERNAME, replied_msg_id) ) if not msg_record: return target_user_id = msg_record["user_id"] user_msg_id = msg_record["user_message_id"] try: sent = await message.copy_to( chat_id = target_user_id, reply_to_message_id = user_msg_id, ) await save_reply( BOT_USERNAME, target_user_id, user_msg_id, sent.message_id, message.message_id ) await increment_sent(BOT_USERNAME) user = await get_user(BOT_USERNAME, target_user_id) name = f"@{user['username']}" if user and user.get("username") else ( user["first_name"] if user else str(target_user_id) ) await message.answer( f"» تم إرسال ردك إلى: {name}", parse_mode="HTML", disable_web_page_preview=True ) except Exception as e: await message.answer( f"⚠️ فشل الإرسال: {e}", parse_mode="HTML", disable_web_page_preview=True ) @dev_router.callback_query(F.data.startswith("dev:reply:")) async def cb_dev_reply(callback: CallbackQuery, state: FSMContext, bot: Bot): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return parts = callback.data.split(":") target_user_id = int(parts[2]) user_msg_id = int(parts[3]) original_text = callback.message.text or callback.message.caption or "" REPLY_SUFFIX = "\n\n» حسناً، قم بإرسال ردك الآن:" clean_text = original_text.replace(REPLY_SUFFIX, "").strip() await state.update_data( reply_to_user_id = target_user_id, reply_to_msg_id = user_msg_id, reply_dev_msg_id = callback.message.message_id, reply_original_text = clean_text, reply_is_caption = bool(callback.message.caption), ) await state.set_state(DevStates.waiting_reply) cancel_kb = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="الغاء الرد", callback_data="dev:cancel_reply", style="primary")] ]) new_text = clean_text + REPLY_SUFFIX try: if callback.message.caption: await callback.message.edit_caption(new_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=cancel_kb) else: await callback.message.edit_text(new_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=cancel_kb) except Exception: pass await callback.answer() @dev_router.callback_query(F.data == "dev:cancel_reply") async def cb_dev_cancel_reply(callback: CallbackQuery, state: FSMContext): data = await state.get_data() user_id = data.get("reply_to_user_id") user_msg_id = data.get("reply_to_msg_id") original = data.get("reply_original_text", "") is_caption = data.get("reply_is_caption", False) await state.clear() kb = build_dev_keyboard(user_id, user_msg_id) if user_id else None try: if is_caption: await callback.message.edit_caption(original, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) else: await callback.message.edit_text(original, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) except Exception: pass await callback.answer() @dev_router.message(DevStates.waiting_reply) async def dev_reply_handler(message: Message, state: FSMContext, bot: Bot): data = await state.get_data() target_user_id = data.get("reply_to_user_id") user_msg_id = data.get("reply_to_msg_id") reply_dev_msg_id = data.get("reply_dev_msg_id") if not target_user_id: await state.clear() return try: sent = await message.copy_to( chat_id = target_user_id, reply_to_message_id = user_msg_id, ) reply_msg_id = sent.message_id reply_dev_msg_id_saved = message.message_id await save_reply( BOT_USERNAME, target_user_id, user_msg_id, reply_msg_id, reply_dev_msg_id_saved ) await increment_sent(BOT_USERNAME) user = await get_user(BOT_USERNAME, target_user_id) name = f"@{user['username']}" if user and user.get("username") else ( user["first_name"] if user else str(target_user_id) ) await message.answer( f"» تم ارسال ردك الى: {name}\n", parse_mode="HTML", disable_web_page_preview=True) if reply_dev_msg_id: try: pass except Exception: pass except Exception as e: await message.answer(f"❌ فشل إرسال الرد: {e}") await state.clear() @dev_router.edited_message() async def dev_edited_reply(message: Message, bot: Bot): if not await is_dev_or_admin(message.from_user.id): return msg_record = await get_message_by_reply_dev_msg_id(BOT_USERNAME, message.message_id) if not msg_record: return try: if message.text: await bot.edit_message_text( chat_id = msg_record["user_id"], message_id = msg_record["reply_message_id"], text = message.text ) elif message.caption: await bot.edit_message_caption( chat_id = msg_record["user_id"], message_id = msg_record["reply_message_id"], caption = message.caption ) except Exception as e: logger.error(f"خطأ في تعديل رد المطور: {e}") @dev_router.callback_query(F.data.startswith("dev:mute:")) async def cb_dev_mute(callback: CallbackQuery, bot: Bot): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return target_id = int(callback.data.split(":")[2]) user = await get_user(BOT_USERNAME, target_id) name = f"@{user['username']}" if user and user.get("username") else ( user["first_name"] if user else str(target_id) ) if user and user["is_muted"]: await callback.message.edit_reply_markup( reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="الغاء الكتم", callback_data=f"dev:unmute:{target_id}", style="danger")], [InlineKeyboardButton(text="رجوع", callback_data=f"dev:restore_kb:{target_id}:{callback.message.message_id}", style="primary")], ]) ) await callback.answer("هذا المستخدم مكتوم بالفعل!", show_alert=True) else: await callback.message.edit_reply_markup( reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"نعم، كتم {name}", callback_data=f"dev:mute_confirm:{target_id}", style="danger")], [InlineKeyboardButton(text="رجوع", callback_data=f"dev:restore_kb:{target_id}:{callback.message.message_id}", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("dev:mute_confirm:")) async def cb_dev_mute_confirm(callback: CallbackQuery, bot: Bot): target_id = int(callback.data.split(":")[2]) await set_user_muted(BOT_USERNAME, target_id, True) try: await bot.send_message(target_id, "🔇 تم كتمك من إرسال الرسائل في هذا البوت.") except Exception: pass user = await get_user(BOT_USERNAME, target_id) name = f"@{user['username']}" if user and user.get("username") else ( user["first_name"] if user else str(target_id) ) try: await callback.message.edit_reply_markup( reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="رد", callback_data=f"dev:reply:{target_id}:0", style="success"), InlineKeyboardButton(text="كتم", callback_data=f"dev:mute:{target_id}", style="danger"), ], [InlineKeyboardButton(text="معلومات المستخدم", callback_data=f"dev:info:{target_id}", style="primary")], ]) ) except Exception: pass await callback.answer(f"✅ تم كتم {name}.", show_alert=True) @dev_router.callback_query(F.data.startswith("dev:unmute:")) async def cb_dev_unmute(callback: CallbackQuery, bot: Bot): target_id = int(callback.data.split(":")[2]) await set_user_muted(BOT_USERNAME, target_id, False) try: await bot.send_message(target_id, "🔊 تم إلغاء كتمك، يمكنك إرسال الرسائل مجدداً.") except Exception: pass await callback.answer("✅ تم إلغاء الكتم.", show_alert=True) try: await callback.message.edit_reply_markup( reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="رد", callback_data=f"dev:reply:{target_id}:0", style="success"), InlineKeyboardButton(text="كتم", callback_data=f"dev:mute:{target_id}", style="danger"), ], [InlineKeyboardButton(text="معلومات المستخدم", callback_data=f"dev:info:{target_id}", style="primary")], ]) ) except Exception: pass @dev_router.callback_query(F.data.startswith("dev:restore_kb:")) async def cb_dev_restore_kb(callback: CallbackQuery): parts = callback.data.split(":") target_id = int(parts[2]) msg_record = await get_message_by_dev_msg_id(BOT_USERNAME, callback.message.message_id) kb = build_dev_keyboard(target_id, msg_record["user_message_id"] if msg_record else 0) try: await callback.message.edit_reply_markup(reply_markup=kb) except Exception: pass await callback.answer() @dev_router.callback_query(F.data.startswith("dev:info:")) async def cb_dev_info(callback: CallbackQuery, bot: Bot): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return target_id = int(callback.data.split(":")[2]) user = await get_user(BOT_USERNAME, target_id) if not user: await callback.answer("⚠️ المستخدم غير موجود!", show_alert=True) return bio = "" try: chat = await bot.get_chat(target_id) bio = chat.bio or "" except Exception: pass name = user["first_name"] or "" username = f"@{user['username']}" if user.get("username") else "لا يوجد" info_msg = await callback.message.answer( f"» معلومات المستخدم:\n" f"» الاسم: {name}\n" f"» اليوزر: {username}\n" f"» الايدي: {target_id}\n" f"» البايو:\n{bio or 'لا يوجد'}", parse_mode="HTML", disable_web_page_preview=True, ) await info_msg.edit_reply_markup( reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data=f"dev:close_info:{info_msg.message_id}", style="primary")] ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("dev:close_info:")) async def cb_dev_close_info(callback: CallbackQuery): try: await callback.message.delete() except Exception: pass await callback.answer() async def build_dev_panel() -> InlineKeyboardMarkup: s = await get_all_settings(BOT_USERNAME) is_open = s.get("is_open", "1") == "1" notify_u = "✅" if s.get("notify_new_users") == "1" else "✖️" notify_b = "✅" if s.get("notify_blocked") == "1" else "✖️" status = "مفتوح 🟢" if is_open else "مقفول 🔴" return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"حالة البوت {status}", callback_data="dp:toggle_open", style="success")], [ InlineKeyboardButton(text=f"اشعار الدخول {notify_u}", callback_data="dp:toggle_notify_u", style="success"), InlineKeyboardButton(text=f"اشعار الحظر {notify_b}", callback_data="dp:toggle_notify_b", style="success"), ], [InlineKeyboardButton(text="قسم الاحصائيات", callback_data="dp:stats", style="primary")], [InlineKeyboardButton(text="قسم الاذاعة", callback_data="dp:broadcast", style="primary")], [InlineKeyboardButton(text="قسم الادمنية", callback_data="dp:admins", style="primary")], [ InlineKeyboardButton(text="قسم التعديل", callback_data="dp:edit", style="danger"), InlineKeyboardButton(text="قسم المكتومين", callback_data="dp:muted", style="danger"), ], [InlineKeyboardButton(text="قسم الاشتراك الاجباري", callback_data="dp:mandatory", style="danger")], [InlineKeyboardButton(text="مكان الاستلام", callback_data="dp:receive", style="danger")], [ InlineKeyboardButton(text="اضافة زر", callback_data="cbtn:add", style="success"), InlineKeyboardButton(text="ادارة الازرار", callback_data="cbtn:manage", style="primary"), ], ]) @dev_router.callback_query(F.data == "dp:panel") async def cb_dp_panel(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return await state.clear() await callback.message.edit_text( "» هذه هي لوحة التحكم وادارة بوت التواصل الخاص بك.\n\n» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات وإدارة الاشتراكات وكل شيء من مكان واحد.\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_dev_panel() ) await callback.answer() @dev_router.callback_query(F.data == "dp:toggle_open") async def cb_dp_toggle_open(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return is_open = await get_setting(BOT_USERNAME, "is_open") == "1" if is_open: await state.set_state(DevStates.waiting_close_msg) await callback.message.edit_text( "» أرسل الرسالة التي ستظهر للمستخدمين عندما يكون البوت مقفولاً:", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")] ]) ) else: await state.set_state(DevStates.waiting_open_msg) await callback.message.edit_text( "» أرسل الرسالة التي ستُرسل للمستخدمين عند فتح البوت:", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")] ]) ) await callback.answer() @dev_router.callback_query(F.data.in_({"dp:toggle_notify_u", "dp:toggle_notify_b"})) async def cb_dp_toggle(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return key_map = { "dp:toggle_notify_u": "notify_new_users", "dp:toggle_notify_b": "notify_blocked", } key = key_map[callback.data] current = await get_setting(BOT_USERNAME, key) await set_setting(BOT_USERNAME, key, "0" if current == "1" else "1") await callback.message.edit_reply_markup(reply_markup=await build_dev_panel()) await callback.answer() @dev_router.callback_query(F.data == "dp:stats") async def cb_dp_stats(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return stats = await get_statistics(BOT_USERNAME) text = ( "📊 إحصائيات المستخدمين:\n" f"[+] عدد مستخدمين البوت: {stats['total_users']} مستخدم\n" f"[+] النشطين آخر 24 ساعة: {stats['active_24h']} مستخدم\n" f"[+] المستخدمين المكتومين: {stats['muted_users']} مستخدم\n" f"[+] قام بحظر البوت: {stats['blocked_users']} مستخدم\n\n" "⚙️ إحصائيات عامة للبوت:\n" f"[+] عدد الرسائل المرسلة: {stats['sent']} رسالة\n" f"[+] عدد الرسائل المستلمة: {stats['received']} رسالة\n" f"[+] عدد الاشتراك الإجباري: {stats['mand_count']} قناة" ) await callback.message.edit_text( text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")] ]) ) await callback.answer() @dev_router.callback_query(F.data == "dp:broadcast") async def cb_dp_broadcast(callback: CallbackQuery, state: FSMContext): await state.clear() if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return await callback.message.edit_text( "» عزيزي المطور هذا هو قسم التحكم بالاذاعة.\n» حدد نوع الاذاعة سواء رسالة موجهة او نصية:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="اذاعة نصية", callback_data="dp:bcast_text", style="success"), InlineKeyboardButton(text="اذاعة موجهة", callback_data="dp:bcast_fwd", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data == "dp:bcast_text") async def cb_dp_bcast_text(callback: CallbackQuery, state: FSMContext): await state.set_state(DevStates.waiting_broadcast_msg) await callback.message.edit_text( "» ارسل رسالة الاذاعة المراد اذاعتها لجميع المستخدمين الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:broadcast", style="primary")] ]) ) await callback.answer() @dev_router.callback_query(F.data == "dp:bcast_fwd") async def cb_dp_bcast_fwd(callback: CallbackQuery, state: FSMContext): await state.set_state(DevStates.waiting_broadcast_fwd) await callback.message.edit_text( "» قم بتوجيه الرسالة المطلوب إذاعتها الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:broadcast", style="primary")] ]) ) await callback.answer() async def _do_broadcast(message: Message, wait: Message | None = None, forward: bool = False): uname = BOT_USERNAME tok = BOT_TOKEN chat_id = message.chat.id users = await get_all_active_users(uname) sent = failed = 0 b = Bot(token=tok) try: for i, user in enumerate(users): try: if forward: await b.forward_message(user["id"], from_chat_id=chat_id, message_id=message.message_id) else: await b.copy_message(user["id"], from_chat_id=chat_id, message_id=message.message_id) sent += 1 except Exception: failed += 1 if (i + 1) % BROADCAST_BATCH == 0: await asyncio.sleep(BROADCAST_DELAY) if wait: await b.edit_message_text( chat_id=chat_id, message_id=wait.message_id, text=( f"» تم ارسال الاذاعة الى كافة مستخدمين البوت.\n\n" f"» نجـح الارسال إلى: {sent} مستخدم\n" f"» فشل الارسال إلى: {failed} مستخدم" ), parse_mode="HTML", disable_web_page_preview=True) except Exception: pass finally: await b.session.close() @dev_router.message(DevStates.waiting_broadcast_msg) async def process_broadcast_msg(message: Message, state: FSMContext, bot: Bot): await state.clear() wait = await message.answer("⏳ جاري الإذاعة...") asyncio.create_task(_do_broadcast(message, wait, forward=False)) @dev_router.message(DevStates.waiting_broadcast_fwd) async def process_broadcast_fwd(message: Message, state: FSMContext, bot: Bot): await state.clear() wait = await message.answer("⏳ جاري الإذاعة...") asyncio.create_task(_do_broadcast(message, wait, forward=True)) @dev_router.callback_query(F.data == "dp:admins") async def cb_dp_admins(callback: CallbackQuery, state: FSMContext): await state.clear() if callback.from_user.id != OWNER_ID: await callback.answer("⛔ هذا القسم للمطور فقط!", show_alert=True) return await callback.message.edit_text( "» عزيزي المطور هذا هو قسم التحكم بالادمنية.\n» يمكن اضافة وحذف وعرض الادمنية من الازرار:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="مسح جميع الادمنية", callback_data="dp:admins_clear", style="danger")], [ InlineKeyboardButton(text="الادمنية المضافة", callback_data="dp:admins_list", style="success"), InlineKeyboardButton(text="اضافة ادمن", callback_data="dp:admins_add", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data == "dp:admins_list") async def cb_dp_admins_list(callback: CallbackQuery): admins = await get_admins(BOT_USERNAME) if not admins: await callback.answer("لا يوجد أدمنية مضافون.", show_alert=True) return buttons = [] for admin in admins: name = f"@{admin['username']}" if admin.get("username") else admin["first_name"] buttons.append([ InlineKeyboardButton(text=name, url=f"tg://user?id={admin['id']}", style="success"), InlineKeyboardButton(text="🗑️", callback_data=f"dp:admin_remove:{admin['id']}", style="danger"), ]) buttons.append([InlineKeyboardButton(text="رجوع", callback_data="dp:admins", style="primary")]) await callback.message.edit_text( "» هذه هي قائمة الادمنية الذين تم اضافتهم مسبقاً:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons) ) await callback.answer() @dev_router.callback_query(F.data.startswith("dp:admin_remove:")) async def cb_dp_admin_remove(callback: CallbackQuery): admin_id = int(callback.data.split(":")[2]) await remove_admin(BOT_USERNAME, admin_id) await callback.answer("✅ تم حذف الادمن.", show_alert=True) await cb_dp_admins_list(callback) @dev_router.callback_query(F.data == "dp:admins_clear") async def cb_dp_admins_clear(callback: CallbackQuery): await callback.message.edit_text( "» هل أنت متأكد من مسح جميع الأدمنية؟", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="نعم", callback_data="dp:admins_clear_confirm", style="danger")], [InlineKeyboardButton(text="رجوع", callback_data="dp:admins", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data == "dp:admins_clear_confirm") async def cb_dp_admins_clear_confirm(callback: CallbackQuery): await clear_admins(BOT_USERNAME) await callback.answer("✅ تم مسح جميع الأدمنية.", show_alert=True) await cb_dp_admins(callback) @dev_router.callback_query(F.data == "dp:admins_add") async def cb_dp_admins_add(callback: CallbackQuery, state: FSMContext): await state.set_state(DevStates.waiting_add_admin) await callback.message.edit_text( "» أرسل أيدي المستخدم الذي تريد اضافته ادمن في البوت:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:admins", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_add_admin) async def process_add_admin(message: Message, state: FSMContext, bot: Bot): try: admin_id = int(message.text.strip()) except (ValueError, AttributeError): await message.answer("⚠️ أرسل ID رقمي صحيح!") return try: chat = await bot.get_chat(admin_id) await add_admin(BOT_USERNAME, admin_id, chat.username, chat.first_name or "") name = f"@{chat.username}" if chat.username else chat.first_name await state.clear() await message.answer( f"» تم اضافة المستخدم {name} كأدمن في البوت.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="قسم الأدمنية", callback_data="dp:admins", style="primary")] ]) ) except Exception: await message.answer("⚠️ لم يتم العثور على المستخدم!") await state.clear() @dev_router.callback_query(F.data == "dp:edit") async def cb_dp_edit(callback: CallbackQuery, state: FSMContext): await state.clear() if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return await callback.message.edit_text( "•⊱ عزيزي المطور هذا هو قسم التعديل.\n" "•⊱ يمكنك تعديل رسالة الترحيب من هنا:", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رسالة الترحيب", callback_data="dp:edit_welcome", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data == "dp:noop") async def cb_dp_noop(callback: CallbackQuery): await callback.answer() @dev_router.callback_query(F.data == "dp:edit_welcome") async def cb_dp_edit_welcome(callback: CallbackQuery, state: FSMContext): welcome = await get_setting(BOT_USERNAME, "welcome_message") await state.set_state(DevStates.waiting_welcome) await callback.message.edit_text( f"» رسالة الترحيب الحالية:\n{welcome or 'افتراضية'}\n\n» استخدم {{name}} لعرض اسم المستخدم في الرسالة الترحيبية.\n» قم بإرسال رسالة الترحيب الجديدة الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:edit", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_welcome) async def process_welcome(message: Message, state: FSMContext): text = message.text or "" await set_setting(BOT_USERNAME, "welcome_message", text) await set_setting(BOT_USERNAME, "welcome_message_entities", _entities_to_json(message.entities) or "") await state.clear() await message.answer( "✅ تم تحديث رسالة الترحيب.", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="قسم التعديل", callback_data="dp:edit", style="primary")] ]) ) @dev_router.callback_query(F.data == "dp:muted") async def cb_dp_muted(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return muted = await get_muted_users(BOT_USERNAME) if not muted: await callback.message.edit_text( "•⊱ لا يوجد مستخدمون مكتومون.", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")] ]) ) await callback.answer() return buttons = [] for user in muted: name = f"@{user['username']}" if user.get("username") else user["first_name"] buttons.append([ InlineKeyboardButton(text=name, url=f"tg://user?id={user['id']}", style="success"), InlineKeyboardButton(text="إلغاء الكتم", callback_data=f"dev:unmute:{user['id']}", style="danger"), ]) buttons.append([InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")]) await callback.message.edit_text( "•⊱ هذا هو قسم المكتومين:", reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons) ) await callback.answer() @dev_router.callback_query(F.data == "dp:mandatory") async def cb_dp_mandatory(callback: CallbackQuery, state: FSMContext): await state.clear() if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return channels = await get_mandatory_channels(BOT_USERNAME) buttons = [] for ch in channels: link = ch.get("invite_link") or ( f"https://t.me/{ch['channel_username']}" if ch.get("channel_username") else None ) if not link: try: link_obj = await callback.bot.create_chat_invite_link(ch["channel_id"]) link = link_obj.invite_link await update_mandatory_channel_link(BOT_USERNAME, ch["id"], link) except Exception: link = None buttons.append([ InlineKeyboardButton(text=ch["channel_title"], url=link, style="success") if link else InlineKeyboardButton(text=ch["channel_title"] + " ⚠️", url="https://t.me", style="success"), InlineKeyboardButton(text="🗑️", callback_data=f"dp:mand_del:{ch['id']}", style="danger"), ]) buttons.append([InlineKeyboardButton(text="إضافة قناة", callback_data="dp:mand_add", style="success")]) buttons.append([InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")]) await callback.message.edit_text( "» عزيزي المطور هذا هو قسم الاشتراك الاجباري.\n» يمكن اضافة وحذف وعرض القنوات من الازرار:\n\n" if channels else "» لا توجد قنوات مضافة حالياً، اضف قناة الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons) ) await callback.answer() @dev_router.callback_query(F.data.startswith("dp:mand_del:")) async def cb_dp_mand_del(callback: CallbackQuery): ch_id = int(callback.data.split(":")[2]) await remove_mandatory_channel(BOT_USERNAME, ch_id) await callback.answer("✅ تم حذف القناة.", show_alert=True) await cb_dp_mandatory(callback) @dev_router.callback_query(F.data == "dp:mand_add") async def cb_dp_mand_add(callback: CallbackQuery, state: FSMContext): await state.set_state(DevStates.waiting_mand_type) await callback.message.edit_text( "» اختر الطريقة التي تريد عرض طلب الاشتراك فيها من المستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="ازرار", callback_data="dp:mand_type:buttons", style="success"), InlineKeyboardButton(text="رسالة نصية", callback_data="dp:mand_type:text", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dp:mandatory", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("dp:mand_type:")) async def cb_dp_mand_type(callback: CallbackQuery, state: FSMContext): display_type = callback.data.split(":")[2] await state.update_data(mand_display_type=display_type) await state.set_state(DevStates.waiting_mand_fwd) await callback.message.edit_text( "» ممتاز، الان قم بتوجيه رسالة من القناة/المجموعة لاضافتها.\n⚠️ يجب ان يكون البوت مشرفاً في القناة/المجموعة.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:mandatory", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_mand_fwd) async def process_mand_fwd(message: Message, state: FSMContext, bot: Bot): import re as _re chat_id = None chat_obj = None is_forward = False if message.forward_from_chat: chat_obj = message.forward_from_chat chat_id = chat_obj.id chat_type = chat_obj.type is_forward = True if chat_type in ("supergroup", "group") and not chat_obj.username: await message.answer("⚠️ لا يمكن قبول رابط المجموعة الخاص أو توجيه رسالة منها!") return elif message.text: text = message.text.strip() if text.startswith("@"): username = text.lstrip("@") try: chat_obj = await bot.get_chat(f"@{username}") chat_id = chat_obj.id except Exception: await message.answer("⚠️ لم يتم العثور على هذا اليوزر!") return elif _re.match(r"https?://t\.me/\+", text): try: chat_obj = await bot.get_chat(text) chat_id = chat_obj.id if chat_obj.type in ("supergroup", "group"): await message.answer("⚠️ لا يمكن قبول رابط المجموعة الخاص!") return except Exception: await message.answer("⚠️ لا يمكن الوصول إلى هذا الرابط الخاص، تأكد أن البوت مشرف في القناة!") return elif _re.match(r"https?://t\.me/[^+]", text): username = text.rstrip("/").split("/")[-1] try: chat_obj = await bot.get_chat(f"@{username}") chat_id = chat_obj.id except Exception: await message.answer("⚠️ لم يتم العثور على هذا الرابط!") return elif _re.match(r"^-?\d+$", text): try: chat_obj = await bot.get_chat(int(text)) chat_id = chat_obj.id except Exception: await message.answer("⚠️ لم يتم العثور على هذا الـ ID!") return else: await message.answer( "⚠️ صيغة غير معروفة. أرسل أحد التالي:\n" "• يوزر: @username\n" "• رابط عام: https://t.me/username\n" "• رابط خاص (قناة فقط): https://t.me/+xxx\n" "• ID: -100xxxxxxxxxx\n" "• أو وجّه رسالة من القناة/المجموعة العامة" ) return else: await message.answer("⚠️ يرجى إرسال نص أو توجيه رسالة!") return try: bot_id = (await bot.get_me()).id member = await bot.get_chat_member(chat_id, bot_id) if member.status not in ("administrator", "creator"): await message.answer("⚠️ يجب إضافة البوت كمشرف في القناة/المجموعة!") await state.clear() return except Exception: await message.answer("⚠️ لا يمكن التحقق من صلاحيات البوت، تأكد أنه مشرف!") await state.clear() return invite_link = None if chat_obj and getattr(chat_obj, "username", None): invite_link = f"https://t.me/{chat_obj.username}" elif not invite_link and chat_obj: try: async with aiohttp.ClientSession() as http: async with http.post( f"https://api.telegram.org/bot{FACTORY_TOKEN}/createChatInviteLink", json={"chat_id": chat_id}, timeout=aiohttp.ClientTimeout(total=10) ) as resp: data = await resp.json() if data.get("ok"): invite_link = data["result"]["invite_link"] except Exception: pass title = getattr(chat_obj, "title", None) or str(chat_id) username_val = getattr(chat_obj, "username", None) data_state = await state.get_data() display_type = data_state.get("mand_display_type", "buttons") await add_mandatory_channel( BOT_USERNAME, chat_id, title, username_val, invite_link, display_type ) await state.clear() await message.answer( f"✅ تم إضافة {title}!", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="الاشتراك الإجباري", callback_data="dp:mandatory", style="primary")] ]) ) @dev_router.callback_query(F.data == "dp:receive") async def cb_dp_receive(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True) return in_group = await get_setting(BOT_USERNAME, "receive_in_group") == "1" priv_icon = "✅" if not in_group else "✖️" grp_icon = "✅" if in_group else "✖️" await callback.message.edit_text( "» عزيزي المطور هذا هو قسم مكان استلام الرسائل\n\n" "في خاص البوت:\n" "• الرسائل تصل إلى محادثتك مع البوت\n" "• الرسائل تصل لك فقط.\n\n" "في مجموعة مشتركة:\n" "• الرسائل تصل إلى مجموعة فريق العمل.\n" "• مناسب للفرق متعددة الأدمنية", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"في الخاص: {priv_icon}", callback_data="dp:receive_private", style="success")], [InlineKeyboardButton(text=f"في المجموعة: {grp_icon}", callback_data="dp:receive_group", style="success")], [InlineKeyboardButton(text="المجموعات المضافة", callback_data="dp:groups_list", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data == "dp:receive_private") async def cb_dp_receive_private(callback: CallbackQuery): await set_setting(BOT_USERNAME, "receive_in_group", "0") await cb_dp_receive(callback) @dev_router.callback_query(F.data == "dp:receive_group") async def cb_dp_receive_group(callback: CallbackQuery): groups = await get_groups(BOT_USERNAME) if not groups: await callback.answer( "» لم تقم بإضافة أي مجموعة! اضغط على 'المجموعات المضافة' لإضافة مجموعة.", show_alert=True ) return await set_setting(BOT_USERNAME, "receive_in_group", "1") await cb_dp_receive(callback) @dev_router.callback_query(F.data == "dp:groups_list") async def cb_dp_groups_list(callback: CallbackQuery): groups = await get_groups(BOT_USERNAME) if not groups: await callback.message.edit_text( "» لم تقم بإضافة مجموعة، اضغط على زر اضافة مجموعة الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="اضافة مجموعة", callback_data="dp:group_add", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dp:receive", style="primary")], ]) ) await callback.answer() return buttons = [] for group in groups: link = group.get("invite_link") or ( f"https://t.me/{group['username']}" if group.get("username") else group.get("invite_link") or "https://t.me" ) buttons.append([ InlineKeyboardButton(text=group["chat_title"], url=link, style="success"), InlineKeyboardButton(text="🗑️", callback_data=f"dp:group_del:{group['id']}", style="danger"), ]) buttons.append([InlineKeyboardButton(text="اضافة مجموعة", callback_data="dp:group_add", style="success")]) buttons.append([InlineKeyboardButton(text="رجوع", callback_data="dp:receive", style="primary")]) await callback.message.edit_text( "» هذه هي قائمة بالمجموعات المضافة حالياً:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons) ) await callback.answer() @dev_router.callback_query(F.data.startswith("dp:group_del:")) async def cb_dp_group_del(callback: CallbackQuery): group_id = int(callback.data.split(":")[2]) await remove_group(BOT_USERNAME, group_id) await callback.answer("✅ تم حذف المجموعة.", show_alert=True) await cb_dp_groups_list(callback) @dev_router.callback_query(F.data == "dp:group_add") async def cb_dp_group_add(callback: CallbackQuery, state: FSMContext): await state.set_state(DevStates.waiting_group_fwd) await callback.message.edit_text( "•⊱ أرسل معلومات المجموعة بأحد الأشكال التالية:\n\n" "1. يوزر المجموعة: @usernamechat\n" "2. رابط عام: https://t.me/usernamechat\n" "3. معرف المجموعة الخاصة: -1001234567890\n\n" "للحصول على معرف المجموعة الخاصة:\n" "• أضف @userinfobot إلى المجموعة\n" "• أرسل /start في المجموعة\n" "• سيعطيك البوت المعرف الرقمي", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:groups_list", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_group_fwd) async def process_group_fwd(message: Message, state: FSMContext, bot: Bot): if message.text and message.text.strip() == "/start": await state.clear() await cmd_start(message, state, bot) return text = message.text.strip() if message.text else "" chat = None username_match = re.match(r'^@([a-zA-Z0-9_]{5,32})$', text) if username_match: username = username_match.group(1) try: chat = await bot.get_chat(f"@{username}") except Exception: await message.answer( "⚠️ لم يتم العثور على المجموعة بهذا اليوزر!", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:groups_list", style="primary")] ]) ) return if not chat: public_match = re.match(r'^https?://t\.me/([a-zA-Z0-9_]{5,32})$', text) if public_match: username = public_match.group(1) try: chat = await bot.get_chat(f"@{username}") except Exception: await message.answer( "⚠️ لم يتم العثور على المجموعة بهذا الرابط!", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:groups_list", style="primary")] ]) ) return if not chat: try: chat_id = int(text) chat = await bot.get_chat(chat_id) except (ValueError, Exception): pass if not chat: await message.answer( "⚠️ الصيغة غير صحيحة أو لم يتم العثور على المجموعة!\n\n" "يرجى إرسال بأحد الأشكال التالية:\n" "• يوزر: @usernamechat\n" "• رابط عام: https://t.me/usernamechat\n" "• معرف رقمي: -1001234567890", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:groups_list", style="primary")] ]) ) return chat_id = chat.id chat_type = getattr(chat, 'type', None) if chat_type not in ("group", "supergroup"): await message.answer( "⚠️ يجب أن يكون المصدر مجموعة وليس قناة!", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:groups_list", style="primary")] ]) ) await state.clear() return try: me = await bot.get_me() member = await bot.get_chat_member(chat_id, me.id) if member.status not in ("administrator", "creator"): await message.answer( "⚠️ يجب إضافة البوت كمشرف في المجموعة أولاً!", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:groups_list", style="primary")] ]) ) await state.clear() return except Exception: await message.answer( "⚠️ لا يمكن الوصول للمجموعة!", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dp:groups_list", style="primary")] ]) ) await state.clear() return invite_link = None if chat.username: invite_link = f"https://t.me/{chat.username}" else: try: link_obj = await bot.create_chat_invite_link(chat_id) invite_link = link_obj.invite_link except Exception: pass await add_group(BOT_USERNAME, chat_id, chat.title or str(chat_id), chat.username, invite_link) await state.clear() await message.answer( f"✅ تم إضافة المجموعة {chat.title}!", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="المجموعات المضافة", callback_data="dp:groups_list", style="primary")] ]) ) @dev_router.message(DevStates.waiting_close_msg) async def process_close_msg(message: Message, state: FSMContext): await set_setting(BOT_USERNAME, "is_open", "0") await set_setting(BOT_USERNAME, "closed_message", message.text or "") await state.clear() await message.answer( "🔴 تم قفل البوت.", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="لوحة التحكم", callback_data="dp:panel", style="primary")] ]) ) @dev_router.message(DevStates.waiting_open_msg) async def process_open_msg(message: Message, state: FSMContext, bot: Bot): open_msg = message.text or "" await set_setting(BOT_USERNAME, "is_open", "1") await set_setting(BOT_USERNAME, "open_message", open_msg) await state.clear() uname = BOT_USERNAME tok = BOT_TOKEN chat_id = message.chat.id async def _run_open(): users = await get_all_active_users(uname) sent = failed = 0 b = Bot(token=tok) try: for i, user in enumerate(users): try: await b.send_message(user["id"], open_msg) sent += 1 except Exception: failed += 1 if (i + 1) % BROADCAST_BATCH == 0: await asyncio.sleep(BROADCAST_DELAY) await b.send_message( chat_id, f"🟢 تم فتح البوت!\n📤 أُرسلت رسالة الفتح إلى: {sent} مستخدم", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="لوحة التحكم", callback_data="dp:panel", style="primary")] ]) ) except Exception: pass finally: await b.session.close() asyncio.create_task(_run_open()) @dev_router.my_chat_member() async def on_my_chat_member(event: ChatMemberUpdated, bot: Bot): new_status = event.new_chat_member.status old_status = event.old_chat_member.status chat = event.chat if chat.type == "private": user_id = event.from_user.id db_user = await get_user(BOT_USERNAME, user_id) if db_user: import aiosqlite if new_status == "kicked": async with aiosqlite.connect(get_db_path(BOT_USERNAME)) as db: await db.execute("UPDATE users SET is_blocked = 1 WHERE id = ?", (user_id,)) await db.commit() settings = await get_all_settings(BOT_USERNAME) if settings.get("notify_blocked") == "1": name = f"@{event.from_user.username}" if event.from_user.username else event.from_user.first_name await notify_owner( bot, f"🚫 مستخدم حظر البوت!\n" f"الاسم: {event.from_user.first_name}\n" f"المعرف: {name}\n" f"الـ ID: {user_id}" ) elif new_status == "member": async with aiosqlite.connect(get_db_path(BOT_USERNAME)) as db: await db.execute("UPDATE users SET is_blocked = 0 WHERE id = ?", (user_id,)) await db.commit() return if new_status in ("administrator", "creator") and old_status not in ("administrator", "creator"): invite_link = None if chat.username: invite_link = f"https://t.me/{chat.username}" else: try: link_obj = await bot.create_chat_invite_link(chat.id) invite_link = link_obj.invite_link except Exception: pass await add_group(BOT_USERNAME, chat.id, chat.title or str(chat.id), chat.username, invite_link) await notify_owner( bot, f"📢 تم إضافة البوت كمشرف في مجموعة!\n" f"المجموعة: {chat.title}\n" f"الرابط: {invite_link or str(chat.id)}" ) elif new_status in ("left", "kicked") and old_status in ("administrator", "creator", "member"): groups = await get_groups(BOT_USERNAME) for group in groups: if group["chat_id"] == chat.id: await remove_group(BOT_USERNAME, group["id"]) break def _render_custom_button(b: dict): label = b.get("label") or "" emoji = b.get("emoji") if emoji: label = f"{emoji} {label}" kwargs = {"text": label} if b["btn_type"] == "url": kwargs["url"] = b["content"] else: kwargs["callback_data"] = f"cbtn:press:{b['id']}" color = b.get("color") if color and color != "default": style_map = {"green": "success", "blue": "primary", "red": "danger"} tg = style_map.get(color) if tg: kwargs["style"] = tg if b.get("custom_emoji_id"): kwargs["icon_custom_emoji_id"] = b["custom_emoji_id"] return InlineKeyboardButton(**kwargs) @dev_router.callback_query(F.data == "cbtn:add") async def cbtn_add(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await state.clear() await callback.message.edit_text( "» قم بأختيار نوع الزر الذي سيظهر للمستخدم، يُمكن ان يكون رابط أو كالباك، اختر من الازرار:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="url - رابط", callback_data="cbtn:type:url", style="success")], [InlineKeyboardButton(text="callback - كالباك", callback_data="cbtn:type:callback", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("cbtn:type:")) async def cbtn_type(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return btn_type = callback.data.split(":")[2] await state.update_data(cbtn_type=btn_type) await state.set_state(DevStates.waiting_cbtn_label) await callback.message.edit_text( "» أرسل اسم الزر الذي سيظهر للمستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="cbtn:add", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_cbtn_label) async def cbtn_process_label(message: Message, state: FSMContext): if not await is_dev_or_admin(message.from_user.id): return if message.text and message.text.startswith("/"): await state.clear(); return label = (message.text or "").strip() if not label: await message.answer("⚠️ أرسل اسماً صحيحاً للزر!"); return data = await state.get_data() btn_type = data.get("cbtn_type") await state.update_data(cbtn_label=label) await state.set_state(DevStates.waiting_cbtn_content) prompt = "» أرسل الرابط الآن:" if btn_type == "url" else "» أرسل الرسالة النصية التي ستظهر عند الضغط على الزر:" await message.answer( prompt, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="cbtn:add", style="primary")] ]) ) @dev_router.message(DevStates.waiting_cbtn_content) async def cbtn_process_content(message: Message, state: FSMContext): if not await is_dev_or_admin(message.from_user.id): return if message.text and message.text.startswith("/"): await state.clear(); return content = (message.text or "").strip() data = await state.get_data() btn_type = data.get("cbtn_type") label = data.get("cbtn_label") if btn_type == "url": if not (content.startswith("http://") or content.startswith("https://") or content.startswith("tg://")): await message.answer("⚠️ أرسل رابطاً صحيحاً يبدأ بـ https:// أو http://") return if not content: await message.answer("⚠️ المحتوى فارغ!"); return await add_custom_button(BOT_USERNAME, btn_type, label, content) await state.clear() await message.answer( "✅ تمت إضافة الزر بنجاح، وسيظهر للمستخدمين الآن.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=" لوحة التحكم", callback_data="dp:panel", style="primary")] ]) ) @dev_router.callback_query(F.data == "cbtn:manage") async def cbtn_manage(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await state.clear() buttons = await get_custom_buttons(BOT_USERNAME) if not buttons: await callback.answer("لا توجد أزرار مضافة.", show_alert=True); return rows = [[InlineKeyboardButton(text=b["label"], callback_data=f"cbtn:view:{b['id']}", style="success")] for b in buttons] rows.append([InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")]) await callback.message.edit_text( "عزيزي المطور اختر الزر المراد ادارته:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) await callback.answer() async def _cbtn_view_render(callback: CallbackQuery, btn_id: int): b = await get_custom_button(BOT_USERNAME, btn_id) if not b: await callback.answer("الزر غير موجود.", show_alert=True); return vis = "ظاهر ✅" if b.get("is_visible", 1) else "مخفي 🚫" await callback.message.edit_text( f"اسم الزر: {b['label']}\nالحالة: {vis}\n\n" "عزيزي المطور يمكنك ادارة الزر المضاف:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="🗑 حذف الزر", callback_data=f"cbtn:del:{btn_id}", style="danger")], [InlineKeyboardButton(text="✏️ تعديل محتوى الزر", callback_data=f"cbtn:edit:{btn_id}", style="success")], [InlineKeyboardButton(text="📝 تعديل اسم الزر", callback_data=f"cbtn:ename:{btn_id}", style="success")], [InlineKeyboardButton(text="🎨 تعديل لون الزر", callback_data=f"cbtn:ecolor:{btn_id}", style="success")], [InlineKeyboardButton(text="😀 تعديل ايموجي الزر", callback_data=f"cbtn:eemoji:{btn_id}", style="success")], [InlineKeyboardButton(text="👁 اخفاء/اظهار الزر", callback_data=f"cbtn:toggle:{btn_id}", style="primary")], [InlineKeyboardButton(text="رجوع", callback_data="cbtn:manage", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("cbtn:view:")) async def cbtn_view(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await _cbtn_view_render(callback, int(callback.data.split(":")[2])) @dev_router.callback_query(F.data.startswith("cbtn:del:")) async def cbtn_del(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return btn_id = int(callback.data.split(":")[2]) await delete_custom_button(BOT_USERNAME, btn_id) await callback.answer("✅ تم حذف الزر.", show_alert=True) await cbtn_manage(callback, state) @dev_router.callback_query(F.data.startswith("cbtn:toggle:")) async def cbtn_toggle(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return btn_id = int(callback.data.split(":")[2]) b = await get_custom_button(BOT_USERNAME, btn_id) if not b: await callback.answer("الزر غير موجود.", show_alert=True); return new_vis = 0 if b.get("is_visible", 1) else 1 await update_custom_button(BOT_USERNAME, btn_id, is_visible=new_vis) await callback.answer("✅ تم.", show_alert=False) await _cbtn_view_render(callback, btn_id) @dev_router.callback_query(F.data.startswith("cbtn:ecolor:")) async def cbtn_ecolor(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return btn_id = int(callback.data.split(":")[2]) await callback.message.edit_text( "🎨 اختر لون الزر:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="🟢 أخضر", callback_data=f"cbtn:setcolor:{btn_id}:green", style="success"), InlineKeyboardButton(text="🔵 أزرق", callback_data=f"cbtn:setcolor:{btn_id}:blue", style="primary")], [InlineKeyboardButton(text="🔴 أحمر", callback_data=f"cbtn:setcolor:{btn_id}:red", style="danger"), InlineKeyboardButton(text="⚪ افتراضي", callback_data=f"cbtn:setcolor:{btn_id}:default")], [InlineKeyboardButton(text="رجوع", callback_data=f"cbtn:view:{btn_id}", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("cbtn:setcolor:")) async def cbtn_setcolor(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return parts = callback.data.split(":") btn_id = int(parts[2]); color = parts[3] await update_custom_button(BOT_USERNAME, btn_id, color=color) await callback.answer("✅ تم تغيير اللون.", show_alert=False) await _cbtn_view_render(callback, btn_id) @dev_router.callback_query(F.data.startswith("cbtn:eemoji:")) async def cbtn_eemoji(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return btn_id = int(callback.data.split(":")[2]) await state.update_data(cbtn_emoji_id=btn_id) await state.set_state(DevStates.waiting_cbtn_emoji) await callback.message.edit_text( "😀 أرسل الإيموجي الجديد (عادي أو متحرك مميز)، أو أرسل - لحذف الإيموجي:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data=f"cbtn:view:{btn_id}", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_cbtn_emoji) async def cbtn_process_emoji(message: Message, state: FSMContext): if not await is_dev_or_admin(message.from_user.id): return if message.text and message.text.startswith("/"): await state.clear(); return data = await state.get_data() btn_id = data.get("cbtn_emoji_id") text = (message.text or "").strip() if text == "-": await update_custom_button(BOT_USERNAME, btn_id, emoji=None, custom_emoji_id=None) await state.clear() await message.answer("✅ تم حذف الإيموجي.", parse_mode="HTML", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="ادارة الازرار", callback_data="cbtn:manage", style="primary")]])) return custom_emoji_id = None if message.entities: for e in message.entities: if e.type == "custom_emoji": custom_emoji_id = e.custom_emoji_id break if custom_emoji_id: await update_custom_button(BOT_USERNAME, btn_id, custom_emoji_id=custom_emoji_id, emoji=None) msg = "✅ تم تعيين الإيموجي المتحرك بنجاح!" else: await update_custom_button(BOT_USERNAME, btn_id, emoji=text, custom_emoji_id=None) msg = f"✅ تم تعيين الإيموجي: {text}" await state.clear() await message.answer(msg, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="ادارة الازرار", callback_data="cbtn:manage", style="primary")]])) @dev_router.callback_query(F.data.startswith("cbtn:ename:")) async def cbtn_ename(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return btn_id = int(callback.data.split(":")[2]) await state.update_data(cbtn_ename_id=btn_id) await state.set_state(DevStates.waiting_cbtn_name) await callback.message.edit_text( "📝 أرسل الاسم الجديد للزر:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data=f"cbtn:view:{btn_id}", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_cbtn_name) async def cbtn_process_name(message: Message, state: FSMContext): if not await is_dev_or_admin(message.from_user.id): return if message.text and message.text.startswith("/"): await state.clear(); return label = (message.text or "").strip() if not label: await message.answer("⚠️ أرسل اسماً صحيحاً!"); return data = await state.get_data() btn_id = data.get("cbtn_ename_id") await update_custom_button(BOT_USERNAME, btn_id, label=label) await state.clear() await message.answer("✅ تم تعديل اسم الزر.", parse_mode="HTML", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="ادارة الازرار", callback_data="cbtn:manage", style="primary")]])) @dev_router.callback_query(F.data.startswith("cbtn:edit:")) async def cbtn_edit(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return btn_id = int(callback.data.split(":")[2]) b = await get_custom_button(BOT_USERNAME, btn_id) if not b: await callback.answer("الزر غير موجود.", show_alert=True); return await state.update_data(cbtn_edit_id=btn_id, cbtn_edit_type=b["btn_type"]) await state.set_state(DevStates.waiting_cbtn_edit_content) prompt = "» أرسل الرابط الجديد:" if b["btn_type"] == "url" else "» أرسل الرسالة النصية الجديدة:" await callback.message.edit_text( prompt, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data=f"cbtn:view:{btn_id}", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_cbtn_edit_content) async def cbtn_process_edit_content(message: Message, state: FSMContext): if not await is_dev_or_admin(message.from_user.id): return if message.text and message.text.startswith("/"): await state.clear(); return content = (message.text or "").strip() data = await state.get_data() btn_type = data.get("cbtn_edit_type") btn_id = data.get("cbtn_edit_id") if btn_type == "url" and not (content.startswith("http://") or content.startswith("https://") or content.startswith("tg://")): await message.answer("⚠️ أرسل رابطاً صحيحاً!"); return if not content: await message.answer("⚠️ المحتوى فارغ!"); return await update_custom_button(BOT_USERNAME, btn_id, content=content) await state.clear() await message.answer("✅ تم تعديل محتوى الزر بنجاح.", parse_mode="HTML", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="ادارة الازرار", callback_data="cbtn:manage", style="primary")]])) @router.callback_query(F.data.startswith("cbtn:press:")) async def cbtn_press(callback: CallbackQuery): btn_id = int(callback.data.split(":")[2]) b = await get_custom_button(BOT_USERNAME, btn_id) if not b: await callback.answer("هذا الزر لم يعد متاحاً.", show_alert=True); return await callback.message.edit_text( b["content"], parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="com:cbtn_back", style="primary")] ]) ) await callback.answer() @router.callback_query(F.data == "com:cbtn_back") async def cbtn_user_back(callback: CallbackQuery, state: FSMContext, bot: Bot): welcome_text, w_ents = await build_welcome_message(callback.from_user) if w_ents: await callback.message.edit_text(welcome_text, entities=w_ents, disable_web_page_preview=True, reply_markup=await _user_custom_kb()) else: await callback.message.edit_text(welcome_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await _user_custom_kb()) await state.set_state(UserStates.waiting_message) await callback.answer() async def _user_custom_kb() -> InlineKeyboardMarkup | None: rows = [] try: for b in await get_custom_buttons(BOT_USERNAME): if not b.get("is_visible", 1): continue rows.append([_render_custom_button(b)]) except Exception: pass return InlineKeyboardMarkup(inline_keyboard=rows) if rows else None