# transcoding.py version: 1.3 # script was programmed by: Kodo # For any inquiries: @DevKodo # Data source and operation: @SourceKodo import asyncio import logging import os import shutil import tempfile from contextlib import asynccontextmanager from datetime import datetime from aiogram import Bot, Dispatcher, Router, F from aiogram.types import ( Message, CallbackQuery, FSInputFile, InlineKeyboardMarkup, InlineKeyboardButton, ) 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 sqlalchemy import select import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from information.database import get_session, is_bot_paid as _is_paid_bot from information.models import FactorySettings from cobots.transcoding.db import ( init_db, get_setting, set_setting, get_all_settings, get_or_create_user, get_user, get_all_active_users, get_statistics, ban_user, unban_user, get_banned_users, get_admins, is_admin, add_admin, remove_admin, get_mandatory_channels, add_mandatory_channel, remove_mandatory_channel, get_funded_channels, add_funded_channel, remove_funded_channel, record_funded_sub, get_funded_channel_count, mark_funded_complete, add_custom_button, get_custom_buttons, get_custom_button, update_custom_button, delete_custom_button, ) logger = logging.getLogger(__name__) BOT_TOKEN = "" BOT_USERNAME = "" OWNER_ID = 0 FACTORY_TOKEN = "" _FACTORY_URL_CACHE: dict = {} # توكن المصنع -> رابطه (يمنع تداخل الحقوق) MAKER_MSG = "🤖 ادري عجبك البوت يلا اصنع واحد مثله مجاني !" MAX_FILE_SIZE = 20 * 1024 * 1024 GIF_MAX_SECONDS = 15 GIF_MAX_WIDTH = 480 BROADCAST_BATCH = 30 BROADCAST_DELAY = 1.0 CONVERSIONS = { "mp3_to_voice": { "label": "بصمة ← MP3", "from": "ملف MP3", "to": "رسالة صوتية", "expect": "audio" }, "voice_to_mp3": { "label": "MP3 ← بصمة", "from": "رسالة صوتية", "to": "ملف MP3", "expect": "voice" }, "image_to_sticker": { "label": "ملصق ← صورة", "from": "صورة", "to": "ملصق + PNG", "expect": "photo" }, "sticker_to_image": { "label": "صورة ← ملصق", "from": "ملصق", "to": "صورة PNG", "expect": "sticker" }, "video_to_audio": { "label": "صوت ← فيديو", "from": "فيديو", "to": "ملف MP3", "expect": "video" }, "video_to_gif": { "label": "متحركة ← فيديو", "from": "فيديو", "to": "صورة متحركة GIF", "expect": "video" }, "vnote_to_video": { "label": "فيديو ← فيديو نوت", "from": "فيديو نوت (دائري)", "to": "فيديو عادي", "expect": "video_note" }, "vnote_to_voice": { "label": "بصمة ← فيديو نوت", "from": "فيديو نوت (دائري)", "to": "رسالة صوتية", "expect": "video_note" }, "file_to_pdf": { "label": "PDF ← ملف ورد/اكسل/إلخ", "from": "ملف Office", "to": "PDF", "expect": "document" } } router = Router() dev_router = Router() class UserStates(StatesGroup): waiting_file = State() class DevStates(StatesGroup): waiting_broadcast = State() waiting_broadcast_fwd = State() waiting_ban_id = State() waiting_unban_id = State() waiting_add_admin = State() waiting_welcome = State() waiting_closed_msg = State() waiting_mand_fwd = State() waiting_funded_fwd = State() waiting_funded_count = 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() async def is_owner_or_admin(user_id: int) -> bool: if user_id == OWNER_ID: return True return await is_admin(BOT_USERNAME, user_id) 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_main_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 async def get_factory_maker_rights() -> bool: try: if await _is_paid_bot(BOT_USERNAME): return False async with get_session() as session: result = await session.execute(select(FactorySettings).limit(1)) settings = result.scalar_one_or_none() return settings.cobots_maker_rights if settings else False except Exception: return False async def get_main_factory_url() -> str: if not FACTORY_TOKEN: return "" cached = _FACTORY_URL_CACHE.get(FACTORY_TOKEN) if cached: return cached try: import aiohttp 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 "" @asynccontextmanager async def temp_dir(user_id: int): path = tempfile.mkdtemp(prefix=f"tc_{user_id}_") try: yield path finally: shutil.rmtree(path, ignore_errors=True) async def run_ffmpeg(*args) -> tuple[bool, str]: proc = await asyncio.create_subprocess_exec( "ffmpeg", "-y", *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) _, stderr = await proc.communicate() return proc.returncode == 0, stderr.decode(errors="replace") async def run_libreoffice(input_path: str, out_dir: str) -> bool: proc = await asyncio.create_subprocess_exec( "libreoffice", "--headless", "--convert-to", "pdf", "--outdir", out_dir, input_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) await proc.communicate() return proc.returncode == 0 def kb_back() -> InlineKeyboardMarkup: return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="tc:back", style="primary")] ]) def kb_cancel() -> InlineKeyboardMarkup: return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="الغاء", callback_data="tc:back", style="primary")] ]) async def build_main_keyboard() -> InlineKeyboardMarkup: rows = [ [ InlineKeyboardButton(text="من MP3 ← بصمة", callback_data="tc:mp3_to_voice", style="success"), InlineKeyboardButton(text="MP3 ← بصمة", callback_data="tc:voice_to_mp3", style="success"), ], [ InlineKeyboardButton(text="صورة ← ملصق", callback_data="tc:image_to_sticker", style="success"), InlineKeyboardButton(text="ملصق ← صورة", callback_data="tc:sticker_to_image", style="success"), ], [ InlineKeyboardButton(text="فيديو ← MP3", callback_data="tc:video_to_audio", style="primary"), InlineKeyboardButton(text="فيديو ← متحركة", callback_data="tc:video_to_gif", style="primary"), ], [ InlineKeyboardButton(text="فيديو ← فيديو نوت", callback_data="tc:vnote_to_video", style="primary"), InlineKeyboardButton(text="فيديو نوت ← بصمة", callback_data="tc:vnote_to_voice", style="primary"), ], [ InlineKeyboardButton(text="PDF ← ملف ورد/اكسل/إلـخ", callback_data="tc:file_to_pdf", style="danger"), ], ] 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) 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_start_text(first_name: str, user_id: int): welcome = await get_setting(BOT_USERNAME, "welcome_message", "") ent_json = await get_setting(BOT_USERNAME, "welcome_message_entities", "") link = f'{first_name}' entities = _entities_from_json(ent_json) if welcome and _has_custom_emoji(entities): ph = "{first_name}"; idx = welcome.find(ph) text = welcome.replace(ph, first_name) if idx >= 0: shift = len(first_name) - len(ph) for e in entities: if e.offset > idx: e.offset += shift return text, entities if welcome: text = welcome.replace("{first_name}", link) else: text = ( f"» أهلاً عزيزي: {link} 👋🏻\n\n" "» مرحباً بك في المنظومة الذكية المتخصصة في تحويل صيغ الملفات والوسائط فورياً.\n" "» يتيح لك البوت تحويل الصور، الأصوات، الفيديوهات، والمستندات بدقة عالية وبأسرع وقت ممكن.\n" "» المنصة تدعم معالجة الملفات الكبيرة مع الحفاظ التام على جودة الملف الأصلي دون أي تغيير.\n\n" "» يرجى الاطلاع على قائمة التحويلات المتاحة أدناه واختيار الصيغة المطلوبة للبدء:" ) if await get_factory_maker_rights(): fac_url = await get_main_factory_url() if fac_url: fac_un = fac_url.replace("https://t.me/", "").replace("/", "") text += f'\n» 🤖 ادري عجبك البوت يلا اصنع واحد مثلة مجاني!\n» @{fac_un}' return text, None async def check_subscription(bot: Bot, user_id: int) -> list[dict]: channels = await get_mandatory_channels(BOT_USERNAME) not_subbed = [] for ch in channels: try: member = await asyncio.wait_for(bot.get_chat_member(int(ch["channel_id"]), user_id), timeout=5) if member.status in ("left", "kicked", "banned"): not_subbed.append(ch) except Exception: not_subbed.append(ch) return not_subbed async def handle_subscription_check(bot: Bot, user_id: int, msg_or_cb) -> bool: is_cb = isinstance(msg_or_cb, CallbackQuery) send = msg_or_cb.message.answer if is_cb else msg_or_cb.answer not_subbed = await check_subscription(bot, user_id) if not not_subbed: funded = await get_funded_channels(BOT_USERNAME) for ch in funded: try: member = await asyncio.wait_for(bot.get_chat_member(int(ch["channel_id"]), user_id), timeout=5) if member.status not in ("left", "kicked", "banned"): await _record_funded(bot, user_id, ch) except Exception: pass return True ch = not_subbed[0] link = ch.get("invite_link") or ( f"https://t.me/{ch['channel_username']}" if ch.get("channel_username") else None ) if ch["display_type"] == "buttons": await send( "عزيزي المستخدم اشترك في القناة التالية قبل استخدام البوت:", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton( text=f"@{ch.get('channel_username') or ch['channel_title']}", url=link or "#" )], [InlineKeyboardButton(text="✅ تحقق من الاشتراك", callback_data="tc:check_sub")], ]) ) else: msg = ( f"عزيزي المستخدم اشترك في القناة التالية:\n" f"- {ch['channel_title']}\n\nاشترك ثم أرسل /start" if link else f"عزيزي المستخدم اشترك في القناة التالية:\n" f"- {ch['channel_title']}\n\nاشترك ثم أرسل /start" ) await send(msg, parse_mode="HTML", disable_web_page_preview=True) return False async def _record_funded(bot: Bot, user_id: int, ch: dict): added = await record_funded_sub(BOT_USERNAME, ch["id"], user_id) if not added: return current, target = await get_funded_channel_count(BOT_USERNAME, ch["id"]) notify = await get_setting(BOT_USERNAME, "fund_notify", "1") if notify == "1": link = ch.get("invite_link") or ( f"https://t.me/{ch['channel_username']}" if ch.get("channel_username") else str(ch["channel_id"]) ) await notify_owner( bot, f"🔔 اشتراك جديد في قناة ممولة:\n" f"المستخدم: {user_id}\n" f"القناة: {ch['channel_title']}\n" f"عدد المشتركين: {current}/{target}" ) if current >= target: await mark_funded_complete(BOT_USERNAME, ch["id"]) link = ch.get("invite_link") or ( f"https://t.me/{ch['channel_username']}" if ch.get("channel_username") else str(ch["channel_id"]) ) await notify_owner( bot, f"✅ اكتملت قناة ممولة!\n" f"القناة: {ch['channel_title']}\n" f"تم الوصول إلى {target} عضو." ) async def convert_voice_to_mp3(input_path: str, output_path: str) -> tuple[bool, str]: return await run_ffmpeg( "-i", input_path, "-acodec", "libmp3lame", "-q:a", "2", output_path ) async def convert_mp3_to_voice(input_path: str, output_path: str) -> tuple[bool, str]: return await run_ffmpeg( "-i", input_path, "-acodec", "libopus", "-ar", "48000", "-b:a", "64k", output_path ) async def convert_image_to_sticker( input_path: str, output_png: str, output_webp: str ) -> tuple[bool, str]: try: from PIL import Image img = Image.open(input_path).convert("RGBA") canvas = Image.new("RGBA", (512, 512), (0, 0, 0, 0)) img.thumbnail((512, 512), Image.LANCZOS) x = (512 - img.width) // 2 y = (512 - img.height) // 2 canvas.paste(img, (x, y), img) canvas.save(output_png, "PNG") canvas.save(output_webp, "WEBP") return True, "" except Exception as e: return False, str(e) async def convert_sticker_to_image( input_path: str, output_path: str, is_animated: bool = False ) -> tuple[bool, str]: if is_animated: try: import rlottie_python as rlottie anim = rlottie.LottieAnimation.from_tgs(input_path) width, height = 512, 512 frame_data = anim.lottie_animation_render(0, width, height) from PIL import Image img = Image.frombytes("RGBA", (width, height), bytes(frame_data)) img.save(output_path, "PNG") return True, "" except ImportError: return False, "rlottie_python_missing" except Exception as e: return False, str(e) else: try: from PIL import Image img = Image.open(input_path).convert("RGBA") img.save(output_path, "PNG") return True, "" except Exception as e: return False, str(e) async def convert_video_to_audio(input_path: str, output_path: str) -> tuple[bool, str]: return await run_ffmpeg( "-i", input_path, "-vn", "-acodec", "libmp3lame", "-q:a", "2", output_path ) async def convert_video_to_gif(input_path: str, output_path: str) -> tuple[bool, str]: palette_path = output_path.replace(".gif", "_palette.png") ok, err = await run_ffmpeg( "-i", input_path, "-t", str(GIF_MAX_SECONDS), "-vf", f"fps=10,scale={GIF_MAX_WIDTH}:-1:flags=lanczos,palettegen", palette_path ) if not ok: return False, err ok, err = await run_ffmpeg( "-i", input_path, "-i", palette_path, "-t", str(GIF_MAX_SECONDS), "-lavfi", f"fps=10,scale={GIF_MAX_WIDTH}:-1:flags=lanczos[x];[x][1:v]paletteuse", "-loop", "0", output_path ) try: os.remove(palette_path) except Exception: pass return ok, err async def convert_vnote_to_video(input_path: str, output_path: str) -> tuple[bool, str]: return await run_ffmpeg( "-i", input_path, "-c:v", "libvpx", "-c:a", "libopus", "-b:v", "0", "-crf", "10", output_path ) async def convert_vnote_to_voice(input_path: str, output_path: str) -> tuple[bool, str]: return await run_ffmpeg( "-i", input_path, "-vn", "-acodec", "libopus", "-ar", "48000", "-b:a", "64k", output_path ) async def convert_file_to_pdf(input_path: str, out_dir: str) -> tuple[bool, str]: ok = await run_libreoffice(input_path, out_dir) if not ok: return False, "libreoffice_error" base = os.path.splitext(os.path.basename(input_path))[0] pdf = os.path.join(out_dir, base + ".pdf") found = os.path.exists(pdf) return found, "" if found else "pdf_not_found" @router.message(CommandStart()) async def cmd_start(message: Message, state: FSMContext, bot: Bot): await state.clear() user = message.from_user db_usr = await get_or_create_user(BOT_USERNAME, user.id, user.username, user.first_name) is_open = await get_setting(BOT_USERNAME, "is_open", "1") if db_usr.get("is_banned"): await message.answer("⛔ أنت محظور من استخدام هذا البوت.") return if is_open != "1" and not await is_owner_or_admin(user.id): msg = await get_setting(BOT_USERNAME, "closed_message", "🔴 البوت مقفول حالياً.") await message.answer(msg) return if not await handle_subscription_check(bot, user.id, message): return notify = await get_setting(BOT_USERNAME, "notify_new_users", "1") if notify == "1" and db_usr.get("is_new"): name = f"@{user.username}" if user.username else user.first_name await notify_owner( bot, f"👤 مستخدم جديد!\n" f"الاسم: {user.first_name}\nالمعرف: {name}\nالـ ID: {user.id}" ) if await is_owner_or_admin(user.id): panel_text = await _build_dev_panel_text("بوت تحويل الصيغ") if await get_factory_maker_rights(): panel_text += f"\n{MAKER_MSG}" await message.answer(panel_text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_dev_panel_keyboard()) text, w_ents = await build_start_text(user.first_name, user.id) if w_ents: await message.answer(text, entities=w_ents, disable_web_page_preview=True, reply_markup=await build_main_keyboard()) else: await message.answer(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_main_keyboard()) @router.callback_query(F.data == "tc:check_sub") async def cb_check_sub(callback: CallbackQuery, bot: Bot, state: FSMContext): if not await handle_subscription_check(bot, callback.from_user.id, callback): return await state.clear() text, w_ents = await build_start_text(callback.from_user.first_name, callback.from_user.id) if w_ents: await callback.message.edit_text(text, entities=w_ents, disable_web_page_preview=True, reply_markup=await build_main_keyboard()) else: await callback.message.edit_text(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_main_keyboard()) await callback.answer() @router.callback_query(F.data == "tc:back") async def cb_back(callback: CallbackQuery, state: FSMContext): await state.clear() text, w_ents = await build_start_text(callback.from_user.first_name, callback.from_user.id) if w_ents: await callback.message.edit_text(text, entities=w_ents, disable_web_page_preview=True, reply_markup=await build_main_keyboard()) else: await callback.message.edit_text(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_main_keyboard()) await callback.answer() @router.callback_query(F.data.startswith("tc:")) async def cb_conversion_select(callback: CallbackQuery, state: FSMContext, bot: Bot): conv_key = callback.data[3:] if conv_key not in CONVERSIONS: await callback.answer() return if not await handle_subscription_check(bot, callback.from_user.id, callback): return conv = CONVERSIONS[conv_key] await state.update_data(conv_key=conv_key) await state.set_state(UserStates.waiting_file) await callback.message.edit_text( f"» حسناً، لقد اخترت تحويل {conv['from']} إلى {conv['to']}\n" f"» ارسل {conv['from']} لتحويله الآن:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb_cancel() ) await callback.answer() @router.message(UserStates.waiting_file) async def handle_file(message: Message, state: FSMContext, bot: Bot): data = await state.get_data() conv_key = data.get("conv_key") if not conv_key: await state.clear() return conv = CONVERSIONS[conv_key] file_id = None file_name = "input" file_size = 0 if conv["expect"] == "voice" and message.voice: file_id = message.voice.file_id file_size = message.voice.file_size or 0 file_name = "voice.ogg" elif conv["expect"] == "audio" and message.audio: file_id = message.audio.file_id file_size = message.audio.file_size or 0 file_name = message.audio.file_name or "audio.mp3" elif conv["expect"] == "audio" and message.document: fname = (message.document.file_name or "").lower() if fname.endswith((".mp3", ".ogg", ".wav", ".flac", ".m4a", ".aac")): file_id = message.document.file_id file_size = message.document.file_size or 0 file_name = message.document.file_name or "audio.mp3" elif conv["expect"] == "photo" and message.photo: file_id = message.photo[-1].file_id file_size = message.photo[-1].file_size or 0 file_name = "image.jpg" elif conv["expect"] == "photo" and message.document: fname = (message.document.file_name or "").lower() if any(fname.endswith(e) for e in (".jpg", ".jpeg", ".png", ".webp", ".bmp")): file_id = message.document.file_id file_size = message.document.file_size or 0 file_name = message.document.file_name or "image.jpg" elif conv["expect"] == "sticker" and message.sticker: file_id = message.sticker.file_id file_size = message.sticker.file_size or 0 file_name = "sticker.tgs" if message.sticker.is_animated else "sticker.webp" elif conv["expect"] == "video" and message.video: file_id = message.video.file_id file_size = message.video.file_size or 0 file_name = message.video.file_name or "video.mp4" elif conv["expect"] == "video" and message.document: fname = (message.document.file_name or "").lower() if any(fname.endswith(e) for e in (".mp4", ".mkv", ".avi", ".mov", ".webm")): file_id = message.document.file_id file_size = message.document.file_size or 0 file_name = message.document.file_name or "video.mp4" elif conv["expect"] == "video_note" and message.video_note: file_id = message.video_note.file_id file_size = message.video_note.file_size or 0 file_name = "videonote.mp4" elif conv["expect"] == "document" and message.document: file_id = message.document.file_id file_size = message.document.file_size or 0 file_name = message.document.file_name or "document" if not file_id: type_hints = { "voice": "رسالة صوتية 🎤", "audio": "ملف صوتي MP3 🎵", "photo": "صورة 🖼", "sticker": "ملصق 🏷", "video": "فيديو 🎬", "video_note": "فيديو نوت (دائري) 📹", "document": "ملف مستند 📄", } await message.answer( f"⚠️ يرجى إرسال {type_hints.get(conv['expect'], conv['from'])}", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb_cancel() ) return if file_size > MAX_FILE_SIZE: await message.answer( f"⚠️ حجم الملف كبير جداً! يتجاوز الحد الاقصى.\n" f"الحد الأقصى: 20 MB\n" f"حجم ملفك: {file_size / 1024 / 1024:.1f} MB", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb_cancel() ) return await state.clear() processing_msg = await message.answer("⏳ جاري التحويل، يرجى الانتظار...") tok = BOT_TOKEN uname = BOT_USERNAME chat_id = message.chat.id user_id = message.from_user.id proc_id = processing_msg.message_id sticker_obj = message.sticker if message.sticker else None async def _run_work(): b = Bot(token=tok) try: async with temp_dir(user_id) as tmpdir: input_path = os.path.join(tmpdir, file_name) try: tg_file = await b.get_file(file_id) await b.download_file(tg_file.file_path, input_path) except Exception as e: try: await b.edit_message_text(f"❌ فشل تحميل الملف: {e}", chat_id=chat_id, message_id=proc_id, reply_markup=kb_back()) except Exception: pass return try: await _process_conversion(chat_id, proc_id, b, uname, conv_key, input_path, tmpdir, sticker_obj) except Exception as e: logger.error(f"خطأ في التحويل {conv_key}: {e}") try: await b.edit_message_text( "❌ حدث خطأ أثناء التحويل. يرجى المحاولة مرة أخرى.", chat_id=chat_id, message_id=proc_id, reply_markup=kb_back() ) except Exception: pass finally: await b.session.close() asyncio.create_task(_run_work()) async def _process_conversion( chat_id: int, proc_id: int, b: Bot, uname: str, conv_key: str, input_path: str, tmpdir: str, sticker_obj=None ): async def fail(reason: str): await b.edit_message_text(f"❌ {reason}", chat_id=chat_id, message_id=proc_id, reply_markup=kb_back()) async def _delete_proc(): try: await b.delete_message(chat_id, proc_id) except Exception: pass async def _back(): await b.send_message(chat_id, "» اضغط زر الرجوع للعودة الى القائمة:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=kb_back()) if conv_key == "voice_to_mp3": out = os.path.join(tmpdir, "output.mp3") ok, err = await convert_voice_to_mp3(input_path, out) if not ok: await fail("فشل التحويل. تأكد أن الملف رسالة صوتية صحيحة.") return await _delete_proc() await b.send_audio( chat_id, FSInputFile(out, filename="SourceKdoo.mp3"), caption="» تم تحويل رسالة صوتية إلى MP3 🎙️.", parse_mode="HTML", ) await _back() elif conv_key == "mp3_to_voice": out = os.path.join(tmpdir, "output.ogg") ok, err = await convert_mp3_to_voice(input_path, out) if not ok: await fail("فشل التحويل. تأكد أن الملف ملف صوتي صحيح.") return await _delete_proc() await b.send_voice( chat_id, FSInputFile(out, filename="SourceKodo.ogg"), caption="» تم تحويل ملف MP3 إلى بصمة صوتية 🎙.", parse_mode="HTML") await _back() elif conv_key == "image_to_sticker": out_png = os.path.join(tmpdir, "sticker.png") out_webp = os.path.join(tmpdir, "sticker.webp") ok, err = await convert_image_to_sticker(input_path, out_png, out_webp) if not ok: await fail(f"فشل تحويل الصورة: {err}") return await _delete_proc() await b.send_document( chat_id, FSInputFile(out_webp, filename="sticker.webp"), caption="" ) await b.send_document( chat_id, FSInputFile(out_png, filename="sticker_512.png"), caption="» تم تحويل صورة إلى ملصق مباشر 🌄.", parse_mode="HTML") await _back() elif conv_key == "sticker_to_image": out = os.path.join(tmpdir, "sticker_out.png") animated = sticker_obj and sticker_obj.is_animated ok, err = await convert_sticker_to_image(input_path, out, is_animated=bool(animated)) if not ok: if err == "rlottie_python_missing": await fail( "الملصقات المتحركة تحتاج مكتبة rlottie_python.\n" "يرجى تثبيتها:\npip install rlottie-python" ) else: await fail(f"فشل تحويل الملصق: {err}") return await _delete_proc() await b.send_document( chat_id, FSInputFile(out, filename="sticker.png"), caption="» تم تحويل الملصق إلى صورة PNG 🖼️.", parse_mode="HTML") await _back() elif conv_key == "video_to_audio": out = os.path.join(tmpdir, "output.mp3") ok, err = await convert_video_to_audio(input_path, out) if not ok: await fail("فشل استخراج الصوت من الفيديو.") return size = os.path.getsize(out) if size > 50 * 1024 * 1024: await fail("الملف الناتج كبير جداً (أكبر من 50 MB).") return await _delete_proc() await b.send_audio( chat_id, FSInputFile(out, filename="audio.mp3"), caption="» تم استخراج الصوت من الفيديو بنجاح 🔊.", parse_mode="HTML") await _back() elif conv_key == "video_to_gif": out = os.path.join(tmpdir, "output.gif") ok, err = await convert_video_to_gif(input_path, out) if not ok: await fail("فشل تحويل الفيديو إلى GIF.") return size = os.path.getsize(out) if size > 50 * 1024 * 1024: await fail( f"ملف GIF الناتج كبير جداً ({size / 1024 / 1024:.1f} MB).\n" "جرّب فيديو أقصر أو بدقة أقل." ) return await _delete_proc() await b.send_animation( chat_id, FSInputFile(out, filename="animation.gif"), caption=f"» تم تحويل الفيديو الى ملصق متحرك 🎥.", parse_mode="HTML") await _back() elif conv_key == "vnote_to_video": out = os.path.join(tmpdir, "output.mp4") ok, err = await convert_vnote_to_video(input_path, out) if not ok: ok, err = await run_ffmpeg("-i", input_path, "-c", "copy", out) if not ok: await fail("فشل تحويل فيديو النوت.") return await _delete_proc() await b.send_video( chat_id, FSInputFile(out, filename="video.mp4"), caption="» تم تحويل فيديو نوت الى فيديو عادي 🎬.", parse_mode="HTML") await _back() elif conv_key == "vnote_to_voice": out = os.path.join(tmpdir, "output.ogg") ok, err = await convert_vnote_to_voice(input_path, out) if not ok: await fail("فشل استخراج الصوت من فيديو النوت.") return await _delete_proc() await b.send_voice( chat_id, FSInputFile(out, filename="voice.ogg"), caption="» تم استخراج الصوت من فيديو نوت بنجاح 🎙️.", parse_mode="HTML") await _back() elif conv_key == "file_to_pdf": ok, err = await convert_file_to_pdf(input_path, tmpdir) if not ok: if err == "libreoffice_error": await fail("فشل التحويل. تأكد أن الملف من نوع مدعوم (Word, Excel, PowerPoint, إلخ).") else: await fail("لم يتم إنشاء ملف PDF. قد يكون الملف تالفاً.") return base = os.path.splitext(os.path.basename(input_path))[0] pdf_out = os.path.join(tmpdir, base + ".pdf") size = os.path.getsize(pdf_out) if size > 50 * 1024 * 1024: await fail("ملف PDF الناتج كبير جداً (أكبر من 50 MB).") return await _delete_proc() await b.send_document( chat_id, FSInputFile(pdf_out, filename=base + ".pdf"), caption="» تم تحويل الملف المرفق الى ملف PDF 📄.", parse_mode="HTML") await _back() async def build_dev_panel_keyboard() -> InlineKeyboardMarkup: s = await get_all_settings(BOT_USERNAME) is_open = "🟢 مفتوح" if s.get("is_open", "1") == "1" else "🔴 مقفول" n_blk = "✅" if s.get("notify_blocked", "1") == "1" else "✖️" n_new = "✅" if s.get("notify_new_users", "1") == "1" else "✖️" n_fund = "✅" if s.get("fund_notify", "1") == "1" else "✖️" return InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"حالة البوت {is_open}", callback_data="dev:toggle_open", style="success")], [ InlineKeyboardButton(text=f"اشعار الحظر {n_blk}", callback_data="dev:toggle_notify_blk", style="success"), InlineKeyboardButton(text=f"اشعار الدخول {n_new}", callback_data="dev:toggle_notify_new", style="success"), ], [InlineKeyboardButton(text=f"اشعار الممولة {n_fund}", callback_data="dev:toggle_notify_fund", style="success")], [ InlineKeyboardButton(text="قسم الاحصائيات", callback_data="dev:stats", style="primary"), InlineKeyboardButton(text="قسم الاذاعة", callback_data="dev:broadcast", style="primary"), ], [ InlineKeyboardButton(text="قسم الادمنية", callback_data="dev:admins", style="primary"), InlineKeyboardButton(text="الاشتراك الاجباري", callback_data="dev:mand", style="primary"), ], [InlineKeyboardButton(text="القنوات الممولة", callback_data="dev:funded", style="danger")], [InlineKeyboardButton(text="تعديل رسالة الترحيب", callback_data="dev: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 == "dev:panel") async def cb_dev_panel(callback: CallbackQuery, state: FSMContext): await state.clear() if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await callback.message.edit_text( "» هذه هي لوحة التحكم وادارة بوت تحويل الصيغ الخاص بك.\n\n» لوحة تحكم متكاملة تتيح لك الإشراف على صلاحيات البوت، متابعة الإحصائيات وإدارة الاشتراكات وكل شيء من مكان واحد.\n\n» قم بأختيار الاجراء المطلوب من القائمة ادناه:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=await build_dev_panel_keyboard() ) await callback.answer() @dev_router.callback_query(F.data == "dev:toggle_open") async def cb_toggle_open(callback: CallbackQuery): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return current = await get_setting(BOT_USERNAME, "is_open", "1") await set_setting(BOT_USERNAME, "is_open", "0" if current == "1" else "1") await callback.answer("✅ تم التغيير.", show_alert=True) await callback.message.edit_reply_markup(reply_markup=await build_dev_panel_keyboard()) @dev_router.callback_query(F.data.in_({ "dev:toggle_notify_blk", "dev:toggle_notify_new", "dev:toggle_notify_fund" })) async def cb_toggle_notify(callback: CallbackQuery): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return key_map = { "dev:toggle_notify_blk": "notify_blocked", "dev:toggle_notify_new": "notify_new_users", "dev:toggle_notify_fund": "fund_notify", } key = key_map[callback.data] current = await get_setting(BOT_USERNAME, key, "1") await set_setting(BOT_USERNAME, key, "0" if current == "1" else "1") await callback.answer("✅ تم التغيير.", show_alert=True) await callback.message.edit_reply_markup(reply_markup=await build_dev_panel_keyboard()) @dev_router.callback_query(F.data == "dev:stats") async def cb_stats(callback: CallbackQuery): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return stats = await get_statistics(BOT_USERNAME) await callback.message.edit_text( f"📊 إحصائيات البوت\n\n" f"👥 إجمالي المستخدمين: {stats['total']}\n" f"✅ النشطون: {stats['active']}\n" f"🚫 المحظورون: {stats['banned']}\n" f"🔕 حظروا البوت: {stats['blocked']}", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")] ]) ) await callback.answer() @dev_router.callback_query(F.data == "dev:welcome") async def cb_welcome(callback: CallbackQuery, state: FSMContext): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await state.set_state(DevStates.waiting_welcome) await callback.message.edit_text( "» قسم تعديل رسالة الترحيب:\n\n» استخدم {first_name} لعرض اسم المستخدم في رسالة الترحيب.\n» ارسل رسالة الترحيب الجديدة او ارسل كلمة 'حذف' لاعادة الرسالة الترحيبية حيثما كانت:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_welcome) async def process_welcome(message: Message, state: FSMContext): if not await is_owner_or_admin(message.from_user.id): return text = message.text.strip() if message.text else "" if text.lower() in ("حذف", "delete", "reset"): await set_setting(BOT_USERNAME, "welcome_message", "") await set_setting(BOT_USERNAME, "welcome_message_entities", "") else: await set_setting(BOT_USERNAME, "welcome_message", text) await set_setting(BOT_USERNAME, "welcome_message_entities", _entities_to_json(message.entities) or "") await state.clear() await message.answer( "✅ تم تحديث رسالة الترحيب.", reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="لوحة التحكم", callback_data="dev:panel", style="primary")] ]) ) @dev_router.callback_query(F.data == "dev:broadcast") async def cb_broadcast_menu(callback: CallbackQuery, state: FSMContext): await state.clear() if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await callback.message.edit_text( "» قسم الاذاعة بالبوت:.\n» في هذا القسم تستطيع ان تقوم بعمل اذاعة من بوتك الى المستخدمين.\n\n» حدد نوع الاذاعة سواء رسالة موجهة او نصية:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="اذاعة نصية", callback_data="bcast:text", style="success"), InlineKeyboardButton(text="اذاعة موجهة", callback_data="bcast:fwd", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data == "bcast:text") async def cb_bcast_text(callback: CallbackQuery, state: FSMContext): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await state.set_state(DevStates.waiting_broadcast) await callback.message.edit_text( "» ارسل رسالة الاذاعة المراد اذاعتها لجميع المستخدمين الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dev:broadcast", style="primary")] ]) ) await callback.answer() @dev_router.callback_query(F.data == "bcast:fwd") async def cb_bcast_fwd(callback: CallbackQuery, state: FSMContext): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return 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="dev: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, u in enumerate(users): try: if forward: await b.forward_message(u["id"], chat_id, message.message_id) elif message.text: await b.send_message(u["id"], message.text, parse_mode="HTML", disable_web_page_preview=True) elif message.photo: await b.send_photo(u["id"], message.photo[-1].file_id, caption=message.caption) elif message.video: await b.send_video(u["id"], message.video.file_id, caption=message.caption) elif message.document: await b.send_document(u["id"], message.document.file_id, caption=message.caption) elif message.voice: await b.send_voice(u["id"], message.voice.file_id) elif message.audio: await b.send_audio(u["id"], message.audio.file_id, caption=message.caption) elif message.sticker: await b.send_sticker(u["id"], message.sticker.file_id) else: await b.send_message(u["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) async def process_broadcast(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 == "dev:admins") async def cb_admins(callback: CallbackQuery, state: FSMContext): await state.clear() if callback.from_user.id != OWNER_ID: await callback.answer("⛔ للمالك فقط!", show_alert=True); return admins = await get_admins(BOT_USERNAME) text = "» عزيزي المطور هذا هو قسم التحكم بالادمنية.\n» يمكن اضافة وحذف الادمنية بالبوت من الازرار:\n\n" for a in admins: name = f"@{a['username']}" if a.get("username") else a.get("first_name", str(a["id"])) text += f"• {name} — {a['id']}\n" if not admins: text += "لا يوجد أدمنية حالياً." await callback.message.edit_text( text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="اضافة ادمن", callback_data="dev:add_admin", style="success"), InlineKeyboardButton(text="حذف ادمن", callback_data="dev:remove_admin", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data == "dev:add_admin") async def cb_add_admin(callback: CallbackQuery, state: FSMContext): if callback.from_user.id != OWNER_ID: await callback.answer("⛔ للمالك فقط!", show_alert=True); return 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="dev:admins", style="primary")] ]) ) await callback.answer() @dev_router.message(DevStates.waiting_add_admin) async def process_add_admin(message: Message, state: FSMContext): if message.from_user.id != OWNER_ID: return try: target_id = int(message.text.strip()) except (ValueError, AttributeError): await message.answer("⚠️ أرسل ID رقمي صحيح!"); return user = await get_user(BOT_USERNAME, target_id) if not user: await message.answer("⚠️ المستخدم غير موجود في قاعدة البيانات.") await state.clear(); return await add_admin(BOT_USERNAME, target_id, user.get("username"), user.get("first_name", "")) await state.clear() await message.answer( f"✅ تم تعيين {target_id} أدمناً في البوت.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="الادمنية", callback_data="dev:admins", style="primary")] ]) ) @dev_router.callback_query(F.data == "dev:remove_admin") async def cb_remove_admin(callback: CallbackQuery): if callback.from_user.id != OWNER_ID: await callback.answer("⛔ للمالك فقط!", show_alert=True); return admins = await get_admins(BOT_USERNAME) if not admins: await callback.answer("لا يوجد أدمنية.", show_alert=True); return rows = [[InlineKeyboardButton( text=f"@{a['username']}" if a.get("username") else a.get("first_name", str(a["id"])), callback_data=f"dev:del_admin:{a['id']}", style="danger" )] for a in admins] rows.append([InlineKeyboardButton(text="رجوع", callback_data="dev:admins", style="primary")]) await callback.message.edit_text( "» اضغط على زر أي أدمن موجود لحذفه من قائمة الادمنية:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) await callback.answer() @dev_router.callback_query(F.data.startswith("dev:del_admin:")) async def cb_del_admin(callback: CallbackQuery): if callback.from_user.id != OWNER_ID: await callback.answer("⛔ للمالك فقط!", show_alert=True); return await remove_admin(BOT_USERNAME, int(callback.data.split(":")[2])) await callback.answer("✅ تم إزالة الأدمن.", show_alert=True) await cb_admins(callback) async def _show_mand_panel(callback: CallbackQuery): channels = await get_mandatory_channels(BOT_USERNAME) text = "» عزيزي المطور هذا هو قسم الاشتراك الاجباري.\n» يمكن اضافة وحذف القنوات من الازرار:\n\n" rows = [] for ch in channels: name = f"@{ch['channel_username']}" if ch.get("channel_username") else ch["channel_title"] text += f"• {name}\n" rows.append([InlineKeyboardButton( text=f"🗑 {name}", callback_data=f"dev:del_mand:{ch['id']}", style="danger" )]) if not channels: text += "لا توجد قنوات." rows.append([InlineKeyboardButton(text="اضافة قناة", callback_data="dev:add_mand", style="success")]) rows.append([InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")]) await callback.message.edit_text( text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) @dev_router.callback_query(F.data == "dev:mand") async def cb_mand(callback: CallbackQuery, state: FSMContext): await state.clear() if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await _show_mand_panel(callback); await callback.answer() @dev_router.callback_query(F.data == "dev:add_mand") async def cb_add_mand(callback: CallbackQuery, state: FSMContext): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await state.update_data(mand_display_type="buttons") await callback.message.edit_text( "» اختر الطريقة التي تريد عرض طلب الاشتراك فيها من المستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="ازرار", callback_data="mand_dtype:buttons", style="success"), InlineKeyboardButton(text="رسالة نصية", callback_data="mand_dtype:text", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dev:mand", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("mand_dtype:")) async def cb_mand_dtype(callback: CallbackQuery, state: FSMContext): await state.update_data(mand_display_type=callback.data.split(":")[1]) 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="dev:mand", 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 await is_owner_or_admin(message.from_user.id): return if not message.forward_from_chat: await message.answer("⚠️ يرجى توجيه رسالة من القناة فقط!"); return chat = message.forward_from_chat chat_id = chat.id try: member = await bot.get_chat_member(chat_id, (await bot.get_me()).id) if member.status not in ("administrator", "creator"): await message.answer("⚠️ يجب إضافة البوت كمشرف في القناة أولاً!") await state.clear(); return except Exception: await message.answer("⚠️ لا يمكن الوصول للقناة. تأكد أن البوت مشرف فيها!") await state.clear(); return invite_link = f"https://t.me/{chat.username}" if chat.username else None if not invite_link: try: link_obj = await bot.create_chat_invite_link(chat_id) invite_link = link_obj.invite_link except Exception: pass data = await state.get_data() await add_mandatory_channel( BOT_USERNAME, str(chat_id), chat.title or str(chat_id), chat.username, invite_link, data.get("mand_display_type", "buttons") ) await state.clear() await message.answer( f"✅ تمت إضافة القناة {chat.title}.", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="الاشتراك الإجباري", callback_data="dev:mand", style="primary")] ]) ) @dev_router.callback_query(F.data.startswith("dev:del_mand:")) async def cb_del_mand(callback: CallbackQuery): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await remove_mandatory_channel(BOT_USERNAME, int(callback.data.split(":")[2])) await callback.answer("✅ تم حذف القناة.", show_alert=True) await _show_mand_panel(callback) async def _show_funded_panel(callback: CallbackQuery): channels = await get_funded_channels(BOT_USERNAME) text = "» عزيزي المطور هذا هو قسم تمويل القنوات.\n\n» هذا القسم مختص في اجراء تمويل لقناة انت من يحددها حيث سيطلب البوت من المستخدمين الاشتراك في هذه القناة، عند اكتمال العدد المطلوب الذي تحدده انت سيتم ازالة القناة من القنوات الممولة.\n\n" rows = [] for ch in channels: name = f"@{ch['channel_username']}" if ch.get("channel_username") else ch["channel_title"] text += f"• {name} ({ch['current_count']}/{ch['target_count']})\n" rows.append([InlineKeyboardButton( text=f"🗑 {name}", callback_data=f"dev:del_funded:{ch['id']}", style="danger" )]) if not channels: text += "لا توجد قنوات." rows.append([InlineKeyboardButton(text="اضافة قناة تمويل", callback_data="dev:add_funded", style="success")]) rows.append([InlineKeyboardButton(text="رجوع", callback_data="dev:panel", style="primary")]) await callback.message.edit_text( text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=rows) ) @dev_router.callback_query(F.data == "dev:funded") async def cb_funded(callback: CallbackQuery, state: FSMContext): await state.clear() if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await _show_funded_panel(callback); await callback.answer() @dev_router.callback_query(F.data == "dev:add_funded") async def cb_add_funded(callback: CallbackQuery, state: FSMContext): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await state.update_data(funded_display_type="buttons") await callback.message.edit_text( "» اختر الطريقة التي تريد عرض طلب الاشتراك فيها من المستخدم:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="ازرار", callback_data="funded_dtype:buttons", style="success"), InlineKeyboardButton(text="رسالة نصية", callback_data="funded_dtype:text", style="success"), ], [InlineKeyboardButton(text="رجوع", callback_data="dev:funded", style="primary")], ]) ) await callback.answer() @dev_router.callback_query(F.data.startswith("funded_dtype:")) async def cb_funded_dtype(callback: CallbackQuery, state: FSMContext): await state.update_data(funded_display_type=callback.data.split(":")[1]) 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="dev: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 await is_owner_or_admin(message.from_user.id): return if not message.forward_from_chat: await message.answer("⚠️ يرجى توجيه رسالة من القناة فقط!"); return chat = message.forward_from_chat chat_id = chat.id try: member = await bot.get_chat_member(chat_id, (await bot.get_me()).id) if member.status not in ("administrator", "creator"): await message.answer("⚠️ يجب إضافة البوت كمشرف في القناة أولاً!") await state.clear(); return except Exception: await message.answer("⚠️ لا يمكن الوصول للقناة. تأكد أن البوت مشرف فيها!") await state.clear(); return invite_link = f"https://t.me/{chat.username}" if chat.username else None if not invite_link: try: link_obj = await bot.create_chat_invite_link(chat_id) invite_link = link_obj.invite_link except Exception: pass await state.update_data( funded_chat_id=str(chat_id), funded_title=chat.title or str(chat_id), funded_username=chat.username, funded_invite=invite_link, ) await state.set_state(DevStates.waiting_funded_count) await message.answer( f"» القناة: {chat.title}\n» قم بإرسال عدد المشتركين المستهدف الان:", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="رجوع", callback_data="dev:funded", style="primary")] ]) ) @dev_router.message(DevStates.waiting_funded_count) async def process_funded_count(message: Message, state: FSMContext): if not await is_owner_or_admin(message.from_user.id): return try: target = int(message.text.strip()) if target <= 0: raise ValueError except (ValueError, AttributeError): await message.answer("⚠️ أرسل رقماً صحيحاً أكبر من صفر!"); return data = await state.get_data() await add_funded_channel( BOT_USERNAME, data["funded_chat_id"], data["funded_title"], data.get("funded_username"), data.get("funded_invite"), data.get("funded_display_type", "buttons"), target, ) await state.clear() await message.answer( f"» تم اضافة القناة: {data['funded_title']} بنجاح.\n» الهدف: {target} عضو", parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="القنوات الممولة", callback_data="dev:funded", style="primary")] ]) ) @dev_router.callback_query(F.data.startswith("dev:del_funded:")) async def cb_del_funded(callback: CallbackQuery): if not await is_owner_or_admin(callback.from_user.id): await callback.answer("⛔ غير مصرح!", show_alert=True); return await remove_funded_channel(BOT_USERNAME, int(callback.data.split(":")[2])) await callback.answer("✅ تم حذف القناة.", show_alert=True) await _show_funded_panel(callback) 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_owner_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="dev: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_owner_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_owner_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_owner_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="dev:panel", style="primary")] ]) ) @dev_router.callback_query(F.data == "cbtn:manage") async def cbtn_manage(callback: CallbackQuery, state: FSMContext): if not await is_owner_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="dev: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_owner_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_owner_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_owner_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_owner_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_owner_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_owner_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_owner_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_owner_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_owner_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_owner_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_owner_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="tc:back", style="primary")] ]) ) await callback.answer() _transcoding_dispatcher: Dispatcher | None = None def get_transcoding_dispatcher() -> Dispatcher: global _transcoding_dispatcher if _transcoding_dispatcher is None: storage = MemoryStorage() _transcoding_dispatcher = Dispatcher(storage=storage) _transcoding_dispatcher.include_router(dev_router) _transcoding_dispatcher.include_router(router) return _transcoding_dispatcher