# script was programmed by: Kodo # For any inquiries: @DevKodo # Data source and operation: @SourceKodo # buttons.py version: 1.3 import asyncio import logging import aiohttp import re from aiogram import Bot, Dispatcher, Router, F from aiogram.types import ( Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton, ) from aiogram.filters import CommandStart, Command from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage from aiogram.exceptions import TelegramBadRequest from sqlalchemy import select import sys, os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from information.database import get_session, is_bot_paid as _is_paid_bot from information.models import FactorySettings from cobots.buttons.db import ( init_db, get_setting, set_setting, get_all_settings, get_active_buttons, get_all_buttons_at_level, get_button, add_button, update_button_label, update_button_content, update_button_columns, update_button_row, toggle_button, delete_button_cascade, move_button, move_button_to, increment_press, get_top_buttons, count_buttons, get_rows_info, get_next_row_number, get_next_sort_in_row, get_all_submenus, build_tree_text, TYPE_ICONS, get_or_create_user, get_all_active_users, ban_user, unban_user, get_statistics, get_admins, is_admin, add_admin, remove_admin, get_mandatory_channels, add_mandatory_channel, remove_mandatory_channel, get_funded_channels, add_funded_channel, remove_funded_channel, record_funded_sub, get_funded_channel_count, mark_funded_complete, ) logger = logging.getLogger(__name__) BOT_TOKEN = "" BOT_USERNAME = "" MAKER_MSG = "🤖 ادري عجبك البوت يلا اصنع واحد مثله مجاني !" OWNER_ID = 0 FACTORY_TOKEN = "" MAX_BUTTONS_PER_ROW = 5 BROADCAST_BATCH = 30 BROADCAST_DELAY = 1.0 router = Router() dev_router = Router() class DevStates(StatesGroup): waiting_broadcast = State() waiting_broadcast_fwd = State() waiting_ban_id = State() waiting_unban_id = State() waiting_add_admin = State() waiting_welcome = State() waiting_welcome_image = State() waiting_mand_fwd = State() waiting_funded_fwd = State() waiting_funded_count = State() waiting_btn_label = State() waiting_btn_url = State() waiting_btn_content = State() waiting_btn_caption = State() waiting_edit_label = State() waiting_edit_content = State() async def _is_owner_or_admin(user_id: int) -> bool: return user_id == OWNER_ID or await is_admin(BOT_USERNAME, user_id) async def _notify_owner(bot: Bot, text: str): try: await bot.send_message(OWNER_ID, text, parse_mode="HTML", disable_web_page_preview=True) except Exception: pass async def _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)) s = result.scalar_one_or_none() return s.cobots_maker_rights if s else False except Exception: return False async def _factory_url() -> str: if not FACTORY_TOKEN: return "" try: async with aiohttp.ClientSession() as sess: async with sess.get( f"https://api.telegram.org/bot{FACTORY_TOKEN}/getMe", timeout=aiohttp.ClientTimeout(total=10) ) as r: data = await r.json() if data.get("ok"): return f"https://t.me/{data['result']['username']}" except Exception: pass return "" def _parse_level(raw: str) -> int | None: return None if raw == "root" else int(raw) def _level_str(parent_id: int | None) -> str: return "root" if parent_id is None else str(parent_id) def _btn_to_ik(btn: dict) -> InlineKeyboardButton: colors = ["success", "primary", "danger"] style = colors[btn["row_number"] % 3] if btn["response_type"] == "url": return InlineKeyboardButton( text=btn["label"], url=btn["response_data"] or "#", style=style ) return InlineKeyboardButton( text=btn["label"], callback_data=f"btn:{btn['id']}", style=style ) def _group_by_row(buttons: list[dict]) -> list[list[InlineKeyboardButton]]: rows: dict[int, list] = {} for btn in buttons: rn = btn["row_number"] if rn not in rows: rows[rn] = [] rows[rn].append(_btn_to_ik(btn)) return [rows[rn] for rn in sorted(rows.keys())] 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 _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 _build_root_keyboard(preview: bool = False) -> InlineKeyboardMarkup | None: buttons = await get_active_buttons(BOT_USERNAME, None) kb_rows = _group_by_row(buttons) if preview: kb_rows.append([InlineKeyboardButton( text="انهاء المعاينة", callback_data="devbtn:end_preview", style="primary" )]) if not kb_rows: return None return InlineKeyboardMarkup(inline_keyboard=kb_rows) async def _build_submenu_keyboard(parent_btn: dict) -> InlineKeyboardMarkup: buttons = await get_active_buttons(BOT_USERNAME, parent_btn["id"]) back_target = parent_btn["parent_id"] if parent_btn["parent_id"] else 0 kb_rows = _group_by_row(buttons) kb_rows.append([InlineKeyboardButton(text="رجوع", callback_data=f"btn:{back_target}", style="primary")]) return InlineKeyboardMarkup(inline_keyboard=kb_rows) def _back_kb(back_target: int) -> InlineKeyboardMarkup: return InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data=f"btn:{back_target}", style="primary") ]]) async def _build_row_selection_kb(bot_username: str, parent_id: int | None, row_cb_prefix: str, cancel_cb: str) -> InlineKeyboardMarkup: if parent_id is None: max_per_row = min(int(await get_setting(bot_username, "root_columns", "5")), MAX_BUTTONS_PER_ROW) else: parent = await get_button(bot_username, parent_id) max_per_row = min(parent["columns"] if parent else 5, MAX_BUTTONS_PER_ROW) rows_info = await get_rows_info(bot_username, parent_id) next_row = await get_next_row_number(bot_username, parent_id) kb_rows = [] for ri in rows_info: if ri["count"] < max_per_row: kb_rows.append([InlineKeyboardButton( text=f"صف {ri['row_number'] + 1} ({ri['count']}/{max_per_row} أزرار)", callback_data=f"{row_cb_prefix}:{ri['row_number']}", style="success" )]) kb_rows.append([InlineKeyboardButton( text="صف جديد", callback_data=f"{row_cb_prefix}:{next_row}", style="success" )]) kb_rows.append([InlineKeyboardButton(text="رجوع", callback_data=cancel_cb, style="primary")]) return InlineKeyboardMarkup(inline_keyboard=kb_rows) async def _check_subscription(bot: Bot, user_id: int, event) -> bool: is_cb = isinstance(event, CallbackQuery) _answer = event.message.answer if is_cb else event.answer for ch in await get_mandatory_channels(BOT_USERNAME): 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"): link = (ch.get("invite_link") or (f"https://t.me/{ch['channel_username']}" if ch.get("channel_username") else None)) if ch["display_type"] == "buttons": await _answer( "عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton( text=f"@{ch.get('channel_username') or ch['channel_title']}", url=link or "#" )], [InlineKeyboardButton(text="✅ تحقق من الاشتراك", callback_data="btn:check_sub")], ]) ) else: ref = f"{ch['channel_title']}" if link else ch["channel_title"] await _answer(f"اشترك في القناة:\n- {ref}\n\nثم أرسل /start", parse_mode="HTML", disable_web_page_preview=True) return False except Exception: pass for ch in await get_funded_channels(BOT_USERNAME): try: member = await asyncio.wait_for( bot.get_chat_member(int(ch["channel_id"]), user_id), timeout=5 ) if member.status not in ("left", "kicked", "banned"): await _handle_funded(bot, user_id, ch) except Exception: pass return True async def _handle_funded(bot: Bot, user_id: int, ch: dict): if not await record_funded_sub(BOT_USERNAME, ch["id"], user_id): return current, target = await get_funded_channel_count(BOT_USERNAME, ch["id"]) link = (ch.get("invite_link") or (f"https://t.me/{ch['channel_username']}" if ch.get("channel_username") else str(ch["channel_id"]))) if await get_setting(BOT_USERNAME, "fund_notify", "1") == "1": await _notify_owner(bot, f"🔔 اشتراك ممول\n" f"المستخدم: {user_id}\n" f"القناة: {ch['channel_title']}\n" f"العداد: {current}/{target}") if current >= target: await mark_funded_complete(BOT_USERNAME, ch["id"]) await _notify_owner(bot, f"✅ اكتملت قناة ممولة!\n" f"القناة: {ch['channel_title']} — {target} عضو.") 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 _show_welcome(event, bot: Bot, preview: bool = False): u = event.from_user welcome = await get_setting(BOT_USERNAME, "welcome_message", "") welcome_img = await get_setting(BOT_USERNAME, "welcome_image", "") ent_json = await get_setting(BOT_USERNAME, "welcome_message_entities", "") link = f'{u.first_name}' w_ents = _entities_from_json(ent_json) use_ents = bool(welcome and _has_custom_emoji(w_ents)) if use_ents: ph = "{first_name}"; idx = welcome.find(ph) text = welcome.replace(ph, u.first_name) if idx >= 0: shift = len(u.first_name) - len(ph) for e in w_ents: if e.offset > idx: e.offset += shift elif welcome: text = welcome.replace("{first_name}", link) w_ents = None else: text = f"» اهلاً عزيزي: {link} 👋🏻.\n» للبدء في استخدام البوت والاستفادة من الخدمات المتاحة، يرجى الضغط على أحد الأزرار في القائمة أدناه." w_ents = None if not use_ents and await _factory_maker_rights(): url = await _factory_url() if url: fac_un = url.replace("https://t.me/", "").replace("/", "") text += f'\n» 🤖 ادري عجبك البوت يلا اصنع واحد مثلة مجاني!\n» @{fac_un}' kb = await _build_root_keyboard(preview=preview) is_cb = isinstance(event, CallbackQuery) async def _send_text(target): if welcome_img: try: if use_ents: await target.answer_photo(photo=welcome_img, caption=text, caption_entities=w_ents, disable_web_page_preview=True, reply_markup=kb) else: await target.answer_photo(photo=welcome_img, caption=text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) return except Exception: pass if use_ents: await target.answer(text, entities=w_ents, disable_web_page_preview=True, reply_markup=kb) else: await target.answer(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) async def _edit_to_text(call): try: await call.message.delete() except Exception: pass await _send_text(call.message) if is_cb: if welcome_img: await _edit_to_text(event) else: try: if use_ents: await event.message.edit_text(text, entities=w_ents, disable_web_page_preview=True, reply_markup=kb) else: await event.message.edit_text(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) except TelegramBadRequest: await _edit_to_text(event) else: await _send_text(event) async def _show_nav(call: CallbackQuery, text: str, keyboard: InlineKeyboardMarkup, parse_mode: str = "HTML"): try: await call.message.edit_text(text, reply_markup=keyboard, parse_mode=parse_mode) except TelegramBadRequest: try: await call.message.delete() except Exception: pass await call.message.answer(text, reply_markup=keyboard, parse_mode=parse_mode) @router.message(CommandStart()) async def cmd_start(message: Message, state: FSMContext, bot: Bot): await state.clear() u = message.from_user db_usr = await get_or_create_user(BOT_USERNAME, u.id, u.username, u.first_name) if db_usr.get("is_banned"): await message.answer("⛔ أنت محظور من استخدام هذا البوت."); return is_open = await get_setting(BOT_USERNAME, "is_open", "1") if is_open != "1" and not await _is_owner_or_admin(u.id): await message.answer( await get_setting(BOT_USERNAME, "closed_message", "🔴 البوت مقفول.") ); return if not await _check_subscription(bot, u.id, message): return if db_usr.get("is_new") and await get_setting(BOT_USERNAME, "notify_new_users", "1") == "1": name = f"@{u.username}" if u.username else u.first_name await _notify_owner(bot, f"👤 مستخدم جديد!\n" f"الاسم: {u.first_name}\nالمعرف: {name}\nID: {u.id}") if await _is_owner_or_admin(u.id): from information.database import get_bot_subscription from information.models import Bot as _BotM from information.database import get_session as _gs from sqlalchemy import select as _sel panel_text = ("» هذه هي لوحة التحكم وادارة بوت الازرار الخاص بك.\n\n" "» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات وإدارة الاشتراكات وتعديل الازرار في البوت وكل شيء من مكان واحد.") fac_url = await _factory_url() fac_un = fac_url.replace("https://t.me/", "") if fac_url else "" deeplink = f"https://t.me/{fac_un}?start=subscription" if fac_un else "" async with _gs() as _s: _r = await _s.execute(_sel(_BotM).where(_BotM.username.ilike(BOT_USERNAME), _BotM.is_active == True)) _br = _r.scalar_one_or_none() if _br and _br.is_paid: _sub = await get_bot_subscription(_br.id) _exp = _sub.expires_at.strftime("%d-%m-%Y") if _sub else "—" panel_text += f"\n» البوت مدفوع حتى: {_exp}\n» يمكنك ادارة الاشتراك من هنا:" + (f'\n@{fac_un}' if deeplink else "") elif _br and not _br.is_paid and deeplink: panel_text += f"\n» اجعل بوتك مدفوعاً من هنا:\n@{fac_un}" panel_text += "\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:" if await _factory_maker_rights(): panel_text += f"\n\n{MAKER_MSG}" await message.answer(panel_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await _build_dev_panel_kb()) await _show_welcome(message, bot) @router.message(Command("help")) async def cmd_help(message: Message): kb = await _build_root_keyboard() await message.answer( "📋 كيفية الاستخدام:\n\nاضغط على الأزرار للتصفح.\n" "أرسل /start للعودة للقائمة الرئيسية.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb ) @router.callback_query(F.data == "btn:check_sub") async def cb_check_sub(call: CallbackQuery, state: FSMContext, bot: Bot): if not await _check_subscription(bot, call.from_user.id, call): return await state.clear() await _show_welcome(call, bot) await call.answer() @router.callback_query(F.data == "btn:0") async def cb_btn_root(call: CallbackQuery, bot: Bot): u = call.from_user db_usr = await get_or_create_user(BOT_USERNAME, u.id, u.username, u.first_name) if db_usr.get("is_banned"): await call.answer("⛔ أنت محظور.", show_alert=True); return if not await _check_subscription(bot, u.id, call): return await _show_welcome(call, bot) await call.answer() @router.callback_query(F.data.startswith("btn:") & ~F.data.in_({"btn:0", "btn:check_sub"})) async def cb_btn(call: CallbackQuery, bot: Bot): raw = call.data[4:] if not raw.isdigit(): await call.answer(); return btn_id = int(raw) btn = await get_button(BOT_USERNAME, btn_id) if not btn or not btn["is_active"]: await call.answer("⚠️ هذا الزر لم يعد متاحاً.", show_alert=True); return u = call.from_user db_usr = await get_or_create_user(BOT_USERNAME, u.id, u.username, u.first_name) if db_usr.get("is_banned"): await call.answer("⛔ أنت محظور.", show_alert=True); return is_open = await get_setting(BOT_USERNAME, "is_open", "1") if is_open != "1" and not await _is_owner_or_admin(u.id): await call.answer( await get_setting(BOT_USERNAME, "closed_message", "🔴 البوت مقفول."), show_alert=True); return if not await _check_subscription(bot, u.id, call): return await increment_press(BOT_USERNAME, btn_id) back_target = btn["parent_id"] if btn["parent_id"] else 0 rtype = btn["response_type"] if rtype == "submenu": children = await get_active_buttons(BOT_USERNAME, btn["id"]) if not children: await call.answer("📂 هذه القائمة فارغة حالياً.", show_alert=True); return kb = await _build_submenu_keyboard(btn) await _show_nav(call, f"📂 {btn['label']}\n\n🔽 اختر:", kb) elif rtype == "text": await _show_nav(call, btn["response_data"] or "—", _back_kb(back_target)) elif rtype == "url": await call.answer() return else: caption = btn["response_caption"] or btn["label"] kb = _back_kb(back_target) try: await call.message.delete() except Exception: pass try: if rtype == "photo": await call.message.answer_photo(btn["response_data"], caption=caption, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) elif rtype == "video": await call.message.answer_video(btn["response_data"], caption=caption, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) elif rtype == "document": await call.message.answer_document(btn["response_data"], caption=caption, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) elif rtype == "audio": await call.message.answer_audio(btn["response_data"], caption=caption, parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) except Exception as e: logger.error(f"خطأ في إرسال الميديا btn={btn_id}: {e}") await call.message.answer("⚠️ تعذّر عرض المحتوى.", reply_markup=_back_kb(back_target)) await call.answer() async def _build_dev_panel_kb() -> InlineKeyboardMarkup: s = await get_all_settings(BOT_USERNAME) is_op = "🟢 مفتوح" if s.get("is_open", "1") == "1" else "🔴 مقفول" n_blk = "✅" if s.get("notify_blocked", "1") == "1" else "✖️" n_new = "✅" if s.get("notify_new_users", "1") == "1" else "✖️" n_fund = "✅" if s.get("fund_notify", "1") == "1" else "✖️" rc = s.get("root_columns", "5") has_img = "🖼 ✅" if s.get("welcome_image", "") else "🖼 إضافة" return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"حالة البوت {is_op}", callback_data="dev:toggle_open", style="success")], [ InlineKeyboardButton(text=f"اشعار الحظر {n_blk}", callback_data="dev:toggle_notify_blk", style="success"), InlineKeyboardButton(text=f"اشعار الدخول {n_new}", callback_data="dev:toggle_notify_new", style="success"), ], [InlineKeyboardButton(text=f"اشعار الممولة {n_fund}", callback_data="dev:toggle_notify_fund", style="success")], [InlineKeyboardButton(text="إدارة الأزرار", callback_data="devbtn:list:root", style="danger")], [InlineKeyboardButton(text="معاينة القائمة", callback_data="devbtn:preview", style="danger")], [ InlineKeyboardButton(text="قسم الاذاعة", callback_data="dev:broadcast", style="primary"), InlineKeyboardButton(text="قسم الاحصائيات", callback_data="dev:stats", style="primary"), ], [ InlineKeyboardButton(text="الاشتراك الاجباري", callback_data="dev:mand", style="primary"), InlineKeyboardButton(text="قسم الادمنية", callback_data="dev:admins", style="primary"), ], [InlineKeyboardButton(text="القنوات الممولة", callback_data="dev:funded", style="danger")], [InlineKeyboardButton(text="تعديل رسالة الترحيب", callback_data="dev:welcome", style="danger")], [InlineKeyboardButton(text=f"أعمدة الجذر: {rc}", callback_data="dev:root_columns", style="danger")], ]) @dev_router.callback_query(F.data == "dev:panel") async def cb_dev_panel(call: CallbackQuery, state: FSMContext): await state.clear() if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔ غير مصرح!", show_alert=True); return await _show_nav(call, "» هذه هي لوحة التحكم وادارة بوت الازرار الخاص بك.\n\n» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات وإدارة الاشتراكات وتعديل الازرار في البوت وكل شيء من مكان واحد.\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:", await _build_dev_panel_kb()) await call.answer() @dev_router.callback_query(F.data == "dev:toggle_open") async def cb_tog_open(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return cur = await get_setting(BOT_USERNAME, "is_open", "1") await set_setting(BOT_USERNAME, "is_open", "0" if cur == "1" else "1") await call.message.edit_reply_markup(reply_markup=await _build_dev_panel_kb()) await call.answer() @dev_router.callback_query(F.data == "dev:toggle_notify_blk") async def cb_tog_blk(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return cur = await get_setting(BOT_USERNAME, "notify_blocked", "1") await set_setting(BOT_USERNAME, "notify_blocked", "0" if cur == "1" else "1") await call.message.edit_reply_markup(reply_markup=await _build_dev_panel_kb()) await call.answer() @dev_router.callback_query(F.data == "dev:toggle_notify_new") async def cb_tog_new(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return cur = await get_setting(BOT_USERNAME, "notify_new_users", "1") await set_setting(BOT_USERNAME, "notify_new_users", "0" if cur == "1" else "1") await call.message.edit_reply_markup(reply_markup=await _build_dev_panel_kb()) await call.answer() @dev_router.callback_query(F.data == "dev:toggle_notify_fund") async def cb_tog_fund(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return cur = await get_setting(BOT_USERNAME, "fund_notify", "1") await set_setting(BOT_USERNAME, "fund_notify", "0" if cur == "1" else "1") await call.message.edit_reply_markup(reply_markup=await _build_dev_panel_kb()) await call.answer() @dev_router.callback_query(F.data == "dev:root_columns") async def cb_root_columns(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return cur = await get_setting(BOT_USERNAME, "root_columns", "5") await call.message.edit_text( f"» أعمدة الجذر هو القسم المختص لتحديد الحد الاقصى للازرار في كل صف بالقائمة.\n» تم اختيار العدد: {cur} ازرار في كل صف.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"{'✅ ' if cur==str(n) else ''}{n}", callback_data=f"dev:set_rc:{n}", style="danger") for n in range(1, 6)], [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")], ]) ) await call.answer() @dev_router.callback_query(F.data.startswith("dev:set_rc:")) async def cb_set_rc(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return n = call.data.split(":")[2] await set_setting(BOT_USERNAME, "root_columns", n) await call.answer(f"✅ تم ضبط الأعمدة على {n}.", show_alert=True) await _show_nav(call, "» هذه هي لوحة التحكم وادارة بوت الازرار الخاص بك.\n\n» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات وإدارة الاشتراكات وتعديل الازرار في البوت وكل شيء من مكان واحد.\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:", await _build_dev_panel_kb()) @dev_router.callback_query(F.data == "devbtn:preview") async def cb_preview(call: CallbackQuery, bot: Bot): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await _show_welcome(call, bot, preview=True) await call.answer("👁 وضع المعاينة — اضغط على الأزرار لاختبارها") @dev_router.callback_query(F.data == "devbtn:end_preview") async def cb_end_preview(call: CallbackQuery, state: FSMContext): await state.clear() if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await _show_nav(call, "» هذه هي لوحة التحكم وادارة بوت الازرار الخاص بك.\n\n» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات وإدارة الاشتراكات وتعديل الازرار في البوت وكل شيء من مكان واحد.\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:", await _build_dev_panel_kb()) await call.answer() async def _render_btn_list(call: CallbackQuery, parent_id: int | None, bot: Bot): buttons = await get_all_buttons_at_level(BOT_USERNAME, parent_id) lvl = _level_str(parent_id) if parent_id is None: title = "📋 القائمة الرئيسية" else: p = await get_button(BOT_USERNAME, parent_id) title = f"📂 {p['label'] if p else 'قائمة فرعية'}" total_p = sum(b["press_count"] for b in buttons) header = ( f"» هذه هي القائمة الرئيسية لادارة الازرار واضافة ازرار جديدة..\n" f"» الازرار المضافة: {len(buttons)} زر\n" f"» اجمالي الضغطات: {total_p} ضغطة.\n\n" f"1. اضافة زر:\n" f"» هذا الزر هو المسؤول عن اضافة الازرار في لوحة المستخدمين يمكن أن يكون كالباك أو رابط.\n\n" f"2. شجرة القائمة:\n" f"» هذا الزر يحدد هيكلة الازرار والنشاط ورقم الصف وعدد الضغطات.\n\n" f"3. بعد اضافة زر جديد:\n" f"» اضغط عليه ستفتح لك قائمة تحتوي على تعديل/نقل/اخفاء/حذف الزر.\n\n" f"4. الاسهم ↑ و ↓ :\n" f"» هذه الاسم مهمتها ازاحة الزر او تبديل مكانه في الصف.\n\n" f"» ابدء بإضافة زر الان ثم اضغط عليه بعد الاضافة:" ) rows = [] for btn in buttons: icon = TYPE_ICONS.get(btn["response_type"], "•") status = "🟢" if btn["is_active"] else "🔴" lbl = btn["label"][:22] + ".." if len(btn["label"]) > 22 else btn["label"] rows.append([ InlineKeyboardButton(text=f"{status} {icon} {lbl}", callback_data=f"devbtn:edit:{btn['id']}", style="success"), InlineKeyboardButton(text="↑", callback_data=f"devbtn:up:{btn['id']}:{lvl}", style="success"), InlineKeyboardButton(text="↓", callback_data=f"devbtn:dn:{btn['id']}:{lvl}", style="success"), ]) rows.append([ InlineKeyboardButton(text="اضافة زر", callback_data=f"devbtn:add:{lvl}", style="danger"), InlineKeyboardButton(text="شجرة القائمة", callback_data="devbtn:tree", style="danger"), ]) if parent_id is not None: p = await get_button(BOT_USERNAME, parent_id) back_lvl = _level_str(p["parent_id"] if p else None) rows.append([InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:list:{back_lvl}", style="primary")]) else: rows.append([InlineKeyboardButton(text="لوحة التحكم", callback_data="dev:panel", style="primary")]) try: await call.message.edit_text(header, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows)) except TelegramBadRequest: await call.message.delete() await call.message.answer(header, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows)) @dev_router.callback_query(F.data.startswith("devbtn:list:")) async def cb_btn_list(call: CallbackQuery, state: FSMContext, bot: Bot): await state.clear() if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await _render_btn_list(call, _parse_level(call.data[len("devbtn:list:"):]), bot) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:up:")) async def cb_btn_up(call: CallbackQuery, bot: Bot): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return p = call.data.split(":") await move_button(BOT_USERNAME, int(p[2]), "up") await _render_btn_list(call, _parse_level(p[3]), bot) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:dn:")) async def cb_btn_dn(call: CallbackQuery, bot: Bot): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return p = call.data.split(":") await move_button(BOT_USERNAME, int(p[2]), "down") await _render_btn_list(call, _parse_level(p[3]), bot) await call.answer() @dev_router.callback_query(F.data == "devbtn:tree") async def cb_tree(call: CallbackQuery, state: FSMContext): await state.clear() if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return tree = await build_tree_text(BOT_USERNAME) text = ( "🌳 هيكل القائمة:\n\n" f"{tree}\n\n" "🟢 نشط | 🔴 مخفي | ┤N├ رقم الصف | ✦ عدد الضغطات" if tree else "🌳 القائمة فارغة حالياً." ) await call.message.edit_text(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="devbtn:list:root", style="primary") ]])) await call.answer() TYPE_LABELS = { "text": "📝 نص", "photo": "🖼 صورة", "video": "🎬 فيديو", "document": "📎 ملف", "audio": "🎵 صوت", "submenu": "📂 قائمة فرعية", } @dev_router.callback_query(F.data.startswith("devbtn:add:")) async def cb_btn_add(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return lvl = call.data[len("devbtn:add:"):] await state.update_data(add_parent_lvl=lvl) await call.message.edit_text( "» قم بأختيار نوع الزر الذي سيضهر للمستخدم، يُمكن ان يكون رابط أو كالباك، اختر من الازرار:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رابط - url", callback_data=f"devbtn:akind:url:{lvl}", style="success")], [InlineKeyboardButton(text="كالباك - callback", callback_data=f"devbtn:akind:cb:{lvl}", style="success")], [InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:list:{lvl}", style="primary")], ]) ) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:akind:")) async def cb_add_kind(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return parts = call.data.split(":") kind = parts[2] lvl = parts[3] await state.update_data(add_parent_lvl=lvl, add_kind=kind) await state.set_state(DevStates.waiting_btn_label) await call.message.edit_text( f"» الان ارسل اسم الزر الذي سيضهر للمستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:add:{lvl}", style="primary") ]]) ) await call.answer() @dev_router.message(DevStates.waiting_btn_label) async def process_btn_label(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return label = (message.text or "").strip() if not label or len(label) > 64: await message.answer("⚠️ الاسم يجب أن يكون بين 1 و64 حرفاً."); return data = await state.get_data() await state.update_data(add_label=label) lvl = data.get("add_parent_lvl", "root") kind = data.get("add_kind", "cb") if kind == "url": await state.set_state(DevStates.waiting_btn_url) await message.answer( f"» الاسم: {label}\n» ارسل الرابط الذي سيكون بداخل الزر الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="الغاء", callback_data=f"devbtn:list:{lvl}", style="primary") ]]) ) else: rows = [] row = [] for key, lbl in TYPE_LABELS.items(): row.append(InlineKeyboardButton(text=lbl, callback_data=f"devbtn:type:{key}", style="success")) if len(row) == 2: rows.append(row); row = [] if row: rows.append(row) await message.answer( f"» الاسم: {label}\n» اختر نوع المحتوى الذي سيضهر عند الضغط على الزر.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) @dev_router.message(DevStates.waiting_btn_url) async def process_btn_url(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return url = (message.text or "").strip() if not re.match(r'^https?://[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}(/.*)?$', url): await message.answer("⚠️ الرابط غير صالح. يرجى إرسال رابط صحيح يبدأ بـ http:// أو https:// ويحتوي على نطاق صحيح (مثال: https://example.com)") return await state.update_data(add_url=url) data = await state.get_data() parent_id = _parse_level(data.get("add_parent_lvl", "root")) kb = await _build_row_selection_kb( BOT_USERNAME, parent_id, "devbtn:rowsel", f"devbtn:list:{_level_str(parent_id)}" ) await message.answer("» اختر الصف الذي سيكون فيه الزر.\n» يمكنك اختيار صف جديد او اضافة الزر الى صف موجود:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) @dev_router.callback_query(F.data.startswith("devbtn:type:")) async def cb_btn_type(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return rtype = call.data[len("devbtn:type:"):] await state.update_data(add_type=rtype) data = await state.get_data() parent_id = _parse_level(data.get("add_parent_lvl", "root")) lvl = _level_str(parent_id) if rtype == "submenu": await call.message.edit_text( "📐 اختر الحد الأقصى للأزرار في كل صف داخل هذه القائمة:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text=str(n), callback_data=f"devbtn:cols_new:{n}", style="primary") for n in range(1, 6) ]]) ) elif rtype == "text": await state.set_state(DevStates.waiting_btn_content) await call.message.edit_text( "» ارسل النص الذي سيضهر عند الضغط على الزر:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="الغاء", callback_data=f"devbtn:list:{lvl}", style="primary") ]]) ) else: await state.set_state(DevStates.waiting_btn_content) lbl = TYPE_LABELS.get(rtype, rtype) await call.message.edit_text( f"» ارسل {lbl} الذي سيضهر عند الضغط على الزر:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="الغاء", callback_data=f"devbtn:list:{lvl}", style="primary") ]]) ) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:cols_new:")) async def cb_cols_new(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return cols = int(call.data.split(":")[2]) await state.update_data(add_columns=cols) data = await state.get_data() parent_id = _parse_level(data.get("add_parent_lvl", "root")) kb = await _build_row_selection_kb( BOT_USERNAME, parent_id, "devbtn:rowsel", f"devbtn:list:{_level_str(parent_id)}" ) await call.message.edit_text("» اختر الصف الذي سيكون فيه الزر.\n» يمكنك اختيار صف جديد او اضافة الزر الى صف موجود:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) await call.answer() @dev_router.message(DevStates.waiting_btn_content) async def process_btn_content(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return data = await state.get_data() rtype = data.get("add_type", "text") lvl = data.get("add_parent_lvl", "root") if rtype == "text": await state.update_data(add_content=message.html_text or message.text or "") elif rtype == "url": url = (message.text or "").strip() if not url.startswith(("http://", "https://", "tg://")): await message.answer("⚠️ رابط غير صحيح."); return await state.update_data(add_url=url) else: file_id = None if rtype == "photo" and message.photo: file_id = message.photo[-1].file_id elif rtype == "video" and message.video: file_id = message.video.file_id elif rtype == "document" and message.document: file_id = message.document.file_id elif rtype == "audio" and message.audio: file_id = message.audio.file_id if not file_id: await message.answer(f"⚠️ يرجى إرسال {TYPE_LABELS.get(rtype, rtype)} صحيح."); return await state.update_data(add_file_id=file_id) await state.set_state(DevStates.waiting_btn_caption) await message.answer( "» ارسل الكابشن الذي سيضهر اسفل المحتوى او اضغط تخطي:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="تخطي", callback_data="devbtn:skip_cap", style="primary") ]]) ) return parent_id = _parse_level(lvl) kb = await _build_row_selection_kb(BOT_USERNAME, parent_id, "devbtn:rowsel", f"devbtn:list:{lvl}") await message.answer("» اختر الصف الذي سيكون فيه الزر.\n» يمكنك اختيار صف جديد او اضافة الزر الى صف موجود:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) @dev_router.callback_query(F.data == "devbtn:skip_cap") async def cb_skip_cap(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await state.update_data(add_caption="") data = await state.get_data() parent_id = _parse_level(data.get("add_parent_lvl", "root")) kb = await _build_row_selection_kb(BOT_USERNAME, parent_id, "devbtn:rowsel", f"devbtn:list:{_level_str(parent_id)}") await call.message.edit_text("» اختر الصف الذي سيكون فيه الزر.\n» يمكنك اختيار صف جديد او اضافة الزر الى صف موجود:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) await call.answer() @dev_router.message(DevStates.waiting_btn_caption) async def process_btn_caption(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return caption = message.html_text or message.text or "" await state.update_data(add_caption=caption) data = await state.get_data() parent_id = _parse_level(data.get("add_parent_lvl", "root")) kb = await _build_row_selection_kb(BOT_USERNAME, parent_id, "devbtn:rowsel", f"devbtn:list:{_level_str(parent_id)}") await message.answer("» اختر الصف الذي سيكون فيه الزر.\n» يمكنك اختيار صف جديد او اضافة الزر الى صف موجود:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb) @dev_router.callback_query(F.data.startswith("devbtn:rowsel:")) async def cb_rowsel(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return row_num = int(call.data[len("devbtn:rowsel:"):]) data = await state.get_data() parent_id = _parse_level(data.get("add_parent_lvl", "root")) lvl = _level_str(parent_id) kind = data.get("add_kind", "cb") label = data.get("add_label", "زر") if kind == "url": await add_button(BOT_USERNAME, parent_id, label, "url", data.get("add_url", ""), "", 5, row_num) elif data.get("add_type") == "submenu": await add_button(BOT_USERNAME, parent_id, label, "submenu", "", "", data.get("add_columns", 5), row_num) elif data.get("add_type") == "text": await add_button(BOT_USERNAME, parent_id, label, "text", data.get("add_content", ""), "", 5, row_num) else: await add_button(BOT_USERNAME, parent_id, label, data.get("add_type", "text"), data.get("add_file_id", ""), data.get("add_caption", ""), 5, row_num) await state.clear() await call.message.edit_text( f"✅ تمت إضافة الزر {label} في الصف {row_num + 1}.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:list:{lvl}", style="primary") ]]) ) await call.answer() async def _render_edit_menu(call: CallbackQuery, btn_id: int): btn = await get_button(BOT_USERNAME, btn_id) if not btn: await call.answer("⚠️ الزر غير موجود.", show_alert=True); return icon = TYPE_ICONS.get(btn["response_type"], "•") status = "🟢 نشط" if btn["is_active"] else "🔴 مخفي" lvl = _level_str(btn["parent_id"]) lines = [ f"» الاسم: {btn['label']}\n» النوع: {icon}{btn['response_type']}\n» الصف: {btn['row_number'] + 1}\n» الحالة: {status}\n» الضغطات: {btn['press_count']}\n\n» يمكنك تعديل كل شيء فيه من خلال الازرار:\n", ] if btn["response_type"] == "submenu": lines.insert(4, f"├ حد الصف: {btn['columns']} أزرار") rows = [ [InlineKeyboardButton(text="تعديل الاسم", callback_data=f"devbtn:editlbl:{btn_id}", style="success")], ] if btn["response_type"] not in ("submenu", "url"): rows.append([InlineKeyboardButton(text="تعديل المحتوى", callback_data=f"devbtn:editcnt:{btn_id}", style="success")]) if btn["response_type"] == "url": rows.append([InlineKeyboardButton(text="تعديل الرابط", callback_data=f"devbtn:editcnt:{btn_id}", style="success")]) if btn["response_type"] == "submenu": rows.append([ InlineKeyboardButton(text="تغيير حد الصف", callback_data=f"devbtn:setcols:{btn_id}", style="success"), InlineKeyboardButton(text="عرض الأبناء", callback_data=f"devbtn:list:{btn_id}", style="success"), ]) rows.append([ InlineKeyboardButton(text="تغيير الصف", callback_data=f"devbtn:chrow:{btn_id}", style="success"), InlineKeyboardButton(text="نقل الزر", callback_data=f"devbtn:mv:{btn_id}", style="success"), ]) rows.append([ InlineKeyboardButton(text="اخفاء" if btn["is_active"] else "تفعيل", callback_data=f"devbtn:toggle:{btn_id}", style="danger"), InlineKeyboardButton(text="حذف", callback_data=f"devbtn:del:{btn_id}:{lvl}", style="danger"), ]) rows.append([InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:list:{lvl}", style="primary")]) await call.message.edit_text( "\n".join(lines), parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) @dev_router.callback_query(F.data.startswith("devbtn:edit:")) async def cb_btn_edit(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await _render_edit_menu(call, int(call.data[len("devbtn:edit:"):])) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:toggle:")) async def cb_toggle(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return btn_id = int(call.data[len("devbtn:toggle:"):]) await toggle_button(BOT_USERNAME, btn_id) await _render_edit_menu(call, btn_id) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:del:")) async def cb_del(call: CallbackQuery, bot: Bot): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return p = call.data.split(":") btn_id = int(p[2]) lvl = p[3] btn = await get_button(BOT_USERNAME, btn_id) lbl = btn["label"] if btn else str(btn_id) await delete_button_cascade(BOT_USERNAME, btn_id) await call.answer(f"🗑 تم حذف «{lbl}»", show_alert=True) await _render_btn_list(call, _parse_level(lvl), bot) @dev_router.callback_query(F.data.startswith("devbtn:setcols:")) async def cb_setcols(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return btn_id = int(call.data[len("devbtn:setcols:"):]) await call.message.edit_text( "📐 اختر الحد الأقصى للأزرار في الصف الواحد:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text=str(n), callback_data=f"devbtn:savecols:{btn_id}:{n}", style="primary") for n in range(1, 6) ]]) ) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:savecols:")) async def cb_savecols(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return p = call.data.split(":") btn_id = int(p[2]) cols = int(p[3]) await update_button_columns(BOT_USERNAME, btn_id, cols) await call.answer(f"✅ تم ضبط حد الصف على {cols}.", show_alert=True) await _render_edit_menu(call, btn_id) @dev_router.callback_query(F.data.startswith("devbtn:chrow:")) async def cb_chrow(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return btn_id = int(call.data[len("devbtn:chrow:"):]) btn = await get_button(BOT_USERNAME, btn_id) if not btn: await call.answer("⚠️ الزر غير موجود.", show_alert=True); return parent_id = btn["parent_id"] kb = await _build_row_selection_kb( BOT_USERNAME, parent_id, f"devbtn:svrow:{btn_id}", f"devbtn:edit:{btn_id}" ) await call.message.edit_text( f"» اختر الصف الذي سيكون فيه الزر «{btn['label']}».\n» يمكنك اختيار صف جديد او اضافة الزر الى صف موجود:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb ) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:svrow:")) async def cb_svrow(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return p = call.data.split(":") btn_id = int(p[2]) row_num = int(p[3]) await update_button_row(BOT_USERNAME, btn_id, row_num) await call.answer(f"✅ تم نقل الزر إلى الصف {row_num + 1}.", show_alert=True) await _render_edit_menu(call, btn_id) @dev_router.callback_query(F.data.startswith("devbtn:editlbl:")) async def cb_editlbl(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return btn_id = int(call.data[len("devbtn:editlbl:"):]) btn = await get_button(BOT_USERNAME, btn_id) await state.update_data(edit_btn_id=btn_id) await state.set_state(DevStates.waiting_edit_label) await call.message.edit_text( f"» الاسم الحالي: «{btn['label'] if btn else ''}»\n» ارسل الاسم الجديد للزر الحالي الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:edit:{btn_id}", style="primary") ]]) ) await call.answer() @dev_router.message(DevStates.waiting_edit_label) async def process_edit_label(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return data = await state.get_data() btn_id = data["edit_btn_id"] label = (message.text or "").strip() if not label or len(label) > 64: await message.answer("⚠️ الاسم يجب أن يكون بين 1 و64 حرفاً."); return await update_button_label(BOT_USERNAME, btn_id, label) await state.clear() await message.answer(f"✅ تم تحديث الاسم إلى {label}.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:edit:{btn_id}", style="primary") ]])) @dev_router.callback_query(F.data.startswith("devbtn:editcnt:")) async def cb_editcnt(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return btn_id = int(call.data[len("devbtn:editcnt:"):]) btn = await get_button(BOT_USERNAME, btn_id) if not btn: await call.answer("⚠️ الزر غير موجود.", show_alert=True); return await state.update_data(edit_btn_id=btn_id, edit_type=btn["response_type"]) await state.set_state(DevStates.waiting_edit_content) prompt = { "text": "» ارسل النص الجديد للزر الحالي الان.", "url": "» ارسل الرابط الجديد للزر الحالي الان.", "photo": "» ارسل الصورة الجديدة للزر الحالي الان.", "video": "» ارسل الفيديو الجديد للزر الحالي الان.", "document": "» ارسل الملف الجديد للزر الحالي الان.", "audio": "» ارسل الصوت الجديد للزر الحالي الان.", }.get(btn["response_type"], "» ارسل المحتوى الجديد للزر الحالي الان.") await call.message.edit_text(prompt, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:edit:{btn_id}", style="primary") ]])) await call.answer() @dev_router.message(DevStates.waiting_edit_content) async def process_edit_content(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return data = await state.get_data() btn_id = data["edit_btn_id"] rtype = data["edit_type"] if rtype == "text": await update_button_content(BOT_USERNAME, btn_id, message.html_text or message.text or "") elif rtype == "url": url = (message.text or "").strip() if not url.startswith(("http://", "https://", "tg://")): await message.answer("⚠️ رابط غير صحيح."); return await update_button_content(BOT_USERNAME, btn_id, url) else: file_id = None if rtype == "photo" and message.photo: file_id = message.photo[-1].file_id elif rtype == "video" and message.video: file_id = message.video.file_id elif rtype == "document" and message.document: file_id = message.document.file_id elif rtype == "audio" and message.audio: file_id = message.audio.file_id if not file_id: await message.answer(f"⚠️ يرجى إرسال {TYPE_LABELS.get(rtype, rtype)} صحيح."); return await update_button_content(BOT_USERNAME, btn_id, file_id, message.caption or "") await state.clear() await message.answer("✅ تم تحديث المحتوى.", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:edit:{btn_id}", style="primary") ]])) @dev_router.callback_query(F.data.startswith("devbtn:mv:") & ~F.data.startswith("devbtn:mv_")) async def cb_mv(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return btn_id = int(call.data[len("devbtn:mv:"):]) btn = await get_button(BOT_USERNAME, btn_id) if not btn: await call.answer("⚠️ الزر غير موجود.", show_alert=True); return submenus = await get_all_submenus(BOT_USERNAME, exclude_id=btn_id) rows = [[InlineKeyboardButton( text="📋 القائمة الرئيسية", callback_data=f"devbtn:mv_dest:{btn_id}:root", style="primary" )]] for sm in submenus: lbl = sm["label"][:25] + ".." if len(sm["label"]) > 25 else sm["label"] rows.append([InlineKeyboardButton( text=f"📂 {lbl}", callback_data=f"devbtn:mv_dest:{btn_id}:{sm['id']}", style="primary" )]) rows.append([InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:edit:{btn_id}")]) await call.message.edit_text( f"» الزر «{btn['label']}»\n» قم بإختيار الوجهة التي تريد نقل الزر لها:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:mv_dest:")) async def cb_mv_dest(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return p = call.data.split(":") btn_id = int(p[2]) dest_lvl = p[3] dest_pid = _parse_level(dest_lvl) btn = await get_button(BOT_USERNAME, btn_id) if not btn: await call.answer("⚠️ الزر غير موجود.", show_alert=True); return rows_info = await get_rows_info(BOT_USERNAME, dest_pid) if dest_pid is None: max_per = min(int(await get_setting(BOT_USERNAME, "root_columns", "5")), MAX_BUTTONS_PER_ROW) else: parent = await get_button(BOT_USERNAME, dest_pid) max_per = min(parent["columns"] if parent else 5, MAX_BUTTONS_PER_ROW) next_row = await get_next_row_number(BOT_USERNAME, dest_pid) kb_rows = [] for ri in rows_info: if ri["count"] < max_per: kb_rows.append([InlineKeyboardButton( text=f"صف {ri['row_number'] + 1} ({ri['count']}/{max_per})", callback_data=f"devbtn:mv_row:{btn_id}:{dest_lvl}:{ri['row_number']}", style="success" )]) kb_rows.append([InlineKeyboardButton( text="صف جديد", callback_data=f"devbtn:mv_row:{btn_id}:{dest_lvl}:{next_row}", style="success" )]) kb_rows.append([InlineKeyboardButton(text="رجوع", callback_data=f"devbtn:mv:{btn_id}", style="primary")]) await call.message.edit_text( f"» الزر «{btn['label']}»\n» قم بإختيار الوجهة التي تريد نقل الزر لها:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=kb_rows) ) await call.answer() @dev_router.callback_query(F.data.startswith("devbtn:mv_row:")) async def cb_mv_row(call: CallbackQuery, bot: Bot): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return p = call.data.split(":") btn_id = int(p[2]) dest_lvl = p[3] row_num = int(p[4]) dest_pid = _parse_level(dest_lvl) btn = await get_button(BOT_USERNAME, btn_id) if not btn: await call.answer("⚠️ الزر غير موجود.", show_alert=True); return await move_button_to(BOT_USERNAME, btn_id, dest_pid, row_num) await call.answer(f"✅ تم نقل «{btn['label']}» بنجاح.", show_alert=True) await _render_btn_list(call, dest_pid, bot) @dev_router.callback_query(F.data == "dev:stats") async def cb_stats(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return s = await get_statistics(BOT_USERNAME) total_b = await count_buttons(BOT_USERNAME) top = await get_top_buttons(BOT_USERNAME, 5) top_txt = "" if top: top_txt = "\n\n» أكثر الأزرار ضغطاً:\n" for i, b in enumerate(top, 1): bar = "█" * min(b["press_count"], 10) icon = TYPE_ICONS.get(b["response_type"], "•") top_txt += f"{i}. {icon} {b['label']} — {b['press_count']} {bar}\n" await call.message.edit_text( f"» عزيزي المطور هذا قسم احصائيات البوت\n\n» عدد المستخدمين: {s['total']} مستخدم\n» النشطون حالياً: {s['active']} مستخدم\n» المحضورين: {s['banned']} مستخدم\n» حظرو البوت: {s['blocked']} مستخدم\n» عدد الازرار: {total_b}{top_txt}", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary") ]]) ) await call.answer() @dev_router.callback_query(F.data == "dev:welcome") async def cb_welcome(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return cur = await get_setting(BOT_USERNAME, "welcome_message", "") await state.set_state(DevStates.waiting_welcome) await call.message.edit_text( f"» رسالة الترحيب الحالية:\n{cur or 'افتراضية'}\n\n» استخدم {{first_name}} لعرض اسم المستخدم في الرسالة الترحيبية.\n» قم بإرسال رسالة الترحيب الجديدة الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary") ]]) ) await call.answer() @dev_router.message(DevStates.waiting_welcome) async def process_welcome(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return await set_setting(BOT_USERNAME, "welcome_message", message.text or message.html_text or "") 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="dev:panel", style="primary") ]])) @dev_router.callback_query(F.data == "dev:broadcast") async def cb_broadcast(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await state.set_state(DevStates.waiting_broadcast) await call.message.edit_text( "» عزيزي المطور هذا هو قسم الاذاعة بالبوت.\n\n» ارسل رسالتك الان لعمل اذاعة للمستخدمين أو قم بالضغط على زر اذاعة بالتوجيه ثم ارسل رسالة موجهة لاذاعتها:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="اذاعة بالتوجيه", callback_data="dev:bcast_fwd", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")], ]) ) await call.answer() @dev_router.message(DevStates.waiting_broadcast) async def process_broadcast(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return await state.clear() uname = BOT_USERNAME tok = BOT_TOKEN chat_id = message.chat.id users = await get_all_active_users(uname) async def _run_background(): sent = failed = 0 b = Bot(token=tok) try: for i, u in enumerate(users): try: await b.copy_message(u["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) await b.send_message( chat_id, f"» تم ارسال الاذاعة الى كافة مستخدمين البوت.\n\n" f"» نجـح الارسال إلى: {sent} مستخدم\n" f"» فشل الارسال إلى: {failed} مستخدم", parse_mode="HTML", disable_web_page_preview=True) except Exception: pass finally: await b.session.close() asyncio.create_task(_run_background()) await message.answer("» جارٍ ارسال الاذاعة... سيتم اعلامك عند الانتهاء.", parse_mode="HTML") @dev_router.callback_query(F.data == "dev:bcast_fwd") async def cb_bcast_fwd(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await state.set_state(DevStates.waiting_broadcast_fwd) await call.message.edit_text("» قم بتوجية الرسالة التي تريد اذاعتها الان 📨 :", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:broadcast", style="primary") ]])) await call.answer() @dev_router.message(DevStates.waiting_broadcast_fwd) async def process_bcast_fwd(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return await state.clear() uname = BOT_USERNAME tok = BOT_TOKEN chat_id = message.chat.id users = await get_all_active_users(uname) async def _run_background(): sent = failed = 0 b = Bot(token=tok) try: for i, u in enumerate(users): try: await b.forward_message(u["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) await b.send_message( chat_id, f"» تم ارسال الاذاعة الى كافة مستخدمين البوت.\n\n" f"» نجـح الارسال إلى: {sent} مستخدم\n" f"» فشل الارسال إلى: {failed} مستخدم", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="لوحة التحكم", callback_data="dev:panel", style="primary") ]]) ) except Exception: pass finally: await b.session.close() asyncio.create_task(_run_background()) await message.answer("» جارٍ ارسال الاذاعة... سيتم اعلامك عند الانتهاء.", parse_mode="HTML") async def _render_admins(call: CallbackQuery): admins = await get_admins(BOT_USERNAME) text = "» عزيزي المطور هذا هو قسم التحكم بالادمنية.\n» يمكن اضافة وحذف وعرض الادمنية من الازرار:\n" rows = [] for a in admins: name = f"@{a['username']}" if a.get("username") else a["first_name"] text += f"• {name} ({a['id']})\n" rows.append([InlineKeyboardButton(text=f"🗑 {name}", callback_data=f"dev:del_admin:{a['id']}", style="danger")]) if not admins: text += "لا يوجد أدمنية." rows += [ [InlineKeyboardButton(text="إضافة أدمن", callback_data="dev:add_admin", style="success")], [InlineKeyboardButton(text="حظر مستخدم", callback_data="dev:ban", style="danger"), InlineKeyboardButton(text="رفع الحظر", callback_data="dev:unban", style="danger")], [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")], ] await call.message.edit_text(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows)) @dev_router.callback_query(F.data == "dev:admins") async def cb_admins(call: CallbackQuery, state: FSMContext): await state.clear() if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await _render_admins(call); await call.answer() @dev_router.callback_query(F.data == "dev:add_admin") async def cb_add_admin(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await state.set_state(DevStates.waiting_add_admin) await call.message.edit_text("» أرسل أيدي المستخدم الذي تريد اضافته ادمن في البوت:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:admins", style="primary") ]])) await call.answer() @dev_router.message(DevStates.waiting_add_admin) async def process_add_admin(message: Message, state: FSMContext, bot: Bot): if not await _is_owner_or_admin(message.from_user.id): return try: admin_id = int(message.text.strip()) except (ValueError, AttributeError): await message.answer("⚠️ أرسل ID رقمي صحيح!"); return try: info = await bot.get_chat(admin_id) fn, un = info.first_name or str(admin_id), info.username except Exception: fn, un = str(admin_id), None await add_admin(BOT_USERNAME, admin_id, un, fn) await state.clear() await message.answer(f"» تم اضافة المستخدم {admin_id} كأدمن في البوت.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="قسم الادمنية", callback_data="dev:admins", style="primary") ]])) @dev_router.callback_query(F.data.startswith("dev:del_admin:")) async def cb_del_admin(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await remove_admin(BOT_USERNAME, int(call.data.split(":")[2])) await call.answer("✅ تم حذف الأدمن.", show_alert=True) await _render_admins(call) @dev_router.callback_query(F.data == "dev:ban") async def cb_ban(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await state.set_state(DevStates.waiting_ban_id) await call.message.edit_text("» أرسل أيدي المستخدم الذي تريد حظره من استخدام البوت:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:admins", style="primary") ]])) await call.answer() @dev_router.message(DevStates.waiting_ban_id) async def process_ban(message: Message, state: FSMContext, bot: Bot): if not await _is_owner_or_admin(message.from_user.id): return try: ban_id = int(message.text.strip()) except (ValueError, AttributeError): await message.answer("⚠️ أرسل ID رقمي صحيح!"); return result = await ban_user(BOT_USERNAME, ban_id) await state.clear() if not result: await message.answer("⚠️ المستخدم غير موجود."); return if await get_setting(BOT_USERNAME, "notify_blocked", "1") == "1": await _notify_owner(bot, f"🚫 تم حظر المستخدم {ban_id}.") await message.answer(f"✅ تم حظر {ban_id}.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:admins", style="primary") ]])) @dev_router.callback_query(F.data == "dev:unban") async def cb_unban(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await state.set_state(DevStates.waiting_unban_id) await call.message.edit_text("» أرسل أيدي المستخدم الذي تريد رفع الحظر عنه في البوت:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:admins", style="primary") ]])) await call.answer() @dev_router.message(DevStates.waiting_unban_id) async def process_unban(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return try: uid = int(message.text.strip()) except (ValueError, AttributeError): await message.answer("⚠️ أرسل ID رقمي صحيح!"); return await unban_user(BOT_USERNAME, uid) await state.clear() await message.answer(f"✅ تم رفع الحظر عن {uid}.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[ InlineKeyboardButton(text="رجوع", callback_data="dev:admins", style="primary") ]])) async def _render_mand(call: CallbackQuery): channels = await get_mandatory_channels(BOT_USERNAME) text = "» عزيزي المطور هذا هو قسم الاشتراك الاجباري.\n» يمكن اضافة وحذف وعرض القنوات من الازرار:\n\n" rows = [] for ch in channels: name = f"@{ch['channel_username']}" if ch.get("channel_username") else ch["channel_title"] link = ch.get("invite_link") if not link and ch.get("channel_username"): link = f"https://t.me/{ch['channel_username']}" if not link: cid = str(ch["channel_id"]).replace("-100", "") link = f"https://t.me/c/{cid}" text += f"• {name}\n" rows.append([ InlineKeyboardButton(text=f"{name}", url=link, style="success"), InlineKeyboardButton(text=f"🗑", callback_data=f"dev:del_mand:{ch['id']}", style="danger") ]) if not channels: text += "لا توجد قنوات." rows += [ [InlineKeyboardButton(text="إضافة قناة", callback_data="dev:add_mand", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")], ] await call.message.edit_text( text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) @dev_router.callback_query(F.data == "dev:mand") async def cb_mand(call: CallbackQuery, state: FSMContext): await state.clear() if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔ غير مصرح!", show_alert=True); return await _render_mand(call); await call.answer() @dev_router.callback_query(F.data == "dev:add_mand") async def cb_add_mand(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔ غير مصرح!", show_alert=True); return await call.message.edit_text( "» اختر الطريقة التي تريد عرض طلب الاشتراك فيها من المستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="ازرار", callback_data="mand_type:buttons", style="success"), InlineKeyboardButton(text="رسالة نصية", callback_data="mand_type:text", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dev:mand", style="primary")], ]) ) await call.answer() @dev_router.callback_query(F.data.startswith("mand_type:")) async def cb_mand_type(call: CallbackQuery, state: FSMContext): await state.update_data(mand_display_type=call.data.split(":")[1]) await state.set_state(DevStates.waiting_mand_fwd) await call.message.edit_text( "» ممتاز، الان قم بتوجيه رسالة من القناة/المجموعة لاضافتها.\n⚠️ يجب ان يكون البوت مشرفاً في القناة/المجموعة.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dev:mand", style="primary")] ]) ) await call.answer() @dev_router.message(DevStates.waiting_mand_fwd) async def process_mand_fwd(message: Message, state: FSMContext, bot: Bot): if not await _is_owner_or_admin(message.from_user.id): return if not message.forward_from_chat: await message.answer("⚠️ يرجى توجيه رسالة من القناة فقط!"); return chat = message.forward_from_chat try: me = await bot.get_me() mb = await bot.get_chat_member(chat.id, me.id) if mb.status not in ("administrator", "creator"): await message.answer("⚠️ يجب إضافة البوت كمشرف في القناة أولاً!") await state.clear(); return except Exception: await message.answer("⚠️ لا يمكن الوصول للقناة."); await state.clear(); return link = f"https://t.me/{chat.username}" if chat.username else None if not link: try: link = (await bot.create_chat_invite_link(chat.id)).invite_link except Exception: pass d = await state.get_data() await add_mandatory_channel( BOT_USERNAME, str(chat.id), chat.title or str(chat.id), chat.username, link, d.get("mand_display_type", "buttons") ) 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="dev:mand", style="primary")] ]) ) @dev_router.callback_query(F.data.startswith("dev:del_mand:")) async def cb_del_mand(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔ غير مصرح!", show_alert=True); return await remove_mandatory_channel(BOT_USERNAME, int(call.data.split(":")[2])) await call.answer("✅ تم حذف القناة.", show_alert=True) await _render_mand(call) async def _render_funded(call: CallbackQuery): channels = await get_funded_channels(BOT_USERNAME) text = "» عزيزي المطور هذا هو قسم تمويل القنوات.\n\n» هذا القسم مختص في اجراء تمويل لقناة انت من يحددها حيث سيطلب البوت من المستخدمين الاشتراك في هذه القناة، عند اكتمال العدد المطلوب الذي تحدده انت سيتم ازالة القناة من القنوات الممولة.\n\n" rows = [] for ch in channels: name = f"@{ch['channel_username']}" if ch.get("channel_username") else ch["channel_title"] link = ch.get("invite_link") if not link and ch.get("channel_username"): link = f"https://t.me/{ch['channel_username']}" if not link: cid = str(ch["channel_id"]).replace("-100", "") link = f"https://t.me/c/{cid}" text += f"• {name} ({ch['current_count']}/{ch['target_count']})\n" rows.append([ InlineKeyboardButton(text=f"{name}", url=link, style="success"), InlineKeyboardButton(text=f"🗑", callback_data=f"dev:del_funded:{ch['id']}", style="danger") ]) if not channels: text += "لا توجد قنوات." rows += [ [InlineKeyboardButton(text="اضافة قناة", callback_data="dev:add_funded", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")], ] await call.message.edit_text( text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) @dev_router.callback_query(F.data == "dev:funded") async def cb_funded(call: CallbackQuery, state: FSMContext): await state.clear() if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔ غير مصرح!", show_alert=True); return await _render_funded(call); await call.answer() @dev_router.callback_query(F.data == "dev:add_funded") async def cb_add_funded(call: CallbackQuery, state: FSMContext): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔ غير مصرح!", show_alert=True); return await call.message.edit_text( "» اختر الطريقة التي تريد عرض طلب الاشتراك فيها من المستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="ازرار", callback_data="funded_type:buttons", style="success"), InlineKeyboardButton(text="رسالة نصية", callback_data="funded_type:text", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dev:funded", style="primary")], ]) ) await call.answer() @dev_router.callback_query(F.data.startswith("funded_type:")) async def cb_funded_type(call: CallbackQuery, state: FSMContext): await state.update_data(funded_display_type=call.data.split(":")[1]) await state.set_state(DevStates.waiting_funded_fwd) await call.message.edit_text( "» ممتاز، الان قم بتوجيه رسالة من القناة/المجموعة لاضافتها.\n⚠️ يجب ان يكون البوت مشرفاً في القناة/المجموعة.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dev:funded", style="primary")] ]) ) await call.answer() @dev_router.message(DevStates.waiting_funded_fwd) async def process_funded_fwd(message: Message, state: FSMContext, bot: Bot): if not await _is_owner_or_admin(message.from_user.id): return if not message.forward_from_chat: await message.answer("⚠️ يرجى توجيه رسالة من القناة فقط!"); return chat = message.forward_from_chat try: me = await bot.get_me() mb = await bot.get_chat_member(chat.id, me.id) if mb.status not in ("administrator", "creator"): await message.answer("⚠️ يجب إضافة البوت كمشرف في القناة أولاً!") await state.clear(); return except Exception: await message.answer("⚠️ لا يمكن الوصول للقناة."); await state.clear(); return link = f"https://t.me/{chat.username}" if chat.username else None if not link: try: link = (await bot.create_chat_invite_link(chat.id)).invite_link except Exception: pass await state.update_data( funded_chat_id=str(chat.id), funded_title=chat.title or str(chat.id), funded_username=chat.username, funded_invite=link, ) await state.set_state(DevStates.waiting_funded_count) await message.answer( f"» القناة: {chat.title}\n» قم بإرسال عدد المشتركين المستهدف الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dev:funded", style="primary")] ]) ) @dev_router.message(DevStates.waiting_funded_count) async def process_funded_count(message: Message, state: FSMContext): if not await _is_owner_or_admin(message.from_user.id): return try: target = int(message.text.strip()) if target <= 0: raise ValueError except (ValueError, AttributeError): await message.answer("⚠️ أرسل رقماً صحيحاً أكبر من صفر!"); return d = await state.get_data() await add_funded_channel( BOT_USERNAME, d["funded_chat_id"], d["funded_title"], d.get("funded_username"), d.get("funded_invite"), d.get("funded_display_type", "buttons"), target, ) await state.clear() await message.answer( f"» تم اضافة القناة: {d['funded_title']} بنجاح.\n» الهدف: {target} عضو", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="القنوات الممولة", callback_data="dev:funded", style="primary")] ]) ) @dev_router.callback_query(F.data.startswith("dev:del_funded:")) async def cb_del_funded(call: CallbackQuery): if not await _is_owner_or_admin(call.from_user.id): await call.answer("⛔", show_alert=True); return await remove_funded_channel(BOT_USERNAME, int(call.data.split(":")[2])) await call.answer("✅ تم حذف القناة.", show_alert=True) await _render_funded(call) _buttons_dispatcher: Dispatcher | None = None def get_buttons_dispatcher() -> Dispatcher: global _buttons_dispatcher if _buttons_dispatcher is None: storage = MemoryStorage() _buttons_dispatcher = Dispatcher(storage=storage) _buttons_dispatcher.include_router(dev_router) _buttons_dispatcher.include_router(router) return _buttons_dispatcher