import telebot from telebot import types import subprocess import os import signal import logging import json import time import re import sys import threading # ================================================================= # === اعدادات التسجيل والملفات # ================================================================= logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("bot_debug.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) STATE_FILE = "bot_state.json" control_data = {} BOT_TOKEN = "8547371020:AAHv7Sp1lJx4K-McNYcMY_MeflAeWSGq1NA" bot = telebot.TeleBot(BOT_TOKEN) # ================================================================= # === دوال حالة البوت والحفظ والتحميل # ================================================================= def save_state(): try: state_to_save = {} for msg_id, data in control_data.items(): state_to_save[msg_id] = { "chat_id": data["chat_id"], "file_path": data["file_path"], "state": data["state"], "proc_pid": data["proc"].pid if data.get("proc") else None, "libs": data.get("libs", []) } with open(STATE_FILE, "w") as f: json.dump(state_to_save, f) except Exception as e: logger.error(f"Error saving state: {e}") def load_state(): try: if os.path.exists(STATE_FILE): with open(STATE_FILE, "r") as f: saved_state = json.load(f) for msg_id, data in saved_state.items(): # ✅ تجاهل أي ملف مش موجود if not os.path.exists(data["file_path"]): continue control_data[int(msg_id)] = { "chat_id": data["chat_id"], "proc": None, "file_path": data["file_path"], "state": data["state"], "libs": data.get("libs", []) } except Exception as e: logger.error(f"Error loading state: {e}") # ================================================================= # === دوال مساعدة # ================================================================= def escape_markdown(text): """تهرب المحارف الخاصة في سلاسل MarkdownV2 لمنع خطأ 'can't parse entities'.""" # المحارف التي يجب الهروب منها في MarkdownV2 escape_chars = r'_*[]()~`>#+-=|{}.!' return re.sub(f'([{re.escape(escape_chars)}])', r'\\\1', text) def extract_libraries(file_path): libs = set() try: with open(file_path, "r", encoding="utf-8") as f: content = f.read() patterns = [ r'^\s*import\s+([a-zA-Z0-9_]+)(?:\s*,\s*([a-zA-Z0-9_]+))*', r'^\s*from\s+([a-zA-Z0-9_]+)\s+import' ] for pattern in patterns: matches = re.finditer(pattern, content, re.MULTILINE) for match in matches: groups = match.groups() for lib in groups: if lib and lib not in ("", " "): libs.add(lib.split('.')[0]) # Exclude common standard Python libraries std_libs = sys.stdlib_module_names libs = {lib for lib in libs if lib not in std_libs and lib not in ('telebot', 'subprocess', 'os', 'signal', 'logging', 'json', 'time', 're', 'sys', 'threading')} except Exception as e: logger.error(f"Error extracting libraries: {e}") return list(libs) def build_keyboard(): keyboard = types.InlineKeyboardMarkup() btn_toggle = types.InlineKeyboardButton("إيقاف/تشغيل", callback_data="toggle") btn_install = types.InlineKeyboardButton("تنصيب المكتبات", callback_data="install") btn_delete = types.InlineKeyboardButton("حذف", callback_data="delete") btn_speed = types.InlineKeyboardButton("سرعة البوت", callback_data="speed") btn_restart = types.InlineKeyboardButton("إعادة تشغيل", callback_data="restart") keyboard.row(btn_toggle, btn_restart) keyboard.row(btn_install, btn_speed) keyboard.row(btn_delete) return keyboard def check_running_files(): """تتأكد من أن الملفات التي حالتها 'running' تعمل، وتعيد تشغيلها إذا توقفت.""" for msg_id, data in list(control_data.items()): # ✅ تجاهل الملفات المحذوفة if not os.path.exists(data["file_path"]): logger.warning(f"الملف غير موجود: {data['file_path']}") control_data.pop(msg_id, None) save_state() continue proc = data.get("proc") # إذا كانت الحالة 'running' والعملية (proc) غير موجودة أو توقفت if data["state"] == "running" and (proc is None or proc.poll() is not None): try: # تشغيل الملف من جديد new_proc = subprocess.Popen([ "/home/batvoidc/py311/bin/python", data["file_path"] ]) control_data[msg_id]["proc"] = new_proc logger.info(f"تم إعادة تشغيل الملف: {data['file_path']} تلقائياً.") raw_file_name = os.path.basename(data["file_path"]) escaped_file_name = escape_markdown(raw_file_name) bot.send_message( data["chat_id"], f"♻️ تم إعادة تشغيل الملف تلقائياً:\n*{escaped_file_name}*", parse_mode="markdown" ) except Exception as e: logger.error(f"فشل إعادة تشغيل الملف: {e}") raw_file_name = os.path.basename(data["file_path"]) escaped_file_name = escape_markdown(raw_file_name) bot.send_message( data["chat_id"], f"❌ فشل إعادة تشغيل الملف:\n*{escaped_file_name}*\nالخطأ: {escape_markdown(str(e))}", parse_mode="markdown" ) # ================================================================= # === معالجات الأوامر # ================================================================= @bot.message_handler(commands=["start"]) def start_handler(message): bot.send_message( message.chat.id, "مرحباً! 👋\n\n" "أرسل لي ملف بايثون *(.py)* وسأقوم بتشغيله وإدارته لك.\n\n" "*الأوامر المتاحة:*\n" "*/start* - عرض هذه الرسالة\n" "*/list* - عرض الملفات الجارية\n" "*/restart* - إعادة تشغيل ملف معين\n" "*/deletefile* - حذف ملف غير نشط أو نشط", parse_mode="markdown" ) @bot.message_handler(commands=["list"]) def list_handler(message): text = "📂 الملفات الجارية:\n\n" for msg_id, data in control_data.items(): raw_file = os.path.basename(data["file_path"]) escaped_file = escape_markdown(raw_file) state = "🟢 قيد التشغيل" if data["state"] == "running" else "⏸ متوقف" text += f"- *{escaped_file}* ({state})\n" if not control_data: text = "لا توجد ملفات نشطة حالياً." bot.send_message(message.chat.id, text, parse_mode="markdown") @bot.message_handler(commands=["restart"]) def restart_handler(message): if not control_data: bot.send_message(message.chat.id, "لا توجد ملفات نشطة لإعادة التشغيل.") return keyboard = types.InlineKeyboardMarkup() for msg_id, data in control_data.items(): raw_file_name = os.path.basename(data["file_path"]) btn = types.InlineKeyboardButton( raw_file_name, callback_data=f"restart_{msg_id}" ) keyboard.add(btn) bot.send_message( message.chat.id, "اختر الملف لإعادة تشغيله:", reply_markup=keyboard ) @bot.message_handler(commands=["deletefile"]) # الأمر الجديد def deletefile_handler(message): if not control_data: bot.send_message(message.chat.id, "لا توجد ملفات مُدارة لحذفها.") return keyboard = types.InlineKeyboardMarkup() for msg_id, data in control_data.items(): raw_file_name = os.path.basename(data["file_path"]) btn = types.InlineKeyboardButton( f"❌ حذف: {raw_file_name}", callback_data=f"confirm_delete_global_{msg_id}" # استخدام callback_data خاص ) keyboard.add(btn) bot.send_message( message.chat.id, "⚠️ *اختر الملف الذي تريد حذفه بالكامل (ملف التشغيل ورسالة التحكم):*", reply_markup=keyboard, parse_mode="markdown" ) @bot.message_handler(content_types=["document"]) def document_handler(message): try: doc = message.document if not doc.file_name.endswith(".py"): bot.send_message(message.chat.id, "⚠️ يجب أن يكون الملف بصيغة *.py*!") return file_info = bot.get_file(doc.file_id) downloaded_file = bot.download_file(file_info.file_path) os.makedirs("./downloads", exist_ok=True) local_file_path = os.path.join("./downloads", doc.file_name) with open(local_file_path, "wb") as file: file.write(downloaded_file) libs = extract_libraries(local_file_path) libs_text = "📚 *المكتبات المستخدمة*:\n" + "\n".join(f"- {escape_markdown(lib)}" for lib in libs) if libs else "لم يتم اكتشاف مكتبات خارجية." # تهريب اسم الملف قبل عرضه escaped_file_name = escape_markdown(doc.file_name) bot.send_message( message.chat.id, f"✅ تم رفع الملف بنجاح:\n*{escaped_file_name}*\n\n{libs_text}\n\nجاري التشغيل...", parse_mode="markdown" ) try: # التشغيل الأولي للملف proc = subprocess.Popen( ["/home/batvoidc/py311/bin/python", local_file_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) except Exception as proc_err: bot.send_message(message.chat.id, f"❌ فشل تشغيل الملف:\n{escape_markdown(str(proc_err))}", parse_mode="markdown") logger.error(proc_err) return keyboard = build_keyboard() control_msg = bot.send_message( message.chat.id, f"🛠 *تحكم في الملف*:\n*{escaped_file_name}*", reply_markup=keyboard, parse_mode="markdown" ) control_data[control_msg.message_id] = { "chat_id": message.chat.id, "proc": proc, "file_path": local_file_path, "state": "running", "libs": libs } save_state() except Exception as e: bot.send_message(message.chat.id, f"❌ حدث خطأ غير متوقع:\n{escape_markdown(str(e))}", parse_mode="markdown") logger.error(e) # ================================================================= # === معالجات أزرار التحكم (Callbacks) # ================================================================= @bot.callback_query_handler(func=lambda call: True) def callback_query_handler(call): try: # إذا كانت data تبدأ بـ confirm_delete_global_X فهذا قادم من أمر /deletefile if call.data.startswith("confirm_delete_global_"): handle_confirm_delete_global(call) return # إذا كانت data تبدأ بـ confirm_delete_X فهذا قادم من زر "تأكيد الحذف" في لوحة التحكم elif call.data.startswith("confirm_delete_"): handle_confirm_delete(call) return elif call.data == "cancel_delete": handle_cancel_delete(call) return elif call.data.startswith("restart_"): handle_restart(call) return elif call.data.startswith("cancel_delete_global"): bot.delete_message(call.message.chat.id, call.message.message_id) bot.answer_callback_query(call.id, "تم إلغاء عملية الحذف.") return # الآن نعالج الأزرار العادية (toggle, install, speed, delete, restart) msg_id = call.message.message_id data = control_data.get(msg_id) # إذا لم تكن البيانات متوفرة هنا، فهذا يعني أننا ضغطنا على زر عادي (وليس زر تأكيد حذف) لرسالة قديمة if not data: bot.answer_callback_query(call.id, "⚠️ البيانات غير متوفرة") return proc = data.get("proc") chat_id = data.get("chat_id") current_state = data.get("state") libs = data.get("libs", []) raw_file_name = os.path.basename(data["file_path"]) file_name = escape_markdown(raw_file_name) if call.data == "toggle": if proc and proc.poll() is None: if current_state == "running": try: proc.send_signal(signal.SIGSTOP) data["state"] = "paused" new_text = f"⏸ تم إيقاف الملف مؤقتاً:\n*{file_name}*" except Exception as e: new_text = f"❌ فشل الإيقاف\. قد تكون الاستضافة لا تدعم الإيقاف المؤقت:\n*{file_name}*\nالخطأ: {escape_markdown(str(e))}" logger.error(e) elif current_state == "paused": try: proc.send_signal(signal.SIGCONT) data["state"] = "running" new_text = f"▶️ تم استئناف التشغيل:\n*{file_name}*" except Exception as e: new_text = f"❌ فشل الاستئناف:\n*{file_name}*\nالخطأ: {escape_markdown(str(e))}" logger.error(e) else: new_text = f"⚠️ لا يوجد عملية نشطة للملف:\n*{file_name}*" bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=new_text, reply_markup=build_keyboard(), parse_mode="markdown" ) save_state() elif call.data == "install": if libs: total = len(libs) install_results = [] bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=f"⏳ *جاري تثبيت/تحديث {total} مكتبة...*\nالتقدم: 0%", reply_markup=build_keyboard(), parse_mode="markdown" ) for index, lib in enumerate(libs, start=1): # استخدام --upgrade لضمان تحديث المكتبة command = ["/home/batvoidc/py311/bin/python", "-m", "pip", "install", "--upgrade", lib] try: result = subprocess.run( command, capture_output=True, text=True ) escaped_lib = escape_markdown(lib) if result.returncode == 0: install_results.append(f"✔️ *{escaped_lib}* (تم التحديث/التثبيت)") else: # تفاصيل الخطأ error_msg = result.stderr.strip() or result.stdout.strip() install_results.append(f"❌ *{escaped_lib}* (فشل: {escape_markdown(error_msg[:100])}\.\.\.)") logger.error(f"Failed to install/upgrade {lib}: {error_msg}") except Exception as e: install_results.append(f"❌ *{escaped_lib}* (خطأ: {escape_markdown(str(e))})") logger.error(e) progress = int((index / total) * 100) bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=f"⏳ *جاري تثبيت/تحديث {total} مكتبة...*\nالتقدم: {progress}%", reply_markup=build_keyboard(), parse_mode="markdown" ) time.sleep(0.5) final_text = f"✅ *تم تثبيت/تحديث المكتبات*:\n\n" + "\n".join(install_results) bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=final_text, reply_markup=build_keyboard(), parse_mode="markdown" ) else: bot.answer_callback_query(call.id, "⚠️ لا توجد مكتبات للتثبيت") elif call.data == "speed": start_time = time.time() bot.answer_callback_query(call.id, "⏱ قياس السرعة...") # إعادة تحرير الرسالة الأصلية قبل قياس السرعة original_text = f"🛠 *تحكم في الملف*:\n*{file_name}*" try: bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=original_text, reply_markup=build_keyboard(), parse_mode="markdown" ) except Exception as e: logger.warning(f"Failed to edit message before speed test: {e}") # حساب وعرض السرعة elapsed = int((time.time() - start_time) * 1000) speed_text = f"⚡ *سرعة استجابة البوت*: {elapsed} مللي ثانية" bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=speed_text, reply_markup=build_keyboard(), parse_mode="markdown" ) elif call.data == "delete": keyboard = types.InlineKeyboardMarkup() # التعديل هنا: نمرر الـ msg_id الأصلي (الذي هو نفس call.message.message_id هنا) ضمن callback_data btn_confirm = types.InlineKeyboardButton("تأكيد الحذف", callback_data=f"confirm_delete_{msg_id}") btn_cancel = types.InlineKeyboardButton("إلغاء", callback_data="cancel_delete") keyboard.row(btn_confirm, btn_cancel) bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=f"⚠️ هل أنت متأكد من حذف الملف:\n*{file_name}*؟", reply_markup=keyboard, parse_mode="markdown" ) elif call.data == "restart": if proc and proc.poll() is None: proc.kill() try: new_proc = subprocess.Popen(["/home/batvoidc/py311/bin/python", data["file_path"]]) control_data[msg_id]["proc"] = new_proc control_data[msg_id]["state"] = "running" save_state() bot.answer_callback_query( call.id, f"♻️ تم إعادة تشغيل الملف:\n{raw_file_name}" ) except Exception as e: bot.answer_callback_query( call.id, f"❌ فشل إعادة التشغيل:\n{str(e)}" ) except Exception as e: logger.error(f"Error in callback: {e}") bot.answer_callback_query(call.id, f"❌ حدث خطأ: {str(e)}") # ================================================================= # === دوال معالجة الحذف وإعادة التشغيل # ================================================================= def delete_file_and_data(msg_id_to_delete, file_path, proc, chat_id, message_id_to_delete_from_tg=None): """دالة مساعدة لحذف الملف وإيقاف العملية وحذف البيانات.""" raw_file_name = os.path.basename(file_path) escaped_file_name = escape_markdown(raw_file_name) # 1. إيقاف العملية if proc and proc.poll() is None: try: proc.kill() except Exception as e: logger.error(f"Failed to kill process {file_path}: {e}") # 2. حذف الملف if os.path.exists(file_path): try: os.remove(file_path) except Exception as e: logger.error(f"Failed to delete file {file_path}: {e}") return False, f"❌ فشل حذف الملف من النظام: {escape_markdown(str(e))}" # 3. حذف رسالة التحكم (إذا كانت موجودة) if message_id_to_delete_from_tg: try: bot.delete_message(chat_id, message_id_to_delete_from_tg) except Exception as e: logger.warning(f"Failed to delete control message {message_id_to_delete_from_tg}: {e}") # 4. حذف البيانات وحفظ الحالة control_data.pop(msg_id_to_delete, None) save_state() return True, f"✅ تم حذف الملف: *{escaped_file_name}* بنجاح\." def handle_confirm_delete(call): """حذف ملف من لوحة التحكم الخاصة به (زر تأكيد الحذف).""" # 💡 التعديل الحاسم هنا: استخراج الـ ID المخزن مباشرة من الـ callback_data try: msg_id_to_confirm = int(call.data.split("_")[-1]) except: bot.answer_callback_query(call.id, "⚠️ خطأ في استخلاص بيانات الرسالة.") return data = control_data.get(msg_id_to_confirm) # نستخدم الـ ID الذي تم تمريره if data: # رسالة تأكيد الحذف الحالية هي التي يجب حذفها من التلغرام current_message_id = call.message.message_id chat_id = call.message.chat.id success, result_msg = delete_file_and_data( msg_id_to_confirm, data.get("file_path"), data.get("proc"), data.get("chat_id"), current_message_id # حذف رسالة التأكيد الحالية ) # نرسل رسالة تأكيد جديدة للمستخدم bot.send_message(chat_id, result_msg, parse_mode="markdown") else: # إذا لم يتم العثور على البيانات، نرد على Callback Query bot.answer_callback_query(call.id, "⚠️ البيانات غير متوفرة") # ونحاول حذف الرسالة الحالية (رسالة التأكيد) try: bot.delete_message(call.message.chat.id, call.message.message_id) except: pass def handle_confirm_delete_global(call): """الخطوة الأولى لحذف ملف من قائمة /deletefile.""" msg_id_to_delete = int(call.data.split("_")[-1]) data = control_data.get(msg_id_to_delete) if data: raw_file_name = os.path.basename(data["file_path"]) escaped_file_name = escape_markdown(raw_file_name) keyboard = types.InlineKeyboardMarkup() btn_confirm = types.InlineKeyboardButton("تأكيد الحذف النهائي", callback_data=f"final_delete_{msg_id_to_delete}") btn_cancel = types.InlineKeyboardButton("إلغاء", callback_data="cancel_delete_global") keyboard.row(btn_confirm, btn_cancel) bot.edit_message_text( chat_id=call.message.chat.id, message_id=call.message.message_id, text=f"⚠️ *تأكيد الحذف*: سيتم حذف الملف *{escaped_file_name}* ورسالة التحكم الخاصة به\. هل أنت متأكد؟", reply_markup=keyboard, parse_mode="markdown" ) else: bot.answer_callback_query(call.id, "⚠️ الملف لم يعد موجوداً في القائمة.") @bot.callback_query_handler(func=lambda call: call.data.startswith("final_delete_")) def handle_final_delete(call): """الخطوة النهائية لحذف الملف من قائمة /deletefile.""" msg_id_to_delete = int(call.data.split("_")[-1]) data = control_data.get(msg_id_to_delete) chat_id = call.message.chat.id if data: success, result_msg = delete_file_and_data( msg_id_to_delete, data.get("file_path"), data.get("proc"), data.get("chat_id"), msg_id_to_delete # رسالة التحكم الأصلية هي التي تحمل هذا الـ ID ) # حذف رسالة قائمة الحذف وإرسال رسالة التأكيد try: bot.delete_message(chat_id, call.message.message_id) except: pass bot.send_message(chat_id, result_msg, parse_mode="markdown") else: bot.answer_callback_query(call.id, "⚠️ فشل الحذف: بيانات الملف مفقودة.") try: bot.delete_message(chat_id, call.message.message_id) except: pass def handle_cancel_delete(call): """إلغاء الحذف من لوحة التحكم (زر إلغاء).""" # هنا نستخدم call.message.message_id للبحث لأن رسالة التأكيد هي تعديل للرسالة الأصلية data = control_data.get(call.message.message_id) if data: raw_file_name = os.path.basename(data["file_path"]) escaped_file_name = escape_markdown(raw_file_name) bot.edit_message_text( chat_id=call.message.chat.id, message_id=call.message.message_id, text=f"🛠 *تحكم في الملف*:\n*{escaped_file_name}*", reply_markup=build_keyboard(), parse_mode="markdown" ) else: bot.answer_callback_query(call.id, "⚠️ تم الإلغاء") def handle_restart(call): """إعادة تشغيل من قائمة /restart.""" msg_id = int(call.data.split("_")[-1]) data = control_data.get(msg_id) if data: raw_file_name = os.path.basename(data["file_path"]) escaped_file_name = escape_markdown(raw_file_name) try: if data["proc"] and data["proc"].poll() is None: data["proc"].kill() new_proc = subprocess.Popen(["/home/batvoidc/py311/bin/python", data["file_path"]]) control_data[msg_id]["proc"] = new_proc control_data[msg_id]["state"] = "running" save_state() bot.answer_callback_query( call.id, f"♻️ تم إعادة تشغيل الملف:\n{raw_file_name}" ) # إرسال رسالة للمستخدم بعد العملية bot.send_message( call.message.chat.id, f"♻️ تم إعادة تشغيل الملف:\n*{escaped_file_name}*", parse_mode="markdown" ) except Exception as e: bot.answer_callback_query( call.id, f"❌ فشل إعادة التشغيل:\n{str(e)}" ) else: bot.answer_callback_query(call.id, "⚠️ الملف غير موجود") # ================================================================= # === نقطة التشغيل الرئيسية # ================================================================= def run_periodic_checks(): while True: try: check_running_files() save_state() time.sleep(60) except Exception as e: logger.error(f"Periodic check error: {e}") time.sleep(30) def run_bot(): load_state() # محاولة تشغيل الملفات التي كانت "running" قبل التحميل threading.Thread(target=run_periodic_checks, daemon=True).start() logger.info("Starting bot...") while True: try: bot.infinity_polling(timeout=10, long_polling_timeout=10) except Exception as e: logger.error(f"Bot crashed: {e}. Restarting in 10 seconds...") time.sleep(10) if __name__ == "__main__": run_bot()