Server Engine

import os, asyncio, collections, shutil, urllib.request, json, time from fastapi import FastAPI, WebSocket, Form, UploadFile, File, HTTPException from fastapi.responses import HTMLResponse, Response from fastapi.middleware.cors import CORSMiddleware import uvicorn # --- CONFIG --- app = FastAPI() app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) BASE_DIR = os.environ.get("SERVER_DIR", os.path.abspath("/app")) PLUGINS_DIR = os.path.join(BASE_DIR, "plugins") mc_process = None output_history = collections.deque(maxlen=300) connected_clients = set() # --- HTML GUI --- HTML_CONTENT = """ Server Engine

Live Stream

Select loader & version, then search.

""" # --- BACKEND LOGIC --- def get_path(p: str): safe = os.path.abspath(os.path.join(BASE_DIR, (p or "").strip("/"))) if not safe.startswith(BASE_DIR): raise HTTPException(403, "Access Denied") return safe async def stream_output(pipe): while True: line = await pipe.readline() if not line: break txt = line.decode('utf-8', errors='replace').rstrip() output_history.append(txt) dead = set() for c in connected_clients: try: await c.send_text(txt) except: dead.add(c) connected_clients.difference_update(dead) async def boot_mc(): global mc_process jar = os.path.join(BASE_DIR, "purpur.jar") if not os.path.exists(jar): output_history.append("\x1b[33m[System] purpur.jar not found in /app. Please upload it via Files tab.\x1b[0m") return # Low resource flags mc_process = await asyncio.create_subprocess_exec( "java", "-Xmx8G", "-Xms6G", "-Dfile.encoding=UTF-8", "-jar", jar, "--nogui", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, cwd=BASE_DIR ) asyncio.create_task(stream_output(mc_process.stdout)) @app.on_event("startup") async def start(): os.makedirs(PLUGINS_DIR, exist_ok=True) asyncio.create_task(boot_mc()) @app.get("/") def index(): return HTMLResponse(HTML_CONTENT) @app.websocket("/ws") async def ws_end(ws: WebSocket): await ws.accept() connected_clients.add(ws) for l in output_history: await ws.send_text(l) try: while True: cmd = await ws.receive_text() if mc_process and mc_process.stdin: mc_process.stdin.write((cmd + "\n").encode()) await mc_process.stdin.drain() except: connected_clients.remove(ws) # FS API @app.get("/api/fs/list") def list_fs(path: str=""): t = get_path(path) if not os.path.exists(t): return [] res = [] for x in os.listdir(t): fp = os.path.join(t, x) res.append({"name": x, "is_dir": os.path.isdir(fp)}) return sorted(res, key=lambda k: (not k["is_dir"], k["name"].lower())) @app.post("/api/fs/upload") async def upload(path: str=Form(""), file: UploadFile=File(...)): t = get_path(path) os.makedirs(t, exist_ok=True) with open(os.path.join(t, file.filename), "wb") as f: shutil.copyfileobj(file.file, f) return "ok" @app.post("/api/fs/delete") def delete(path: str=Form(...)): t = get_path(path) if os.path.isdir(t): shutil.rmtree(t) else: os.remove(t) return "ok" @app.get("/api/fs/read") def read(path: str): try: with open(get_path(path), "r", encoding="utf-8") as f: return json.load(f) if path.endswith(".json") else Response(f.read()) except: raise HTTPException(404) # PLUGIN INSTALLER @app.post("/api/plugins/install") def install_pl(url: str=Form(...), filename: str=Form(...), project_id: str=Form(...), version_id: str=Form(...), name: str=Form(...)): try: # Download dest = os.path.join(PLUGINS_DIR, filename) req = urllib.request.Request(url, headers={'User-Agent': 'HF-Panel/1.0'}) with urllib.request.urlopen(req) as r, open(dest, 'wb') as f: shutil.copyfileobj(r, f) # Update JSON Record j_path = os.path.join(PLUGINS_DIR, "plugins.json") data = {} if os.path.exists(j_path): try: with open(j_path, 'r') as f: data = json.load(f) except: pass data[project_id] = { "name": name, "filename": filename, "version_id": version_id, "installed_at": time.time() } with open(j_path, 'w') as f: json.dump(data, f, indent=2) return "ok" except Exception as e: raise HTTPException(500, str(e)) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), log_level="error")