ctfy.sdk.resources.challenges

  1"""``client.challenges`` — global catalog, attachments, facets, feedback chips."""
  2
  3from __future__ import annotations
  4
  5import builtins
  6from typing import Any
  7
  8from ctfy.sdk._helpers import _extract_items, _raise_for_status
  9from ctfy.sdk.base import BaseHttpClient
 10from ctfy.server.models import (
 11    AttachmentList,
 12    ChallengeFacets,
 13    ChallengeInfo,
 14    ChallengeSolveAttemptsResponse,
 15    FeedbackStats,
 16    MyReactionsResponse,
 17)
 18
 19
 20class ChallengesResource:
 21    """Global challenge catalog + per-challenge attachments and solve feedback."""
 22
 23    def __init__(self, http: BaseHttpClient) -> None:
 24        self._http = http
 25
 26    def list(
 27        self,
 28        difficulty: str | None = None,
 29        tag: str = "",
 30        category: str = "",
 31        q: str = "",
 32        offset: int = 0,
 33        limit: int = 50,
 34    ) -> builtins.list[ChallengeInfo]:
 35        params: dict[str, Any] = {"offset": offset, "limit": limit}
 36        if difficulty:
 37            params["difficulty"] = difficulty
 38        if tag:
 39            params["tag"] = tag
 40        if category:
 41            params["category"] = category
 42        if q:
 43            params["q"] = q
 44        resp = self._http.request("GET", "/challenges", params=params)
 45        _raise_for_status(resp)
 46        return _extract_items(resp.json(), ChallengeInfo)
 47
 48    def get(self, challenge_id: str, *, competition_id: str = "") -> ChallengeInfo:
 49        """One challenge's catalog row.
 50
 51        ``competition_id`` scopes the tallies to that competition's
 52        teams; without it they span every event on the platform.
 53        """
 54        params: dict[str, Any] = {}
 55        if competition_id:
 56            params["competition_id"] = competition_id
 57        resp = self._http.request("GET", f"/challenges/{challenge_id}", params=params)
 58        _raise_for_status(resp)
 59        return ChallengeInfo.model_validate(resp.json())
 60
 61    def attachments(self, challenge_id: str) -> AttachmentList:
 62        """List the files shipped under the challenge's ``attachments/`` dir.
 63
 64        Same data as :attr:`ChallengeInfo.attachments` but addressable
 65        directly. Empty list when the challenge ships nothing.
 66        """
 67        resp = self._http.request("GET", f"/challenges/{challenge_id}/attachments")
 68        _raise_for_status(resp)
 69        return AttachmentList.model_validate(resp.json())
 70
 71    def download_attachment(self, challenge_id: str, filename: str) -> bytes:
 72        """Download one attachment, returning the raw bytes.
 73
 74        ``filename`` is the relative path under the challenge's
 75        ``attachments/`` directory (the name from
 76        :meth:`attachments`). Streaming for very large files lives on
 77        the underlying httpx client; this wrapper materialises the
 78        whole body so the common case stays simple.
 79        """
 80        resp = self._http.request("GET", f"/challenges/{challenge_id}/attachments/{filename}")
 81        _raise_for_status(resp)
 82        return resp.content
 83
 84    def solve_attempts(
 85        self,
 86        challenge_id: str,
 87        *,
 88        include_unsolved: bool = False,
 89        limit: int = 500,
 90        competition_id: str = "",
 91    ) -> ChallengeSolveAttemptsResponse:
 92        """Per-team archived-instance breakdown for one challenge
 93        (attempts, solve time, solved flag count).
 94
 95        ``competition_id`` restricts the rows to that competition's
 96        teams; without it the breakdown spans every event the challenge
 97        has ever appeared in.
 98        """
 99        params: dict[str, Any] = {"include_unsolved": include_unsolved, "limit": limit}
100        if competition_id:
101            params["competition_id"] = competition_id
102        resp = self._http.request(
103            "GET",
104            f"/challenges/{challenge_id}/solve-attempts",
105            params=params,
106        )
107        _raise_for_status(resp)
108        return ChallengeSolveAttemptsResponse.model_validate(resp.json())
109
110    def facets(self, *, competition_id: str = "") -> ChallengeFacets:
111        """Available difficulty / tag facets (with counts) for the
112        challenge browser. Scope to one competition's catalog by
113        passing ``competition_id``."""
114        params = {"competition_id": competition_id} if competition_id else {}
115        resp = self._http.request("GET", "/challenges/facets", params=params)
116        _raise_for_status(resp)
117        return ChallengeFacets.model_validate(resp.json())
118
119    def feedback_stats(self, challenge_id: str) -> FeedbackStats:
120        """Dense per-reaction count aggregate for one challenge."""
121        resp = self._http.request("GET", f"/challenges/{challenge_id}/feedback/stats")
122        _raise_for_status(resp)
123        return FeedbackStats.model_validate(resp.json())
124
125    def my_reactions(self, challenge_id: str) -> MyReactionsResponse:
126        """The calling user's active reactions on a challenge (empty if unset)."""
127        resp = self._http.request("GET", f"/challenges/{challenge_id}/feedback/me")
128        _raise_for_status(resp)
129        return MyReactionsResponse.model_validate(resp.json())
130
131    def add_reaction(self, challenge_id: str, reaction: str) -> MyReactionsResponse:
132        """Add one reaction chip to the user's set for this challenge.
133
134        Multi-select: stacks alongside any other reactions the user
135        already has. Idempotent — calling twice with the same triple
136        is a no-op. ``reaction`` must be one of the nine values
137        declared in :data:`ctfy.core.state.models.REACTIONS`.
138
139        Returns the user's full active reaction set after the write.
140        """
141        resp = self._http.request("PUT", f"/challenges/{challenge_id}/feedback/{reaction}")
142        _raise_for_status(resp)
143        return MyReactionsResponse.model_validate(resp.json())
144
145    def remove_reaction(self, challenge_id: str, reaction: str) -> None:
146        """Remove one reaction chip from the user's set (204, idempotent)."""
147        resp = self._http.request("DELETE", f"/challenges/{challenge_id}/feedback/{reaction}")
148        _raise_for_status(resp)
149
150    def clear_reactions(self, challenge_id: str) -> None:
151        """Clear every active reaction the user has on this challenge (204)."""
152        resp = self._http.request("DELETE", f"/challenges/{challenge_id}/feedback")
153        _raise_for_status(resp)