ctfy.sdk.competition

  1"""``client.competition(id)`` — a handle scoped to one competition.
  2
  3Collects the operations that are *inherently* competition-scoped — the ones
  4that used to repeat ``competition_id`` on every call: registration, your team
  5(captain tooling, invites, join-requests), the competition's challenge list,
  6team search, and standings. ``competition_id`` is named once
  7(``client.competition(id)``).
  8
  9Discovery of competitions stays on ``client.competitions`` (``.list()`` /
 10``.get(id)``). Instance lifecycle (``client.instances``) and answer submission
 11(``client.submissions``) stay flat: they're keyed by ``instance_id`` and the
 12server infers the competition, so they don't belong to a comp handle.
 13
 14    comp = client.competition("spring-ctf")
 15    comp.register(mode="solo")
 16    board = comp.scoreboard()
 17    comp.team.invites.create(max_uses=3)
 18"""
 19
 20from __future__ import annotations
 21
 22from functools import cached_property
 23from typing import Any, Literal
 24
 25from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
 26from ctfy.sdk.base import BaseHttpClient
 27from ctfy.sdk.competition.team import CompetitionTeam
 28from ctfy.server.models import (
 29    AdminSolveMatrix,
 30    ChallengeInfo,
 31    CompetitionChallengeBreakdown,
 32    CompetitionDetail,
 33    CompetitionRoster,
 34    CompetitionScoreDistribution,
 35    CompetitionScoreHistory,
 36    ScoreboardEntry,
 37    TeamDetail,
 38    TeamInfo,
 39    TeamInviteInfo,
 40)
 41
 42__all__ = ["Competition"]
 43
 44
 45class Competition:
 46    """A competition-scoped view: registration, your team, the comp's
 47    challenges, team search, and standings — all within one competition."""
 48
 49    def __init__(self, http: BaseHttpClient, competition_id: str) -> None:
 50        self._http = http
 51        self._cid = competition_id
 52
 53    @property
 54    def id(self) -> str:
 55        """The competition id this handle is bound to."""
 56        return self._cid
 57
 58    # -- info -------------------------------------------------------------
 59
 60    def detail(self) -> CompetitionDetail:
 61        """Full competition detail including resolved challenge summaries
 62        (``.challenges``) in the admin's curated order."""
 63        resp = self._http.request("GET", f"/competitions/{self._cid}")
 64        _raise_for_status(resp)
 65        return CompetitionDetail.model_validate(resp.json())
 66
 67    def challenges(
 68        self,
 69        difficulty: str | None = None,
 70        tag: str = "",
 71        q: str = "",
 72        offset: int = 0,
 73        limit: int = 50,
 74    ) -> PagedList[ChallengeInfo]:
 75        """Challenges scoped to this competition, curated-order sorted."""
 76        params: dict[str, Any] = {"offset": offset, "limit": limit}
 77        if difficulty:
 78            params["difficulty"] = difficulty
 79        if tag:
 80            params["tag"] = tag
 81        if q:
 82            params["q"] = q
 83        resp = self._http.request("GET", f"/competitions/{self._cid}/challenges", params=params)
 84        _raise_for_status(resp)
 85        return _extract_items(resp.json(), ChallengeInfo)
 86
 87    def search_teams(self, q: str = "", offset: int = 0, limit: int = 50) -> PagedList[TeamInfo]:
 88        """Substring search over team names registered in this competition
 89        (the join-a-team picker)."""
 90        params: dict[str, Any] = {"offset": offset, "limit": limit}
 91        if q:
 92            params["q"] = q
 93        resp = self._http.request("GET", f"/competitions/{self._cid}/teams/search", params=params)
 94        _raise_for_status(resp)
 95        return _extract_items(resp.json(), TeamInfo)
 96
 97    # -- register / join --------------------------------------------------
 98
 99    def register(
100        self,
101        mode: Literal["solo", "create", "join"] = "solo",
102        *,
103        name: str = "",
104        description: str = "",
105        code: str = "",
106    ) -> TeamDetail:
107        """Get on a team for this competition.
108
109        - ``mode="solo"`` — mint a 1-person team auto-named after you
110          (idempotent on (you, competition)).
111        - ``mode="create"`` — mint a multi-user team you captain; requires
112          ``name``. 409 if you already have a team for this comp.
113        - ``mode="join"`` — redeem a captain's invite ``code`` to join their
114          team. 409 if the code was for a different comp or you already have
115          a team here.
116        """
117        if mode == "join":
118            if not code:
119                raise ValueError('register(mode="join") requires a `code`')
120            resp = self._http.request(
121                "POST", f"/competitions/{self._cid}/teams/redeem", json={"code": code}
122            )
123        else:
124            if mode == "create" and not name:
125                raise ValueError('register(mode="create") requires a `name`')
126            body: dict[str, str] = {"mode": mode}
127            if mode == "create":
128                body["name"] = name
129                body["description"] = description
130            resp = self._http.request("POST", f"/competitions/{self._cid}/teams", json=body)
131        _raise_for_status(resp)
132        return TeamDetail.model_validate(resp.json())
133
134    def request_to_join(self, team_id: str) -> TeamInviteInfo:
135        """Ask to join ``team_id`` in this competition. The captain sees it
136        in their inbox and approves / rejects (``team.requests``)."""
137        resp = self._http.request("POST", f"/competitions/{self._cid}/teams/{team_id}/join-request")
138        _raise_for_status(resp)
139        return TeamInviteInfo.model_validate(resp.json())
140
141    def accept_invite(self, invite_id: str) -> TeamDetail:
142        """Accept a direct invite sent to you → join the captain's team."""
143        resp = self._http.request("POST", f"/competitions/{self._cid}/invites/{invite_id}/accept")
144        _raise_for_status(resp)
145        return TeamDetail.model_validate(resp.json())
146
147    def decline_invite(self, invite_id: str) -> None:
148        """Decline a direct invite sent to you (only the named target may)."""
149        resp = self._http.request("POST", f"/competitions/{self._cid}/invites/{invite_id}/decline")
150        _raise_for_status(resp)
151
152    # -- standings --------------------------------------------------------
153
154    def scoreboard(self, offset: int = 0, limit: int = 50) -> PagedList[ScoreboardEntry]:
155        """Ranked standings for this competition."""
156        resp = self._http.request(
157            "GET",
158            f"/competitions/{self._cid}/scoreboard",
159            params={"offset": offset, "limit": limit},
160        )
161        _raise_for_status(resp)
162        return _extract_items(resp.json(), ScoreboardEntry)
163
164    def score_history(self, *, top: int = 10) -> CompetitionScoreHistory:
165        """Score-progression series for the top ``top`` teams."""
166        resp = self._http.request(
167            "GET", f"/competitions/{self._cid}/score-history", params={"top": top}
168        )
169        _raise_for_status(resp)
170        return CompetitionScoreHistory.model_validate(resp.json())
171
172    def score_distribution(self) -> CompetitionScoreDistribution:
173        """Score histogram buckets for this competition."""
174        resp = self._http.request("GET", f"/competitions/{self._cid}/scoreboard/distribution")
175        _raise_for_status(resp)
176        return CompetitionScoreDistribution.model_validate(resp.json())
177
178    def solve_matrix(self) -> AdminSolveMatrix:
179        """Per-team × per-challenge solve grid scoped to this competition."""
180        resp = self._http.request("GET", f"/competitions/{self._cid}/solve-matrix")
181        _raise_for_status(resp)
182        return AdminSolveMatrix.model_validate(resp.json())
183
184    def roster(self) -> CompetitionRoster:
185        """Every registered team in this competition with its members.
186
187        One request whatever the field size — the per-team
188        :meth:`ctfy.sdk.resources.teams.TeamsResource.get` answers the
189        same question one team at a time, which for a 400-team event is
190        400 round trips. Carries public identity only (display name +
191        avatar), never the entrant dossier.
192        """
193        resp = self._http.request("GET", f"/competitions/{self._cid}/roster")
194        _raise_for_status(resp)
195        return CompetitionRoster.model_validate(resp.json())
196
197    def challenge_breakdown(self) -> CompetitionChallengeBreakdown:
198        """Per-challenge solve / attempt counts for this competition."""
199        resp = self._http.request("GET", f"/competitions/{self._cid}/challenge-breakdown")
200        _raise_for_status(resp)
201        return CompetitionChallengeBreakdown.model_validate(resp.json())
202
203    # -- my team ----------------------------------------------------------
204
205    @cached_property
206    def team(self) -> CompetitionTeam:
207        """My team in this competition (rename / leave / kick + invites + requests)."""
208        return CompetitionTeam(self._http, self._cid)