# script was programmed by: Kodo # For any inquiries: @DevKodo # Data source and operation: @SourceKodo # decor.py version: 1.1 import asyncio import logging import os import sys import aiohttp 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.decor.db import ( init_db, get_or_create_user, get_user, get_setting, set_setting, get_all_settings, get_all_active_users, get_banned_users, ban_user, unban_user, get_statistics, record_activity, get_admins, is_admin, add_admin, remove_admin, clear_admins, get_mandatory_channels, add_mandatory_channel, remove_mandatory_channel, 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_arabic_text = State() waiting_english_text = State() class DevStates(StatesGroup): waiting_broadcast_msg = State() waiting_broadcast_fwd = State() waiting_add_admin = State() waiting_ban_user = State() waiting_mand_type = State() waiting_mand_fwd = State() waiting_close_msg = State() waiting_open_msg = State() waiting_welcome_msg = 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 = "" MAKER_MSG = "🤖 ادري عجبك البوت يلا اصنع واحد مثله مجاني !" OWNER_ID : int = 0 FACTORY_TOKEN: str = "" BROADCAST_BATCH = 5 BROADCAST_DELAY = 2 _FACTORY_URL_CACHE: dict = {} BOLD_UPPER = "𝐀𝐁𝐂𝐃𝐄𝐅𝐆𝐇𝐈𝐉𝐊𝐋𝐌𝐍𝐎𝐏𝐐𝐑𝐒𝐓𝐔𝐕𝐖𝐗𝐘𝐙" BOLD_LOWER = "𝐚𝐛𝐜𝐝𝐞𝐟𝐠𝐡𝐢𝐣𝐤𝐥𝐦𝐧𝐨𝐩𝐪𝐫𝐬𝐭𝐮𝐯𝐰𝐱𝐲𝐳" ITAL_UPPER = "𝐴𝐵𝐶𝐷𝐸𝐹𝐺𝐻𝐼𝐽𝐾𝐿𝑀𝑁𝑂𝑃𝑄𝑅𝑆𝑇𝑈𝑉𝑊𝑋𝑌𝑍" ITAL_LOWER = "𝑎𝑏𝑐𝑑𝑒𝑓𝑔ℎ𝑖𝑗𝑘𝑙𝑚𝑛𝑜𝑝𝑞𝑟𝑠𝑡𝑢𝑣𝑤𝑥𝑦𝑧" BDIT_UPPER = "𝑨𝑩𝑪𝑫𝑬𝑭𝑮𝑯𝑰𝑱𝑲𝑳𝑴𝑵𝑶𝑷𝑸𝑹𝑺𝑻𝑼𝑽𝑾𝑿𝒀𝒁" BDIT_LOWER = "𝒂𝒃𝒄𝒅𝒆𝒇𝒈𝒉𝒊𝒋𝒌𝒍𝒎𝒏𝒐𝒑𝒒𝒓𝒔𝒕𝒖𝒗𝒘𝒙𝒚𝒛" SCRP_UPPER = "𝒜ℬ𝒞𝒟ℰℱ𝒢ℋℐ𝒥𝒦ℒℳ𝒩𝒪𝒫𝒬ℛ𝒮𝒯𝒰𝒱𝒲𝒳𝒴𝒵" SCRP_LOWER = "𝒶𝒷𝒸𝒹ℯ𝒻ℊ𝒽𝒾𝒿𝓀𝓁𝓂𝓃ℴ𝓅𝓆𝓇𝓈𝓉𝓊𝓋𝓌𝓍𝓎𝓏" BSCP_UPPER = "𝓐𝓑𝓒𝓓𝓔𝓕𝓖𝓗𝓘𝓙𝓚𝓛𝓜𝓝𝓞𝓟𝓠𝓡𝓢𝓣𝓤𝓥𝓦𝓧𝓨𝓩" BSCP_LOWER = "𝓪𝓫𝓬𝓭𝓮𝓯𝓰𝓱𝓲𝓳𝓴𝓵𝓶𝓷𝓸𝓹𝓺𝓻𝓼𝓽𝓾𝓿𝔀𝔁𝔂𝔃" FRAK_UPPER = "𝔄𝔅ℭ𝔇𝔈𝔉𝔊ℌℑ𝔍𝔎𝔏𝔐𝔑𝔒𝔓𝔔ℜ𝔖𝔗𝔘𝔙𝔚𝔛𝔜ℨ" FRAK_LOWER = "𝔞𝔟𝔠𝔡𝔢𝔣𝔤𝔥𝔦𝔧𝔨𝔩𝔪𝔫𝔬𝔭𝔮𝔯𝔰𝔱𝔲𝔳𝔴𝔵𝔶𝔷" DSTR_UPPER = "𝔸𝔹ℂ𝔻𝔼𝔽𝔾ℍ𝕀𝕁𝕂𝕃𝕄ℕ𝕆ℙℚℝ𝕊𝕋𝕌𝕍𝕎𝕏𝕐ℤ" DSTR_LOWER = "𝕒𝕓𝕔𝕕𝕖𝕗𝕘𝕙𝕚𝕛𝕜𝕝𝕞𝕟𝕠𝕡𝕢𝕣𝕤𝕥𝕦𝕧𝕨𝕩𝕪𝕫" SANS_UPPER = "𝖠𝖡𝖢𝖣𝖤𝖥𝖦𝖧𝖨𝖩𝖪𝖫𝖬𝖭𝖮𝖯𝖰𝖱𝖲𝖳𝖴𝖵𝖶𝖷𝖸𝖹" SANS_LOWER = "𝖺𝖻𝖼𝖽𝖾𝖿𝗀𝗁𝗂𝗃𝗄𝗅𝗆𝗇𝗈𝗉𝗊𝗋𝗌𝗍𝗎𝗏𝗐𝗑𝗒𝗓" SNBD_UPPER = "𝗔𝗕𝗖𝗗𝗘𝗙𝗚𝗛𝗜𝗝𝗞𝗟𝗠𝗡𝗢𝗣𝗤𝗥𝗦𝗧𝗨𝗩𝗪𝗫𝗬𝗭" SNBD_LOWER = "𝗮𝗯𝗰𝗱𝗲𝗳𝗴𝗵𝗶𝗷𝗸𝗹𝗺𝗻𝗼𝗽𝗾𝗿𝘀𝘁𝘂𝘃𝘄𝘅𝘆𝘇" SNIT_UPPER = "𝘈𝘉𝘊𝘋𝘌𝘍𝘎𝘏𝘐𝘑𝘒𝘓𝘔𝘕𝘖𝘗𝘘𝘙𝘚𝘛𝘜𝘝𝘞𝘟𝘠𝘡" SNIT_LOWER = "𝘢𝘣𝘤𝘥𝘦𝘧𝘨𝘩𝘪𝘫𝘬𝘭𝘮𝘯𝘰𝘱𝘲𝘳𝘴𝘵𝘶𝘷𝘸𝘹𝘺𝘻" SNBI_UPPER = "𝘼𝘽𝘾𝘿𝙀𝙁𝙂𝙃𝙄𝙅𝙆𝙇𝙈𝙉𝙊𝙋𝙌𝙍𝙎𝙏𝙐𝙑𝙒𝙓𝙔𝙕" SNBI_LOWER = "𝙖𝙗𝙘𝙙𝙚𝙛𝙜𝙝𝙞𝙟𝙠𝙡𝙢𝙣𝙤𝙥𝙦𝙧𝙨𝙩𝙪𝙫𝙬𝙭𝙮𝙯" MONO_UPPER = "𝙰𝙱𝙲𝙳𝙴𝙵𝙶𝙷𝙸𝙹𝙺𝙻𝙼𝙽𝙾𝙿𝚀𝚁𝚂𝚃𝚄𝚅𝚆𝚇𝚈𝚉" MONO_LOWER = "𝚊𝚋𝚌𝚍𝚎𝚏𝚐𝚑𝚒𝚓𝚔𝚕𝚖𝚗𝚘𝚙𝚚𝚛𝚜𝚝𝚞𝚟𝚠𝚡𝚢𝚣" CIRC_UPPER = "ⒶⒷⒸⒹⒺⒻⒼⒽⒾⒿⓀⓁⓂⓃⓄⓅⓆⓇⓈⓉⓊⓋⓌⓍⓎⓏ" CIRC_LOWER = "ⓐⓑⓒⓓⓔⓕⓖⓗⓘⓙⓚⓛⓜⓝⓞⓟⓠⓡⓢⓣⓤⓥⓦⓧⓨⓩ" FULL_UPPER = "ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺ" FULL_LOWER = "ａｂｃｄｅｆｇｈｉｊｋｌｍｎｏｐｑｒｓｔｕｖｗｘｙｚ" NORM_UPPER = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" NORM_LOWER = "abcdefghijklmnopqrstuvwxyz" SMCP = { 'a':'ᴀ','b':'ʙ','c':'ᴄ','d':'ᴅ','e':'ᴇ','f':'ꜰ','g':'ɢ','h':'ʜ', 'i':'ɪ','j':'ᴊ','k':'ᴋ','l':'ʟ','m':'ᴍ','n':'ɴ','o':'ᴏ','p':'ᴘ', 'q':'Q','r':'ʀ','s':'ꜱ','t':'ᴛ','u':'ᴜ','v':'ᴠ','w':'ᴡ','x':'x', 'y':'ʏ','z':'ᴢ', } SYMBOLS_TEXT = """𓆡𓆏𓆦𓃥𓃠𓃰𓃱𓃯𓃭𓃸𓃵𓃗𓃘𓃙𓃟𓄀𓄃𓃚𓃛𓃜𓃝𓃞𓃒𓃓𓃔𓃕𓃖𓃡𓃢𓃦𓃩𓃫𓃮𓃲𓃴𓃶𓃹𓃻𓃽𓃾𓃿𓄄𓄅𓄆𓆌𓆗𓆓𓆔𓆣𓆤𓆥𓅈𓅏𓅌𓅓𓅛𓅢𓅙𓅡𓅲𓀥𓀧𓀬𓀾𓁁𓁧𓁵𓊖𒆜𒆙𒈞𒋨𒈒𒁂𒈔𒅒𒇫𒄆𒂝𒈓𒁎𒌧𒅌﷽𒐫𖣘𖤍𖠄𖧷𖠇𖠂𖠍𖠑𖠚𖠰𖠢𖡑𖡆𖡛𖢇𖢌𖢠𖢳𖢸𖣋𖣎𖤇𖤎𖥔𖥡𖧝𖦤☝︎➶➩⇗꧁ඩාඞ꫞𐂂𐂃𐂄𐂅✱꥟⁑꙳✾✼ 𝟏 𝟐 𝟑 𝟒 𝟓 𝟔 𝟕 𝟖 𝟗 𝟎 🇮🇶 ﷼ ﷻ ﷽ ✞ ッ 彡 Ω ۞ ۩ ✟ 『 』۝ Ξ 道 凸 父 个 ¤ 品 〠 ๛ 𖤍 ᶠᶸᶜᵏᵧₒᵤ ࿐ ⍆ ⍅ ⇭ ༒ 𖠃 𖠅 𖠆 𖠊 𖡒 𖡗 𖣩 ꧁ ꧂ 〰 𖥓 𖥏 𖥎 𖥌 𖥋 𖥊 𖥈 𖥅 𖥃 𖥂 𖥀 𖤼 𖤹 𖤸 𖤷 𖤶 𖤭 𖤫 𖤪 𖤨 𖤧 𖤥 𖤤 𖤣 𖤢 𖤡 𖤟 𖤞 𖤝 𖤜 𖤛 𖤚 𖤘 𖤙 𖤗 𖤕 𖤓 𖤒 𖤐 ဏ ࿘ ࿗ ࿖ ࿕ ࿑ ࿌ ࿋ ࿊ ࿉ ࿈ ࿇ ࿅ ࿄ ࿃ ࿂ ༼ ༽ ༺ ༻ ༗ ༖ ༕ ⏝ ⏜ ⏎ ၄ ߷ ܛ ׀ 𝟢 𝟣 𝟤 𝟥 𝟦 𝟧 𝟨 𝟩 𝟪 𝟫 𝟶 𝟷 𝟸 𝟹 𝟺 𝟻 𝟼 𝟽 𝟾 𝟿 𝟘 𝟙 𝟚 𝟛 𝟜 𝟝 𝟞 𝟟 𝟠 𝟡 𝟎 𝟏 𝟐 𝟑 𝟒 𝟓 𝟔 𝟕 𝟖 𝟗 ０ １ ２ ３ ４ ５ ６ ７ ８ ９""" def _translate(text: str, src: str, dst: str) -> str: return text.translate(str.maketrans(src, dst)) def en_font(text: str, upper: str, lower: str) -> str: return _translate(text, NORM_UPPER + NORM_LOWER, upper + lower) def en_small_caps(text: str) -> str: return "".join(SMCP.get(c.lower(), c) for c in text) def get_english_decorations(text: str) -> list[str]: return [ f"1⃣ {en_font(text, BOLD_UPPER, BOLD_LOWER)}", f"2⃣ {en_font(text, ITAL_UPPER, ITAL_LOWER)}", f"3⃣ {en_font(text, BDIT_UPPER, BDIT_LOWER)}", f"4⃣ {en_font(text, SCRP_UPPER, SCRP_LOWER)}", f"5⃣ {en_font(text, BSCP_UPPER, BSCP_LOWER)}", f"6⃣ {en_font(text, FRAK_UPPER, FRAK_LOWER)}", f"7⃣ {en_font(text, DSTR_UPPER, DSTR_LOWER)}", f"8⃣ {en_font(text, SANS_UPPER, SANS_LOWER)}", f"9⃣ {en_font(text, SNBD_UPPER, SNBD_LOWER)}", f"🔟 {en_font(text, SNIT_UPPER, SNIT_LOWER)}", f"1️⃣1️⃣ {en_font(text, SNBI_UPPER, SNBI_LOWER)}", f"1️⃣2️⃣ {en_font(text, MONO_UPPER, MONO_LOWER)}", f"1️⃣3️⃣ {en_font(text, CIRC_UPPER, CIRC_LOWER)}", f"1️⃣4️ {en_small_caps(text)}", f"1️⃣5️⃣ {en_font(text, FULL_UPPER, FULL_LOWER)}", ] def get_arabic_decorations(text: str) -> list[str]: t = text return [ f"1⃣ {' · '.join(t)}", f"2 {' ★ '.join(t)}", f"3⃣ {' ♡ '.join(t)}", f"4⃣ {' ◆ '.join(t)}", f"5⃣ {' ⋆ '.join(t)}", f"6⃣ {' ~ '.join(t)}", f"7⃣ {' ➳ '.join(t)}", f"8⃣ {' ꕥ '.join(t)}", f"9⃣ ꧁ {t} ꧂", f"🔟 •⊱ {t} ⊰•", f"1️⃣1️⃣ 『 {t} 』", f"1️⃣2️⃣ ༺ {t} ༻", f"1️⃣3️⃣ 𓆡 {t} 𓆡", f"1️⃣4️⃣ ❝ {t} ❞", f"1️⃣5️⃣ ┤ {t} ├", ] async def get_factory_maker_rights() -> bool: try: async with get_session() as session: from sqlalchemy import select from information.models import FactorySettings 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 get_factory_url() -> str: if not FACTORY_TOKEN: return "" if FACTORY_TOKEN in _FACTORY_URL_CACHE: return _FACTORY_URL_CACHE[FACTORY_TOKEN] try: import aiohttp as _aio async with _aio.ClientSession() as _sess: async with _sess.get( f"https://api.telegram.org/bot{FACTORY_TOKEN}/getMe", timeout=_aio.ClientTimeout(total=10) ) as r: data = await r.json() if data.get("ok"): url = f"https://t.me/{data['result']['username']}" _FACTORY_URL_CACHE[FACTORY_TOKEN] = url return url except Exception: pass return "" 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 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 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_decor_welcome(user, settings): welcome = settings.get("welcome_message", "") ent_json = settings.get("welcome_message_entities", "") name_link = f'{user.first_name}' entities = _entities_from_json(ent_json) default = ( f"» مرحباً عزيزي: {name_link} 👋🏻\n» للبدء في زخرفة نصوصك، يرجى اتباع الآتي:\n🌐 ¦ أولاً: اضغط على زر اللغة التي تريدها (عربي / إنجليزي).\n✍️ ¦ ثانياً: أرسل الاسم أو الكلمة المراد زخرفتها.\n✨ استمتع بتشكيلة واسعة من الخطوط الراقية!" ) 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 text = welcome.replace("{name}", name_link) if welcome else default if await get_factory_maker_rights(): fac_url = await get_factory_url() if fac_url: fac_un = fac_url.replace("https://t.me/", "").replace("/", "") text += f'\n» {MAKER_MSG}\n» @{fac_un}' return text, None 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) async def check_subscription(bot: Bot, user_id: int) -> list[dict]: channels = await get_mandatory_channels(BOT_USERNAME) not_subbed = [] for ch in channels: chat_id = ch["channel_id"] if str(chat_id).lstrip("-").isdigit(): chat_id = int(chat_id) try: member = await asyncio.wait_for( bot.get_chat_member(chat_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 "#" ) 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) async def main_keyboard() -> InlineKeyboardMarkup: rows = [ [ InlineKeyboardButton(text="English", callback_data="decor:english", style="success"), InlineKeyboardButton(text="عربي", callback_data="decor:arabic", style="success"), ], [InlineKeyboardButton(text="الرموز", callback_data="decor:symbols", style="primary")], ] 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) @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): await message.answer(await get_setting(BOT_USERNAME, "closed_message") or "🔴 البوت مقفول حالياً.") return db_user = await get_or_create_user(BOT_USERNAME, user.id, user.username, user.first_name) if db_user["is_banned"]: 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 await record_activity(BOT_USERNAME, user.id, "message") settings = await get_all_settings(BOT_USERNAME) if settings.get("notify_new_users") == "1" and db_user.get("is_new"): name = f"@{user.username}" if user.username else user.first_name await notify_owner(bot, f"👤 مستخدم جديد!\nالاسم: {user.first_name}\nالمعرف: {name}\nالـ ID: {user.id}") text, w_ents = await _build_decor_welcome(user, settings) if await is_dev_or_admin(user.id): await message.answer( await _build_dev_panel_text("بوت الزخرفة"), parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_dev_panel() ) if w_ents: await message.answer(text, entities=w_ents, disable_web_page_preview=True, reply_markup=await main_keyboard()) else: await message.answer(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await main_keyboard()) @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 name_link = f'{user.first_name}' await callback.message.answer( f"» مرحباً عزيزي: {name_link} 👋🏻\n» للبدء في زخرفة نصوصك، يرجى اتباع الآتي:\n🌐 ¦ أولاً: اضغط على زر اللغة التي تريدها (عربي / إنجليزي).\n✍️ ¦ ثانياً: أرسل الاسم أو الكلمة المراد زخرفتها.\n✨ استمتع بتشكيلة واسعة من الخطوط الراقية!", parse_mode="HTML", disable_web_page_preview=True, reply_markup=await main_keyboard() ) await state.clear() await callback.answer() @router.callback_query(F.data == "decor:arabic") async def cb_decor_arabic(callback: CallbackQuery, state: FSMContext): await state.set_state(UserStates.waiting_arabic_text) await callback.message.edit_text( "» حسناً، قم بإرسال اسمك باللغة العربية فقط الان.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="decor:back", style="primary")] ]) ) await callback.answer() @router.callback_query(F.data == "decor:english") async def cb_decor_english(callback: CallbackQuery, state: FSMContext): await state.set_state(UserStates.waiting_english_text) await callback.message.edit_text( "» حسناً، قم بإرسال اسمك باللغة الانكليزية فقط الان.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="decor:back", style="primary")] ]) ) await callback.answer() @router.callback_query(F.data == "decor:symbols") async def cb_decor_symbols(callback: CallbackQuery): await callback.message.edit_text( SYMBOLS_TEXT + "\n\n•⊱ اضغط على رجوع للعودة:", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="decor:back", style="primary")] ]) ) await callback.answer() @router.callback_query(F.data == "decor:back") async def cb_decor_back(callback: CallbackQuery, state: FSMContext): await state.clear() await callback.message.edit_text( "» مرحباً بعودتك 👋🏻\n» للبدء في زخرفة نصوصك، يرجى اتباع الآتي:\n🌐 ¦ أولاً: اضغط على زر اللغة التي تريدها (عربي / إنجليزي).\n✍️ ¦ ثانياً: أرسل الاسم أو الكلمة المراد زخرفتها.\n✨ استمتع بتشكيلة واسعة من الخطوط الراقية!", parse_mode="HTML", disable_web_page_preview=True, reply_markup=await main_keyboard() ) await callback.answer() @router.message(UserStates.waiting_arabic_text) async def process_arabic_text(message: Message, state: FSMContext, bot: Bot): text = (message.text or "").strip() if not text: await message.answer("⚠️ الرجاء إرسال نص."); return await record_activity(BOT_USERNAME, message.from_user.id, "message") await record_activity(BOT_USERNAME, message.from_user.id, "decorate") decos = get_arabic_decorations(text) tok = BOT_TOKEN chat_id = message.chat.id async def _run_decos(): b = Bot(token=tok) try: for deco in decos: await b.send_message(chat_id, deco) await asyncio.sleep(0.15) await b.send_message( chat_id, f'» تم زخرفة "{text}" بنجاح، اضغط رجوع للعودة.', parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="decor:back", style="primary")] ]) ) except Exception: pass finally: await b.session.close() await state.clear() asyncio.create_task(_run_decos()) @router.message(UserStates.waiting_english_text) async def process_english_text(message: Message, state: FSMContext, bot: Bot): text = (message.text or "").strip() if not text: await message.answer("⚠️ Please send some text."); return await record_activity(BOT_USERNAME, message.from_user.id, "message") await record_activity(BOT_USERNAME, message.from_user.id, "decorate") decos = get_english_decorations(text) tok = BOT_TOKEN chat_id = message.chat.id async def _run_decos(): b = Bot(token=tok) try: for deco in decos: await b.send_message(chat_id, deco) await asyncio.sleep(0.15) await b.send_message( chat_id, f'» تم زخرفة "{text}" بنجاح، اضغط رجوع للعودة.', parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="decor:back", style="primary")] ]) ) except Exception: pass finally: await b.session.close() await state.clear() asyncio.create_task(_run_decos()) @router.message(F.text) async def handle_any_message(message: Message, bot: Bot): 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): await message.answer(await get_setting(BOT_USERNAME, "closed_message") or "🔴 البوت مقفول حالياً.") return db_user = await get_or_create_user(BOT_USERNAME, user.id, user.username, user.first_name) if db_user["is_banned"]: await message.answer("⛔ أنت محظور من استخدام هذا البوت."); return await record_activity(BOT_USERNAME, user.id, "message") await message.answer( "» مرحباً بعودتك 👋🏻\n» للبدء في زخرفة نصوصك، يرجى اتباع الآتي:\n🌐 ¦ أولاً: اضغط على زر اللغة التي تريدها (عربي / إنجليزي).\n✍️ ¦ ثانياً: أرسل الاسم أو الكلمة المراد زخرفتها.\n✨ استمتع بتشكيلة واسعة من الخطوط الراقية!", parse_mode="HTML", disable_web_page_preview=True, reply_markup=await main_keyboard() ) 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_b}", callback_data="dp:toggle_notify_b", style="success"), InlineKeyboardButton(text=f"اشعار الدخول {notify_u}", callback_data="dp:toggle_notify_u", 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:banned", style="danger"), InlineKeyboardButton(text=" تعديل الترحيب", callback_data="dp:edit_welcome", style="danger"), ], [InlineKeyboardButton(text=" الاشتراك الاجباري", callback_data="dp:mandatory", style="danger")], [InlineKeyboardButton(text=" حظر مستخدم", callback_data="dp:ban_user", 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("» ارسل الرسالة التي ستضهر للمستخدمين عندما يكون البوت مقفولاً:", parse_mode="HTML", disable_web_page_preview=True, 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("» ارسل الرسالة التي ستُرسل للمستخدمين عند فتح البوت.", 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.in_({"dp:toggle_notify_u", "dp:toggle_notify_b"})) async def cb_dp_toggle_notify(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) await callback.message.edit_text( "📊 إحصائيات المستخدمين:\n" f"[+] عدد مستخدمين البوت: {stats['total_users']} مستخدم\n" f"[+] النشطين آخر 24 ساعة: {stats['active_24h']} مستخدم\n" f"[+] المستخدمين المحظورين: {stats['banned_users']} مستخدم\n" f"[+] قام بحظر البوت: {stats['blocked_bot']} مستخدم\n\n" "⚙️ إحصائيات عامة للبوت:\n" f"[+] عدد الرسائل المرسلة للبوت: {stats['messages']} رسالة\n" f"[+] عدد طلبات الزخرفة: {stats['decorations']} طلب\n" f"[+] عدد الاشتراك الإجباري: {stats['mand_count']} قناة", 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 = [[ InlineKeyboardButton(text=f"@{a['username']}" if a.get("username") else a["first_name"], url=f"tg://user?id={a['id']}", style="success"), InlineKeyboardButton(text="🗑️", callback_data=f"dp:admin_remove:{a['id']}", style="danger") ] for a in admins] 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): await remove_admin(BOT_USERNAME, int(callback.data.split(":")[2])) 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()) 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:banned") async def cb_dp_banned(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 users = await get_banned_users(BOT_USERNAME) if not users: await callback.answer("لا يوجد مستخدمون محظورون.", show_alert=True); return buttons = [[ InlineKeyboardButton(text=f"@{u['username']}" if u.get("username") else (u["first_name"] or str(u["id"])), url=f"tg://user?id={u['id']}", style="success"), InlineKeyboardButton(text="🗑️", callback_data=f"dp:unban:{u['id']}", style="danger") ] for u in users] buttons.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=buttons)) await callback.answer() @dev_router.callback_query(F.data.startswith("dp:unban:")) async def cb_dp_unban(callback: CallbackQuery, bot: Bot): user_id = int(callback.data.split(":")[2]) await unban_user(BOT_USERNAME, user_id) try: await bot.send_message(user_id, "✅ عزيزي المستخدم، تم رفع الحظر عنك ويمكنك استخدام البوت الآن.") except Exception: pass await callback.answer("✅ تم رفع الحظر عن المستخدم.", show_alert=True) await cb_dp_banned(callback) @dev_router.callback_query(F.data == "dp:ban_user") async def cb_dp_ban_user(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await state.set_state(DevStates.waiting_ban_user) await callback.message.edit_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.message(DevStates.waiting_ban_user) async def process_ban_user(message: Message, state: FSMContext, bot: Bot): try: user_id = int(message.text.strip()) except (ValueError, AttributeError): await message.answer("⚠️ أرسل ID رقمي صحيح!"); return success = await ban_user(BOT_USERNAME, user_id) await state.clear() if not success: await message.answer("⚠️ المستخدم غير موجود في قاعدة البيانات.", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="لوحة التحكم", callback_data="dp:panel", style="primary")]])); return try: await bot.send_message(user_id, "⛔ عزيزي المستخدم، تم حظرك من استخدام هذا البوت من قبل الإدارة.") except Exception: pass await message.answer(f"✅ تم حظر المستخدم {user_id} بنجاح.", 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 == "dp:edit_welcome") async def cb_dp_edit_welcome(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return welcome = await get_setting(BOT_USERNAME, "welcome_message") await state.set_state(DevStates.waiting_welcome_msg) 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:panel", style="primary")]])) await callback.answer() @dev_router.message(DevStates.waiting_welcome_msg) async def process_welcome_msg(message: Message, state: FSMContext): await set_setting(BOT_USERNAME, "welcome_message", message.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="dp:panel", 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 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.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 "#") buttons.append([InlineKeyboardButton(text=ch["channel_title"], url=link, 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» يمكن اضافة وحذف وعرض القنوات من الازرار:" if channels else "» عزيزي المطور هذا هو قسم الاشتراك الاجباري.\n» » لا توجد قنوات مضافة حالياً، اضف قناة الان:", 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): await remove_mandatory_channel(BOT_USERNAME, int(callback.data.split(":")[2])) 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): await state.update_data(mand_display_type=callback.data.split(":")[2]) 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): if not message.forward_from_chat: await message.answer("⚠️ يرجى توجيه رسالة من القناة فقط!"); return chat = message.forward_from_chat chat_id = chat.id try: async with aiohttp.ClientSession() as http: async with http.get(f"https://api.telegram.org/bot{BOT_TOKEN}/getChatMember", params={"chat_id": chat_id, "user_id": (await bot.get_me()).id}, timeout=aiohttp.ClientTimeout(total=10)) as resp: data = await resp.json() if data.get("result", {}).get("status", "") not in ("administrator", "creator"): await message.answer("⚠️ يجب إضافة البوت كمشرف في القناة!") await state.clear(); return except Exception: await message.answer("⚠️ لا يمكن التحقق من صلاحيات القناة!") await state.clear(); return invite_link = f"https://t.me/{chat.username}" if chat.username else None if not invite_link: try: async with aiohttp.ClientSession() as http: async with http.post(f"https://api.telegram.org/bot{BOT_TOKEN}/createChatInviteLink", json={"chat_id": chat_id}, timeout=aiohttp.ClientTimeout(total=10)) as resp: d = await resp.json() if d.get("ok"): invite_link = d["result"]["invite_link"] except Exception: pass data_state = await state.get_data() await add_mandatory_channel(BOT_USERNAME, str(chat_id), chat.title or str(chat_id), chat.username, invite_link, data_state.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="dp:mandatory", style="primary")]])) @dev_router.my_chat_member() async def on_my_chat_member(event: ChatMemberUpdated, bot: Bot): if event.chat.type != "private": return user_id = event.from_user.id new_status = event.new_chat_member.status import aiosqlite from cobots.decor.db import get_db_path 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الاسم: {event.from_user.first_name}\nالمعرف: {name}\nالـ 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() 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="decor:back", style="primary")] ]) ) await callback.answer() _dispatcher: Dispatcher | None = None def get_decor_dispatcher() -> Dispatcher: global _dispatcher if _dispatcher is None: storage = MemoryStorage() _dispatcher = Dispatcher(storage=storage) _dispatcher.include_router(dev_router) _dispatcher.include_router(router) return _dispatcher