Delete app.py

app.py

@@ -1,413 +0,0 @@
-"""
-╔══════════════════════════════════════════════════════════════╗
-║         LINUX SYSTEM SERVER — by BU THE ZYZ STUDIO          ║
-║    🖥️  سيرفر نظام لينكس — يُتحكم فيه عبر السيرفر الأول     ║
-╚══════════════════════════════════════════════════════════════╝
-
-هذا الملف خاص بـ: سيرفر 2 (سيرفر نظام Linux)
-This file belongs to: SERVER 2 (Linux System Server)
-"""
-
-import os
-import subprocess
-import json
-import threading
-import time
-import shutil
-import signal
-from pathlib import Path
-from flask import Flask, request, jsonify, Response, stream_with_context
-from flask_cors import CORS
-
-app = Flask(__name__)
-CORS(app, resources={r"/api/*": {"origins": "*"}})
-
-# ══════════════════════════════════════════
-#  🔐 Security Token — رمز الأمان
-#  Set as environment variable: LINUX_API_TOKEN
-#  عيّن هذا المتغير في إعدادات Hugging Face Space
-# ══════════════════════════════════════════
-API_TOKEN = os.environ.get("LINUX_API_TOKEN", "zyz-linux-secret-2025")
-
-# ══════════════════════════════════════════
-#  📁 Workspace Directory
-# ══════════════════════════════════════════
-WORKSPACE = "/workspace"
-os.makedirs(WORKSPACE, exist_ok=True)
-
-# Active processes store (for killing long-running processes)
-active_processes = {}
-process_lock = threading.Lock()
-
-
-# ══════════════════════════════════════════
-#  🔒 Auth Helper
-# ══════════════════════════════════════════
-def verify_token(req) -> bool:
-    """التحقق من رمز الأمان في الطلب"""
-    # Check header first
-    token_header = req.headers.get("X-API-Token", "")
-    if token_header == API_TOKEN:
-        return True
-    # Check JSON body
-    if req.is_json:
-        body = req.get_json(silent=True) or {}
-        if body.get("token", "") == API_TOKEN:
-            return True
-    return False
-
-
-# ══════════════════════════════════════════
-#  ✅ Status Endpoint
-# ══════════════════════════════════════════
-@app.route("/")
-def root():
-    return jsonify({
-        "name": "Linux System Server",
-        "version": "1.0.0",
-        "by": "BU THE ZYZ STUDIO",
-        "status": "online",
-        "note": "Use /api/status with your token to verify authentication."
-    })
-
-
-@app.route("/api/status")
-def status():
-    """حالة السيرفر — يعمل بدون توثيق للتحقق من الاتصال"""
-    try:
-        uname = subprocess.run("uname -a", shell=True, capture_output=True, text=True, timeout=5)
-        disk   = subprocess.run("df -h /workspace 2>/dev/null || df -h /", shell=True, capture_output=True, text=True, timeout=5)
-        mem    = subprocess.run("free -h", shell=True, capture_output=True, text=True, timeout=5)
-        python = subprocess.run("python3 --version", shell=True, capture_output=True, text=True, timeout=5)
-    except Exception:
-        uname = disk = mem = python = type("R", (), {"stdout": "N/A", "stderr": ""})()
-
-    authenticated = verify_token(request)
-    return jsonify({
-        "status": "online",
-        "authenticated": authenticated,
-        "system": uname.stdout.strip(),
-        "disk": disk.stdout.strip(),
-        "memory": mem.stdout.strip(),
-        "python": python.stdout.strip(),
-        "workspace": WORKSPACE,
-        "timestamp": time.time()
-    })
-
-
-# ══════════════════════════════════════════
-#  ⚡ Execute Command — تنفيذ أوامر الترمينال
-# ══════════════════════════════════════════
-@app.route("/api/execute", methods=["POST"])
-def execute_command():
-    """
-    تنفيذ أمر ترمينال واحد والحصول على النتيجة كاملة
-    Execute a single terminal command and get full output
-    """
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized — Invalid or missing API token"}), 401
-
-    data    = request.get_json(silent=True) or {}
-    command = data.get("command", "").strip()
-    cwd     = data.get("cwd", WORKSPACE)
-    timeout = min(int(data.get("timeout", 30)), 300)  # max 5 minutes
-
-    if not command:
-        return jsonify({"error": "No command provided"}), 400
-
-    # Validate working directory
-    if not os.path.exists(cwd):
-        cwd = WORKSPACE
-
-    try:
-        result = subprocess.run(
-            command,
-            shell=True,
-            capture_output=True,
-            text=True,
-            cwd=cwd,
-            timeout=timeout,
-            env={**os.environ, "HOME": "/root", "TERM": "xterm-256color"}
-        )
-        return jsonify({
-            "success": True,
-            "stdout":     result.stdout,
-            "stderr":     result.stderr,
-            "returncode": result.returncode,
-            "command":    command,
-            "cwd":        cwd
-        })
-    except subprocess.TimeoutExpired:
-        return jsonify({
-            "success": False,
-            "error": f"Command timed out after {timeout} seconds",
-            "returncode": -1
-        }), 408
-    except Exception as e:
-        return jsonify({
-            "success": False,
-            "error": str(e),
-            "returncode": -1
-        }), 500
-
-
-# ══════════════════════════════════════════
-#  📡 Stream Execute — تنفيذ مع بث النتائج
-# ══════════════════════════════════════════
-@app.route("/api/stream", methods=["POST"])
-def stream_execute():
-    """
-    تنفيذ أمر مع بث المخرجات سطراً بسطر (مفيد للأوامر الطويلة)
-    Stream command output line by line (useful for long-running commands)
-    """
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-
-    data    = request.get_json(silent=True) or {}
-    command = data.get("command", "").strip()
-    cwd     = data.get("cwd", WORKSPACE)
-    pid_key = data.get("pid_key", f"proc_{int(time.time())}")
-
-    if not command:
-        return jsonify({"error": "No command provided"}), 400
-
-    if not os.path.exists(cwd):
-        cwd = WORKSPACE
-
-    @stream_with_context
-    def generate():
-        try:
-            process = subprocess.Popen(
-                command,
-                shell=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,
-                text=True,
-                cwd=cwd,
-                env={**os.environ, "HOME": "/root", "TERM": "xterm-256color"},
-                bufsize=1
-            )
-            with process_lock:
-                active_processes[pid_key] = process
-
-            yield f"data: {json.dumps({'type': 'start', 'pid': process.pid, 'command': command})}\n\n"
-
-            for line in iter(process.stdout.readline, ""):
-                yield f"data: {json.dumps({'type': 'output', 'line': line})}\n\n"
-
-            process.wait()
-            with process_lock:
-                active_processes.pop(pid_key, None)
-
-            yield f"data: {json.dumps({'type': 'done', 'returncode': process.returncode})}\n\n"
-            yield "data: [DONE]\n\n"
-
-        except Exception as e:
-            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
-
-    return Response(
-        generate(),
-        mimetype="text/event-stream",
-        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
-    )
-
-
-# ══════════════════════════════════════════
-#  🛑 Kill Process — إيقاف عملية جارية
-# ══════════════════════════════════════════
-@app.route("/api/kill", methods=["POST"])
-def kill_process():
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-    data = request.get_json(silent=True) or {}
-    pid_key = data.get("pid_key", "")
-    with process_lock:
-        proc = active_processes.get(pid_key)
-    if proc:
-        try:
-            proc.terminate()
-            time.sleep(0.5)
-            if proc.poll() is None:
-                proc.kill()
-            return jsonify({"success": True, "message": "Process terminated"})
-        except Exception as e:
-            return jsonify({"error": str(e)}), 500
-    return jsonify({"error": "Process not found"}), 404
-
-
-# ══════════════════════════════════════════
-#  📁 File System Operations
-# ══════════════════════════════════════════
-@app.route("/api/files/list", methods=["GET", "POST"])
-def list_files():
-    """قائمة الملفات في مسار معين"""
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-
-    if request.method == "POST":
-        data = request.get_json(silent=True) or {}
-        path = data.get("path", WORKSPACE)
-    else:
-        path = request.args.get("path", WORKSPACE)
-
-    try:
-        items = []
-        for name in sorted(os.listdir(path)):
-            full = os.path.join(path, name)
-            try:
-                stat = os.stat(full)
-                items.append({
-                    "name":     name,
-                    "path":     full,
-                    "is_dir":   os.path.isdir(full),
-                    "size":     stat.st_size,
-                    "modified": stat.st_mtime
-                })
-            except Exception:
-                pass
-        return jsonify({"success": True, "path": path, "items": items})
-    except Exception as e:
-        return jsonify({"error": str(e)}), 500
-
-
-@app.route("/api/files/read", methods=["POST"])
-def read_file():
-    """قراءة محتوى ملف"""
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-    data = request.get_json(silent=True) or {}
-    path = data.get("path", "").strip()
-    if not path:
-        return jsonify({"error": "No path provided"}), 400
-    try:
-        with open(path, "r", encoding="utf-8", errors="replace") as f:
-            content = f.read()
-        return jsonify({"success": True, "path": path, "content": content, "size": len(content)})
-    except Exception as e:
-        return jsonify({"error": str(e)}), 500
-
-
-@app.route("/api/files/write", methods=["POST"])
-def write_file():
-    """كتابة محتوى إلى ملف"""
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-    data    = request.get_json(silent=True) or {}
-    path    = data.get("path", "").strip()
-    content = data.get("content", "")
-    if not path:
-        return jsonify({"error": "No path provided"}), 400
-    try:
-        dir_ = os.path.dirname(path)
-        if dir_:
-            os.makedirs(dir_, exist_ok=True)
-        with open(path, "w", encoding="utf-8") as f:
-            f.write(content)
-        return jsonify({"success": True, "path": path, "bytes_written": len(content.encode())})
-    except Exception as e:
-        return jsonify({"error": str(e)}), 500
-
-
-@app.route("/api/files/delete", methods=["POST"])
-def delete_file():
-    """حذف ملف أو مجلد"""
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-    data = request.get_json(silent=True) or {}
-    path = data.get("path", "").strip()
-    if not path:
-        return jsonify({"error": "No path provided"}), 400
-    try:
-        if os.path.isdir(path):
-            shutil.rmtree(path)
-        else:
-            os.remove(path)
-        return jsonify({"success": True, "deleted": path})
-    except Exception as e:
-        return jsonify({"error": str(e)}), 500
-
-
-@app.route("/api/files/mkdir", methods=["POST"])
-def make_dir():
-    """إنشاء مجلد جديد"""
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-    data = request.get_json(silent=True) or {}
-    path = data.get("path", "").strip()
-    try:
-        os.makedirs(path, exist_ok=True)
-        return jsonify({"success": True, "created": path})
-    except Exception as e:
-        return jsonify({"error": str(e)}), 500
-
-
-# ══════════════════════════════════════════
-#  📦 Package Manager — إدارة الحزم
-# ══════════════════════════════════════════
-@app.route("/api/packages/install", methods=["POST"])
-def install_package():
-    """تثبيت حزمة Python أو apt"""
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-    data    = request.get_json(silent=True) or {}
-    package = data.get("package", "").strip()
-    manager = data.get("manager", "pip")  # "pip" or "apt"
-    if not package:
-        return jsonify({"error": "No package name"}), 400
-    try:
-        if manager == "apt":
-            cmd = f"apt-get install -y {package}"
-        else:
-            cmd = f"pip install {package}"
-        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=120)
-        return jsonify({
-            "success": result.returncode == 0,
-            "stdout": result.stdout,
-            "stderr": result.stderr
-        })
-    except subprocess.TimeoutExpired:
-        return jsonify({"error": "Installation timed out"}), 408
-    except Exception as e:
-        return jsonify({"error": str(e)}), 500
-
-
-# ══════════════════════════════════════════
-#  🔧 System Info — معلومات النظام
-# ══════════════════════════════════════════
-@app.route("/api/sysinfo", methods=["GET", "POST"])
-def sysinfo():
-    """معلومات تفصيلية عن النظام"""
-    if not verify_token(request):
-        return jsonify({"error": "Unauthorized"}), 401
-
-    commands = {
-        "os":       "cat /etc/os-release 2>/dev/null | head -5",
-        "cpu":      "lscpu 2>/dev/null | grep -E 'Architecture|CPU\\(s\\)|Model name' | head -5",
-        "memory":   "free -h",
-        "disk":     "df -h",
-        "network":  "ip addr show 2>/dev/null | grep 'inet ' | head -5",
-        "python":   "python3 --version && pip3 list 2>/dev/null | head -20",
-        "processes":"ps aux --sort=-%cpu 2>/dev/null | head -10",
-        "uptime":   "uptime"
-    }
-    info = {}
-    for key, cmd in commands.items():
-        try:
-            r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
-            info[key] = r.stdout.strip()
-        except Exception:
-            info[key] = "N/A"
-
-    return jsonify({"success": True, "info": info})
-
-
-# ══════════════════════════════════════════
-#  🚀 Run — تشغيل السيرفر
-# ══════════════════════════════════════════
-if __name__ == "__main__":
-    print("=" * 60)
-    print("  🖥️  Linux System Server — BU THE ZYZ STUDIO")
-    print("  Port: 7860")
-    print("  Token: Set LINUX_API_TOKEN env variable")
-    print("=" * 60)
-    app.run(host="0.0.0.0", port=7860, debug=False, threaded=True)