ctfy.sdk.resources.me

  1"""``client.me`` — the calling user's profile, inbox, account + progress."""
  2
  3from __future__ import annotations
  4
  5from typing import Any
  6
  7from ctfy.sdk._helpers import _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    InboxResponse,
 11    MeResponse,
 12    MilestoneProgress,
 13    MyAchievementsResponse,
 14    MySolveSummary,
 15    NotificationPreferencesResponse,
 16    StarGazerVerifyResponse,
 17)
 18
 19
 20class MeResource:
 21    """Self-service surface for the authenticated caller."""
 22
 23    def __init__(self, http: BaseHttpClient) -> None:
 24        self._http = http
 25
 26    def get(self) -> MeResponse:
 27        """Get the calling user's profile + auth context.
 28
 29        Per-competition team memberships ride along on
 30        ``competition_teams``; there is no global "current team" any
 31        more — pick the row whose ``competition_id`` matches the
 32        scope you care about.
 33        """
 34        resp = self._http.request("GET", "/me")
 35        _raise_for_status(resp)
 36        return MeResponse.model_validate(resp.json())
 37
 38    def inbox(self) -> InboxResponse:
 39        """Pending team invites, join requests, direct invites, announcements."""
 40        resp = self._http.request("GET", "/me/inbox")
 41        _raise_for_status(resp)
 42        return InboxResponse.model_validate(resp.json())
 43
 44    def notification_preferences(self) -> NotificationPreferencesResponse:
 45        """Which categories reach the caller by email.
 46
 47        Each cell reports ``effective`` (what happens now), ``default``
 48        (what reverting would restore) and ``explicit`` (whether the user
 49        actually chose it, or is merely inheriting). Storage is sparse, so
 50        an untouched cell has no row — ``explicit`` is how you tell.
 51        """
 52        resp = self._http.request("GET", "/me/notification-preferences")
 53        _raise_for_status(resp)
 54        return NotificationPreferencesResponse.model_validate(resp.json())
 55
 56    def set_notification_preference(
 57        self, category: str, enabled: bool, *, channel: str = "email"
 58    ) -> NotificationPreferencesResponse:
 59        """Turn one category on or off for one channel.
 60
 61        ``security`` is refused with a 400 — sign-in codes and account
 62        alerts exist to protect the account and are not switchable.
 63        """
 64        resp = self._http.request(
 65            "PATCH",
 66            "/me/notification-preferences",
 67            json={"category": category, "channel": channel, "enabled": enabled},
 68        )
 69        _raise_for_status(resp)
 70        return NotificationPreferencesResponse.model_validate(resp.json())
 71
 72    def clear_notification_preference(
 73        self, category: str, *, channel: str = "email"
 74    ) -> NotificationPreferencesResponse:
 75        """Drop the explicit cell so the default applies again.
 76
 77        This is *revert*, not *disable*: for most categories the default
 78        is on, so clearing an explicit ``False`` turns mail back on.
 79        """
 80        resp = self._http.request("DELETE", f"/me/notification-preferences/{category}/{channel}")
 81        _raise_for_status(resp)
 82        return NotificationPreferencesResponse.model_validate(resp.json())
 83
 84    def read_announcement(self, announcement_id: str) -> None:
 85        """Mark one announcement as read for the calling user."""
 86        resp = self._http.request("POST", f"/me/announcements/{announcement_id}/read")
 87        _raise_for_status(resp)
 88
 89    def read_all_announcements(self) -> None:
 90        """Mark every relevant announcement as read."""
 91        resp = self._http.request("POST", "/me/announcements/read-all")
 92        _raise_for_status(resp)
 93
 94    def update_profile(self, **fields: Any) -> MeResponse:
 95        """PATCH the caller's profile.
 96
 97        Pass any subset of ``bio`` / ``country`` / ``website_url`` /
 98        ``timezone`` / ``social_links``. Omit a key to leave it unchanged;
 99        pass ``None`` to clear it. Returns the updated :class:`MeResponse`.
100
101        Example::
102
103            client.me.update_profile(bio="hacking goblins", country="JP")
104            client.me.update_profile(website_url=None)  # clear it
105        """
106        resp = self._http.request("PATCH", "/me/profile", json=fields)
107        _raise_for_status(resp)
108        return MeResponse.model_validate(resp.json())
109
110    def update_visibility(self, overrides: dict[str, bool]) -> MeResponse:
111        """PATCH per-field privacy flags on the caller's profile.
112
113        Keys outside :data:`PROFILE_VISIBILITY_KEYS` are silently dropped
114        server-side. Returns the updated :class:`MeResponse`.
115        """
116        resp = self._http.request("PATCH", "/me/profile/visibility", json=overrides)
117        _raise_for_status(resp)
118        return MeResponse.model_validate(resp.json())
119
120    def export_account(self) -> bytes:
121        """Download a ZIP archive of every piece of data the platform
122        holds about the caller (user, identities, tokens, memberships,
123        submissions, solves, activity). Returns raw bytes — write to a
124        file with ``Path(...).write_bytes(client.me.export_account())``."""
125        resp = self._http.request("GET", "/me/export")
126        _raise_for_status(resp)
127        return resp.content
128
129    def delete_account(self, confirm_display_name: str) -> None:
130        """Self-delete the calling user. ``confirm_display_name`` must
131        match the caller's display name (falls back to email when empty)
132        exactly — typo gate against fat-fingered destruction."""
133        resp = self._http.request(
134            "DELETE",
135            "/me",
136            json={"confirm_display_name": confirm_display_name},
137        )
138        _raise_for_status(resp)
139
140    def achievements(self, competition_id: str = "") -> MyAchievementsResponse:
141        """Caller's unlocked + locked badges (with progress hints)."""
142        params: dict[str, str] = {}
143        if competition_id:
144            params["competition_id"] = competition_id
145        resp = self._http.request("GET", "/me/achievements", params=params)
146        _raise_for_status(resp)
147        return MyAchievementsResponse.model_validate(resp.json())
148
149    def solves(self, *, competition_id: str = "") -> list[MySolveSummary]:
150        """Per-challenge solve summary for the caller.
151
152        Unscoped this spans every team the caller has ever played for.
153        Pass ``competition_id`` to ask the narrower question a
154        competition-scoped surface means: have I solved this *here*.
155        """
156        params: dict[str, str] = {}
157        if competition_id:
158            params["competition_id"] = competition_id
159        resp = self._http.request("GET", "/me/solves", params=params)
160        _raise_for_status(resp)
161        return [MySolveSummary.model_validate(s) for s in resp.json()]
162
163    def milestone_progress(self, *, competition_id: str = "") -> list[MilestoneProgress]:
164        """Partial-solve progress per challenge for the caller.
165
166        One row per challenge with at least one captured question;
167        ``solved_question_ids`` enumerates which milestones the user
168        has cleared, ``total_questions`` is the spec's declared count.
169        With ``competition_id`` set the result narrows to solves
170        stamped against the caller's team in that comp.
171        """
172        params: dict[str, str] = {}
173        if competition_id:
174            params["competition_id"] = competition_id
175        resp = self._http.request("GET", "/me/milestone-progress", params=params)
176        _raise_for_status(resp)
177        return [MilestoneProgress.model_validate(s) for s in resp.json()]
178
179    def verify_star_gazer(self, competition_id: str = "") -> StarGazerVerifyResponse:
180        """Verify the caller starred the ctfy GitHub repo and grant the
181        ``star_gazer`` badge if so.
182
183        Friendly failures (no GitHub identity, repo not configured, not
184        yet starred, GitHub rate-limited) come back as 200 with
185        ``verified=False`` + an actionable ``reason``."""
186        params = {"competition_id": competition_id} if competition_id else None
187        resp = self._http.request("POST", "/me/star-gazer/verify", params=params)
188        _raise_for_status(resp)
189        return StarGazerVerifyResponse.model_validate(resp.json())