386 lines
14 KiB
Python
386 lines
14 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from fastapi import Depends, FastAPI, Form, Request, WebSocket, WebSocketDisconnect
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy.orm import Session
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
|
|
from app.config import settings
|
|
from app.database import Base, SessionLocal, engine, get_db
|
|
from app.game import (
|
|
WORK_LABELS,
|
|
advance,
|
|
current_round,
|
|
maybe_close_round,
|
|
playable_jobs,
|
|
public_job,
|
|
start_game,
|
|
submit_answer,
|
|
)
|
|
from app.models import FetchTask, GameSession, Job, Player, Round
|
|
from app.ral import hide_salary
|
|
from app.realtime import hub
|
|
from app.seed import seed_jobs
|
|
|
|
BASE = Path(__file__).resolve().parent
|
|
app = FastAPI(title="Guess the RAL")
|
|
app.add_middleware(SessionMiddleware, secret_key=settings.secret_key)
|
|
app.mount("/static", StaticFiles(directory=BASE / "static"), name="static")
|
|
templates = Jinja2Templates(directory=str(BASE / "templates"))
|
|
|
|
|
|
@app.on_event("startup")
|
|
def startup() -> None:
|
|
Path("data").mkdir(exist_ok=True)
|
|
Base.metadata.create_all(bind=engine)
|
|
db = SessionLocal()
|
|
seed_jobs(db)
|
|
db.close()
|
|
|
|
|
|
def render(request: Request, name: str, **ctx):
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse(name, ctx)
|
|
|
|
|
|
def require_admin(request: Request):
|
|
if not request.session.get("admin"):
|
|
return False
|
|
return True
|
|
|
|
|
|
def snapshot(db: Session, session: GameSession) -> dict:
|
|
rnd = current_round(db, session) if session.status != "lobby" else None
|
|
players = [
|
|
{"id": p.id, "name": p.name, "score": p.score, "connected": p.connected, "host": p.is_host}
|
|
for p in sorted(session.players, key=lambda x: (-x.score, x.name.lower()))
|
|
]
|
|
payload = {
|
|
"type": "state",
|
|
"session": {
|
|
"id": session.id,
|
|
"status": session.status,
|
|
"current_round": session.current_round,
|
|
"num_rounds": session.num_rounds,
|
|
"max_players": session.max_players,
|
|
"extra_guesses": session.extra_guesses,
|
|
"hide_company": session.hide_company,
|
|
"hide_subject": session.hide_subject,
|
|
},
|
|
"players": players,
|
|
}
|
|
if rnd:
|
|
answered = {a.player_id for a in rnd.answers}
|
|
payload["round"] = {
|
|
"number": rnd.round_number,
|
|
"bonus": rnd.is_bonus,
|
|
"status": rnd.status,
|
|
"options": json.loads(rnd.options_json),
|
|
"job": public_job(session, rnd.job, rnd.is_bonus),
|
|
"answered_ids": list(answered),
|
|
"answered": [p.name for p in session.players if p.id in answered],
|
|
"waiting": [p.name for p in session.players if p.connected and p.id not in answered],
|
|
"work_types": WORK_LABELS,
|
|
}
|
|
if rnd.status == "revealed":
|
|
payload["round"]["correct"] = rnd.correct_option
|
|
payload["round"]["ral_exact"] = rnd.job.ral_exact
|
|
payload["round"]["ral_min"] = rnd.job.ral_min
|
|
payload["round"]["ral_max"] = rnd.job.ral_max
|
|
payload["round"]["work_type"] = rnd.job.work_type
|
|
payload["round"]["company"] = rnd.job.company
|
|
payload["round"]["answers"] = [
|
|
{
|
|
"name": a.player.name,
|
|
"choice": a.choice,
|
|
"precise": a.precise_guess,
|
|
"work_type": a.work_type_guess,
|
|
"points": a.points,
|
|
"correct": a.choice == rnd.correct_option,
|
|
}
|
|
for a in rnd.answers
|
|
]
|
|
if session.status == "finished":
|
|
payload["winner"] = players[0]["name"] if players else ""
|
|
return payload
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def home(request: Request):
|
|
return render(request, "index.html")
|
|
|
|
|
|
@app.post("/create")
|
|
def create_session(
|
|
request: Request,
|
|
name: str = Form(...),
|
|
max_players: int = Form(8),
|
|
num_rounds: int = Form(8),
|
|
extra_guesses: str | None = Form(None),
|
|
hide_company: str | None = Form(None),
|
|
hide_subject: str | None = Form(None),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
name = name.strip()[:24] or "Host"
|
|
session = GameSession(
|
|
max_players=max(2, min(12, max_players)),
|
|
num_rounds=max(4, min(32, num_rounds)),
|
|
extra_guesses=bool(extra_guesses),
|
|
hide_company=bool(hide_company),
|
|
hide_subject=bool(hide_subject),
|
|
)
|
|
db.add(session)
|
|
db.flush()
|
|
player = Player(session_id=session.id, name=name, is_host=True)
|
|
db.add(player)
|
|
db.commit()
|
|
request.session["player_token"] = player.token
|
|
request.session["player_name"] = player.name
|
|
return RedirectResponse(f"/play/{session.id}", status_code=303)
|
|
|
|
|
|
@app.get("/join/{session_id}", response_class=HTMLResponse)
|
|
def join_page(request: Request, session_id: str, db: Session = Depends(get_db)):
|
|
session = db.get(GameSession, session_id)
|
|
if not session:
|
|
return render(request, "index.html", error="Session not found")
|
|
return render(request, "join.html", session=session)
|
|
|
|
|
|
@app.post("/join/{session_id}")
|
|
def join_session(
|
|
request: Request,
|
|
session_id: str,
|
|
name: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
session = db.get(GameSession, session_id)
|
|
if not session:
|
|
return render(request, "index.html", error="Session not found")
|
|
name = name.strip()[:24] or "Player"
|
|
existing = next((p for p in session.players if p.name.lower() == name.lower()), None)
|
|
if existing:
|
|
player = existing
|
|
player.connected = True
|
|
else:
|
|
if session.status != "lobby":
|
|
return render(request, "join.html", session=session, error="Match already started")
|
|
connected = [p for p in session.players if p.connected]
|
|
if len(connected) >= session.max_players:
|
|
return render(request, "join.html", session=session, error="Lobby is full")
|
|
player = Player(session_id=session.id, name=name)
|
|
db.add(player)
|
|
db.commit()
|
|
request.session["player_token"] = player.token
|
|
request.session["player_name"] = player.name
|
|
return RedirectResponse(f"/play/{session.id}", status_code=303)
|
|
|
|
|
|
@app.get("/play/{session_id}", response_class=HTMLResponse)
|
|
def play(request: Request, session_id: str, db: Session = Depends(get_db)):
|
|
session = db.get(GameSession, session_id)
|
|
if not session:
|
|
return render(request, "index.html", error="Session not found")
|
|
token = request.session.get("player_token")
|
|
player = next((p for p in session.players if p.token == token), None)
|
|
if not player:
|
|
return RedirectResponse(f"/join/{session_id}", status_code=303)
|
|
return render(request, "play.html", session=session, player=player)
|
|
|
|
|
|
@app.post("/play/{session_id}/settings")
|
|
def update_settings(
|
|
request: Request,
|
|
session_id: str,
|
|
max_players: int = Form(8),
|
|
num_rounds: int = Form(8),
|
|
extra_guesses: str | None = Form(None),
|
|
hide_company: str | None = Form(None),
|
|
hide_subject: str | None = Form(None),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
session = db.get(GameSession, session_id)
|
|
player = next((p for p in session.players if p.token == request.session.get("player_token")), None)
|
|
if not session or not player or not player.is_host or session.status != "lobby":
|
|
return RedirectResponse(f"/play/{session_id}", status_code=303)
|
|
session.max_players = max(2, min(12, max_players))
|
|
session.num_rounds = max(4, min(32, num_rounds))
|
|
session.extra_guesses = bool(extra_guesses)
|
|
session.hide_company = bool(hide_company)
|
|
session.hide_subject = bool(hide_subject)
|
|
db.commit()
|
|
return RedirectResponse(f"/play/{session_id}", status_code=303)
|
|
|
|
|
|
@app.websocket("/ws/{session_id}")
|
|
async def ws_play(websocket: WebSocket, session_id: str):
|
|
token = websocket.query_params.get("token", "")
|
|
db = SessionLocal()
|
|
try:
|
|
session = db.get(GameSession, session_id)
|
|
player = next((p for p in (session.players if session else []) if p.token == token), None)
|
|
if not session or not player:
|
|
await websocket.close()
|
|
return
|
|
player.connected = True
|
|
db.commit()
|
|
await hub.join(session_id, websocket)
|
|
await hub.broadcast(session_id, snapshot(db, session))
|
|
while True:
|
|
raw = await websocket.receive_text()
|
|
try:
|
|
msg = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
db.expire_all()
|
|
session = db.get(GameSession, session_id)
|
|
player = db.get(Player, player.id)
|
|
action = msg.get("action")
|
|
if action == "start" and player.is_host and session.status == "lobby":
|
|
try:
|
|
start_game(db, session)
|
|
except ValueError as exc:
|
|
await websocket.send_text(json.dumps({"type": "error", "message": str(exc)}))
|
|
elif action == "answer" and session.status == "playing":
|
|
rnd = current_round(db, session)
|
|
if rnd and rnd.status == "open":
|
|
precise = msg.get("precise")
|
|
precise_val = int(precise) if precise not in (None, "", False) else None
|
|
try:
|
|
submit_answer(
|
|
db,
|
|
player,
|
|
rnd,
|
|
str(msg.get("choice") or ""),
|
|
precise_val,
|
|
str(msg.get("work_type") or ""),
|
|
)
|
|
except ValueError:
|
|
pass
|
|
db.refresh(rnd)
|
|
db.refresh(session)
|
|
if maybe_close_round(db, session, rnd):
|
|
db.refresh(session)
|
|
elif action == "next" and player.is_host:
|
|
rnd = current_round(db, session)
|
|
if rnd and rnd.status == "revealed":
|
|
advance(db, session)
|
|
await hub.broadcast(session_id, snapshot(db, session))
|
|
except WebSocketDisconnect:
|
|
pass
|
|
finally:
|
|
hub.leave(session_id, websocket)
|
|
try:
|
|
player = next((p for p in db.get(GameSession, session_id).players if p.token == token), None)
|
|
if player:
|
|
player.connected = False
|
|
db.commit()
|
|
session = db.get(GameSession, session_id)
|
|
await hub.broadcast(session_id, snapshot(db, session))
|
|
except Exception:
|
|
pass
|
|
db.close()
|
|
|
|
|
|
@app.get("/admin", response_class=HTMLResponse)
|
|
def admin(request: Request, db: Session = Depends(get_db)):
|
|
if not require_admin(request):
|
|
return render(request, "admin_login.html")
|
|
jobs = db.query(Job).order_by(Job.id.desc()).limit(80).all()
|
|
tasks = db.query(FetchTask).order_by(FetchTask.id.desc()).limit(40).all()
|
|
queued = db.query(FetchTask).filter(FetchTask.status.in_(["queued", "running"])).count()
|
|
return render(
|
|
request,
|
|
"admin.html",
|
|
jobs=jobs,
|
|
tasks=tasks,
|
|
queued=queued,
|
|
playable=len(playable_jobs(db)),
|
|
)
|
|
|
|
|
|
@app.post("/admin/login")
|
|
def admin_login(request: Request, password: str = Form(...)):
|
|
if password != settings.admin_password:
|
|
return render(request, "admin_login.html", error="Wrong password")
|
|
request.session["admin"] = True
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
|
|
@app.post("/admin/logout")
|
|
def admin_logout(request: Request):
|
|
request.session.pop("admin", None)
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
|
|
@app.post("/admin/fetch")
|
|
def admin_fetch(request: Request, url: str = Form(...), db: Session = Depends(get_db)):
|
|
if not require_admin(request):
|
|
return RedirectResponse("/admin", status_code=303)
|
|
db.add(FetchTask(url=url.strip(), status="queued"))
|
|
db.commit()
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
|
|
@app.post("/admin/job")
|
|
def admin_add_job(
|
|
request: Request,
|
|
title: str = Form(...),
|
|
company: str = Form(""),
|
|
subject: str = Form("other"),
|
|
location: str = Form(""),
|
|
work_type: str = Form("hybrid"),
|
|
ral_exact: str = Form(""),
|
|
ral_min: str = Form(""),
|
|
ral_max: str = Form(""),
|
|
extra_info: str = Form(""),
|
|
description: str = Form(""),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
if not require_admin(request):
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
def num(v):
|
|
v = (v or "").replace(".", "").replace(" ", "")
|
|
return int(v) if v.isdigit() else None
|
|
|
|
exact, low, high = num(ral_exact), num(ral_min), num(ral_max)
|
|
db.add(
|
|
Job(
|
|
title=title,
|
|
company=company,
|
|
subject=subject,
|
|
location=location,
|
|
work_type=work_type,
|
|
ral_exact=exact,
|
|
ral_min=low,
|
|
ral_max=high,
|
|
extra_info=extra_info,
|
|
description=description,
|
|
description_clean=hide_salary(description),
|
|
status="ready" if (exact or low) else "needs_salary",
|
|
source="manual",
|
|
)
|
|
)
|
|
db.commit()
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
|
|
@app.post("/admin/job/{job_id}/status")
|
|
def admin_job_status(
|
|
request: Request,
|
|
job_id: int,
|
|
status: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
if not require_admin(request):
|
|
return RedirectResponse("/admin", status_code=303)
|
|
job = db.get(Job, job_id)
|
|
if job and status in {"ready", "needs_salary", "hidden"}:
|
|
job.status = status
|
|
db.commit()
|
|
return RedirectResponse("/admin", status_code=303)
|