ctfy.sdk.competition.team

  1"""``client.competition(id).team`` — my team within one competition.
  2
  3Captain + member management for the caller's team in a single competition:
  4rename / leave / kick, plus the outgoing invite codes (``team.invites``) and
  5the incoming join-request queue (``team.requests``).
  6"""
  7
  8from __future__ import annotations
  9
 10from functools import cached_property
 11
 12from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
 13from ctfy.sdk.base import BaseHttpClient
 14from ctfy.server.models import (
 15    DirectInviteSent,
 16    RiskAppealView,
 17    RiskNotice,
 18    TeamAffiliationSelection,
 19    TeamDetail,
 20    TeamInviteInfo,
 21    TeamMachineLogin,
 22    TeamMachineStatus,
 23    TeamPracticeSummary,
 24)
 25
 26
 27class CompetitionTeam:
 28    """The caller's team in this competition."""
 29
 30    def __init__(self, http: BaseHttpClient, competition_id: str) -> None:
 31        self._http = http
 32        self._cid = competition_id
 33
 34    def rename(self, *, name: str | None = None, description: str | None = None) -> TeamDetail:
 35        """Captain-only: rename / re-describe the team."""
 36        body: dict[str, str] = {}
 37        if name is not None:
 38            body["name"] = name
 39        if description is not None:
 40            body["description"] = description
 41        resp = self._http.request("PATCH", f"/competitions/{self._cid}/team", json=body)
 42        _raise_for_status(resp)
 43        return TeamDetail.model_validate(resp.json())
 44
 45    def affiliation(self) -> TeamAffiliationSelection:
 46        """Captain-only: the squad's institution badge and its options.
 47
 48        Three facts in one answer — what the captain chose, what it
 49        resolves to *now*, and every institution a current member
 50        publishes. The middle one can be ``None`` while the first is
 51        set: that is a choice whose backing has gone, and it reads
 52        differently from having chosen nothing.
 53        """
 54        resp = self._http.request("GET", f"/competitions/{self._cid}/team/affiliation")
 55        _raise_for_status(resp)
 56        return TeamAffiliationSelection.model_validate(resp.json())
 57
 58    def set_affiliation(self, domain: str) -> TeamAffiliationSelection:
 59        """Captain-only: publish ``domain`` beside the squad's name.
 60
 61        Refused (``409 affiliation_not_published``) unless a current
 62        member already publishes that institution on their own profile —
 63        a team cannot claim one nobody on it has shown.
 64        """
 65        resp = self._http.request(
 66            "PUT", f"/competitions/{self._cid}/team/affiliation", json={"domain": domain}
 67        )
 68        _raise_for_status(resp)
 69        return TeamAffiliationSelection.model_validate(resp.json())
 70
 71    def clear_affiliation(self) -> TeamAffiliationSelection:
 72        """Captain-only: stop publishing a badge. Never refused, and
 73        takes no domain — so it also escapes the state where the choice
 74        outlived the member who backed it."""
 75        resp = self._http.request("DELETE", f"/competitions/{self._cid}/team/affiliation")
 76        _raise_for_status(resp)
 77        return TeamAffiliationSelection.model_validate(resp.json())
 78
 79    def machine(self) -> TeamMachineStatus:
 80        """Whether the team has a machine and whether *you* may open it.
 81
 82        Any member may ask; the answer carries no host and no username.
 83        """
 84        resp = self._http.request("GET", f"/competitions/{self._cid}/team/machine")
 85        _raise_for_status(resp)
 86        return TeamMachineStatus.model_validate(resp.json())
 87
 88    def practice(self) -> TeamPracticeSummary:
 89        """What your team answered right in practice here, after a challenge
 90        closed (#2183): ``challenge_count`` and each first right answer.
 91        Never a solve and never on a board; ``404 team_not_found`` without a
 92        team here."""
 93        resp = self._http.request("GET", f"/competitions/{self._cid}/my-team/practice")
 94        _raise_for_status(resp)
 95        return TeamPracticeSummary.model_validate(resp.json())
 96
 97    def view_machine(self) -> TeamMachineLogin:
 98        """Captain-only: open the team's machine login (``403 not_captain``
 99        for anyone else, ``404 machine_not_assigned`` when there is none).
100
101        Each call writes an audit row naming you. The login is a secret:
102        do not log or store the result.
103        """
104        resp = self._http.request("POST", f"/competitions/{self._cid}/team/machine/view")
105        _raise_for_status(resp)
106        return TeamMachineLogin.model_validate(resp.json())
107
108    def risk_notices(self) -> list[RiskNotice]:
109        """What the organisers did about your team here, newest first (#2068),
110        with whether you may appeal each. Empty for a team nothing was done to."""
111        resp = self._http.request("GET", f"/competitions/{self._cid}/risk/notices")
112        _raise_for_status(resp)
113        return [RiskNotice.model_validate(row) for row in resp.json()]
114
115    def appeal(self, action_id: str, body: str) -> RiskAppealView:
116        """Captain-only: appeal an action. One appeal per action, within seven
117        days of it or until the report locks (whichever is earlier). **Plain
118        text only**, at most 2000 characters: no Markdown, no HTML, no
119        attachments, and URLs are not made into links."""
120        resp = self._http.request(
121            "POST",
122            f"/competitions/{self._cid}/risk/actions/{action_id}/appeal",
123            json={"body": body},
124        )
125        _raise_for_status(resp)
126        return RiskAppealView.model_validate(resp.json())
127
128    def supplement_appeal(self, action_id: str, body: str) -> RiskAppealView:
129        """Captain-only: add the one supplement to your appeal, on the same rules."""
130        resp = self._http.request(
131            "POST",
132            f"/competitions/{self._cid}/risk/actions/{action_id}/appeal/supplement",
133            json={"body": body},
134        )
135        _raise_for_status(resp)
136        return RiskAppealView.model_validate(resp.json())
137
138    def leave(self) -> None:
139        """Drop your membership for this competition. If captain and others
140        remain, captaincy transfers to the longest-tenured member; if you
141        were the lone member the team is deleted."""
142        resp = self._http.request("POST", f"/competitions/{self._cid}/team/leave")
143        _raise_for_status(resp)
144
145    def kick(self, user_id: str) -> None:
146        """Captain-only: remove ``user_id`` from the team. Self-kick is
147        rejected (use :meth:`leave` instead)."""
148        resp = self._http.request("DELETE", f"/competitions/{self._cid}/team/members/{user_id}")
149        _raise_for_status(resp)
150
151    @cached_property
152    def invites(self) -> CompetitionInvites:
153        """Invite codes + direct email invites the captain hands out."""
154        return CompetitionInvites(self._http, self._cid)
155
156    @cached_property
157    def requests(self) -> CompetitionRequests:
158        """Pending join requests the captain approves / rejects."""
159        return CompetitionRequests(self._http, self._cid)
160
161
162class CompetitionInvites:
163    """Captain-only: the team's outgoing invites for this competition."""
164
165    def __init__(self, http: BaseHttpClient, competition_id: str) -> None:
166        self._http = http
167        self._cid = competition_id
168
169    def create(self, *, max_uses: int = 1, ttl_seconds: int = 24 * 3600) -> TeamInviteInfo:
170        """Mint an invite code scoped to this comp. ``max_uses=0`` is
171        unlimited; ``ttl_seconds=0`` never expires. Others redeem it via
172        ``client.competition(id).register(mode="join", code=…)``."""
173        resp = self._http.request(
174            "POST",
175            f"/competitions/{self._cid}/invites",
176            json={"max_uses": max_uses, "ttl_seconds": ttl_seconds},
177        )
178        _raise_for_status(resp)
179        return TeamInviteInfo.model_validate(resp.json())
180
181    def list(self) -> PagedList[TeamInviteInfo]:
182        """All outstanding invites for the caller's team in this comp."""
183        resp = self._http.request("GET", f"/competitions/{self._cid}/invites")
184        _raise_for_status(resp)
185        return _extract_items(resp.json(), TeamInviteInfo)
186
187    def revoke(self, invite_id: str) -> None:
188        """Invalidate a previously-minted invite."""
189        resp = self._http.request("DELETE", f"/competitions/{self._cid}/invites/{invite_id}")
190        _raise_for_status(resp)
191
192    def send(self, email: str) -> DirectInviteSent:
193        """Invite one exact email address. The lookup is case-insensitive
194        but otherwise unforgiving (no fuzzy match).
195
196        ⚠️ The answer is the same whether or not the address has an account,
197        is already on a team, or was already invited (#2863): it says the
198        invite was sent *if the address can receive one*. The recipient
199        sees it in their inbox and accepts via
200        :meth:`Competition.accept_invite`."""
201        resp = self._http.request(
202            "POST",
203            f"/competitions/{self._cid}/invites/direct",
204            json={"email": email},
205        )
206        _raise_for_status(resp)
207        return DirectInviteSent.model_validate(resp.json())
208
209
210class CompetitionRequests:
211    """Captain-only: incoming join requests for this competition."""
212
213    def __init__(self, http: BaseHttpClient, competition_id: str) -> None:
214        self._http = http
215        self._cid = competition_id
216
217    def approve(self, invite_id: str) -> TeamDetail:
218        """Approve a pending join request → the requester joins the team."""
219        resp = self._http.request("POST", f"/competitions/{self._cid}/invites/{invite_id}/approve")
220        _raise_for_status(resp)
221        return TeamDetail.model_validate(resp.json())
222
223    def reject(self, invite_id: str) -> None:
224        """Reject a pending join request → invite revoked. The requester
225        can re-open a fresh request afterwards."""
226        resp = self._http.request("POST", f"/competitions/{self._cid}/invites/{invite_id}/reject")
227        _raise_for_status(resp)