# down.py version: 1.2 # script was programmed by: Kodo # For any inquiries: @DevKodo # Data source and operation: @SourceKodo import asyncio import logging import os import re import sys import tempfile 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.download.db import ( init_db, get_or_create_user, get_user, get_setting, set_setting, get_all_settings, get_mandatory_channels, get_all_active_users, get_admins, is_admin, add_admin, remove_admin, clear_admins, remove_mandatory_channel, add_mandatory_channel, get_statistics, record_download, get_user_daily_count, get_funded_channels, add_funded_channel, remove_funded_channel, record_funded_subscription, 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 DevStates(StatesGroup): waiting_broadcast_msg = State() waiting_broadcast_fwd = State() waiting_add_admin = State() waiting_mand_type = State() waiting_mand_fwd = State() waiting_welcome = State() waiting_close_msg = State() waiting_open_msg = State() waiting_daily_limit = State() waiting_max_duration = State() waiting_funded_type = State() waiting_funded_fwd = State() waiting_funded_target = 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 = "" _FACTORY_URL_CACHE: dict = {} # توكن المصنع -> رابطه (يمنع تداخل الحقوق) COOKIES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cookies") BROADCAST_BATCH = 5 BROADCAST_DELAY = 2 PLATFORMS = { "youtube": "YouTube", "tiktok": "TikTok", "instagram": "Instagram", "facebook": "Facebook", "snapchat": "Snapchat", "pinterest": "Pinterest", } URL_PATTERNS = { "youtube": re.compile(r"(https?://)?(www\.)?(youtube\.com|youtu\.be|music\.youtube\.com)/\S+"), "tiktok": re.compile(r"(https?://)?(www\.|vm\.)?tiktok\.com/\S+"), "instagram": re.compile(r"(https?://)?(www\.)?instagram\.com/\S+"), "facebook": re.compile(r"(https?://)?(www\.|m\.)?facebook\.com/\S+|https?://fb\.watch/\S+"), "snapchat": re.compile(r"(https?://)?(www\.)?snapchat\.com/\S+"), "pinterest": re.compile(r"(https?://)?(www\.|[a-z]+\.)?pinterest\.(com|[a-z]+)/\S+|https?://(www\.)?pin\.it/\S+"), } async def _build_dev_panel_text(bot_label: str) -> str: """يبني نص لوحة المطور مع معلومات الاشتراك""" from information.database import get_bot_subscription from information.models import Bot as _BotModel from information.database import get_session as _get_session from sqlalchemy import select as _select text = ( f"» هذه هي لوحة التحكم وادارة {bot_label} الخاص بك.\n\n" "» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، " "متابعة الإحصائيات الفورية وإدارة القنوات وكل شيء من مكان واحد." ) factory_url = await get_factory_url() factory_username = factory_url.replace("https://t.me/", "") if factory_url else "" sub_deeplink = f"https://t.me/{factory_username}?start=subscription" if factory_username else "" async with _get_session() as _sess: _res = await _sess.execute( _select(_BotModel).where(_BotModel.username.ilike(BOT_USERNAME), _BotModel.is_active == True) ) _bot_rec = _res.scalar_one_or_none() if _bot_rec and _bot_rec.is_paid: _sub = await get_bot_subscription(_bot_rec.id) _expires = _sub.expires_at.strftime("%d-%m-%Y") if _sub else "—" text += ( f"\n» البوت مدفوع حتى: {_expires}" f"\n» يمكنك ادارة الاشتراك من هنا:" + (f'\n@{factory_username}' if sub_deeplink else "") ) elif _bot_rec and not _bot_rec.is_paid and sub_deeplink: text += ( f"\n» اجعل بوتك مدفوعاً من هنا:" f'\n@{factory_username}' ) text += "\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:" return text async def get_factory_url() -> str: # cache لكل توكن على حدة (يمنع تداخل الحقوق بين البوتات المختلفة) if not FACTORY_TOKEN: return "" cached = _FACTORY_URL_CACHE.get(FACTORY_TOKEN) if cached: return cached try: async with aiohttp.ClientSession() as session: async with session.get( f"https://api.telegram.org/bot{FACTORY_TOKEN}/getMe", timeout=aiohttp.ClientTimeout(total=10) ) as resp: data = await resp.json() if data.get("ok"): url = f"https://t.me/{data['result']['username']}" _FACTORY_URL_CACHE[FACTORY_TOKEN] = url return url except Exception: pass return "" async def notify_owner(bot: Bot, text: str, reply_markup=None): try: await bot.send_message(OWNER_ID, text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=reply_markup) except Exception: pass async def is_dev_or_admin(user_id: int) -> bool: if user_id == OWNER_ID: return True return await is_admin(BOT_USERNAME, user_id) async def get_factory_maker_rights() -> bool: try: if await _is_paid_bot(BOT_USERNAME): return False async with get_session() as session: result = await session.execute(select(FactorySettings).limit(1)) settings = result.scalar_one_or_none() return settings.cobots_maker_rights if settings else False except Exception: return False async def check_subscription(bot: Bot, user_id: int) -> list[dict]: channels = await get_mandatory_channels(BOT_USERNAME) not_subbed = [] for ch in channels: try: member = await asyncio.wait_for(bot.get_chat_member(ch["channel_id"], user_id), timeout=5) if member.status in ("left", "kicked"): not_subbed.append(ch) except Exception: not_subbed.append(ch) return not_subbed async def check_funded_channels(bot: Bot, user_id: int) -> list[dict]: channels = await get_funded_channels(BOT_USERNAME) not_subbed = [] for ch in channels: try: member = await asyncio.wait_for(bot.get_chat_member(ch["channel_id"], user_id), timeout=5) if member.status not in ("left", "kicked"): completed = await record_funded_subscription(BOT_USERNAME, ch["id"], user_id) if completed: settings = await get_all_settings(BOT_USERNAME) if settings.get("notify_funded_sub") == "1": await notify_owner(bot, f"🎉 اكتملت القناة الممولة {ch['channel_title']}!") else: 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, check_cb: str = "check_sub"): 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_cb)], ]) ) else: await bot.send_message( chat_id, f"عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:\n" f"- {channel['channel_title']}\n\nاشترك ثم أرسل /start", parse_mode="HTML", disable_web_page_preview=True) def detect_platform(url: str) -> str | None: for platform, pattern in URL_PATTERNS.items(): if pattern.search(url): return platform return None def _entities_to_json(entities): import json as _json if not entities: return None return _json.dumps([ {"type": e.type, "offset": e.offset, "length": e.length, "custom_emoji_id": getattr(e, "custom_emoji_id", None), "url": getattr(e, "url", None)} for e in entities ]) def _entities_from_json(s): import json as _json if not s: return None try: from aiogram.types import MessageEntity as _ME data = _json.loads(s) ents = [] for d in data: kwargs = {"type": d["type"], "offset": d["offset"], "length": d["length"]} if d.get("custom_emoji_id"): kwargs["custom_emoji_id"] = d["custom_emoji_id"] if d.get("url"): kwargs["url"] = d["url"] ents.append(_ME(**kwargs)) return ents except Exception: return None def _has_custom_emoji(entities): if not entities: return False return any(getattr(e, "type", None) == "custom_emoji" for e in entities) async def build_welcome_message(user): settings = await get_all_settings(BOT_USERNAME) welcome = settings.get("welcome_message", "") ent_json = settings.get("welcome_message_entities", "") factory_url = await get_factory_url() maker_rights = await get_factory_maker_rights() name_link = f'{user.first_name}' entities = _entities_from_json(ent_json) if welcome and _has_custom_emoji(entities): ph = "{name}"; idx = welcome.find(ph) text = welcome.replace(ph, user.first_name) if idx >= 0: shift = len(user.first_name) - len(ph) for e in entities: if e.offset > idx: e.offset += shift return text, entities if welcome: text = welcome.replace("{name}", name_link) else: text = ( f"» اهلاً عزيزي: {name_link} 👋🏻\n" "» مرحباً بك في نظام التحميل الذكي والفوري.\n" "» الشبكات المدعومة بالكامل 🌐:\n\n" "┌─ ❖ ⁠YouTube⁠ ┋ يوتيوب\n" "├─ ❖ ⁠TikTok⁠ ┋ تيك توك\n" "├─ ❖ ⁠Instagram⁠ ┋ انستقرام\n" "├─ ❖ ⁠Facebook⁠ ┋ فيسبوك\n" "├─ ❖ ⁠Snapchat⁠ ┋ سنابشات\n" "└─ ❖ ⁠Pinterest⁠ ┋ بينتريست\n\n" "» المنظومة جاهزة للعمل ✨:\n" "» كل ما عليك هو إرسال الرابط المطلوب الان." ) if maker_rights and factory_url: fac_un = factory_url.replace("https://t.me/", "").replace("/", "") text += f'\n» 🤖 ادري عجبك البوت يلا اصنع واحد مثلة مجاني!\n» @{fac_un}' return text, None async def build_download_keyboard(url: str, platform: str) -> InlineKeyboardMarkup: return InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="📥 تحميل فيديو", callback_data=f"dl:video:{platform}", style="success"), InlineKeyboardButton(text="🎵 تحميل صوت", callback_data=f"dl:audio:{platform}", style="success"), ], [InlineKeyboardButton(text="إلغاء", callback_data="dl:cancel", style="primary")], ]) async def download_video(url: str, platform: str, audio_only: bool = False) -> str | list[str] | None: tmp_dir = tempfile.mkdtemp(prefix="kodo_dl_") try: if platform == "youtube": return await _dl_ytdlp(url, tmp_dir, audio_only) elif platform == "tiktok": return await _dl_tiktok(url, tmp_dir, audio_only) elif platform in ("instagram", "pinterest"): return await _dl_gallery(url, tmp_dir, audio_only) elif platform in ("facebook", "snapchat"): return await _dl_cobalt(url, tmp_dir, audio_only) else: return await _dl_ytdlp(url, tmp_dir, audio_only) except Exception as e: logger.error(f"download_video error [{platform}]: {e}") return None async def _dl_ytdlp(url: str, out_dir: str, audio_only: bool) -> str | None: try: import yt_dlp except ImportError: return None proxy = "socks5://127.0.0.1:40000" if audio_only: ydl_opts = { "format": "bestaudio/best", "outtmpl": os.path.join(out_dir, "%(id)s.%(ext)s"), "proxy": proxy, "quiet": True, "no_warnings": True, "postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "mp3"}], } else: ydl_opts = { "format": "bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best", "outtmpl": os.path.join(out_dir, "%(id)s.%(ext)s"), "proxy": proxy, "quiet": True, "no_warnings": True, "merge_output_format": "mp4", "postprocessor_args": ["-movflags", "faststart"], } try: loop = asyncio.get_event_loop() def _run(): with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=True) return ydl.prepare_filename(info) filename = await loop.run_in_executor(None, _run) if audio_only: mp3 = os.path.splitext(filename)[0] + ".mp3" if os.path.exists(mp3): return mp3 if os.path.exists(filename): return filename for f in os.listdir(out_dir): fp = os.path.join(out_dir, f) if os.path.isfile(fp): return fp except Exception as e: logger.error(f"yt-dlp error: {e}") return None async def _dl_tiktok(url: str, out_dir: str, audio_only: bool) -> str | list[str] | None: try: api_url = f"https://www.tikwm.com/api/?url={url}" async with aiohttp.ClientSession() as session: async with session.get(api_url, timeout=aiohttp.ClientTimeout(total=30)) as resp: data = await resp.json() if data.get("code") == 0: d = data["data"] images = d.get("images") if images and not audio_only: paths = [] for i, img_url in enumerate(images): out_path = os.path.join(out_dir, f"photo_{i}.jpg") async with session.get(img_url, timeout=aiohttp.ClientTimeout(total=60)) as r: if r.status == 200: with open(out_path, "wb") as f: f.write(await r.read()) paths.append(out_path) if paths: return paths dl_url = d.get("music") if audio_only else d.get("play") or d.get("wmplay") if dl_url: ext = "mp3" if audio_only else "mp4" out_path = os.path.join(out_dir, f"tiktok.{ext}") async with session.get(dl_url, timeout=aiohttp.ClientTimeout(total=120)) as r: if r.status == 200: with open(out_path, "wb") as f: f.write(await r.read()) return out_path except Exception as e: logger.error(f"TikWM error: {e}") return await _dl_gallery(url, out_dir, audio_only) async def _dl_gallery(url: str, out_dir: str, audio_only: bool) -> str | list[str] | None: try: import gallery_dl.job import gallery_dl.config gallery_dl.config.clear() gallery_dl.config.set((), "directory", []) gallery_dl.config.set((), "filename", "{filename}.{extension}") gallery_dl.config.set((), "base-directory", out_dir) gallery_dl.config.set(("extractor",), "proxy", "socks5://127.0.0.1:40000") cookie_file = os.path.join(COOKIES_DIR, "instagram.txt") if os.path.exists(cookie_file): gallery_dl.config.set(("extractor", "instagram"), "cookies", cookie_file) loop = asyncio.get_event_loop() def _run(): job = gallery_dl.job.DownloadJob(url) job.run() await loop.run_in_executor(None, _run) files = [] for root, dirs, fnames in os.walk(out_dir): for fname in sorted(fnames): fp = os.path.join(root, fname) if os.path.isfile(fp) and os.path.getsize(fp) > 0: files.append(fp) if not files: return None images = [f for f in files if f.lower().endswith((".jpg", ".jpeg", ".png", ".webp", ".gif"))] videos = [f for f in files if f.lower().endswith((".mp4", ".mkv", ".webm", ".mov"))] audios = [f for f in files if f.lower().endswith((".mp3", ".m4a", ".ogg", ".opus"))] if audio_only: return audios[0] if audios else (files[0] if files else None) if images and not videos: return images if len(images) > 1 else images[0] if videos: return videos[0] return files[0] except Exception as e: logger.error(f"gallery-dl error: {e}") return None async def _dl_cobalt(url: str, out_dir: str, audio_only: bool) -> str | None: try: payload = { "url": url, "downloadMode": "audio" if audio_only else "auto", "videoQuality": "1080", "filenameStyle": "basic", } async with aiohttp.ClientSession() as session: async with session.post( "http://127.0.0.1:9000/", json=payload, headers={"Accept": "application/json", "Content-Type": "application/json"}, timeout=aiohttp.ClientTimeout(total=30) ) as resp: data = await resp.json() if data.get("status") in ("tunnel", "redirect", "stream"): dl_url = data.get("url") or data.get("tunnel") if dl_url: ext = "mp3" if audio_only else "mp4" out_path = os.path.join(out_dir, f"cobalt.{ext}") async with session.get(dl_url, timeout=aiohttp.ClientTimeout(total=120)) as r: if r.status == 200: with open(out_path, "wb") as f: f.write(await r.read()) return out_path except Exception as e: logger.error(f"Cobalt error: {e}") return None async def get_media_duration(file_path: str) -> int: try: import subprocess result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", file_path], capture_output=True, text=True, timeout=15 ) return int(float(result.stdout.strip())) except Exception: return 0 def _get_video_codec(file_path: str) -> str: try: import subprocess result = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name", "-of", "default=noprint_wrappers=1:nokey=1", file_path], capture_output=True, text=True, timeout=15 ) return (result.stdout or "").strip().lower() except Exception: return "" async def ensure_h264(file_path: str) -> str: # يعيد ترميز الفيديو الى H.264/AAC فقط اذا لم يكن متوافقاً مع تليجرام codec = _get_video_codec(file_path) if codec == "h264": return file_path if not codec: return file_path out_path = file_path.rsplit(".", 1)[0] + "_h264.mp4" try: proc = await asyncio.create_subprocess_exec( "ffmpeg", "-y", "-i", file_path, "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-c:a", "aac", "-b:a", "128k", "-pix_fmt", "yuv420p", "-movflags", "+faststart", out_path, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, ) await proc.wait() if proc.returncode == 0 and os.path.exists(out_path) and os.path.getsize(out_path) > 0: cleanup_file(file_path) return out_path except Exception as e: logger.error(f"ensure_h264 error: {e}") return file_path def cleanup_file(file_path: str): try: if file_path and os.path.exists(file_path): os.remove(file_path) parent = os.path.dirname(file_path) if os.path.isdir(parent) and not os.listdir(parent): os.rmdir(parent) except Exception: pass @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 check_funded_channels(bot, user.id) 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}") if await is_dev_or_admin(user.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 fac_url = await get_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 "" panel_text = ("» هذه هي لوحة التحكم وادارة بوت التحميل الخاص بك.\n\n" "» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات الفورية وإدارة القنوات وكل شيء من مكان واحد.") 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 get_factory_maker_rights(): factory_url = await get_factory_url() if factory_url: fac_un2 = factory_url.replace("https://t.me/", "").replace("/", "") panel_text += f'\n» 🤖 ادري عجبك البوت يلا اصنع واحد مثلة مجاني!\n» @{fac_un2}' await message.answer(panel_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_dev_panel()) _wtxt, _wents = await build_welcome_message(user) if _wents: await message.answer(_wtxt, entities=_wents, disable_web_page_preview=True, reply_markup=await _user_custom_kb()) else: await message.answer(_wtxt, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await _user_custom_kb()) @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 await check_funded_channels(bot, user.id) _wtxt, _wents = await build_welcome_message(user) if _wents: await callback.message.answer(_wtxt, entities=_wents, disable_web_page_preview=True, reply_markup=await _user_custom_kb()) else: await callback.message.answer(_wtxt, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await _user_custom_kb()) await state.clear() await callback.answer() @router.callback_query(F.data == "check_funded_sub") async def cb_check_funded_sub(callback: CallbackQuery, bot: Bot): not_funded = await check_funded_channels(bot, callback.from_user.id) if not_funded: await send_sub_request(bot, callback.message.chat.id, not_funded[0], check_cb="check_funded_sub") await callback.answer() return await callback.answer("✅ شكراً على الاشتراك!", show_alert=True) @router.message(F.text) async def handle_url(message: Message, state: FSMContext, 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 not_subbed = await check_subscription(bot, user.id) if not_subbed: await send_sub_request(bot, message.chat.id, not_subbed[0]) return not_funded = await check_funded_channels(bot, user.id) if not_funded: await send_sub_request(bot, message.chat.id, not_funded[0], check_cb="check_funded_sub") return daily_limit = await get_setting(BOT_USERNAME, "daily_limit") if daily_limit != "unlimited": try: limit = int(daily_limit) if await get_user_daily_count(BOT_USERNAME, user.id) >= limit: await message.answer(f"⚠️ لقد وصلت لحد التحميل اليومي ({limit} تحميل).\nيتجدد الحد عند منتصف الليل.") return except ValueError: pass url = platform = None for word in message.text.strip().split(): p = detect_platform(word) if p: url, platform = word, p break if not url: await message.answer( "» ⚠️ لم أتمكن من التعرف على الرابط، ارسل رابط صحيح.", parse_mode="HTML", disable_web_page_preview=True, ) return await state.update_data(pending_url=url, pending_platform=platform) await message.answer( f"» تم التعرف على رابط {PLATFORMS.get(platform, platform.title())} اختر طريقة التحميل الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_download_keyboard(url, platform) ) @router.callback_query(F.data == "dl:cancel") async def cb_dl_cancel(callback: CallbackQuery, state: FSMContext): await state.clear() await callback.message.edit_text("❌ تم إلغاء التحميل.") await callback.answer() @router.callback_query(F.data.startswith("dl:")) async def cb_dl_start(callback: CallbackQuery, state: FSMContext, bot: Bot): parts = callback.data.split(":") dl_type = parts[1] platform = parts[2] data = await state.get_data() url = data.get("pending_url") if not url: await callback.answer("⚠️ انتهت صلاحية الطلب، أرسل الرابط مجدداً.", show_alert=True) return audio_only = dl_type == "audio" await state.clear() max_dur_str = await get_setting(BOT_USERNAME, "max_duration") try: max_duration = int(max_dur_str) * 60 except ValueError: max_duration = 600 platform_name = PLATFORMS.get(platform, platform.title()) type_label = "الصوت" if audio_only else "الفيديو" wait_msg = await callback.message.edit_text(f"⏳ جاري تحميل {type_label} من {platform_name} انتضر قليلاً...", parse_mode="HTML", disable_web_page_preview=True,) uname = BOT_USERNAME tok = BOT_TOKEN chat_id = callback.message.chat.id user_id = callback.from_user.id wait_id = wait_msg.message_id async def _run_download(): file_path = None b = Bot(token=tok) try: file_path = await download_video(url, platform, audio_only) if not file_path: await b.edit_message_text("❌ فشل التحميل. تأكد من صحة الرابط أو حاول مرة أخرى لاحقاً.", chat_id=chat_id, message_id=wait_id) return duration = await get_media_duration(file_path) if max_duration > 0 and duration > max_duration: cleanup_file(file_path); file_path = None await b.edit_message_text(f"⚠️ المحتوى يتجاوز الحد المسموح به ({max_dur_str} دقيقة).", chat_id=chat_id, message_id=wait_id) return caption = f"» Download from: {platform_name} Source: @{uname}" try: from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo if isinstance(file_path, list): media = [] for i, fp in enumerate(file_path): ext = fp.lower().split(".")[-1] if ext in ("jpg", "jpeg", "png", "webp"): media.append(InputMediaPhoto( media = FSInputFile(fp), caption = caption if i == 0 else None, )) else: media.append(InputMediaVideo( media = FSInputFile(fp), caption = caption if i == 0 else None, )) for i in range(0, len(media), 10): await b.send_media_group(chat_id, media[i:i+10]) elif audio_only: await b.send_audio( chat_id, audio = FSInputFile(file_path), caption = caption, ) elif file_path.lower().split(".")[-1] in ("jpg", "jpeg", "png", "webp", "gif"): await b.send_photo( chat_id, photo = FSInputFile(file_path), caption = caption, ) else: if platform == "instagram": file_path = await ensure_h264(file_path) await b.send_video( chat_id, video = FSInputFile(file_path), caption = caption, supports_streaming = True, ) await record_download(uname, user_id, platform) try: await b.delete_message(chat_id, wait_id) except Exception: pass except Exception as e: logger.error(f"Error sending file: {e}") try: await b.edit_message_text("❌ فشل في إرسال الملف. حجم الملف قد يكون كبيراً.", chat_id=chat_id, message_id=wait_id) except Exception: pass except Exception as e: logger.error(f"download task error: {e}") finally: if file_path: if isinstance(file_path, list): for fp in file_path: cleanup_file(fp) else: cleanup_file(file_path) await b.session.close() asyncio.create_task(_run_download()) 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 "✖️" notify_f = "✅" if s.get("notify_funded_sub") == "1" else "✖️" status = "مفتوح 🟢" if is_open else "مقفول 🔴" daily_lim = s.get("daily_limit", "unlimited") max_dur = s.get("max_duration", "10") lim_label = "لا حد" if daily_lim == "unlimited" else f"{daily_lim} تحميل" return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"حالة البوت {status}", callback_data="dp:toggle_open", style="success")], [ InlineKeyboardButton(text=f"اشعار الدخول {notify_u}", callback_data="dp:toggle_notify_u", style="success"), InlineKeyboardButton(text=f"اشعار الحظر {notify_b}", callback_data="dp:toggle_notify_b", style="success"), ], [InlineKeyboardButton(text=f"اشعار الممولة {notify_f}", callback_data="dp:toggle_notify_f", 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:mandatory", style="primary")], [InlineKeyboardButton(text=" القنوات الممولة", callback_data="dp:funded", style="danger")], [ InlineKeyboardButton(text=f" الحد اليومي: {lim_label}", callback_data="dp:set_limit", style="danger"), InlineKeyboardButton(text=f"المدة: {max_dur} دقيقة", callback_data="dp:set_duration", style="danger"), ], [InlineKeyboardButton(text=" رسالة الترحيب", callback_data="dp:edit_welcome", 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", "dp:toggle_notify_f"})) async def cb_dp_toggle(callback: CallbackQuery): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return key_map = {"dp:toggle_notify_u": "notify_new_users", "dp:toggle_notify_b": "notify_blocked", "dp:toggle_notify_f": "notify_funded_sub"} 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) dl = stats["dl_by_platform"] 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['total_dl']}\n" f" ▸ YouTube: {dl.get('youtube',0)}\n" f" ▸ TikTok: {dl.get('tiktok',0)}\n" f" ▸ Instagram: {dl.get('instagram',0)}\n" f" ▸ Facebook: {dl.get('facebook',0)}\n" f" ▸ Snapchat: {dl.get('snapchat',0)}\n" f" ▸ Pinterest: {dl.get('pinterest',0)}\n\n" f"[+] الاشتراك الإجباري: {stats['mand_count']} قناة\n" f"[+] القنوات الممولة: {stats['fund_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="success")] 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: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» يمكن اضافة وحذف وعرض القنوات من الازرار:\n\n" if channels else "» لا توجد قنوات مضافة حالياً، اضف قناة الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons)) await callback.answer() @dev_router.callback_query(F.data.startswith("dp:mand_del:")) async def cb_dp_mand_del(callback: CallbackQuery): 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.callback_query(F.data == "dp:funded") async def cb_dp_funded(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_funded_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=f"{ch['channel_title']} ({ch['current_count']}/{ch['target_count']})", url=link, style="success"), InlineKeyboardButton(text="🗑️", callback_data=f"dp:fund_del:{ch['id']}", style="danger")]) buttons.append([InlineKeyboardButton(text="إضافة قناة ممولة", callback_data="dp:fund_add", style="success")]) buttons.append([InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")]) await callback.message.edit_text( "» عزيزي المطور هذا هو قسم تمويل القنوات.\n\n» هذا القسم مختص في اجراء تمويل لقناة انت من يحددها حيث سيطلب البوت من المستخدمين الاشتراك في هذه القناة، عند اكتمال العدد المطلوب الذي تحدده انت سيتم ازالة القناة من القنوات الممولة." if channels else "» عزيزي المطور هذا هو قسم تمويل القنوات.\n\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:fund_del:")) async def cb_dp_fund_del(callback: CallbackQuery): await remove_funded_channel(BOT_USERNAME, int(callback.data.split(":")[2])) await callback.answer("✅ تم حذف القناة.", show_alert=True) await cb_dp_funded(callback) @dev_router.callback_query(F.data == "dp:fund_add") async def cb_dp_fund_add(callback: CallbackQuery, state: FSMContext): await state.set_state(DevStates.waiting_funded_type) await callback.message.edit_text("» اختر الطريقة التي تريد عرض طلب الاشتراك فيها من المستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="أزرار", callback_data="dp:fund_type:buttons", style="success"), InlineKeyboardButton(text="رسالة نصية", callback_data="dp:fund_type:text", style="success")], [InlineKeyboardButton(text="رجوع", callback_data="dp:funded", style="primary")], ])) await callback.answer() @dev_router.callback_query(F.data.startswith("dp:fund_type:")) async def cb_dp_fund_type(callback: CallbackQuery, state: FSMContext): await state.update_data(fund_display_type=callback.data.split(":")[2]) await state.set_state(DevStates.waiting_funded_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:funded", style="primary")]])) await callback.answer() @dev_router.message(DevStates.waiting_funded_fwd) async def process_funded_fwd(message: Message, state: FSMContext, bot: Bot): if not message.forward_from_chat: await message.answer("⚠️ يرجى توجيه رسالة من القناة فقط!"); return chat = message.forward_from_chat 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 await state.update_data(fund_chat_id=chat.id, fund_chat_title=chat.title or str(chat.id), fund_username=chat.username, fund_invite=invite_link) await state.set_state(DevStates.waiting_funded_target) await message.answer(f"» القناة: {chat.title}\n» قم بإرسال عدد المشتركين المستهدف الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="رجوع", callback_data="dp:funded", style="primary")]])) @dev_router.message(DevStates.waiting_funded_target) async def process_funded_target(message: Message, state: FSMContext): try: target = int(message.text.strip()) if target <= 0: raise ValueError except (ValueError, AttributeError): await message.answer("⚠️ أرسل رقماً صحيحاً أكبر من صفر!"); return data_state = await state.get_data() await add_funded_channel(BOT_USERNAME, str(data_state["fund_chat_id"]), data_state["fund_chat_title"], data_state.get("fund_username"), data_state.get("fund_invite"), target, data_state.get("fund_display_type", "buttons")) await state.clear() await message.answer(f"» تم اضافة القناة الممولة بنجاح.\n» الهدف: {target} عضو", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="القنوات الممولة", callback_data="dp:funded", style="primary")]])) @dev_router.callback_query(F.data == "dp:set_limit") async def cb_dp_set_limit(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return current = await get_setting(BOT_USERNAME, "daily_limit") label = "لا حد" if current == "unlimited" else f"{current} تحميل" await state.set_state(DevStates.waiting_daily_limit) await callback.message.edit_text( f"» الحد اليومي الحالي: {label}\n» ارسل الحد الجديد أو ارسل 0 لالغاء الحد.", 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_daily_limit) async def process_daily_limit(message: Message, state: FSMContext): try: val = int(message.text.strip()) except (ValueError, AttributeError): await message.answer("⚠️ أرسل رقماً صحيحاً!"); return new_val = "unlimited" if val <= 0 else str(val) await set_setting(BOT_USERNAME, "daily_limit", new_val) await state.clear() label = "لا حد" if new_val == "unlimited" else f"{val} تحميل/يوم" await message.answer(f"✅ تم تحديث الحد اليومي إلى: {label}", 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:set_duration") async def cb_dp_set_duration(callback: CallbackQuery, state: FSMContext): if not await is_dev_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return current = await get_setting(BOT_USERNAME, "max_duration") await state.set_state(DevStates.waiting_max_duration) await callback.message.edit_text( f"» الحد اليومي للمدة: {current} دقيقة\n» ارسل الحد الجديد بالدقائق أو ارسل 0 للالغاء.", 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_max_duration) async def process_max_duration(message: Message, state: FSMContext): try: val = int(message.text.strip()) if val < 0: raise ValueError except (ValueError, AttributeError): await message.answer("⚠️ أرسل رقماً صحيحاً!"); return await set_setting(BOT_USERNAME, "max_duration", str(val)) await state.clear() await message.answer(f"✅ تم تحديث حد المدة إلى: {'لا حد' if val == 0 else f'{val} دقيقة'}", 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): welcome = await get_setting(BOT_USERNAME, "welcome_message") await state.set_state(DevStates.waiting_welcome) await callback.message.edit_text( f"» رسالة الترحيب الحالية:\n{welcome or 'افتراضية'}\n\n» استخدم {{name}} لعرض اسم المستخدم في الرسالة الترحيبية.\n» قم بإرسال رسالة الترحيب الجديدة الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="رجوع", callback_data="dp:panel", style="primary")]])) await callback.answer() @dev_router.message(DevStates.waiting_welcome) async def process_welcome(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 set_setting(BOT_USERNAME, "open_message", open_msg) await state.clear() uname = BOT_USERNAME tok = BOT_TOKEN chat_id = message.chat.id async def _run_open(): users = await get_all_active_users(uname) sent = failed = 0 b = Bot(token=tok) try: for i, user in enumerate(users): try: await b.send_message(user["id"], open_msg); sent += 1 except Exception: failed += 1 if (i + 1) % BROADCAST_BATCH == 0: await asyncio.sleep(BROADCAST_DELAY) await b.send_message(chat_id, f"🟢 تم فتح البوت!\n📤 أُرسلت رسالة الفتح إلى: {sent} مستخدم", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="لوحة التحكم", callback_data="dp:panel", style="primary")]])) except Exception: pass finally: await b.session.close() asyncio.create_task(_run_open()) @dev_router.my_chat_member() async def on_my_chat_member(event: ChatMemberUpdated, bot: Bot): if event.chat.type != "private": return user_id = event.from_user.id new_status = event.new_chat_member.status import aiosqlite from cobots.download.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() _dispatcher: Dispatcher | None = None def get_down_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 # ============================================================ # الأزرار المخصصة - يضيفها مطور البوت للوحة المستخدمين # ============================================================ def _render_custom_button(b: dict): """يبني InlineKeyboardButton من زر مخصص مع إيموجي/لون""" 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="down:cbtn_back", style="primary")] ]) ) await callback.answer() @router.callback_query(F.data == "down:cbtn_back") async def cbtn_user_back(callback: CallbackQuery, state: FSMContext, bot: Bot): _wtxt, _wents = await build_welcome_message(callback.from_user) if _wents: await callback.message.edit_text(_wtxt, entities=_wents, disable_web_page_preview=True, reply_markup=await _user_custom_kb()) else: await callback.message.edit_text(_wtxt, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await _user_custom_kb()) await callback.answer() async def _user_custom_kb(): rows = [] try: for b in await get_custom_buttons(BOT_USERNAME): if not b.get("is_visible", 1): continue rows.append([_render_custom_button(b)]) except Exception: pass return InlineKeyboardMarkup(inline_keyboard=rows) if rows else None