Helpers
There are lots of helper functions in Phillip as it tries to have as little dependencies as possible. This will help you access requests data from osu! and process them.
osu!v v1 API¶
In this section we will only bring functions to help you with osu! API. (v1, obviously.)
phillip.osu.old.api.APIClient
¶
get_api(self, endpoint, **kwargs)
async
¶
Request something based on endpoint. This function is a coroutine.
Parameters:
- endpoint -
str
-- The API endpoint, reference could be found in osu!wiki. - **kwargs -
dict
| optional -- Keyword arguments that will be passed as a query string.
Raises:
Exception
-- If API key is not assigned.
Returns
List[dict]
-- API response.
Source code in phillip\osu\old\api.py
async def get_api(self, endpoint: str, **kwargs) -> List[dict]:
"""Request something based on endpoint. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*
**Parameters:**
* endpoint - `str` -- The API endpoint, reference could be found in [osu!wiki](https://github.com/ppy/osu-api/wiki).
* \*\*kwargs - `dict` | optional -- Keyword arguments that will be passed as a query string.
**Raises:**
* `Exception` -- If API key is not assigned.
**Returns**
* `List[dict]` -- API response.
"""
kwargs["k"] = self._key
api_args = urlencode(kwargs)
api_url = self.BASE_URL + endpoint + "?" + api_args
async with self._session.get(api_url) as api_res:
return await api_res.json()
get_beatmaps(self, **kwargs)
async
¶
Get beatmapset from osu! API. This function is a coroutine.
Returns
List[Beatmap]
-- Beatmapsets fetched from API.
Source code in phillip\osu\old\api.py
async def get_beatmaps(self, **kwargs) -> List[Beatmap]:
"""Get beatmapset from osu! API. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*
**Returns**
* `List[Beatmap]` -- Beatmapsets fetched from API.
"""
return [Beatmap(map) for map in await self.get_api("get_beatmaps", **kwargs)]
osu!web¶
In this section, we will scrape the site in order to get data from osu!
Warning
A throttler is automatically initiated to rate limit your request to 2 requests/minute in order to avoid getting IP banned. You may change this but in general this is highly discouraged.
phillip.osu.new.abstract.ABCClient
¶
get_events(self, nominate=True, rank=True, love=True, nomination_reset=True, disqualify=True, **kwargs)
¶
Get events of from osu!website. This function is a coroutine.
Parameters:
- nominate -
bool
-- Whether to get nomination events or not. This impliesqualify
event. - rank -
bool
-- Whether to get ranked events or not. - love -
bool
-- Whetger to get loved events or not. - nomination_reset -
bool
-- Whether to get nomination reset events or not. - disqualify --
bool
-- Whether to get disqualification events or not.
Yields:
- list
abstract.EventBase
-- List of events resulted from fetching osu!web, with next index as next event that will be processed.
Source code in phillip\osu\new\abstract.py
async def get_events(
self,
nominate: bool = True,
rank: bool = True,
love: bool = True,
nomination_reset: bool = True,
disqualify: bool = True,
**kwargs,
) -> AsyncGenerator[Type[abstract.EventBase], None]:
"""Get events of from osu!website. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*
**Parameters:**
* nominate - `bool` -- Whether to get nomination events or not. **This implies `qualify` event.**
* rank - `bool` -- Whether to get ranked events or not.
* love - `bool` -- Whetger to get loved events or not.
* nomination_reset - `bool` -- Whether to get nomination reset events or not.
* disqualify -- `bool` -- Whether to get disqualification events or not.
**Yields:**
* list `abstract.EventBase` -- List of events resulted from fetching osu!web, \
with next index as next event that will be processed.
"""
additions = list()
types_val = [nominate, rank, love, nomination_reset, disqualify]
for i in range(5):
additions.append(types_val[i] and self.TYPES[i] or str())
if types_val[0]:
additions.append("qualify")
extras = urlencode(kwargs)
url = self.events_url + "&types%5B%5D=".join(additions) + "&" + extras
events = await self.get_json(url, "json-events")
events.reverse() # type: ignore
event_cases = {
"nominate": classes.Nominated,
"disqualify": classes.Disqualified,
"nomination_reset": classes.Popped,
"rank": classes.Ranked,
"love": classes.Loved,
}
for i, event in enumerate(events):
action = event["type"] # type: ignore
if action == "qualify":
continue # Skip qualified event news
next_map = None
if i + 1 != len(events):
next_map = events[i + 1] # type: ignore
yield event_cases[action](event, next_map, app=self._app) # type: ignore
get_users(self, group_id)
async
¶
Get users inside of a group. This function is a coroutine.
Parameters:
- group_id -
int
-- The group id.
Returns
List[dict]
-- A dictionary containing users' data.
Source code in phillip\osu\new\abstract.py
async def get_users(self, group_id: int) -> List[GroupUser]:
"""Get users inside of a group. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*
**Parameters:**
* group_id - `int` -- The group id.
**Returns**
* `List[dict]` -- A dictionary containing users' data.
"""
uri = self.groups_url + str(group_id)
users_json = await self.get_json(uri, "json-users")
out = []
for user in users_json:
out.append(GroupUser(user))
return out
nomination_history(self, mapid)
async
¶
Get nomination history of a beatmap. This function is a coroutine.
Parameters:
- mapid -
int
-- Beatmapset ID to gather.
Returns
- parent -
List[child]
-- A list containing child tuples.child
-Tuple[str, int]
-- A tuple with a string of event type and user id of user triggering the event.
Source code in phillip\osu\new\abstract.py
async def nomination_history(self, mapid: int) -> List[Tuple[str, int]]:
"""Get nomination history of a beatmap. *This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*
**Parameters:**
* mapid - `int` -- Beatmapset ID to gather.
**Returns**
* parent - `List[child]` -- A list containing child tuples.
* `child` - `Tuple[str, int]` -- A tuple with a string of event type and user id of user triggering the event.
"""
uri = f"https://osu.ppy.sh/beatmapsets/{str(mapid)}/discussion"
set_json = await self.get_json(uri, "json-beatmapset-discussion")
js = set_json["beatmapset"]["events"] # type: ignore
history = []
for i, event in enumerate(js):
if i + 1 != len(js):
next_event = js[i + 1]
if event["type"] in self.EVENTS:
event_name = self.EVENTS[event["type"]]
if next_event["type"] == "qualify":
event_name = "Qualified"
history.append((event_name, event["user_id"]))
return history
phillip.osu.new.web.WebClient
¶
get_html(self, uri)
async
¶
Receive html from uri with rate limit. This function is a coroutine.
Parameters:
- uri -
str
-- URL of discussion page.
Returns
- BeautifulSoup -- Soup'd html response.
Source code in phillip\osu\new\web.py
async def get_html(self, uri: str) -> BeautifulSoup:
"""Receive html from uri with rate limit.
*This function is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutine).*
**Parameters:**
* uri - `str` -- URL of discussion page.
**Returns**
* BeautifulSoup -- Soup'd html response.
"""
async with self._throttler:
async with self._session.get(uri, cookies={"locale": "en"}) as site_html:
return BeautifulSoup(await site_html.text(), features="html.parser")