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) -> ChallengeInfo:
 49        resp = self._http.request("GET", f"/challenges/{challenge_id}")
 50        _raise_for_status(resp)
 51        return ChallengeInfo.model_validate(resp.json())
 52
 53    def attachments(self, challenge_id: str) -> AttachmentList:
 54        """List the files shipped under the challenge's ``attachments/`` dir.
 55
 56        Same data as :attr:`ChallengeInfo.attachments` but addressable
 57        directly. Empty list when the challenge ships nothing.
 58        """
 59        resp = self._http.request("GET", f"/challenges/{challenge_id}/attachments")
 60        _raise_for_status(resp)
 61        return AttachmentList.model_validate(resp.json())
 62
 63    def download_attachment(self, challenge_id: str, filename: str) -> bytes:
 64        """Download one attachment, returning the raw bytes.
 65
 66        ``filename`` is the relative path under the challenge's
 67        ``attachments/`` directory (the name from
 68        :meth:`attachments`). Streaming for very large files lives on
 69        the underlying httpx client; this wrapper materialises the
 70        whole body so the common case stays simple.
 71        """
 72        resp = self._http.request("GET", f"/challenges/{challenge_id}/attachments/{filename}")
 73        _raise_for_status(resp)
 74        return resp.content
 75
 76    def solve_attempts(
 77        self,
 78        challenge_id: str,
 79        *,
 80        include_unsolved: bool = False,
 81        limit: int = 500,
 82    ) -> ChallengeSolveAttemptsResponse:
 83        """Per-team archived-instance breakdown for one challenge
 84        (attempts, solve time, solved flag count)."""
 85        resp = self._http.request(
 86            "GET",
 87            f"/challenges/{challenge_id}/solve-attempts",
 88            params={"include_unsolved": include_unsolved, "limit": limit},
 89        )
 90        _raise_for_status(resp)
 91        return ChallengeSolveAttemptsResponse.model_validate(resp.json())
 92
 93    def facets(self, *, competition_id: str = "") -> ChallengeFacets:
 94        """Available difficulty / tag facets (with counts) for the
 95        challenge browser. Scope to one competition's catalog by
 96        passing ``competition_id``."""
 97        params = {"competition_id": competition_id} if competition_id else {}
 98        resp = self._http.request("GET", "/challenges/facets", params=params)
 99        _raise_for_status(resp)
100        return ChallengeFacets.model_validate(resp.json())
101
102    def feedback_stats(self, challenge_id: str) -> FeedbackStats:
103        """Dense per-reaction count aggregate for one challenge."""
104        resp = self._http.request("GET", f"/challenges/{challenge_id}/feedback/stats")
105        _raise_for_status(resp)
106        return FeedbackStats.model_validate(resp.json())
107
108    def my_reactions(self, challenge_id: str) -> MyReactionsResponse:
109        """The calling user's active reactions on a challenge (empty if unset)."""
110        resp = self._http.request("GET", f"/challenges/{challenge_id}/feedback/me")
111        _raise_for_status(resp)
112        return MyReactionsResponse.model_validate(resp.json())
113
114    def add_reaction(self, challenge_id: str, reaction: str) -> MyReactionsResponse:
115        """Add one reaction chip to the user's set for this challenge.
116
117        Multi-select: stacks alongside any other reactions the user
118        already has. Idempotent — calling twice with the same triple
119        is a no-op. ``reaction`` must be one of the nine values
120        declared in :data:`ctfy.core.state.models.REACTIONS`.
121
122        Returns the user's full active reaction set after the write.
123        """
124        resp = self._http.request("PUT", f"/challenges/{challenge_id}/feedback/{reaction}")
125        _raise_for_status(resp)
126        return MyReactionsResponse.model_validate(resp.json())
127
128    def remove_reaction(self, challenge_id: str, reaction: str) -> None:
129        """Remove one reaction chip from the user's set (204, idempotent)."""
130        resp = self._http.request("DELETE", f"/challenges/{challenge_id}/feedback/{reaction}")
131        _raise_for_status(resp)
132
133    def clear_reactions(self, challenge_id: str) -> None:
134        """Clear every active reaction the user has on this challenge (204)."""
135        resp = self._http.request("DELETE", f"/challenges/{challenge_id}/feedback")
136        _raise_for_status(resp)