340 lines
13 KiB
Python
340 lines
13 KiB
Python
"""Server-rendered admin WebUI (Jinja2, plain HTML forms — no JS build)."""
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from fastapi.responses import RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from app import store as store_mod
|
|
from app.db import get_connection
|
|
|
|
from . import config
|
|
from .api import _find_cover, _metadata, _release_system, _store_root, _SAFE_NAME_RE
|
|
from .security import get_user_by_token, hash_password, issue_token, now_iso, require_role, revoke_token, verify_password
|
|
|
|
router = APIRouter(include_in_schema=False)
|
|
templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
|
|
|
|
|
|
def _render(request, name, **ctx):
|
|
ctx.setdefault("user", getattr(request.state, "user", None))
|
|
return templates.TemplateResponse(request, name, ctx)
|
|
|
|
|
|
def web_user(request: Request) -> dict:
|
|
user = get_user_by_token(request.cookies.get("psppt_session", ""))
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_303_SEE_OTHER, headers={"Location": "/login"})
|
|
request.state.user = user
|
|
return user
|
|
|
|
|
|
def web_admin(request: Request, user=Depends(web_user)) -> dict:
|
|
if user["role"] != "admin":
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin only")
|
|
return user
|
|
|
|
|
|
def web_editor(request: Request, user=Depends(web_user)) -> dict:
|
|
if user["role"] not in ("admin", "editor"):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Editor or admin required")
|
|
return user
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# auth pages
|
|
# --------------------------------------------------------------------------- #
|
|
@router.get("/login")
|
|
def login_page(request: Request):
|
|
return _render(request, "login.html")
|
|
|
|
|
|
@router.post("/login")
|
|
async def login_submit(request: Request):
|
|
form = await request.form()
|
|
username = (form.get("username") or "").strip()
|
|
password = form.get("password") or ""
|
|
conn = get_connection()
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT id, username, password_hash, role, is_active FROM users WHERE username = ?", (username,)
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
if row is None or not row["is_active"] or not verify_password(password, row["password_hash"]):
|
|
return _render(request, "login.html", error="Invalid username or password")
|
|
token = issue_token(row["id"], kind="session", label="web session")
|
|
conn = get_connection()
|
|
try:
|
|
conn.execute("UPDATE users SET last_login_at = ? WHERE id = ?", (now_iso(), row["id"]))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
resp = RedirectResponse("/dashboard", status_code=status.HTTP_303_SEE_OTHER)
|
|
resp.set_cookie("psppt_session", token, httponly=True, samesite="lax", path="/")
|
|
return resp
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(request: Request):
|
|
token = request.cookies.get("psppt_session", "")
|
|
if token:
|
|
revoke_token(token)
|
|
resp = RedirectResponse("/login", status_code=status.HTTP_303_SEE_OTHER)
|
|
resp.delete_cookie("psppt_session", path="/")
|
|
return resp
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# dashboard
|
|
# --------------------------------------------------------------------------- #
|
|
@router.get("/")
|
|
def root():
|
|
return RedirectResponse("/dashboard", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.get("/dashboard")
|
|
def dashboard(request: Request, user=Depends(web_user)):
|
|
conn = get_connection()
|
|
try:
|
|
systems = conn.execute(
|
|
"SELECT d.system AS system, COUNT(*) AS releases FROM releases r "
|
|
"JOIN dats d ON d.id = r.dat_id GROUP BY d.system ORDER BY d.system"
|
|
).fetchall()
|
|
users = conn.execute("SELECT COUNT(*) AS n FROM users").fetchone()["n"]
|
|
downloads = conn.execute("SELECT COUNT(*) AS n FROM download_events").fetchone()["n"]
|
|
covers = sum(1 for _ in config.COVERS_DIR.rglob("*") if _.is_file()) if config.COVERS_DIR.is_dir() else 0
|
|
recent = conn.execute(
|
|
"SELECT e.system, e.name, e.bytes, e.created_at, u.username FROM download_events e "
|
|
"LEFT JOIN users u ON u.id = e.user_id ORDER BY e.id DESC LIMIT 15"
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
store_games = len(store_mod.scan(config.STORE_ROOT)) if config.STORE_ROOT.is_dir() else 0
|
|
store_videos = len(store_mod.scan(config.STORE_VIDEOS_ROOT)) if config.STORE_VIDEOS_ROOT.is_dir() else 0
|
|
return _render(request, "dashboard.html", systems=[dict(r) for r in systems], users=users,
|
|
downloads=downloads, covers=covers, recent=[dict(r) for r in recent],
|
|
store_games=store_games, store_videos=store_videos)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# releases (games / videos browser)
|
|
# --------------------------------------------------------------------------- #
|
|
@router.get("/releases")
|
|
def releases_page(request: Request, system: str = "", q: str = "", user=Depends(web_user)):
|
|
sql = "SELECT r.id, d.system, r.name, r.region, r.serial FROM releases r JOIN dats d ON d.id = r.dat_id"
|
|
where, params = [], []
|
|
if system:
|
|
where.append("d.system = ?")
|
|
params.append(system)
|
|
if q:
|
|
where.append("r.name LIKE ?")
|
|
params.append(f"%{q}%")
|
|
if where:
|
|
sql += " WHERE " + " AND ".join(where)
|
|
sql += " ORDER BY r.name LIMIT 200"
|
|
conn = get_connection()
|
|
try:
|
|
rows = conn.execute(sql, params).fetchall()
|
|
items = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
d["has_cover"] = _find_cover(r["id"], r["system"]).exists()
|
|
items.append(d)
|
|
finally:
|
|
conn.close()
|
|
return _render(request, "releases.html", items=items, system=system, q=q)
|
|
|
|
|
|
@router.get("/releases/{release_id}/edit")
|
|
def release_edit_page(release_id: int, request: Request, user=Depends(web_editor)):
|
|
conn = get_connection()
|
|
try:
|
|
rel = conn.execute(
|
|
"SELECT r.id, d.system, r.name, r.description, r.region, r.serial "
|
|
"FROM releases r JOIN dats d ON d.id = r.dat_id WHERE r.id = ?",
|
|
(release_id,),
|
|
).fetchone()
|
|
if rel is None:
|
|
raise HTTPException(status_code=404, detail="not found")
|
|
meta = _metadata(conn, release_id, rel["system"])
|
|
finally:
|
|
conn.close()
|
|
has_cover = _find_cover(release_id, rel["system"]).exists()
|
|
return _render(request, "release_edit.html", rel=dict(rel), meta=meta, has_cover=has_cover)
|
|
|
|
|
|
@router.post("/releases/{release_id}/edit")
|
|
async def release_edit_save(release_id: int, request: Request, user=Depends(web_editor)):
|
|
form = await request.form()
|
|
conn = get_connection()
|
|
try:
|
|
rel = conn.execute(
|
|
"SELECT d.system FROM releases r JOIN dats d ON d.id = r.dat_id WHERE r.id = ?", (release_id,)
|
|
).fetchone()
|
|
if rel is None:
|
|
raise HTTPException(status_code=404, detail="not found")
|
|
conn.execute(
|
|
"UPDATE releases SET name = ?, region = ?, serial = ?, description = ? WHERE id = ?",
|
|
(form.get("name") or "", form.get("region") or "", form.get("serial") or "",
|
|
form.get("description") or "", release_id),
|
|
)
|
|
meta = {}
|
|
for k in ("title", "release_date", "developer", "publisher", "genre", "synopsis"):
|
|
v = form.get(k)
|
|
if v is not None and str(v).strip():
|
|
meta[k] = str(v).strip()
|
|
if meta:
|
|
from .api import _save_metadata
|
|
_save_metadata(conn, release_id, rel["system"], meta)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
return RedirectResponse(f"/releases/{release_id}/edit", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post("/releases/{release_id}/cover")
|
|
async def release_cover_upload(release_id: int, request: Request, user=Depends(web_editor)):
|
|
from .api import _save_cover
|
|
form = await request.form()
|
|
file = form.get("cover")
|
|
if file is None or not getattr(file, "filename", ""):
|
|
raise HTTPException(status_code=400, detail="no file")
|
|
data = await file.read()
|
|
conn = get_connection()
|
|
try:
|
|
system = _release_system(conn, release_id)
|
|
finally:
|
|
conn.close()
|
|
if system is None:
|
|
raise HTTPException(status_code=404, detail="not found")
|
|
_save_cover(release_id, system, data)
|
|
return RedirectResponse(f"/releases/{release_id}/edit", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post("/releases/{release_id}/cover/delete")
|
|
def release_cover_delete(release_id: int, user=Depends(web_editor)):
|
|
from .api import _find_cover
|
|
conn = get_connection()
|
|
try:
|
|
system = _release_system(conn, release_id)
|
|
finally:
|
|
conn.close()
|
|
if system:
|
|
p = _find_cover(release_id, system)
|
|
if p.exists():
|
|
p.unlink()
|
|
return RedirectResponse(f"/releases/{release_id}/edit", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# users (admin)
|
|
# --------------------------------------------------------------------------- #
|
|
@router.get("/users")
|
|
def users_page(request: Request, user=Depends(web_admin)):
|
|
conn = get_connection()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT id, username, role, is_active, created_at, last_login_at FROM users ORDER BY username"
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
return _render(request, "users.html", users=[dict(r) for r in rows])
|
|
|
|
|
|
@router.post("/users")
|
|
async def users_create(request: Request, user=Depends(web_admin)):
|
|
form = await request.form()
|
|
username = (form.get("username") or "").strip()
|
|
password = form.get("password") or ""
|
|
role = form.get("role") or "user"
|
|
if not username or not password:
|
|
raise HTTPException(status_code=400, detail="username and password required")
|
|
conn = get_connection()
|
|
try:
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO users (username, password_hash, role, is_active, created_at) VALUES (?, ?, ?, 1, ?)",
|
|
(username, hash_password(password), role, now_iso()),
|
|
)
|
|
conn.commit()
|
|
except Exception:
|
|
raise HTTPException(status_code=409, detail="username already exists")
|
|
finally:
|
|
conn.close()
|
|
return RedirectResponse("/users", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post("/users/{user_id}")
|
|
async def users_update(user_id: int, request: Request, user=Depends(web_admin)):
|
|
form = await request.form()
|
|
action = form.get("action") or ""
|
|
conn = get_connection()
|
|
try:
|
|
if action == "delete":
|
|
if user_id != user["id"]:
|
|
conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
|
else:
|
|
role = form.get("role")
|
|
if role in ("admin", "editor", "user"):
|
|
conn.execute("UPDATE users SET role = ? WHERE id = ?", (role, user_id))
|
|
conn.execute("UPDATE users SET is_active = ? WHERE id = ?", (1 if form.get("is_active") else 0, user_id))
|
|
pw = form.get("password") or ""
|
|
if pw:
|
|
conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (hash_password(pw), user_id))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
return RedirectResponse("/users", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# store management (editor+)
|
|
# --------------------------------------------------------------------------- #
|
|
@router.get("/store")
|
|
def store_page(request: Request, user=Depends(web_editor)):
|
|
def _rows(root):
|
|
out = []
|
|
for it in store_mod.scan(root):
|
|
out.append({"name": it["name"], "filename": Path(it["path"]).name, "size": it["size"]})
|
|
return out
|
|
games = _rows(config.STORE_ROOT) if config.STORE_ROOT.is_dir() else []
|
|
videos = _rows(config.STORE_VIDEOS_ROOT) if config.STORE_VIDEOS_ROOT.is_dir() else []
|
|
return _render(request, "store.html", games=games, videos=videos)
|
|
|
|
|
|
@router.post("/store/upload")
|
|
async def store_upload(request: Request, user=Depends(web_editor)):
|
|
form = await request.form()
|
|
system = form.get("system") or "games"
|
|
file = form.get("file")
|
|
if file is None or not getattr(file, "filename", ""):
|
|
raise HTTPException(status_code=400, detail="no file")
|
|
data = await file.read()
|
|
root = _store_root(system)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
safe = _SAFE_NAME_RE.sub("", file.filename or "")
|
|
if not safe:
|
|
raise HTTPException(status_code=400, detail="invalid filename")
|
|
(root / safe).write_bytes(data)
|
|
return RedirectResponse("/store", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@router.post("/store/{system}/{name}/delete")
|
|
def store_delete(system: str, name: str, user=Depends(web_editor)):
|
|
root = _store_root(system).resolve()
|
|
target = (root / name).resolve()
|
|
if target.is_file() and root in target.parents:
|
|
target.unlink()
|
|
return RedirectResponse("/store", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# settings (admin)
|
|
# --------------------------------------------------------------------------- #
|
|
@router.get("/settings")
|
|
def settings_page(request: Request, user=Depends(web_admin)):
|
|
return _render(request, "settings.html",
|
|
store_root=config.STORE_ROOT, store_videos_root=config.STORE_VIDEOS_ROOT,
|
|
covers_dir=config.COVERS_DIR)
|