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.core.attachment_layout import attachment_url_path
  8from ctfy.sdk._helpers import PagedList, _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    ) -> PagedList[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(
 81            "GET", f"/challenges/{challenge_id}/attachments/{attachment_url_path(filename)}"
 82        )
 83        _raise_for_status(resp)
 84        return resp.content
 85
 86    def solve_attempts(
 87        self,
 88        challenge_id: str,
 89        *,
 90        include_unsolved: bool = False,
 91        limit: int = 500,
 92        competition_id: str = "",
 93    ) -> ChallengeSolveAttemptsResponse:
 94        """Per-team archived-instance breakdown for one challenge
 95        (attempts, solve time, solved flag count).
 96
 97        ``competition_id`` restricts the rows to that competition's
 98        teams; without it the breakdown spans every event the challenge
 99        has ever appeared in.
100        """
101        params: dict[str, Any] = {"include_unsolved": include_unsolved, "limit": limit}
102        if competition_id:
103            params["competition_id"] = competition_id
104        resp = self._http.request(
105            "GET",
106            f"/challenges/{challenge_id}/solve-attempts",
107            params=params,
108        )
109        _raise_for_status(resp)
110        return ChallengeSolveAttemptsResponse.model_validate(resp.json())
111
112    def facets(self, *, competition_id: str = "") -> ChallengeFacets:
113        """Available difficulty / tag facets (with counts) for the
114        challenge browser. Scope to one competition's catalog by
115        passing ``competition_id``."""
116        params = {"competition_id": competition_id} if competition_id else {}
117        resp = self._http.request("GET", "/challenges/facets", params=params)
118        _raise_for_status(resp)
119        return ChallengeFacets.model_validate(resp.json())
120
121    def feedback_stats(self, challenge_id: str) -> FeedbackStats:
122        """Dense per-reaction count aggregate for one challenge."""
123        resp = self._http.request("GET", f"/challenges/{challenge_id}/feedback/stats")
124        _raise_for_status(resp)
125        return FeedbackStats.model_validate(resp.json())
126
127    def my_reactions(self, challenge_id: str) -> MyReactionsResponse:
128        """The calling user's active reactions on a challenge (empty if unset)."""
129        resp = self._http.request("GET", f"/challenges/{challenge_id}/feedback/me")
130        _raise_for_status(resp)
131        return MyReactionsResponse.model_validate(resp.json())
132
133    def add_reaction(self, challenge_id: str, reaction: str) -> MyReactionsResponse:
134        """Add one reaction chip to the user's set for this challenge.
135
136        Multi-select: stacks alongside any other reactions the user
137        already has. Idempotent — calling twice with the same triple
138        is a no-op. ``reaction`` must be one of the nine values
139        declared in :data:`ctfy.core.state.models.REACTIONS`.
140
141        Returns the user's full active reaction set after the write.
142        """
143        resp = self._http.request("PUT", f"/challenges/{challenge_id}/feedback/{reaction}")
144        _raise_for_status(resp)
145        return MyReactionsResponse.model_validate(resp.json())
146
147    def remove_reaction(self, challenge_id: str, reaction: str) -> None:
148        """Remove one reaction chip from the user's set (204, idempotent)."""
149        resp = self._http.request("DELETE", f"/challenges/{challenge_id}/feedback/{reaction}")
150        _raise_for_status(resp)
151
152    def clear_reactions(self, challenge_id: str) -> None:
153        """Clear every active reaction the user has on this challenge (204)."""
154        resp = self._http.request("DELETE", f"/challenges/{challenge_id}/feedback")
155        _raise_for_status(resp)