146 lines
3.9 KiB
Python
146 lines
3.9 KiB
Python
# backend/app/routes/tournaments/report.py
|
|
from typing import List, Optional
|
|
|
|
from fastapi import Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.orm.attributes import flag_modified
|
|
|
|
from ... import crud, logic, models, schemas
|
|
from ...constants import SUCCESS, MatchStatus
|
|
from ...core.auth import get_optional_user
|
|
from ...core.websocket_manager import send_ws_update
|
|
from ...database import get_db
|
|
from . import router
|
|
|
|
|
|
# --- Helper: Centralize Auth Logic ---
|
|
def _check_auth(t: models.Tournament, user: Optional[str], code: Optional[str]):
|
|
is_admin = user is not None
|
|
code_matches = code is not None and str(code).strip() == str(t.code).strip()
|
|
|
|
if not is_admin and not code_matches:
|
|
raise HTTPException(403, "Invalid tournament code or admin privileges required")
|
|
|
|
|
|
@router.post("/{id}/matches/{match_id}/score")
|
|
async def report_score(
|
|
id: str,
|
|
match_id: str,
|
|
report: schemas.ScoreReport,
|
|
db: Session = Depends(get_db),
|
|
user: Optional[str] = Depends(get_optional_user),
|
|
):
|
|
t = crud.get_tournament(db, id)
|
|
if not t:
|
|
raise HTTPException(404, "Tournament not found")
|
|
|
|
_check_auth(t, user, report.code)
|
|
|
|
match = crud.get_match(db, id, match_id)
|
|
if not match:
|
|
raise HTTPException(404, "Match not found")
|
|
|
|
if not report.sets:
|
|
raise HTTPException(400, "No sets submitted")
|
|
|
|
_apply_score(match, report.sets)
|
|
|
|
flag_modified(match, "sets")
|
|
logic.refresh_bracket(t)
|
|
logic.update_schedule(t)
|
|
db.commit()
|
|
|
|
await send_ws_update(id)
|
|
return SUCCESS
|
|
|
|
|
|
@router.patch("/{id}/matches/{match_id}/score")
|
|
async def edit_score(
|
|
id: str,
|
|
match_id: str,
|
|
report: schemas.ScoreReport,
|
|
db: Session = Depends(get_db),
|
|
user: Optional[str] = Depends(get_optional_user),
|
|
):
|
|
"""
|
|
Allows correcting a score without resetting the match status logic entirely,
|
|
or just re-applying the new sets.
|
|
"""
|
|
t = crud.get_tournament(db, id)
|
|
if not t:
|
|
raise HTTPException(404, "Tournament not found")
|
|
|
|
_check_auth(t, user, report.code)
|
|
|
|
match = crud.get_match(db, id, match_id)
|
|
if not match:
|
|
raise HTTPException(404, "Match not found")
|
|
|
|
if report.sets:
|
|
_apply_score(match, report.sets)
|
|
|
|
flag_modified(match, "sets")
|
|
logic.refresh_bracket(t)
|
|
logic.update_schedule(t)
|
|
db.commit()
|
|
|
|
await send_ws_update(id)
|
|
return SUCCESS
|
|
|
|
|
|
@router.delete("/{id}/matches/{match_id}/score")
|
|
async def clear_score(
|
|
id: str,
|
|
match_id: str,
|
|
code: Optional[str] = Query(None),
|
|
db: Session = Depends(get_db),
|
|
user: Optional[str] = Depends(get_optional_user),
|
|
):
|
|
t = crud.get_tournament(db, id)
|
|
if not t:
|
|
raise HTTPException(404, "Tournament not found")
|
|
|
|
_check_auth(t, user, code)
|
|
|
|
match = crud.get_match(db, id, match_id)
|
|
if not match:
|
|
raise HTTPException(404, "Match not found")
|
|
|
|
match.winner = None
|
|
match.status = MatchStatus.PENDING.value
|
|
match.sets = []
|
|
|
|
flag_modified(match, "sets")
|
|
|
|
logic.refresh_bracket(t)
|
|
logic.update_schedule(t)
|
|
db.commit()
|
|
|
|
await send_ws_update(id)
|
|
return SUCCESS
|
|
|
|
|
|
def _apply_score(match: models.Match, sets: List[schemas.SetScore]):
|
|
"""
|
|
Calculates winner based on sets and updates the match object.
|
|
Does NOT commit to DB.
|
|
"""
|
|
p1_wins = sum(1 for s in sets if s.p1 > s.p2)
|
|
p2_wins = sum(1 for s in sets if s.p2 > s.p1)
|
|
|
|
if p1_wins > p2_wins:
|
|
match.winner = match.p1_name
|
|
elif p2_wins > p1_wins:
|
|
match.winner = match.p2_name
|
|
else:
|
|
p1_points = sum(s.p1 for s in sets)
|
|
p2_points = sum(s.p2 for s in sets)
|
|
|
|
if p1_points == p2_points:
|
|
raise HTTPException(400, "Absolute tie: Sets and Points are equal.")
|
|
|
|
match.winner = match.p1_name if p1_points > p2_points else match.p2_name
|
|
|
|
match.status = MatchStatus.FINISHED.value
|
|
match.sets = [s.model_dump() for s in sets]
|