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    StarGazerVerifyResponse,
 16)
 17
 18
 19class MeResource:
 20    """Self-service surface for the authenticated caller."""
 21
 22    def __init__(self, http: BaseHttpClient) -> None:
 23        self._http = http
 24
 25    def get(self) -> MeResponse:
 26        """Get the calling user's profile + auth context.
 27
 28        Per-competition team memberships ride along on
 29        ``competition_teams``; there is no global "current team" any
 30        more — pick the row whose ``competition_id`` matches the
 31        scope you care about.
 32        """
 33        resp = self._http.request("GET", "/me")
 34        _raise_for_status(resp)
 35        return MeResponse.model_validate(resp.json())
 36
 37    def inbox(self) -> InboxResponse:
 38        """Pending team invites, join requests, direct invites, announcements."""
 39        resp = self._http.request("GET", "/me/inbox")
 40        _raise_for_status(resp)
 41        return InboxResponse.model_validate(resp.json())
 42
 43    def read_announcement(self, announcement_id: str) -> None:
 44        """Mark one announcement as read for the calling user."""
 45        resp = self._http.request("POST", f"/me/announcements/{announcement_id}/read")
 46        _raise_for_status(resp)
 47
 48    def read_all_announcements(self) -> None:
 49        """Mark every relevant announcement as read."""
 50        resp = self._http.request("POST", "/me/announcements/read-all")
 51        _raise_for_status(resp)
 52
 53    def update_profile(self, **fields: Any) -> MeResponse:
 54        """PATCH the caller's profile.
 55
 56        Pass any subset of ``bio`` / ``country`` / ``website_url`` /
 57        ``timezone`` / ``social_links``. Omit a key to leave it unchanged;
 58        pass ``None`` to clear it. Returns the updated :class:`MeResponse`.
 59
 60        Example::
 61
 62            client.me.update_profile(bio="hacking goblins", country="JP")
 63            client.me.update_profile(website_url=None)  # clear it
 64        """
 65        resp = self._http.request("PATCH", "/me/profile", json=fields)
 66        _raise_for_status(resp)
 67        return MeResponse.model_validate(resp.json())
 68
 69    def update_visibility(self, overrides: dict[str, bool]) -> MeResponse:
 70        """PATCH per-field privacy flags on the caller's profile.
 71
 72        Keys outside :data:`PROFILE_VISIBILITY_KEYS` are silently dropped
 73        server-side. Returns the updated :class:`MeResponse`.
 74        """
 75        resp = self._http.request("PATCH", "/me/profile/visibility", json=overrides)
 76        _raise_for_status(resp)
 77        return MeResponse.model_validate(resp.json())
 78
 79    def export_account(self) -> bytes:
 80        """Download a ZIP archive of every piece of data the platform
 81        holds about the caller (user, identities, tokens, memberships,
 82        submissions, solves, activity). Returns raw bytes — write to a
 83        file with ``Path(...).write_bytes(client.me.export_account())``."""
 84        resp = self._http.request("GET", "/me/export")
 85        _raise_for_status(resp)
 86        return resp.content
 87
 88    def delete_account(self, confirm_display_name: str) -> None:
 89        """Self-delete the calling user. ``confirm_display_name`` must
 90        match the caller's display name (falls back to email when empty)
 91        exactly — typo gate against fat-fingered destruction."""
 92        resp = self._http.request(
 93            "DELETE",
 94            "/me",
 95            json={"confirm_display_name": confirm_display_name},
 96        )
 97        _raise_for_status(resp)
 98
 99    def achievements(self, competition_id: str = "") -> MyAchievementsResponse:
100        """Caller's unlocked + locked badges (with progress hints)."""
101        params: dict[str, str] = {}
102        if competition_id:
103            params["competition_id"] = competition_id
104        resp = self._http.request("GET", "/me/achievements", params=params)
105        _raise_for_status(resp)
106        return MyAchievementsResponse.model_validate(resp.json())
107
108    def solves(self) -> list[MySolveSummary]:
109        """Per-challenge solve summary across every team the caller has played for."""
110        resp = self._http.request("GET", "/me/solves")
111        _raise_for_status(resp)
112        return [MySolveSummary.model_validate(s) for s in resp.json()]
113
114    def milestone_progress(self, *, competition_id: str = "") -> list[MilestoneProgress]:
115        """Partial-solve progress per challenge for the caller.
116
117        One row per challenge with at least one captured question;
118        ``solved_question_ids`` enumerates which milestones the user
119        has cleared, ``total_questions`` is the spec's declared count.
120        With ``competition_id`` set the result narrows to solves
121        stamped against the caller's team in that comp.
122        """
123        params: dict[str, str] = {}
124        if competition_id:
125            params["competition_id"] = competition_id
126        resp = self._http.request("GET", "/me/milestone-progress", params=params)
127        _raise_for_status(resp)
128        return [MilestoneProgress.model_validate(s) for s in resp.json()]
129
130    def verify_star_gazer(self, competition_id: str = "") -> StarGazerVerifyResponse:
131        """Verify the caller starred the ctfy GitHub repo and grant the
132        ``star_gazer`` badge if so.
133
134        Friendly failures (no GitHub identity, repo not configured, not
135        yet starred, GitHub rate-limited) come back as 200 with
136        ``verified=False`` + an actionable ``reason``."""
137        params = {"competition_id": competition_id} if competition_id else None
138        resp = self._http.request("POST", "/me/star-gazer/verify", params=params)
139        _raise_for_status(resp)
140        return StarGazerVerifyResponse.model_validate(resp.json())