import gradio as gr import os import subprocess import threading import shutil import spaces WORKDIR = "/home/user/app/server" os.makedirs(WORKDIR, exist_ok=True) log_output = [] mc_process = None # الحفاظ على استمرار تشغيل الـ Space وتنشيط الـ GPU @spaces.GPU def activate_gpu(): return "[SYSTEM] GPU Activated." def log(message): log_output.append(message) if len(log_output) > 120: log_output.pop(0) # --- نظام إدارة تبعيات السيرفر تشغيل/إيقاف --- def download_dependencies(): if not os.path.exists(os.path.join(WORKDIR, "jdk")): log("[SYSTEM] Downloading Java 21...") os.system(f"wget -qO {WORKDIR}/jdk.tar.gz https://github.com/adoptium/temurin21-binaries/releases/download/jdk-21.0.4%2B7/OpenJDK21U-jre_x64_linux_hotspot_21.0.4_7.tar.gz") os.system(f"tar -xzf {WORKDIR}/jdk.tar.gz -C {WORKDIR}") for extracted in os.listdir(WORKDIR): if extracted.startswith("jdk-21"): os.rename(os.path.join(WORKDIR, extracted), os.path.join(WORKDIR, "jdk")) os.system(f"rm {WORKDIR}/jdk.tar.gz") if not os.path.exists(os.path.join(WORKDIR, "paper.jar")): log("[SYSTEM] Downloading Purpur 1.21.1...") os.system(f"wget -qO {WORKDIR}/paper.jar https://api.purpurmc.org/v2/purpur/1.21.1/latest/download") plugins_dir = os.path.join(WORKDIR, "plugins") os.makedirs(plugins_dir, exist_ok=True) if not os.path.exists(os.path.join(plugins_dir, "playit.jar")): log("[SYSTEM] Downloading Playit Plugin...") os.system(f"wget -qO {plugins_dir}/playit.jar https://github.com/playit-cloud/playit-minecraft-plugin/releases/latest/download/playit-minecraft-plugin.jar") if not os.path.exists(os.path.join(plugins_dir, "ViaVersion.jar")): log("[SYSTEM] Downloading ViaVersion Plugin...") via_url = "https://cdn.modrinth.com/data/P1OZGk5p/versions/ruzmiBqe/ViaVersion-5.10.0.jar?mr_download_reason=standalone&mr_game_version=1.21.1&mr_loader=paper" os.system(f'wget -qO {plugins_dir}/ViaVersion.jar "{via_url}"') if not os.path.exists(os.path.join(plugins_dir, "ViaBackwards.jar")): log("[SYSTEM] Downloading ViaBackwards Plugin...") via_b_url = "https://cdn.modrinth.com/data/NpvuJQoq/versions/YjpKsm6j/ViaBackwards-5.10.0.jar?mr_download_reason=standalone&mr_game_version=1.21.1&mr_loader=paper" os.system(f'wget -qO {plugins_dir}/ViaBackwards.jar "{via_b_url}"') eula_p = os.path.join(WORKDIR, "eula.txt") if not os.path.exists(eula_p): with open(eula_p, "w") as f: f.write("eula=true\n") props_p = os.path.join(WORKDIR, "server.properties") if not os.path.exists(props_p): log("[SYSTEM] Configuring server.properties for cracked mode...") with open(props_p, "w") as f: f.write("online-mode=false\nspawn-protection=0\n") def start_server_backend(): global mc_process if mc_process and mc_process.poll() is None: log("[SYSTEM] Server is already running!") return download_dependencies() log("[SYSTEM] Starting Minecraft Server (Purpur 1.21.1)...") java_path = os.path.join(WORKDIR, "jdk/bin/java") mc_process = subprocess.Popen( [java_path, "-Xms2G", "-Xmx4G", "-jar", "paper.jar", "nogui"], cwd=WORKDIR, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) for line in iter(mc_process.stdout.readline, ''): if line: log(f"[MINECRAFT] {line.strip()}") mc_process.stdout.close() def control_server(action): global mc_process if action == "START": if mc_process and mc_process.poll() is None: return "❌ السيرفر يعمل بالفعل الآن!" threading.Thread(target=start_server_backend, daemon=True).start() return "⚡ جاري تشغيل خادم ماين كرافت الآن..." elif action == "STOP": if mc_process and mc_process.poll() is None: mc_process.terminate() mc_process = None log("[SYSTEM] Server stopped by user.") return "🛑 تم إرسال أمر إيقاف السيرفر بنجاح." return "⚠️ السيرفر متوقف بالفعل." elif action == "RESTART": if mc_process and mc_process.poll() is None: mc_process.terminate() mc_process = None log("[SYSTEM] Restarting server...") threading.Thread(target=start_server_backend, daemon=True).start() return "🔄 جاري إعادة تشغيل السيرفر..." def get_logs(): return "\n".join(log_output) # --- محرك إدارة الملفات المتقدم الحصري --- def get_safe_path(rel_path): abs_path = os.path.abspath(os.path.join(WORKDIR, rel_path)) if abs_path.startswith(os.path.abspath(WORKDIR)): return abs_path return WORKDIR def scan_directory(current_rel): target_dir = get_safe_path(current_rel) folders = [] files = [] try: for item in os.listdir(target_dir): full_path = os.path.join(target_dir, item) if os.path.isdir(full_path): folders.append(item) else: files.append(item) except Exception as e: log(f"[FILE MGR ERROR] {e}") folders.sort() files.sort() folder_choices = [".. (المجلد الأعلى)"] + folders if current_rel else folders file_choices = files if files else ["لا توجد ملفات في هذا المسار"] path_display = f"📁 المسار الحالي المفتوح: /root/{current_rel}" if current_rel else "📁 المسار الحالي المفتوح: /root (المجلد الرئيسي)" return path_display, gr.Dropdown(choices=folder_choices, value=None), gr.Dropdown(choices=file_choices, value=None) def enter_folder(selected_folder, current_rel): if not selected_folder: return current_rel, *scan_directory(current_rel) if selected_folder == ".. (المجلد الأعلى)": if not current_rel: return "", *scan_directory("") parts = current_rel.strip("/").split("/") parent_rel = "/".join(parts[:-1]) return parent_rel, *scan_directory(parent_rel) else: new_rel = os.path.join(current_rel, selected_folder).strip("/") return new_rel, *scan_directory(new_rel) # امتدادات ثنائية معروفة يمنع فتحها كنص إطلاقاً (تتسبب بتجمد الصفحة) BINARY_EXTENSIONS = { ".jar", ".zip", ".gz", ".tar", ".rar", ".7z", ".class", ".dat", ".mca", ".mcr", ".png", ".jpg", ".jpeg", ".gif", ".ico", ".exe", ".bin", ".db", ".so", ".dll" } # الحد الأقصى لحجم الملف المسموح بفتحه في المحرر (بالبايت) - 1 ميجابايت MAX_EDITABLE_SIZE = 1 * 1024 * 1024 def is_probably_binary(file_path, chunk_size=1024): """يفحص أول جزء من الملف بحثاً عن بايتات null تدل على أنه ملف ثنائي.""" try: with open(file_path, "rb") as f: chunk = f.read(chunk_size) return b"\x00" in chunk except Exception: return True def load_file_content(selected_file, current_rel): """ يقرأ محتوى الملف المحدد ويعرضه في المحرر. تم إصلاح المشكلة: الآن يتم فحص نوع الملف وحجمه قبل القراءة لمنع تجمد الصفحة عند فتح ملفات ثنائية (jar, mca, png...) أو ملفات كبيرة جداً. """ if not selected_file or selected_file == "لا توجد ملفات في هذا المسار": return gr.update(value="الرجاء تحديد ملف متاح للقراءة.") file_path = os.path.join(get_safe_path(current_rel), selected_file) if not os.path.isfile(file_path): return gr.update(value=f"⚠️ لا يمكن فتح [{selected_file}] لأنه مجلد وليس ملفاً.") ext = os.path.splitext(selected_file)[1].lower() # 1) تحقق من الامتداد الثنائي المعروف if ext in BINARY_EXTENSIONS: return gr.update( value=f"🚫 لا يمكن عرض [{selected_file}] في المحرر لأنه ملف ثنائي (Binary) وليس ملف نصي.\n" f"هذا النوع من الملفات (jar/zip/png/mca...) لا يُعدّل كنص، بل يُرفع أو يُستبدل فقط عبر خانة الرفع." ) # 2) تحقق من الحجم لتجنب تجميد المتصفح try: size = os.path.getsize(file_path) except Exception as e: return gr.update(value=f"خطأ أثناء قراءة حجم الملف: {e}") if size > MAX_EDITABLE_SIZE: return gr.update( value=f"🚫 الملف [{selected_file}] حجمه كبير جداً ({size/1024/1024:.2f} MB) وفتحه سيجمّد الصفحة.\n" f"الحد المسموح للتحرير هو 1 ميجابايت. استخدم زر الرفع لتبديله بدلاً من فتحه هنا." ) # 3) فحص إضافي: بايتات null تدل على محتوى ثنائي حتى لو الامتداد نصي if is_probably_binary(file_path): return gr.update( value=f"🚫 الملف [{selected_file}] يحتوي على بيانات ثنائية غير قابلة للعرض كنص." ) try: with open(file_path, "r", encoding="utf-8", errors="ignore") as f: content = f.read() lang_map = {".yml": "yaml", ".yaml": "yaml", ".json": "json", ".properties": "yaml", ".txt": None, ".log": None} return gr.update(value=content, language=lang_map.get(ext, None)) except Exception as e: return gr.update(value=f"خطأ أثناء قراءة الملف: {e}") def save_file_content(selected_file, content, current_rel): if not selected_file or selected_file == "لا توجد ملفات في هذا المسار": return "❌ لم يتم تحديد ملف برمجى صالح لحفظه." file_path = os.path.join(get_safe_path(current_rel), selected_file) try: with open(file_path, "w", encoding="utf-8") as f: f.write(content) return f"✅ تم حفظ الملف [{selected_file}] بنجاح في المسار الحالي!" except Exception as e: return f"❌ فشل الحفظ: {e}" def create_item(name, item_type, current_rel): if not name: return "⚠️ الرجاء إدخال اسم المجلد أو الملف أولاً.", *scan_directory(current_rel) target_path = os.path.join(get_safe_path(current_rel), name) try: if item_type == "مجلد Folder": os.makedirs(target_path, exist_ok=True) msg = f"📁 تم إنشاء المجلد [{name}] بنجاح." else: with open(target_path, "w", encoding="utf-8") as f: f.write("") msg = f"📄 تم إنشاء الملف الفارغ [{name}] بنجاح." return msg, *scan_directory(current_rel) except Exception as e: return f"❌ فشل الإنشاء: {e}", *scan_directory(current_rel) def rename_item(old_name, new_name, current_rel): if not old_name or not new_name: return "⚠️ حدد العنصر واكتب الاسم الجديد.", *scan_directory(current_rel) old_p = os.path.join(get_safe_path(current_rel), old_name) new_p = os.path.join(get_safe_path(current_rel), new_name) try: os.rename(old_p, new_p) return f"✏️ تم إعادة التسمية بنجاح إلى [{new_name}].", *scan_directory(current_rel) except Exception as e: return f"❌ خطأ أثناء التسمية: {e}", *scan_directory(current_rel) def delete_item(item_name, is_folder, current_rel): if not item_name or item_name == "لا توجد ملفات في هذا المسار": return "⚠️ لم يتم تحديد عنصر لحذفه.", *scan_directory(current_rel) target_p = os.path.join(get_safe_path(current_rel), item_name) try: if is_folder: shutil.rmtree(target_p) msg = f"🗑️ تم حذف المجلد ومحتوياته [{item_name}] بالكامل." else: os.remove(target_p) msg = f"🗑️ تم حذف الملف [{item_name}]." return msg, *scan_directory(current_rel) except Exception as e: return f"❌ فشل الحذف: {e}", *scan_directory(current_rel) def upload_to_current_dir(file_objs, current_rel): if not file_objs: return "❌ لم يتم اختيار ملفات للرفع.", *scan_directory(current_rel) target_dir = get_safe_path(current_rel) uploaded_list = [] if not isinstance(file_objs, list): file_objs = [file_objs] for f_obj in file_objs: original_name = getattr(f_obj, 'orig_name', os.path.basename(f_obj.name)) shutil.copy(f_obj.name, os.path.join(target_dir, original_name)) uploaded_list.append(original_name) return f"✅ تم رفع {len(uploaded_list)} ملفات مباشرة إلى الدليل الحالي ومستعدة للاستخدام!", *scan_directory(current_rel) # تشغيل تلقائي للسيرفر في البداية عند بناء الـ Space لأول مرة threading.Thread(target=start_server_backend, daemon=True).start() custom_theme = gr.themes.Default( primary_hue="green", secondary_hue="emerald", neutral_hue="slate" ) with gr.Blocks(title="MC Server Host") as demo: demo.load(fn=activate_gpu, inputs=None, outputs=None) current_path_state = gr.State("") gr.Markdown("# 🟢 MC Server Host - Elite Control Panel") gr.Markdown("مرحباً بك في لوحة التحكم المتكاملة. يمكنك الآن التحكم في تشغيل السيرفر وإدارة كافة ملفات خادمك وتحديثها مباشرة.") with gr.Tab("🖥️ الكونسول والتحكم الحى"): with gr.Row(): btn_start = gr.Button("▶️ تشغيل السيرفر (Start)", variant="primary") btn_stop = gr.Button("🛑 إيقاف السيرفر (Stop)", variant="stop") btn_restart = gr.Button("🔄 إعادة تشغيل (Restart)", variant="secondary") sys_status = gr.Markdown("⚡ حالة السيرفر الحالية تظهر في السجلات بالأسفل...") console_logs = gr.Textbox(label="Server Live Console", lines=22, max_lines=22, value="Loading Server System...", interactive=False) refresh_btn = gr.Button("🔄 تحديث شاشة السجلات (Refresh Logs)", variant="primary") btn_start.click(fn=control_server, inputs=gr.State("START"), outputs=sys_status) btn_stop.click(fn=control_server, inputs=gr.State("STOP"), outputs=sys_status) btn_restart.click(fn=control_server, inputs=gr.State("RESTART"), outputs=sys_status) refresh_btn.click(fn=get_logs, outputs=console_logs) with gr.Tab("📂 مدير الملفات الاحترافي (File Manager)"): path_text = gr.Markdown("📁 المسار الحالي المفتوح: /root") with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 🗂️ التصفح والمجلدات") folder_dropdown = gr.Dropdown(label="المجلدات الفرعية (اضغط للاختيار والدخول)", choices=[]) btn_refresh_dir = gr.Button("🔄 تحديث الدليل الحالي", variant="secondary") gr.Markdown("### 📄 الملفات المتاحة") file_dropdown = gr.Dropdown(label="الملفات في هذا المجلد (سيتم فتح الملف تلقائياً عند اختياره)", choices=[]) with gr.Column(scale=2): gr.Markdown("### 📝 محرر نصوص الملف المفتوح") editor_content = gr.Code(label="محتويات الملف الحالي", language=None, lines=18, interactive=True) with gr.Row(): btn_load_f = gr.Button("📂 إعادة تحميل الملف المحدد", variant="secondary") btn_save_f = gr.Button("💾 حفظ التعديلات النصية", variant="primary") file_op_status = gr.Markdown("") gr.Markdown("---") with gr.Row(): with gr.Column(): gr.Markdown("### ➕ إنشاء عناصر جديدة") new_item_name = gr.Textbox(label="اسم الملف أو المجلد الجديد (بالامتداد مثل world أو spigot.yml)") new_item_type = gr.Radio(choices=["مجلد Folder", "ملف نصي File"], value="ملف نصي File", label="نوع العنصر") btn_create = gr.Button("➕ إنشاء الآن", variant="primary") gr.Markdown("### ✏️ تعديل الاسم وإعادة التسمية") rename_new_name = gr.Textbox(label="الاسم الجديد للعنصر المختار") btn_rename = gr.Button("✏️ تطبيق تغيير الاسم", variant="secondary") with gr.Column(): gr.Markdown("### 📤 رفع ملفات مباشرة هنا") file_uploader = gr.File(label="اسحب وأفلت الملفات لرفعها للمسار المفتوح حالياً", file_count="multiple") btn_upload = gr.Button("🚀 رفع وحفظ الملفات المحددة", variant="primary") gr.Markdown("### 🗑️ منطقة الحذف السريع") delete_is_folder = gr.Checkbox(label="العنصر المراد حذفه عبارة عن مجلد كامل", value=False) btn_delete = gr.Button("❌ حذف العنصر المحدد نهائياً", variant="stop") def run_initial_scan(): return scan_directory("") demo.load(fn=run_initial_scan, outputs=[path_text, folder_dropdown, file_dropdown]) demo.load(fn=get_logs, outputs=console_logs) btn_refresh_dir.click(fn=scan_directory, inputs=current_path_state, outputs=[path_text, folder_dropdown, file_dropdown]) folder_dropdown.change( fn=enter_folder, inputs=[folder_dropdown, current_path_state], outputs=[current_path_state, path_text, folder_dropdown, file_dropdown] ) # ✅ الإصلاح الأساسي: فتح الملف تلقائياً فوراً عند اختياره من القائمة file_dropdown.change(fn=load_file_content, inputs=[file_dropdown, current_path_state], outputs=editor_content) # الزر يبقى متاحاً أيضاً لإعادة التحميل اليدوي (مثلاً بعد تعديل الملف من مكان آخر) btn_load_f.click(fn=load_file_content, inputs=[file_dropdown, current_path_state], outputs=editor_content) btn_save_f.click(fn=save_file_content, inputs=[file_dropdown, editor_content, current_path_state], outputs=file_op_status) btn_create.click(fn=create_item, inputs=[new_item_name, new_item_type, current_path_state], outputs=[file_op_status, path_text, folder_dropdown, file_dropdown]) btn_rename.click(fn=rename_item, inputs=[file_dropdown, rename_new_name, current_path_state], outputs=[file_op_status, path_text, folder_dropdown, file_dropdown]) btn_delete.click(fn=delete_item, inputs=[file_dropdown, delete_is_folder, current_path_state], outputs=[file_op_status, path_text, folder_dropdown, file_dropdown]) btn_upload.click(fn=upload_to_current_dir, inputs=[file_uploader, current_path_state], outputs=[file_op_status, path_text, folder_dropdown, file_dropdown]) demo.launch(theme=custom_theme)