Refactor for smaller context

This commit is contained in:
2026-06-10 15:29:03 -07:00
parent 2d57973cd8
commit 2ddcb86cc0
26 changed files with 3373 additions and 3188 deletions

View File

@@ -1,35 +1,26 @@
import json
from typing import List, Optional
from typing import Optional
from fastapi import FastAPI, Request, Form, Depends, HTTPException, status
from fastapi.responses import RedirectResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from sqlmodel import Session
import uvicorn
from pathlib import Path
from . import crud
from . import cards
from .database import create_db_and_tables, get_session
from .models import Game, Player, Obstacle, Vote
from .templates import templates, BASE_DIR
app = FastAPI(title="Rats with Gats Remote Play")
# Get the directory of this module
BASE_DIR = Path(__file__).resolve().parent
# Mount static files
app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")
# Templates setup
templates = Jinja2Templates(directory=BASE_DIR / "templates")
# Register startup event to create tables
@app.on_event("startup")
def on_startup():
create_db_and_tables()
# --- HTTP Route Handlers ---
# --- HTTP Core Route Handlers ---
@app.get("/", response_class=HTMLResponse)
def home(request: Request):
@@ -80,23 +71,7 @@ def player_dashboard(request: Request, game_id: str, player_id: str, db: Session
"player": player
})
# Creator Admin Panel
@app.get("/game/{game_id}/admin", response_class=HTMLResponse)
def creator_admin_panel(request: Request, game_id: str, key: str, db: Session = Depends(get_session)):
game = crud.get_game(db, game_id)
if not game:
raise HTTPException(status_code=404, detail="Game not found")
if game.admin_key != key:
raise HTTPException(status_code=403, detail="Forbidden: Invalid admin key")
base_url = str(request.base_url).rstrip("/")
return templates.TemplateResponse("admin.html", {
"request": request,
"game": game,
"base_url": base_url
})
# --- HTMX Snippet and Phase Polling Routes ---
# --- HTMX Core Polling Route ---
@app.get("/game/{game_id}/player/{player_id}/phase-check")
def phase_check_route(game_id: str, player_id: str, current: str, db: Session = Depends(get_session)):
@@ -108,559 +83,21 @@ def phase_check_route(game_id: str, player_id: str, current: str, db: Session =
return HTMLResponse(status_code=200, headers={"HX-Trigger": "phase-changed"})
return HTMLResponse(status_code=204) # No content, no reload
@app.get("/game/{game_id}/lobby/players", response_class=HTMLResponse)
def lobby_players_snippet(request: Request, game_id: str, player_id: Optional[str] = None, db: Session = Depends(get_session)):
game = crud.get_game(db, game_id)
if not game:
return HTMLResponse("")
return templates.TemplateResponse("lobby_players_snippet.html", {
"request": request,
"game": game,
"player_id": player_id
})
# --- Register Modular Phase Routers ---
@app.get("/game/{game_id}/scene/roster", response_class=HTMLResponse)
def scene_roster_snippet(request: Request, game_id: str, db: Session = Depends(get_session)):
game = crud.get_game(db, game_id)
if not game:
return HTMLResponse("")
return templates.TemplateResponse("scene_roster_snippet.html", {
"request": request,
"game": game,
"is_captain": crud.is_player_captain
})
from .phase_view import router as phase_view_router
from .routes_lobby import router as lobby_router
from .routes_character import router as character_router
from .routes_scene import router as scene_router
from .routes_upkeep import router as upkeep_router
from .routes_admin import router as admin_router
@app.get("/game/{game_id}/scene/obstacles", response_class=HTMLResponse)
def scene_obstacles_snippet(request: Request, game_id: str, player_id: str, db: Session = Depends(get_session)):
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
return HTMLResponse("")
hand = crud.get_player_hand(player)
return templates.TemplateResponse("scene_obstacles_snippet.html", {
"request": request,
"game": game,
"player": player,
"hand": hand,
"parse_card": cards.parse_card,
"json_loads": json.loads,
"is_captain": crud.is_player_captain
})
@app.get("/game/{game_id}/between/roster", response_class=HTMLResponse)
def between_scenes_roster_snippet(request: Request, game_id: str, db: Session = Depends(get_session)):
game = crud.get_game(db, game_id)
if not game:
return HTMLResponse("")
return templates.TemplateResponse("voting_roster_snippet.html", {
"request": request,
"game": game
})
# --- HTMX View Polling Render Route ---
@app.get("/game/{game_id}/player/{player_id}/view", response_class=HTMLResponse)
def get_game_phase_view(
request: Request,
game_id: str,
player_id: str,
edit: bool = False,
db: Session = Depends(get_session)
):
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
return HTMLResponse("<p class='alert alert-danger'>Session disconnected or expired.</p>")
base_url = str(request.base_url).rstrip("/")
# Context helper functions to run inside Jinja templates
def get_player_name_by_id(*args) -> str:
pid = args[1] if len(args) == 2 else (args[0] if len(args) == 1 else None)
if not pid:
return "None"
p = db.get(Player, pid)
return p.name if p else "Unknown"
def get_obstacle_by_id(obs_id: str) -> Optional[Obstacle]:
return db.get(Obstacle, obs_id)
# Core variables
hand = crud.get_player_hand(player)
max_hand_size = crud.calculate_max_hand_size(player, game.players)
# Compute inbox tasks for current player
inbox_tasks = []
for p in game.players:
if p.id != player.id:
if p.other_like_from_player_id == player.id and not p.other_like:
inbox_tasks.append({"player": p, "type": "like"})
if p.other_hate_from_player_id == player.id and not p.other_hate:
inbox_tasks.append({"player": p, "type": "hate"})
# Freshly fetch votes to avoid cache stale
from sqlmodel import select
votes = db.exec(select(Vote).where(Vote.game_id == game_id)).all()
# Calculate eligible deep players
eligible_deeps = [p for p in game.players if p.previous_role != "deep"]
context = {
"request": request,
"game": game,
"player": player,
"hand": hand,
"max_hand_size": max_hand_size,
"base_url": base_url,
"db": db,
"get_player_name": get_player_name_by_id,
"get_obstacle_by_id": get_obstacle_by_id,
"parse_card": cards.parse_card,
"is_captain": crud.is_player_captain,
"json_loads": json.loads,
"deck_count": len(crud.get_game_deck(game)),
"inbox_tasks": inbox_tasks,
"votes": votes,
"edit_mode": edit,
"eligible_deeps": eligible_deeps
}
# Select template based on current game phase
if game.phase == "lobby":
return templates.TemplateResponse("lobby_partial.html", context)
elif game.phase in ["character_creation", "swap_techniques"]:
return templates.TemplateResponse("character_creation_partial.html", context)
elif game.phase == "scene_setup":
return templates.TemplateResponse("scene_setup_partial.html", context)
elif game.phase == "scene":
return templates.TemplateResponse("scene_partial.html", context)
elif game.phase == "between_scenes":
return templates.TemplateResponse("between_scenes_partial.html", context)
elif game.phase == "deep_upkeep":
return templates.TemplateResponse("deep_upkeep_partial.html", context)
return HTMLResponse("<p>Unknown game state.</p>")
# --- HTMX Action Handlers ---
# 1. Start character creation from lobby
@app.post("/game/{game_id}/lobby/start")
def lobby_start_character_creation(game_id: str, db: Session = Depends(get_session)):
game = crud.get_game(db, game_id)
if not game:
raise HTTPException(status_code=404, detail="Game not found")
game.phase = "character_creation"
db.add(game)
db.commit()
return HTMLResponse(status_code=204) # Return No Content, polling redirects
# 2. Save basic character details
@app.post("/game/{game_id}/player/{player_id}/save-basic")
def player_save_basic_details(
game_id: str,
player_id: str,
avatar_look: str = Form(...),
avatar_smell: str = Form(...),
first_words: str = Form(...),
good_at_math: str = Form(...),
db: Session = Depends(get_session)
):
player = crud.get_player(db, player_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
player.avatar_look = avatar_look.strip()
player.avatar_smell = avatar_smell.strip()
player.first_words = first_words.strip()
player.good_at_math = good_at_math
db.add(player)
db.commit()
return RedirectResponse(url=f"/game/{game_id}/player/{player_id}/view", status_code=status.HTTP_303_SEE_OTHER)
# 3. Edit basic character details (reset values to trigger form view)
@app.post("/game/{game_id}/player/{player_id}/edit-basic")
def player_edit_basic_details(
game_id: str,
player_id: str,
db: Session = Depends(get_session)
):
# Trigger HTMX reload with edit mode active
return RedirectResponse(url=f"/game/{game_id}/player/{player_id}/view?edit=true", status_code=status.HTTP_303_SEE_OTHER)
# Auto-assign unassigned delegated tasks (Like and Hate) randomly to crewmates
@app.post("/game/{game_id}/player/{player_id}/delegate/auto")
def player_auto_delegate(
game_id: str,
player_id: str,
db: Session = Depends(get_session)
):
import random
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
raise HTTPException(status_code=404, detail="Not found")
other_players = [p for p in game.players if p.id != player.id]
if other_players:
# Assign 'like' if not assigned
if not player.other_like_from_player_id:
like_target = random.choice(other_players)
player.other_like_from_player_id = like_target.id
player.other_like = ""
# Assign 'hate' if not assigned
if not player.other_hate_from_player_id:
if len(other_players) >= 2:
# Try to pick a different target than the like target
remaining = [op for op in other_players if op.id != player.other_like_from_player_id]
hate_target = random.choice(remaining)
else:
hate_target = random.choice(other_players)
player.other_hate_from_player_id = hate_target.id
player.other_hate = ""
db.add(player)
db.commit()
return RedirectResponse(url=f"/game/{game_id}/player/{player_id}/view", status_code=status.HTTP_303_SEE_OTHER)
# Get Crew Creation Status (polling endpoint)
@app.get("/game/{game_id}/player/{player_id}/crew-status", response_class=HTMLResponse)
def player_crew_status(
request: Request,
game_id: str,
player_id: str,
db: Session = Depends(get_session)
):
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
raise HTTPException(status_code=404, detail="Not found")
def get_player_name_by_id(pid: Optional[str]) -> str:
if not pid:
return "None"
p = db.get(Player, pid)
return p.name if p else "Unknown"
return templates.TemplateResponse("crew_status_snippet.html", {
"request": request,
"game": game,
"player": player,
"json_loads": json.loads,
"get_player_name": get_player_name_by_id
})
# 4. Delegate question 5/6 to another player
@app.post("/game/{game_id}/player/{player_id}/delegate/{question_type}", response_class=HTMLResponse)
def player_delegate_question(
request: Request,
game_id: str,
player_id: str,
question_type: str, # "like" or "hate"
target_player_id: str = Form(...),
db: Session = Depends(get_session)
):
crud.delegate_question(db, player_id, question_type, target_player_id)
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
raise HTTPException(status_code=404, detail="Not found")
def get_player_name_by_id(pid: Optional[str]) -> str:
if not pid:
return "None"
p = db.get(Player, pid)
return p.name if p else "Unknown"
return templates.TemplateResponse("delegation_snippet.html", {
"request": request,
"game": game,
"player": player,
"question_type": question_type,
"get_player_name": get_player_name_by_id
})
# Revoke a delegated question task
@app.post("/game/{game_id}/player/{player_id}/revoke-delegate/{question_type}", response_class=HTMLResponse)
def player_revoke_delegation(
request: Request,
game_id: str,
player_id: str,
question_type: str, # "like" or "hate"
db: Session = Depends(get_session)
):
player = crud.get_player(db, player_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
if question_type == "like":
player.other_like_from_player_id = None
player.other_like = ""
elif question_type == "hate":
player.other_hate_from_player_id = None
player.other_hate = ""
db.add(player)
db.commit()
game = crud.get_game(db, game_id)
if not game:
raise HTTPException(status_code=404, detail="Game not found")
def get_player_name_by_id(pid: Optional[str]) -> str:
if not pid:
return "None"
p = db.get(Player, pid)
return p.name if p else "Unknown"
return templates.TemplateResponse("delegation_snippet.html", {
"request": request,
"game": game,
"player": player,
"question_type": question_type,
"get_player_name": get_player_name_by_id
})
# Fetch the delegation card status (for background polling when waiting)
@app.get("/game/{game_id}/player/{player_id}/delegation/{question_type}", response_class=HTMLResponse)
def player_get_delegation_status(
request: Request,
game_id: str,
player_id: str,
question_type: str, # "like" or "hate"
db: Session = Depends(get_session)
):
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
raise HTTPException(status_code=404, detail="Not found")
def get_player_name_by_id(pid: Optional[str]) -> str:
if not pid:
return "None"
p = db.get(Player, pid)
return p.name if p else "Unknown"
return templates.TemplateResponse("delegation_snippet.html", {
"request": request,
"game": game,
"player": player,
"question_type": question_type,
"get_player_name": get_player_name_by_id
})
# 5. Answer a delegated question for another player
@app.post("/game/{game_id}/player/{player_id}/submit-delegate/{target_player_id}/{question_type}")
def player_submit_delegated_answer(
game_id: str,
player_id: str, # The answering player
target_player_id: str, # The player whose sheet we are filling out
question_type: str, # "like" or "hate"
answer: str = Form(...),
db: Session = Depends(get_session)
):
crud.submit_delegated_answer(db, target_player_id, question_type, answer.strip(), player_id)
return HTMLResponse("")
# 6. Submit 3 Secret Pirate Techniques
@app.post("/game/{game_id}/player/{player_id}/submit-techniques")
def player_submit_techniques(
request: Request,
game_id: str,
player_id: str,
tech1: str = Form(...),
tech2: str = Form(...),
tech3: str = Form(...),
db: Session = Depends(get_session)
):
crud.submit_techniques(db, player_id, tech1.strip(), tech2.strip(), tech3.strip())
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
raise HTTPException(status_code=404, detail="Not found")
return templates.TemplateResponse("techniques_snippet.html", {
"request": request,
"game": game,
"player": player,
"json_loads": json.loads
})
# 7. Assign Swapped Techniques to Face Cards (Jack, Queen, King)
@app.post("/game/{game_id}/player/{player_id}/assign-face-techniques")
def player_assign_face_techniques(
request: Request,
game_id: str,
player_id: str,
jack: str = Form(...),
queen: str = Form(...),
king: str = Form(...),
db: Session = Depends(get_session)
):
if len({jack, queen, king}) < 3:
# Prevent assigning the same technique to multiple face cards
return HTMLResponse("<p class='alert alert-danger'>Error: Each card must be assigned a unique technique.</p>", status_code=400)
crud.assign_face_card_techniques(db, player_id, jack, queen, king)
game = crud.get_game(db, game_id)
player = crud.get_player(db, player_id)
if not game or not player:
raise HTTPException(status_code=404, detail="Not found")
return templates.TemplateResponse("techniques_snippet.html", {
"request": request,
"game": game,
"player": player,
"json_loads": json.loads
})
# 8. Set Role during Scene Setup
@app.post("/game/{game_id}/player/{player_id}/set-role", response_class=HTMLResponse)
def player_set_role(
request: Request,
game_id: str,
player_id: str,
role: str,
db: Session = Depends(get_session)
):
crud.update_scene_role(db, player_id, role)
return get_game_phase_view(request, game_id, player_id, db=db)
# 9. Start Scene (Verify Setup & Shuffle & Draw Obstacles)
@app.post("/game/{game_id}/scene/start")
def start_scene_route(request: Request, game_id: str, db: Session = Depends(get_session)):
success, msg = crud.confirm_scene_setup(db, game_id)
if not success:
# Render the setup partial again with the error message
game = crud.get_game(db, game_id)
# We need a player to render the partial, let's grab any player or the one who requested it.
# But we don't have the player_id in this path. Let's make sure we pass the error message,
# or we just return an HTTP 400 with the error text so HTMX displays it!
return HTMLResponse(f"<div class='alert alert-danger'>{msg}</div>", status_code=400)
return HTMLResponse(status_code=204)
# 11. Pi-Rat plays card against an obstacle
@app.post("/game/{game_id}/player/{player_id}/play-card")
def play_card_route(
game_id: str,
player_id: str,
obstacle_id: str = Form(...),
card_code: str = Form(...),
db: Session = Depends(get_session)
):
ok, msg, res = crud.play_card_on_obstacle(db, player_id, obstacle_id, card_code)
if not ok:
return HTMLResponse(f"<p class='alert alert-danger'>{msg}</p>", status_code=400)
# Return a success notification and HX-Trigger header to refresh the hand.
color_prefix = "🟩" if res["success"] else "🟥"
drew_text = f" (Drew card: {res['drew_card']})" if res["drew_card"] else ""
drew_attr = "true" if res["drew_card"] else "false"
html = (
f"<div class='play-notification' data-drew='{drew_attr}' style='position:fixed;bottom:20px;right:20px;background:#152238;border:1px solid #d4af37;padding:15px;border-radius:8px;z-index:1000;' onclick='this.remove()'>"
f"<p>{color_prefix} {res['details']}{drew_text}</p>"
f"<span style='font-size:0.8em;color:#aaa;'>Click to dismiss</span>"
f"</div>"
)
return HTMLResponse(html, headers={"HX-Trigger": "hand-changed"})
# 12. Pi-Rat plays a Joker to replace an obstacle
@app.post("/game/{game_id}/player/{player_id}/play-joker")
def play_joker_route(
game_id: str,
player_id: str,
card_code: str = Form(...),
obstacle_id: str = Form(...),
db: Session = Depends(get_session)
):
ok, msg, res = crud.play_card_on_obstacle(db, player_id, obstacle_id, card_code)
if not ok:
return HTMLResponse(f"<p class='alert alert-danger'>{msg}</p>", status_code=400)
html = (
f"<div class='play-notification' data-drew='false' style='position:fixed;bottom:20px;right:20px;background:#152238;border:1px solid #d4af37;padding:15px;border-radius:8px;z-index:1000;' onclick='this.remove()'>"
f"<p>🃏 Play Joker: Discarded obstacle and drew a new replacement!</p>"
f"<span style='font-size:0.8em;color:#aaa;'>Click to dismiss</span>"
f"</div>"
)
return HTMLResponse(html, headers={"HX-Trigger": "hand-changed"})
# 13. Toggle Objective Completion Checklist
@app.post("/game/{game_id}/player/{player_id}/objective/toggle")
def toggle_objective_route(
game_id: str,
player_id: str,
type: str, # "personal_1", "personal_2", "personal_3"
db: Session = Depends(get_session)
):
player = crud.get_player(db, player_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
# Toggle logic
current_status = False
if type == "personal_1":
current_status = player.completed_personal_1
elif type == "personal_2":
current_status = player.completed_personal_2
elif type == "personal_3":
current_status = player.completed_personal_3
crud.toggle_objective(db, player_id, type, not current_status)
return HTMLResponse(status_code=204)
# 14. Deep ends the scene
@app.post("/game/{game_id}/scene/end")
def end_scene_route(game_id: str, db: Session = Depends(get_session)):
crud.end_scene_and_transition(db, game_id)
return HTMLResponse(status_code=204)
# 15. Cast vote to Rank Up a player
@app.post("/game/{game_id}/player/{player_id}/submit-vote")
def submit_vote_route(
game_id: str,
player_id: str,
nominated_id: str = Form(...),
db: Session = Depends(get_session)
):
crud.submit_rank_vote(db, game_id, player_id, nominated_id)
return RedirectResponse(url=f"/game/{game_id}/player/{player_id}/view", status_code=status.HTTP_303_SEE_OTHER)
# 16. Player Ready for Next Scene
@app.post("/game/{game_id}/player/{player_id}/ready-next")
def ready_next_route(game_id: str, player_id: str, db: Session = Depends(get_session)):
player = crud.get_player(db, player_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
player.is_ready = True
db.add(player)
db.commit()
# Check if all players in the game are ready. If so, transition to deep upkeep!
game = crud.get_game(db, game_id)
if game and all(p.is_ready for p in game.players):
crud.transition_to_deep_upkeep(db, game_id)
return RedirectResponse(url=f"/game/{game_id}/player/{player_id}/view", status_code=status.HTTP_303_SEE_OTHER)
# 17. Confirm Hand Refresh for Deep player
@app.post("/game/{game_id}/player/{player_id}/confirm-refresh")
def confirm_deep_refresh_route(
game_id: str,
player_id: str,
discard_cards: str = Form("[]"),
db: Session = Depends(get_session)
):
try:
discards = json.loads(discard_cards)
except Exception:
discards = []
crud.confirm_deep_refresh(db, player_id, discards)
return RedirectResponse(url=f"/game/{game_id}/player/{player_id}/view", status_code=status.HTTP_303_SEE_OTHER)
app.include_router(phase_view_router)
app.include_router(lobby_router)
app.include_router(character_router)
app.include_router(scene_router)
app.include_router(upkeep_router)
app.include_router(admin_router)
def main():
import argparse