| 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 |
|
|
| |
| @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" |
| } |
|
|
| |
| 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() |
|
|
| |
| if ext in BINARY_EXTENSIONS: |
| return gr.update( |
| value=f"🚫 لا يمكن عرض [{selected_file}] في المحرر لأنه ملف ثنائي (Binary) وليس ملف نصي.\n" |
| f"هذا النوع من الملفات (jar/zip/png/mca...) لا يُعدّل كنص، بل يُرفع أو يُستبدل فقط عبر خانة الرفع." |
| ) |
|
|
| |
| 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 ميجابايت. استخدم زر الرفع لتبديله بدلاً من فتحه هنا." |
| ) |
|
|
| |
| 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) |
|
|
| |
| 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) |
