""" Riaslink Port Builder — FastAPI backend. Replaces an earlier Gradio-based UI entirely. Gradio 5.9.1 turned out to have several internal route/schema bugs on this deployment (a localhost- reachability probe that crashed under HF Spaces' container networking, a JSON-schema-to-Python-type conversion bug that 500'd on every page load, and a mount_gradio_app() path that silently broke form submission) — each fix uncovered a new failure in a different internal code path. Rather than keep patching Gradio internals we don't control, this is a small, self-contained FastAPI app: a handful of plain routes, no framework-level "magic" deciding how the browser and server talk to each other. Routes: GET / the build form (static/index.html) POST /api/build start a build, returns {build_id} GET /api/build/{id}/stream Server-Sent Events: live log lines GET /api/build/{id}/download the finished APK, once ready All the actual build logic (zip handling, plugin injection, Gradle invocation) lives in build_pipeline.py and is untouched by this rewrite. """ from __future__ import annotations import asyncio import json import shutil import threading import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Literal from fastapi import FastAPI, Form, UploadFile from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse from fastapi.staticfiles import StaticFiles from build_pipeline import BuildRequest, run_build STATIC_DIR = Path(__file__).parent / "static" UPLOAD_DIR = Path("/tmp/riaslink-uploads") UPLOAD_DIR.mkdir(parents=True, exist_ok=True) app = FastAPI(title="Riaslink Port Builder") app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") # --------------------------------------------------------------------------- # In-memory build tracking # --------------------------------------------------------------------------- # One process, no external queue/database — builds are tracked in a plain # dict guarded by a lock. This matches the previous Gradio version's own # approach (an in-memory list capping kept build directories) and is # sufficient for a single-instance Space; it does not survive a restart, # which is fine since an in-progress build wouldn't survive one either. BuildStatus = Literal["running", "success", "failed"] @dataclass class BuildState: status: BuildStatus = "running" log_lines: list[str] = field(default_factory=list) apk_path: str | None = None error_message: str | None = None log_tail: str = "" _builds: dict[str, BuildState] = {} _builds_lock = threading.Lock() # Same cap/eviction idea as the previous version: keep disk usage bounded # across a long-running process by deleting the oldest finished builds' # working directories once there are too many, rather than never cleaning # up. Tracks work_dir roots, not the state dict entries themselves (state # stays cheap — it's just log strings — so it isn't evicted). _MAX_KEPT_BUILD_DIRS = 20 _kept_build_dirs: list[Path] = [] def _remember_and_trim_build_dir(apk_path: str) -> None: # app/build/outputs/apk//x.apk -> work_dir is 5 levels up. build_dir = Path(apk_path).parents[5] _kept_build_dirs.append(build_dir) while len(_kept_build_dirs) > _MAX_KEPT_BUILD_DIRS: oldest = _kept_build_dirs.pop(0) shutil.rmtree(oldest, ignore_errors=True) def _run_build_in_background(build_id: str, req: BuildRequest, upload_path: Path) -> None: def on_log(msg: str) -> None: with _builds_lock: _builds[build_id].log_lines.append(msg) result = run_build(req, log_callback=on_log) with _builds_lock: state = _builds[build_id] state.log_tail = result.log_tail if result.success: state.status = "success" state.apk_path = result.apk_path _remember_and_trim_build_dir(result.apk_path) else: state.status = "failed" state.error_message = result.error_message if result.log_tail: state.log_lines.append("\n--- Gradle günlüğünün son satırları ---") state.log_lines.extend(result.log_tail.splitlines()) # The uploaded zip is only needed while run_build() is extracting it; # safe to remove once the build has fully finished either way. upload_path.unlink(missing_ok=True) # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- def _sse_event(payload: dict) -> str: return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" @app.get("/", response_class=HTMLResponse) async def index() -> HTMLResponse: html_path = STATIC_DIR / "index.html" return HTMLResponse(html_path.read_text(encoding="utf-8")) @app.post("/api/build") async def start_build( www_zip: UploadFile, # Blank by default: build_pipeline derives the app name from the game's # own index.html and the package id from that name. Anything # typed into the form still wins — see _resolve_identity() there. app_name: str = Form(""), application_id: str = Form(""), version_name: str = Form("1.0.0"), version_code: int = Form(1), fps_default_visible: bool = Form(False), accent_hex: str = Form("#e8a343"), build_variant: str = Form("debug"), # Licensing / protection. All optional — leaving them blank builds a # perfectly working app whose premium sections stay locked, which is what # a quick test build wants. See BuildRequest in build_pipeline.py. license_server: str = Form(""), license_public_key: str = Form(""), expected_signature: str = Form(""), encrypt_assets: bool = Form(False), ) -> dict: build_id = uuid.uuid4().hex upload_path = UPLOAD_DIR / f"{build_id}.zip" with upload_path.open("wb") as f: shutil.copyfileobj(www_zip.file, f) req = BuildRequest( www_zip_path=str(upload_path), # Passed through empty on purpose when unset — run_build() treats an # empty string as "derive this from the game itself". app_name=app_name.strip(), application_id=application_id.strip(), version_name=version_name.strip() or "1.0.0", version_code=version_code, fps_default_visible=fps_default_visible, accent_hex=accent_hex.strip() or "#e8a343", build_variant=build_variant, license_server=license_server.strip(), license_public_key=license_public_key.strip(), expected_signature=expected_signature.strip(), encrypt_assets=encrypt_assets, ) with _builds_lock: _builds[build_id] = BuildState() thread = threading.Thread( target=_run_build_in_background, args=(build_id, req, upload_path), daemon=True, ) thread.start() return {"build_id": build_id} @app.get("/api/build/{build_id}/stream") async def stream_build_log(build_id: str) -> StreamingResponse: async def event_generator(): sent = 0 while True: with _builds_lock: state = _builds.get(build_id) if state is None: yield _sse_event({"type": "error", "message": "Bilinmeyen build id."}) return new_lines = state.log_lines[sent:] sent = len(state.log_lines) status = state.status apk_ready = state.apk_path is not None error_message = state.error_message for line in new_lines: yield _sse_event({"type": "log", "line": line}) if status != "running": yield _sse_event({ "type": "done", "status": status, "download_url": f"/api/build/{build_id}/download" if apk_ready else None, "error_message": error_message, }) return await asyncio.sleep(0.5) return StreamingResponse(event_generator(), media_type="text/event-stream") @app.get("/api/build/{build_id}/download") async def download_apk(build_id: str): with _builds_lock: state = _builds.get(build_id) if state is None or state.apk_path is None: return HTMLResponse("APK bulunamadı ya da build henüz tamamlanmadı.", status_code=404) apk_path = Path(state.apk_path) if not apk_path.exists(): return HTMLResponse("APK dosyası artık mevcut değil (temizlenmiş olabilir).", status_code=404) return FileResponse( apk_path, media_type="application/vnd.android.package-archive", filename=apk_path.name, ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)