Add broadcast endpoints

CHANGELOG.rst

@@ -8,8 +8,9 @@ To be released
 * Added ``sheet`` optional parameter to ``Tournaments.stream_results``, and fix returned typed dict.
 * Added ``studies.import_pgn`` to import PGN to study
 * Added ``tv.stream_current_game_of_channel`` to stream the current TV game of a channel
+* Added ``broadcast.get_top``, ``broadcast.get_by_user``, ``broadcast.reset_round`` to get broadcast information and reset rounds
 
-Thanks to @nicvagn, @tors42, @fitztrev and @trevorbayless for their contributions to this release.
+Thanks to @nicvagn, @tors42, @fitztrev, @trevorbayless, @friedrichtenhagen for their contributions to this release.
 
 v0.13.2 (2023-12-04)
 --------------------

README.rst

@@ -98,13 +98,16 @@ Most of the API is available:
     client.bots.decline_challenge
 
     client.broadcasts.get_official
+    client.broadcasts.get_top
+    client.broadcasts.get_by_user
     client.broadcasts.create
     client.broadcasts.get
     client.broadcasts.update
     client.broadcasts.push_pgn_update
     client.broadcasts.create_round
     client.broadcasts.get_round
     client.broadcasts.update_round
+    client.broadcasts.reset_round
     client.broadcasts.get_round_pgns
     client.broadcasts.get_pgns
     client.broadcasts.stream_round

berserk/__init__.py

@@ -14,6 +14,7 @@
 from .types import (
     ArenaResult,
     BroadcastPlayer,
+    PaginatedTopBroadcasts,
     Team,
     LightUser,
     ChapterIdName,
@@ -37,6 +38,7 @@
 __all__ = [
     "ArenaResult",
     "BroadcastPlayer",
+    "PaginatedTopBroadcasts",
     "ChapterIdName",
     "Client",
     "JSON",

berserk/clients/broadcasts.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from typing import Iterator, Any, Dict, List
+from typing import Iterator, Any, Dict, List, Literal, cast
 
 from .. import models
 from ..formats import PGN
 from .base import BaseClient
 
-from ..types.broadcast import BroadcastPlayer
+from ..types.broadcast import BroadcastPlayer, PaginatedTopBroadcasts, BroadcastByUser
 from ..utils import to_str
 
 
@@ -27,6 +27,36 @@ def get_official(
         params = {"nb": nb, "leaderboard": leaderboard}
         yield from self._r.get(path, params=params, stream=True)
 
+    def get_top(
+        self,
+        html: bool | None = None,
+        page: Literal[
+            1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20
+        ]
+        | None = 1,
+    ) -> PaginatedTopBroadcasts:
+        """Get the top broadcast previews as seen on https://lichess.org/broadcast.
+
+        :param html: Convert the "description" field from markdown to HTML
+        :param page: Which page to fetch. Only page 1 has "active" and "upcoming" broadcasts.
+        :return: page of top broadcasts
+        """
+        path = f"api/broadcast/top"
+        params = {"html": html, "page": page}
+        return cast(PaginatedTopBroadcasts, self._r.get(path, params=params))
+
+    def get_by_user(self, username: str, html: bool, page: int = 1) -> BroadcastByUser:
+        """Get all incoming, ongoing, and finished official broadcasts.
+            The broadcasts are sorted by created date, most recent first.
+
+        :param username: username
+        :param page: page number. Defaults to 1
+        :param html: Convert the "description" field from markdown to HTML
+        """
+        path = f"https://lichess.org/api/broadcast/by/{username}"
+        params = {"html": html, "page": page}
+        return cast(BroadcastByUser, self._r.get(path, params=params))
+
     def create(
         self,
         name: str,
@@ -198,6 +228,15 @@ def update_round(
         }
         return self._r.post(path, json=payload, converter=models.Broadcast.convert)
 
+    def reset_round(self, broadcast_round_id: str) -> Dict[str, Any]:
+        """Remove any games from the broadcast round and reset it to its initial state.
+
+        :param broadcast_round_id: eight character broadcast round ID
+        :return: reset successful status
+        """
+        path = f"/api/broadcast/round/{broadcast_round_id}/reset"
+        return self._r.post(path)
+
     def get_round_pgns(self, broadcast_round_id: str) -> Iterator[str]:
         """Get all games of a single round of a broadcast in PGN format.
 

berserk/types/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from .account import AccountInformation, Perf, Preferences, Profile, StreamerInfo
-from .broadcast import BroadcastPlayer
+from .broadcast import BroadcastPlayer, PaginatedTopBroadcasts
 from .bulk_pairings import BulkPairing, BulkPairingGame
 from .challenges import Challenge
 from .common import ClockConfig, ExternalEngine, LightUser, OnlineLightUser, Variant
@@ -20,6 +20,7 @@
     "AccountInformation",
     "ArenaResult",
     "BroadcastPlayer",
+    "PaginatedTopBroadcasts",
     "BulkPairing",
     "BulkPairingGame",
     "Challenge",

berserk/types/broadcast.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-
+from typing import List, Dict, Optional, Any
 from typing_extensions import NotRequired, TypedDict
 
 from .common import Title
@@ -14,3 +14,102 @@ class BroadcastPlayer(TypedDict):
     rating: NotRequired[int]
     # Title, optional
     title: NotRequired[Title]
+
+
+# top broadcasts start here
+
+
+class PaginatedTopBroadcasts(TypedDict):
+    # active broadcasts
+    active: List[BroadcastWithLastRound]
+    # upcoming broadcasts
+    upcoming: List[BroadcastWithLastRound]
+    # pagination meta data
+    past: BroadcastPaginationMetadata
+
+
+class BroadcastWithLastRound(TypedDict, total=False):
+    # group of broadcast
+    group: str
+    # broadcast round
+    round: BroadcastRoundInfo
+    # broadcast tour
+    tour: BroadcastTour
+    # not listed in docs currently
+    roundToLink: Dict[str, Any]
+
+
+class BroadcastPaginationMetadata(TypedDict, total=False):
+    currentPage: int
+    maxPerPage: int
+    currentPageResults: List[Dict[str, Any]]
+    previousPage: Optional[int]
+    nextPage: Optional[int]
+
+
+class BroadcastTour(TypedDict, total=False):
+    id: str
+    name: str
+    slug: str
+    createdAt: int
+    # Start and end dates of the tournament, as Unix timestamps in milliseconds
+    dates: Optional[List[int]]
+    # Additional display information about the tournament
+    info: BroadcastTourInfo
+    # Used to designate featured tournaments on Lichess
+    tier: Optional[int]
+    image: Optional[str]
+    # Full tournament description in markdown format, or in HTML if the html=1 query parameter is set.
+    description: Optional[str]
+    leaderboard: Optional[bool]
+    teamTable: Optional[bool]
+    url: str
+
+
+class BroadcastTourInfo(TypedDict, total=False):
+    # Official website. External website URL
+    website: str
+    # Featured players
+    players: str
+    # Tournament location
+    location: str
+    # Time control
+    tc: str
+    # FIDE rating category
+    fideTc: str
+    # Timezone of the tournament. Example: America/New_York.
+    timeZone: str
+    # Official standings website. External website URL
+    standings: str
+    # Tournament format
+    format: str
+
+
+class BroadcastRoundInfo(TypedDict, total=False):
+    id: str
+    name: str
+    slug: str
+    createdAt: int
+    ongoing: Optional[bool]
+    startsAt: Optional[int]
+    # The start date/time is unknown and the round will start automatically when the previous round completes
+    startsAfterPrevious: Optional[bool]
+    finishedAt: Optional[int]
+    url: str
+    delay: Optional[int]
+
+
+# broadcasts by user
+class BroadcastByUser(TypedDict, total=False):
+    currentPage: int
+    currentPageResults: List[BroadcastByUserCurrentPageResult]
+    maxPerPage: int
+    nbPages: int
+    nbResults: int
+    nextPage: Optional[int]
+    previousPage: Optional[int]
+
+
+class BroadcastByUserCurrentPageResult(TypedDict, total=False):
+    round: BroadcastRoundInfo
+    tour: BroadcastTourInfo

tests/clients/test_broadcasts.py

@@ -0,0 +1,77 @@
+import pytest
+import requests_mock
+
+from berserk import Client, PaginatedTopBroadcasts
+from utils import validate, skip_if_older_3_dot_10
+
+import requests_mock
+from typing import List, Dict
+from berserk.types.broadcast import (
+    BroadcastWithLastRound,
+    BroadcastPaginationMetadata,
+    BroadcastByUser,
+)
+
+
+class TestTopBroadcasts:
+    # test complete response
+    @skip_if_older_3_dot_10
+    def test_response_type_total(self):
+        """Verify that the response matches the total model"""
+        res = Client().broadcasts.get_top(html=True, page=1)
+        validate(PaginatedTopBroadcasts, res)
+
+    # In the following we test the subparts of the response.
+    # This makes it easier to locate errors.
+
+    # test response['active']
+    @skip_if_older_3_dot_10
+    def test_response_type_active_model(self):
+        """Verify that the response matches the typed-dict"""
+        res = Client().broadcasts.get_top(html=True, page=1)
+        validate(List[BroadcastWithLastRound], res.get("active", []))
+
+    # test response['past']
+    @skip_if_older_3_dot_10
+    def test_response_type_past_model(self):
+        """Verify that the response matches the typed-dict"""
+        res = Client().broadcasts.get_top(html=True, page=1)
+        validate(BroadcastPaginationMetadata, res.get("past", []))
+
+    # test response['upcoming']
+    @skip_if_older_3_dot_10
+    def test_response_type_upcoming_model(self):
+        """Verify that the response matches the typed-dict"""
+        res = Client().broadcasts.get_top(html=True, page=1)
+        validate(List[BroadcastWithLastRound], res.get("upcoming", []))
+
+
+class TestResetBroadcastRound:
+    def test_broadcast_round_id_param(self):
+        """The test verifies that the broadcast round id parameter is passed correctly in the query params."""
+        with requests_mock.Mocker() as m:
+            broadcast_round_id = "12345678"
+            m.post(
+                f"https://lichess.org/api/broadcast/round/{broadcast_round_id}/reset",
+                json={"ok": True},
+            )
+            res = Client().broadcasts.reset_round(broadcast_round_id=broadcast_round_id)
+
+
+class TestBroadcastByUsername:
+    # test complete response
+    @skip_if_older_3_dot_10
+    def test_response_type_total(self):
+        """Verify that the response matches the total model"""
+        res = Client().broadcasts.get_by_user(username="STL_Carlsen", html=True, page=1)
+        validate(BroadcastByUser, res)
+
+    def test_broadcast_round_id_param(self):
+        """The test verifies that the username parameter is passed correctly in the query params."""
+        with requests_mock.Mocker() as m:
+            username = "STL_Carlsen"
+            m.get(
+                f"https://lichess.org/api/broadcast/by/{username}",
+                json={"ok": True},
+            )
+            res = Client().broadcasts.get_by_user(username=username, html=False, page=1)