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.core.enums import ApplicationPurpose
 26from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
 27from ctfy.sdk.base import BaseHttpClient
 28from ctfy.sdk.competition.team import CompetitionTeam
 29from ctfy.server.models import (
 30    AcceptTeamAgreementRequest,
 31    AdminSolveMatrix,
 32    ChallengeInfo,
 33    CompetitionApplicationInfo,
 34    CompetitionBoardsResponse,
 35    CompetitionChallengeBreakdown,
 36    CompetitionDetail,
 37    CompetitionRoster,
 38    CompetitionScoreDistribution,
 39    CompetitionScoreHistory,
 40    DefenceMatrixResponse,
 41    ScoreboardEntry,
 42    TeamDetail,
 43    TeamInfo,
 44    TeamInviteInfo,
 45    TeamNameCheck,
 46)
 47
 48__all__ = ["Competition"]
 49
 50
 51class Competition:
 52    """A competition-scoped view: registration, your team, the comp's
 53    challenges, team search, and standings — all within one competition."""
 54
 55    def __init__(self, http: BaseHttpClient, competition_id: str) -> None:
 56        self._http = http
 57        self._cid = competition_id
 58
 59    @property
 60    def id(self) -> str:
 61        """The competition id this handle is bound to."""
 62        return self._cid
 63
 64    # -- info -------------------------------------------------------------
 65
 66    def detail(self) -> CompetitionDetail:
 67        """Full competition detail including resolved challenge summaries
 68        (``.challenges``) in the admin's curated order."""
 69        resp = self._http.request("GET", f"/competitions/{self._cid}")
 70        _raise_for_status(resp)
 71        return CompetitionDetail.model_validate(resp.json())
 72
 73    def challenges(
 74        self,
 75        difficulty: str | None = None,
 76        tag: str = "",
 77        q: str = "",
 78        offset: int = 0,
 79        limit: int = 50,
 80    ) -> PagedList[ChallengeInfo]:
 81        """Challenges scoped to this competition, curated-order sorted."""
 82        params: dict[str, Any] = {"offset": offset, "limit": limit}
 83        if difficulty:
 84            params["difficulty"] = difficulty
 85        if tag:
 86            params["tag"] = tag
 87        if q:
 88            params["q"] = q
 89        resp = self._http.request("GET", f"/competitions/{self._cid}/challenges", params=params)
 90        _raise_for_status(resp)
 91        return _extract_items(resp.json(), ChallengeInfo)
 92
 93    def search_teams(self, q: str = "", offset: int = 0, limit: int = 50) -> PagedList[TeamInfo]:
 94        """Substring search over team names registered in this competition
 95        (the join-a-team picker)."""
 96        params: dict[str, Any] = {"offset": offset, "limit": limit}
 97        if q:
 98            params["q"] = q
 99        resp = self._http.request("GET", f"/competitions/{self._cid}/teams/search", params=params)
100        _raise_for_status(resp)
101        return _extract_items(resp.json(), TeamInfo)
102
103    # -- register / join --------------------------------------------------
104
105    def register(
106        self,
107        mode: Literal["solo", "create", "join"] = "solo",
108        *,
109        name: str = "",
110        description: str = "",
111        code: str = "",
112        agreement: AcceptTeamAgreementRequest | None = None,
113    ) -> TeamDetail:
114        """Get on a team for this competition.
115
116        ``agreement`` is the contributor agreement, accepted for the new team
117        by you — its captain (#1965 B). A competition that takes uploads and
118        names an agreement (``authoring_terms_version`` on its detail)
119        refuses a ``solo`` or ``create`` registration without it (422
120        ``registration_agreement_required``). Only a person's own session can
121        send it: a fine-grained token is refused 403
122        ``registration_agreement_needs_a_person``. ``join`` takes none — the
123        team's captain already accepted.
124
125        - ``mode="solo"`` — mint a 1-person team (idempotent on (you,
126          competition)). Named ``name`` when given — the entry name an
127          individual event shows on its board — else after your display name.
128        - ``mode="create"`` — mint a multi-user team you captain; requires
129          ``name``. 409 if you already have a team for this comp.
130        - ``mode="join"`` — redeem a captain's invite ``code`` to join their
131          team. 409 if the code was for a different comp or you already have
132          a team here.
133        """
134        if mode == "join":
135            if not code:
136                raise ValueError('register(mode="join") requires a `code`')
137            resp = self._http.request(
138                "POST", f"/competitions/{self._cid}/teams/redeem", json={"code": code}
139            )
140        else:
141            if mode == "create" and not name:
142                raise ValueError('register(mode="create") requires a `name`')
143            body: dict[str, object] = {"mode": mode}
144            if name:
145                body["name"] = name
146            if mode == "create":
147                body["description"] = description
148            if agreement is not None:
149                body["agreement"] = agreement.model_dump(mode="json")
150            resp = self._http.request("POST", f"/competitions/{self._cid}/teams", json=body)
151        _raise_for_status(resp)
152        return TeamDetail.model_validate(resp.json())
153
154    def check_team_name(self, name: str) -> TeamNameCheck:
155        """May ``name`` be a team's name in this competition? (#2238)
156
157        Asks the rule every registration and rename applies, without writing:
158        ``name`` is what would be stored, ``available`` the verdict, and
159        ``refusal`` a coded reason (``team_name_has_link`` / ``_reserved`` /
160        ``_blocked`` / ``_too_long`` / ``_too_short`` / ``_taken``) or ``""``.
161        Advisory -- the registration itself decides, since a free name can be
162        taken before you submit.
163        """
164        resp = self._http.request(
165            "GET", f"/competitions/{self._cid}/team-name-check", params={"name": name}
166        )
167        _raise_for_status(resp)
168        return TeamNameCheck.model_validate(resp.json())
169
170    def request_to_join(self, team_id: str) -> TeamInviteInfo:
171        """Ask to join ``team_id`` in this competition. The captain sees it
172        in their inbox and approves / rejects (``team.requests``)."""
173        resp = self._http.request("POST", f"/competitions/{self._cid}/teams/{team_id}/join-request")
174        _raise_for_status(resp)
175        return TeamInviteInfo.model_validate(resp.json())
176
177    def accept_invite(self, invite_id: str) -> TeamDetail:
178        """Accept a direct invite sent to you → join the captain's team."""
179        resp = self._http.request("POST", f"/competitions/{self._cid}/invites/{invite_id}/accept")
180        _raise_for_status(resp)
181        return TeamDetail.model_validate(resp.json())
182
183    def decline_invite(self, invite_id: str) -> None:
184        """Decline a direct invite sent to you (only the named target may)."""
185        resp = self._http.request("POST", f"/competitions/{self._cid}/invites/{invite_id}/decline")
186        _raise_for_status(resp)
187
188    # -- asking to be let in (a private, listed competition) ---------------
189
190    def application(self) -> CompetitionApplicationInfo | None:
191        """Your application to this competition, or ``None`` if you have
192        not applied.
193
194        ``None`` is the server's ``null`` at 200 — a hidden competition
195        already answers 404, so "never applied" is not an error.
196        """
197        resp = self._http.request("GET", f"/competitions/{self._cid}/application")
198        _raise_for_status(resp)
199        body = resp.json()
200        return None if body is None else CompetitionApplicationInfo.model_validate(body)
201
202    def apply(
203        self,
204        message: str = "",
205        *,
206        organization: str,
207        purpose: ApplicationPurpose,
208        contact_email: str,
209        purpose_detail: str = "",
210    ) -> CompetitionApplicationInfo:
211        """Ask the organisers of a private competition or suite to invite you.
212
213        For an event or suite that is listed but invitation-only. The form
214        asks who is applying and why (#2306): ``organization`` and
215        ``contact_email`` are required, ``purpose`` is ``vendor_evaluation`` /
216        ``academic_research`` / ``teaching`` / ``other`` (``other`` needs
217        ``purpose_detail``: ``application_purpose_detail_required``). Calling
218        it again while the application is pending replaces the answers and
219        keeps its place in the queue. An approval arrives as an invitation,
220        after which :meth:`register` works -- and a suite's challenges open --
221        as for anybody invited.
222
223        Refused (409) with a ``code`` saying why: ``application_not_needed``
224        (public — just register), ``already_invited`` (you may already take
225        part), ``registration_closed``, ``application_rejected`` (final;
226        the organisers can still invite you). Not a team join request —
227        that is :meth:`request_to_join`, and asks a captain.
228        """
229        resp = self._http.request(
230            "PUT",
231            f"/competitions/{self._cid}/application",
232            json={
233                "message": message,
234                "organization": organization,
235                "purpose": purpose,
236                "purpose_detail": purpose_detail,
237                "contact_email": contact_email,
238            },
239        )
240        _raise_for_status(resp)
241        return CompetitionApplicationInfo.model_validate(resp.json())
242
243    def withdraw_application(self) -> None:
244        """Withdraw a pending application. A decided one stays, and asking
245        is ``application_already_decided`` (409)."""
246        resp = self._http.request("DELETE", f"/competitions/{self._cid}/application")
247        _raise_for_status(resp)
248
249    # -- standings --------------------------------------------------------
250
251    def scoreboard(
252        self, offset: int = 0, limit: int = 50, division: str = "", board: str = ""
253    ) -> PagedList[ScoreboardEntry]:
254        """Ranked standings for this competition.
255
256        ``board`` names which of the competition's boards (#1965 G) — a
257        slug :meth:`boards` lists. Empty (or ``"main"``) is the primary,
258        the competition's own rule, which is the board this method has
259        always returned. An unknown slug, or a shadow board for anybody
260        but an organiser, raises ``404 competition_board_not_found``.
261
262        ``division`` narrows to one side of a mixed board: ``"human"``
263        or ``"agent"``. Empty — the default — is everybody, which is the
264        board this method has always returned. ⚠️ Split here rather than
265        by filtering the result: the rows are paginated server-side, so a
266        client-side filter yields whichever of one page matched, still
267        carrying their *merged* ranks.
268        """
269        params: dict[str, Any] = {"offset": offset, "limit": limit}
270        if division:
271            params["division"] = division
272        if board:
273            params["board"] = board
274        resp = self._http.request(
275            "GET",
276            f"/competitions/{self._cid}/scoreboard",
277            params=params,
278        )
279        _raise_for_status(resp)
280        return _extract_items(resp.json(), ScoreboardEntry)
281
282    def boards(self) -> CompetitionBoardsResponse:
283        """This competition's boards (#1965 G): the primary first
284        (``primary=True``, slug ``main``), then each published extra — a
285        separate standing under its own rule, e.g. an authoring board beside
286        the solving one. An organiser also receives the shadow boards.
287        Pass a ``slug`` to :meth:`scoreboard` to read one."""
288        resp = self._http.request("GET", f"/competitions/{self._cid}/boards")
289        _raise_for_status(resp)
290        return CompetitionBoardsResponse.model_validate(resp.json())
291
292    def score_history(self, *, top: int = 10) -> CompetitionScoreHistory:
293        """Score-progression series for the top ``top`` teams."""
294        resp = self._http.request(
295            "GET", f"/competitions/{self._cid}/score-history", params={"top": top}
296        )
297        _raise_for_status(resp)
298        return CompetitionScoreHistory.model_validate(resp.json())
299
300    def score_distribution(self) -> CompetitionScoreDistribution:
301        """Score histogram buckets for this competition."""
302        resp = self._http.request("GET", f"/competitions/{self._cid}/scoreboard/distribution")
303        _raise_for_status(resp)
304        return CompetitionScoreDistribution.model_validate(resp.json())
305
306    def solve_matrix(self) -> AdminSolveMatrix:
307        """Per-team × per-challenge solve grid scoped to this competition."""
308        resp = self._http.request("GET", f"/competitions/{self._cid}/solve-matrix")
309        _raise_for_status(resp)
310        return AdminSolveMatrix.model_validate(resp.json())
311
312    def defence_matrix(self) -> DefenceMatrixResponse:
313        """Per-team × per-challenge **defence** verdicts for this competition.
314
315        The defence-side twin of :meth:`solve_matrix`, and deliberately a
316        separate call rather than a flag on it: a solve cell is binary
317        and monotone, a defence cell is one of five verdicts, and three
318        of them describe a team that tried and is not done.
319
320        A team that fixed a challenge and later submitted a worse patch
321        still reads ``fixed`` — that is what the board scores them on, so
322        this and the standings cannot disagree about who is holding what.
323        A pair with no submission is **absent** from ``cells``, never a
324        present empty string.
325        """
326        resp = self._http.request("GET", f"/competitions/{self._cid}/defence/matrix")
327        _raise_for_status(resp)
328        return DefenceMatrixResponse.model_validate(resp.json())
329
330    def roster(self) -> CompetitionRoster:
331        """Every registered team in this competition with its members.
332
333        One request whatever the field size — the per-team
334        :meth:`ctfy.sdk.resources.teams.TeamsResource.get` answers the
335        same question one team at a time, which for a 400-team event is
336        400 round trips. Carries public identity only (display name +
337        avatar), never the entrant dossier.
338        """
339        resp = self._http.request("GET", f"/competitions/{self._cid}/roster")
340        _raise_for_status(resp)
341        return CompetitionRoster.model_validate(resp.json())
342
343    def challenge_breakdown(self) -> CompetitionChallengeBreakdown:
344        """Per-challenge solve / attempt counts for this competition."""
345        resp = self._http.request("GET", f"/competitions/{self._cid}/challenge-breakdown")
346        _raise_for_status(resp)
347        return CompetitionChallengeBreakdown.model_validate(resp.json())
348
349    # -- my team ----------------------------------------------------------
350
351    @cached_property
352    def team(self) -> CompetitionTeam:
353        """My team in this competition (rename / leave / kick + invites + requests)."""
354        return CompetitionTeam(self._http, self._cid)