158 lines
5.0 KiB
Python
158 lines
5.0 KiB
Python
"""FastAPI-App. Lokale Web-UI für aubox.
|
|
|
|
Start:
|
|
uvicorn aubox.web.app:app --host 127.0.0.1 --port 8080
|
|
oder über CLI:
|
|
python -m aubox web
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from .. import filebrowse, p10lite, usb
|
|
from ..library import db as fwdb
|
|
from ..library import identify as fwid
|
|
|
|
WEB_ROOT = Path(__file__).resolve().parent
|
|
FIRMWARE_ROOT = Path(os.environ.get("AUBOX_FIRMWARE_ROOT", "./firmware")).resolve()
|
|
LOADER_ROOT = Path(os.environ.get("AUBOX_LOADER_ROOT", "./loaders")).resolve()
|
|
DB_PATH = FIRMWARE_ROOT / "firmware.db"
|
|
|
|
app = FastAPI(title="aubox")
|
|
app.mount("/static", StaticFiles(directory=WEB_ROOT / "static"), name="static")
|
|
templates = Jinja2Templates(directory=WEB_ROOT / "templates")
|
|
|
|
|
|
def _human_size(n: int) -> str:
|
|
for unit in ("B", "KB", "MB", "GB", "TB"):
|
|
if n < 1024:
|
|
return f"{n:.1f} {unit}" if unit != "B" else f"{n} B"
|
|
n /= 1024
|
|
return f"{n:.1f} PB"
|
|
|
|
|
|
templates.env.filters["humansize"] = _human_size
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request):
|
|
devices = usb.scan()
|
|
with fwdb.open_db(DB_PATH) as conn:
|
|
fw_count = conn.execute("SELECT COUNT(*) FROM firmware").fetchone()[0]
|
|
return templates.TemplateResponse(request, "index.html", {
|
|
"devices": devices,
|
|
"fw_count": fw_count,
|
|
"firmware_root": FIRMWARE_ROOT,
|
|
"loader_root": LOADER_ROOT,
|
|
})
|
|
|
|
|
|
# ---------- Devices --------------------------------------------------------
|
|
|
|
@app.get("/devices", response_class=HTMLResponse)
|
|
async def devices_page(request: Request):
|
|
return templates.TemplateResponse(request, "devices.html", {})
|
|
|
|
|
|
@app.get("/api/devices/html", response_class=HTMLResponse)
|
|
async def devices_partial(request: Request):
|
|
return templates.TemplateResponse(request, "_devices.html", {
|
|
"devices": usb.scan(),
|
|
})
|
|
|
|
|
|
# ---------- Firmware Library ----------------------------------------------
|
|
|
|
@app.get("/firmware", response_class=HTMLResponse)
|
|
async def firmware_page(request: Request):
|
|
with fwdb.open_db(DB_PATH) as conn:
|
|
rows = fwdb.list_all(conn)
|
|
return templates.TemplateResponse(request, "firmware.html", {
|
|
"firmware": rows,
|
|
"firmware_root": FIRMWARE_ROOT,
|
|
})
|
|
|
|
|
|
@app.post("/firmware/scan")
|
|
async def firmware_scan():
|
|
"""Walkt FIRMWARE_ROOT, identifiziert jede Datei, schreibt in DB."""
|
|
if not FIRMWARE_ROOT.is_dir():
|
|
raise HTTPException(404, f"Firmware-Root {FIRMWARE_ROOT} nicht gefunden")
|
|
|
|
seen: set[str] = set()
|
|
added = updated = 0
|
|
with fwdb.open_db(DB_PATH) as conn:
|
|
with fwdb.transaction(conn):
|
|
for path in FIRMWARE_ROOT.rglob("*"):
|
|
if not path.is_file():
|
|
continue
|
|
if path.name == "firmware.db" or path.name.startswith("."):
|
|
continue
|
|
rec = fwid.identify(path, FIRMWARE_ROOT)
|
|
seen.add(rec["rel_path"])
|
|
existed = conn.execute(
|
|
"SELECT 1 FROM firmware WHERE rel_path = ?",
|
|
(rec["rel_path"],),
|
|
).fetchone()
|
|
fwdb.upsert(conn, rec)
|
|
if existed:
|
|
updated += 1
|
|
else:
|
|
added += 1
|
|
removed = fwdb.delete_missing(conn, seen)
|
|
|
|
return JSONResponse({
|
|
"scanned": len(seen),
|
|
"added": added,
|
|
"updated": updated,
|
|
"removed": removed,
|
|
})
|
|
|
|
|
|
@app.get("/firmware/{fw_id}", response_class=HTMLResponse)
|
|
async def firmware_detail(request: Request, fw_id: int):
|
|
with fwdb.open_db(DB_PATH) as conn:
|
|
row = fwdb.get_by_id(conn, fw_id)
|
|
if row is None:
|
|
raise HTTPException(404)
|
|
return templates.TemplateResponse(request, "firmware_detail.html", {
|
|
"fw": row,
|
|
"firmware_root": FIRMWARE_ROOT,
|
|
})
|
|
|
|
|
|
# ---------- File Browser (sandboxed auf FIRMWARE_ROOT) --------------------
|
|
|
|
@app.get("/browse", response_class=HTMLResponse)
|
|
async def browse(request: Request, path: str = ""):
|
|
try:
|
|
target, entries = filebrowse.list_dir(FIRMWARE_ROOT, path)
|
|
except (filebrowse.PathEscapeError, FileNotFoundError, NotADirectoryError) as e:
|
|
raise HTTPException(400, str(e))
|
|
return templates.TemplateResponse(request, "browse.html", {
|
|
"rel": path,
|
|
"entries": entries,
|
|
"crumbs": filebrowse.breadcrumbs(path),
|
|
"firmware_root": FIRMWARE_ROOT,
|
|
})
|
|
|
|
|
|
# ---------- Workflows ------------------------------------------------------
|
|
|
|
@app.get("/workflows/p10lite", response_class=HTMLResponse)
|
|
async def workflow_p10lite(request: Request):
|
|
return templates.TemplateResponse(request, "p10lite.html", {
|
|
"instructions": p10lite.erecovery_instructions(),
|
|
})
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"ok": True}
|