ctfy.sdk.resources.challenges

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