Рефактор

This commit is contained in:
2025-11-26 11:59:35 +03:00
parent c3b4602ed1
commit 5510f85d94
4 changed files with 279 additions and 759 deletions

View File

@ -1,6 +1,6 @@
[project]
name = "osuclient"
version = "0.2.1"
version = "0.3.0"
description = "Client for osu! API"
readme = "README.md"
authors = [

View File

@ -12,278 +12,34 @@ class osuAPIClient(AioHTTPXClient):
redis_url: str,
client_id: str,
client_secret: str,
callback_url: str,
redirect_uri: str,
):
self.base_uri = 'https://osu.ppy.sh/api/v2'
self.client_id = client_id
self.client_secret = client_secret
self.callback_url = callback_url
self.redirect_uri = redirect_uri
super().__init__(
base_url=self.base_uri,
redis_url=redis_url,
key='osu',
limit=10,
limit=1200,
logger='osu! API',
)
async def get_beatmap_favorites(
self, access_token: str, cache_time: int | None = None
):
req = await self.get(
'/me/beatmapset-favourites',
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return s.FavoriteBeatmaps.model_validate(req.json())
case st.UNAUTHORIZED:
raise s.Error(req.status_code, 'Unauthorized')
case _:
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
async def get_beatmap_packs(
self,
access_token: str,
packs_type: Literal[
'standard',
'featured',
'tournament',
'loved',
'chart',
'theme',
'artist',
] = 'standard',
cursor_string: str | None = None,
cache_time: int | None = None,
):
req = await self.get(
'/beatmaps/packs',
params=self.clean_dict(
{'type': packs_type, 'cursor_string': cursor_string}
),
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return s.BeatmapPacks.model_validate(req.json())
case st.UNAUTHORIZED:
raise s.Error(req.status_code, 'Unauthorized')
case _:
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
async def get_beatmap_pack(
self,
access_token: str,
pack: str,
legacy_only: int = 0,
cache_time: int | None = None,
):
req = await self.get(
f'/beatmaps/packs/{pack}',
params=self.clean_dict({'legacy_only': legacy_only}),
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return s.BeatmapPack.model_validate(req.json())
case st.UNAUTHORIZED:
raise s.Error(req.status_code, 'Unauthorized')
case st.NOT_FOUND:
raise s.Error(req.status_code, 'Not Found')
case _:
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
async def lookup_beatmap(
self,
access_token: str,
checksum: str | None = None,
filename: str | None = None,
beatmap_id: int | None = None,
cache_time: int | None = None,
):
req = await self.get(
'/beatmaps/lookup',
params=self.clean_dict(
{
'checksum': checksum,
'filename': filename,
'beatmap_id': beatmap_id,
}
),
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return s.BeatmapExtended.model_validate(req.json())
case st.UNAUTHORIZED:
raise s.Error(req.status_code, 'Unauthorized')
case st.NOT_FOUND:
raise s.Error(req.status_code, 'Not Found')
case _:
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
async def get_user_beatmap_score(
self,
access_token: str,
beatmap: int,
user: int,
cache_time: int | None = None,
):
req = await self.get(
f'/beatmaps/{beatmap}/scores/users/{user}',
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return s.BeatmapUserScore.model_validate(req.json())
case st.UNAUTHORIZED:
raise s.Error(req.status_code, 'Unauthorized')
case st.NOT_FOUND:
raise s.Error(req.status_code, 'Not Found')
case _:
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
async def get_user_beatmap_scores(
self,
access_token: str,
beatmap: int,
user: int,
legacy_only: int = 0,
ruleset: Literal['fruits', 'mania', 'osu', 'taiko'] | None = None,
cache_time: int | None = None,
):
req = await self.get(
f'/beatmaps/{beatmap}/scores/users/{user}/all',
params=self.clean_dict(
{'legacy_only': legacy_only, 'ruleset': ruleset}
),
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return s.UserBeatmapScores.model_validate(req.json())
case st.UNAUTHORIZED:
raise s.Error(req.status_code, 'Unauthorized')
case st.NOT_FOUND:
raise s.Error(req.status_code, 'Not Found')
case _:
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
async def get_beatmap_scores(
self,
access_token: str,
beatmap: int,
legacy_only: int = 0,
mode: Literal['fruits', 'mania', 'osu', 'taiko'] | None = None,
mods: str | None = None,
ranking_type: str | None = None,
cache_time: int | None = None,
):
req = await self.get(
f'/beatmaps/{beatmap}/scores',
params=self.clean_dict(
{
'legacy_only': legacy_only,
'mode': mode,
'mods': mods,
'ranking_type': ranking_type,
}
),
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return s.BeatmapScores.model_validate(req.json())
case st.UNAUTHORIZED:
raise s.Error(req.status_code, 'Unauthorized')
case _:
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
# TODO: implement endpoints
# https://osu.ppy.sh/docs/index.html#get-beatmaps
async def get_user_scores(
self,
access_token: str,
user: int,
score_type: Literal['best', 'firsts', 'recent'],
legacy_only: int = 0,
include_fails: int = 0,
user_id: int,
score_type: Literal['best', 'recent', 'firsts'],
legacy_only: Literal[0, 1] = 0,
include_fails: Literal[0, 1] = 0,
mode: Literal['fruits', 'mania', 'osu', 'taiko'] | None = None,
limit: int | None = None,
offset: int | None = None,
cache_time: int | None = None,
offset: str | None = None,
):
req = await self.get(
f'/users/{user}/scores/{score_type}',
f'/users/{user_id}/scores/{score_type}',
params=self.clean_dict(
{
'legacy_only': legacy_only,
@ -296,7 +52,6 @@ class osuAPIClient(AioHTTPXClient):
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
@ -315,9 +70,6 @@ class osuAPIClient(AioHTTPXClient):
self.logger.error(req.text)
raise s.Error(500, 'Internal Server Error')
# TODO: implement other endpoints
# https://osu.ppy.sh/docs/index.html#get-user-beatmaps
async def get_user(
self,
access_token: str,
@ -342,7 +94,7 @@ class osuAPIClient(AioHTTPXClient):
match req.status_code:
case st.OK:
return s.GetUserSchema.model_validate(req.json())
return s.GetUser.model_validate(req.json())
case st.NOT_FOUND:
raise s.Error(req.status_code, 'Not Found')

View File

@ -114,7 +114,7 @@ class osuAuthClient(AioHTTPXClient):
return s.Token.model_validate(req.json())
case st.BAD_REQUEST:
raise s.Error(req.status_code, req.json()['error'])
raise s.Error(req.status_code, req.json()['message'])
case _:
raise s.Error(req.status_code, 'Internal Server Error')

View File

@ -1,7 +1,7 @@
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field
class Error(Exception):
@ -25,432 +25,146 @@ class UserToken(Token):
class Country(BaseModel):
model_config = ConfigDict(extra='forbid')
code: str
name: str
class Beatmap(BaseModel):
beatmapset_id: int
difficulty_rating: float
id: int
mode: Literal['fruits', 'mania', 'osu', 'taiko']
status: int
total_length: int
user_id: int
version: str
class Cover(BaseModel):
model_config = ConfigDict(extra='forbid')
custom_url: str
url: str
id: int | None
class BeatmapDifficultyAttributes(BaseModel):
star_rating: float
max_combo: int
class DailyChallengeUserStats(BaseModel):
model_config = ConfigDict(extra='forbid')
class osuBeatmapDifficultyAttributes(BeatmapDifficultyAttributes):
aim_difficulty: float
aim_difficult_slider_count: float
speed_difficulty: float
speed_note_count: float
slider_factor: float
aim_difficult_strain_count: float
speed_difficult_strain_count: float
class taikoBeatmapDifficultyAttributes(BeatmapDifficultyAttributes):
mono_stamina_factor: float
class BeatmapExtended(Beatmap):
accuracy: float
ar: float
beatmapset_id: int
bpm: float | None = None
convert: bool
count_circles: int
count_sliders: int
count_spinners: int
cs: float
deleted_at: datetime | None = None
drain: float
hit_length: int
is_scoreable: bool
last_updated: datetime
mode_int: int
passcount: int
daily_streak_best: int
daily_streak_current: int
last_update: str
last_weekly_streak: str
playcount: int
ranked: int
url: str
top_10p_placements: int
top_50p_placements: int
user_id: int
weekly_streak_best: int
weekly_streak_current: int
class BeatmapOwner(BaseModel):
id: int
username: str
class MonthlyPlaycount(BaseModel):
model_config = ConfigDict(extra='forbid')
class UserCompletionData(BaseModel):
beatmapset_ids: list[int]
completed: bool
class BeatmapPack(BaseModel):
author: str
date: datetime
name: str
no_diff_reduction: bool
ruleset_id: int | None
tag: str
url: str
class BeatmapPlaycount(BaseModel):
beatmap_id: int
beatmap: 'Beatmap | None' = None
beatmapset: 'Beatmapset | None' = None
start_date: str
count: int
class BeatmapScores(BaseModel):
scores: list['Score']
userScore: 'BeatmapUserScore | None' = None
class Page(BaseModel):
model_config = ConfigDict(extra='forbid')
html: str
raw: str
class BeatmapUserScore(BaseModel):
position: int
score: 'Score'
class RankHighest(BaseModel):
model_config = ConfigDict(extra='forbid')
rank: int
updated_at: str
class Beatmapset(BaseModel):
artist: str
artist_unicode: str
covers: 'Covers'
creator: str
favourite_count: int
id: int
nsfw: bool
offset: int
class Kudosu(BaseModel):
model_config = ConfigDict(extra='forbid')
available: int
total: int
class ReplaysWatchedCount(BaseModel):
model_config = ConfigDict(extra='forbid')
start_date: str
count: int
class Level(BaseModel):
model_config = ConfigDict(extra='forbid')
current: int
progress: int
class GradeCounts(BaseModel):
model_config = ConfigDict(extra='forbid')
ss: int
ssh: int
s: int
sh: int
a: int
class Rank(BaseModel):
model_config = ConfigDict(extra='forbid')
country: int
class UserStatistics(BaseModel):
model_config = ConfigDict(extra='forbid')
count_100: int
count_300: int
count_50: int
count_miss: int
level: Level
global_rank: int
global_rank_percent: float
global_rank_exp: None
pp: float
pp_exp: int
ranked_score: int
hit_accuracy: float
play_count: int
preview_url: str
source: str
status: str
spotlight: bool
title: str
title_unicode: str
user_id: int
video: bool
play_time: int
total_score: int
total_hits: int
maximum_combo: int
replays_watched_by_others: int
is_ranked: bool
grade_counts: GradeCounts
country_rank: int
rank: Rank
class Covers(BaseModel):
cover: str
cover_2x: str = Field(alias='cover@2x')
card: str
card_2x: str = Field(alias='card@2x')
list: str
list_2x: str = Field(alias='list@2x')
slimcover: str
slimcover_2x: str = Field(alias='slimcover@2x')
class UserAchievement(BaseModel):
model_config = ConfigDict(extra='forbid')
achieved_at: str
achievement_id: int
class BeatmapsetDiscussion(BaseModel):
beatmap: 'Beatmap | None' = None
beatmap_id: int | None = None
beatmapset: 'Beatmapset | None' = None
beatmapset_id: int
can_be_resolved: bool
can_grant_kudosu: bool
created_at: datetime
current_user_attributes: 'CurrentUserAttributes'
deleted_at: datetime | None = None
deleted_by_id: int | None = None
id: int
kudosu_denied: bool
last_post_at: datetime
message_type: Literal[
'hype', 'mapper_note', 'praise', 'problem', 'review', 'suggestion'
]
parent_id: int | None = None
posts: list['BeatmapsetDiscussionPost']
timestamp: int | None = None
updated_at: datetime | None = None
user_id: int
class RankHistory(BaseModel):
model_config = ConfigDict(extra='forbid')
class BeatmapsetDiscussionPost(BaseModel):
beatmapset_discussion_id: int
created_at: datetime
deleted_at: datetime | None = None
deleted_by_id: int | None = None
id: int
last_editor_id: int | None = None
message: str
system: bool
updated_at: datetime | None = None
user_id: int
mode: str
data: list[int]
class BeatmapsetDiscussionVote(BaseModel):
beatmapset_discussion_id: int
created_at: datetime
id: int
score: int
updated_at: datetime
user_id: int
class ActiveTournamentBanner(BaseModel):
model_config = ConfigDict(extra='forbid')
class BeatmapsetAvailability(BaseModel):
download_disabled: bool
more_information: str | None = None
class BeatmapsetHype(BaseModel):
current: int
required: int
class BeatmapsetNominationsSummary(BaseModel):
current: int
required: int
class BeatmapsetExtended(Beatmapset):
availability: BeatmapsetAvailability
bpm: float
can_be_hyped: float
deleted_at: datetime | None = None
discussion_enabled: bool
discussion_locked: bool
hype: BeatmapsetHype
is_scoreable: bool
last_updated: datetime
legacy_thread_url: str | None = None
nominations_summary: BeatmapsetNominationsSummary
ranked: int
ranked_date: datetime | None = None
rating: float
source: str
storyboard: bool
submitted_date: datetime | None = None
tags: str
class Build(BaseModel):
created_at: datetime
display_version: str
id: int
update_stream: 'UpdateStream | None' = None
users: int
version: str | None = None
youtube_id: str | None = None
class Versions(BaseModel):
next: Build | None = None
previous: Build | None = None
class ChangelogEntry(BaseModel):
category: str
created_at: datetime
github_pull_request_id: int | None = None
github_url: str | None = None
id: int | None = None
major: bool
repository_url: str | None = None
title: str | None = None
entry_type: str = Field(alias='type')
url: str | None = None
class ChatChannel(BaseModel):
channel_id: int
name: str
description: str | None = None
icon: str | None = None
channel_type: Literal[
'PUBLIC',
'PRIVATE',
'MULTIPLAYER',
'SPECTATOR',
'TEMPORARY',
'PM',
'GROUP',
'ANNOUNCE',
] = Field(alias='type')
message_length_limit: int
moderated: bool
uuid: str | None = None
class ChatMessage(BaseModel):
channel_id: int
content: str
is_action: bool
message_id: int
sender_id: int
timestamp: datetime
message_type: str = Field(alias='type')
uuid: str | None = None
class Comment(BaseModel):
pass
class CommentBundle(BaseModel):
pass
class CommentSort(BaseModel):
pass
class CommentableMeta(BaseModel):
pass
class CurrentUserAttributes(BaseModel):
pass
class Event(BaseModel):
pass
class Forum(BaseModel):
pass
class ForumPost(BaseModel):
pass
class ForumTopic(BaseModel):
pass
class GithubUser(BaseModel):
pass
class Group(BaseModel):
pass
class KudosuHistory(BaseModel):
pass
class Match(BaseModel):
pass
class MatchEvent(BaseModel):
pass
class MatchGame(BaseModel):
pass
class MultiplayerScores(BaseModel):
pass
class MultiplayerScoresAroundUser(BaseModel):
pass
class MultiplayerScoresCursor(BaseModel):
pass
class MultiplayerScoresSort(BaseModel):
pass
class NewsPost(BaseModel):
pass
class Nomination(BaseModel):
pass
class Notification(BaseModel):
pass
class RankingType(BaseModel):
pass
class Rankings(BaseModel):
pass
class Score(BaseModel):
accuracy: float
beatmap_id: int | None = None
best_id: int | None = None
build_id: int | None = None
classic_total_score: int | None = None
ended_at: datetime | None = None
has_replay: bool | None = None
id: int
is_perfect_combo: bool | None = None
legacy_perfect: bool | None = None
legacy_score_id: int | None = None
legacy_total_score: int | None = None
max_combo: int
maximum_statistics: 'ScoreStatistics | None' = None
mods: list[str]
passed: bool
playlist_item_id: int | None = None
pp: float | None = None
preserve: bool | None = None
processed: bool | None = None
rank: str
ranked: bool | None = None
room_id: int | None = None
ruleset_id: int | None = None
started_at: datetime | None = None
statistics: 'ScoreStatistics'
total_score: int | None = None
type: str
user_id: int
class ScoreStatistics(BaseModel):
pass
class Spotlight(BaseModel):
pass
class Spotlights(BaseModel):
spotlights: list[Spotlight]
class UpdateStream(BaseModel):
pass
class UserAccountHistory(BaseModel):
description: str | None = None
id: int
length: int
permanent: bool
timestamp: datetime
type: Literal['note', 'restriction', 'silence']
class ProfileBanner(BaseModel):
id: int
tournament_id: int
image: str | None = None
image_2x: str | None = Field(alias='image@2x')
image: str
image_2x: str = Field(alias='image@2x')
class UserBadge(BaseModel):
class Badge(BaseModel):
model_config = ConfigDict(extra='forbid')
awarded_at: datetime
description: str
image_2x_url: str = Field(alias='image@2x_url')
@ -458,146 +172,200 @@ class UserBadge(BaseModel):
url: str
class UserKudosu(BaseModel):
available: int
total: int
class Team(BaseModel):
model_config = ConfigDict(extra='forbid')
id: int
name: str
short_name: str
flag_url: str
class UserRankHighest(BaseModel):
rank: int
updated_at: datetime
class ScoreStatistics(BaseModel):
count_100: int
count_300: int
count_50: int
count_geki: None
count_katu: None
count_miss: int
class UserMonthlyPlaycount(BaseModel):
start_date: datetime
count: int
class CurrentUserAttributes(BaseModel):
pin: None
class Covers(BaseModel):
cover: str
cover_2x: str = Field(..., alias='cover@2x')
card: str
card_2x: str = Field(..., alias='card@2x')
list: str
list_2x: str = Field(..., alias='list@2x')
slimcover: str
slimcover_2x: str = Field(..., alias='slimcover@2x')
class Beatmap(BaseModel):
beatmapset_id: int
difficulty_rating: float
id: int
mode: str
status: str
total_length: int
user_id: int
version: str
accuracy: float
ar: float
bpm: float
convert: bool
count_circles: int
count_sliders: int
count_spinners: int
cs: float
deleted_at: datetime | None
drain: float
hit_length: int
is_scoreable: bool
last_updated: str
mode_int: int
passcount: int
playcount: int
ranked: int
url: str
checksum: str
class Beatmapset(BaseModel):
anime_cover: bool
artist: str
artist_unicode: str
covers: Covers
creator: str
favourite_count: int
genre_id: int
hype: None
id: int
language_id: int
nsfw: bool
offset: int
play_count: int
preview_url: str
source: str
spotlight: bool
status: str
title: str
title_unicode: str
track_id: int | None
user_id: int
video: bool
class User(BaseModel):
avatar_url: str | None = None
model_config = ConfigDict(extra='forbid')
avatar_url: str
country_code: str
default_group: str | None = None
default_group: str
id: int
is_active: bool
is_bot: bool
is_deleted: bool
is_online: bool
is_supporter: bool
last_visit: datetime | None = None
last_visit: datetime | None
pm_friends_only: bool
profile_colour: str | None = None
profile_colour: None
username: str
class UserExtended(User):
discord: str | None = None
class Weight(BaseModel):
percentage: float
pp: float
class Score(BaseModel):
model_config = ConfigDict(extra='forbid')
accuracy: float
best_id: int
created_at: str
id: int
max_combo: int
mode: str
mode_int: int
mods: list[str]
passed: bool
perfect: bool
pp: float | None
rank: str
replay: bool
score: int
statistics: ScoreStatistics
type: str
user_id: int
current_user_attributes: CurrentUserAttributes
beatmap: Beatmap
beatmapset: Beatmapset
user: User
weight: Weight | None = None
class GetUser(User):
model_config = ConfigDict(extra='forbid')
cover_url: str
discord: str | None
has_supported: bool
interests: str | None = None
join_date: datetime
location: str | None = None
interests: None
join_date: str
location: str | None
max_blocks: int
max_friends: int
occupation: str | None = None
playmode: Literal['osu', 'taiko', 'fruits', 'mania']
playstyle: list[str]
occupation: None
playmode: str
playstyle: list[str] | None
post_count: int
profile_hue: int | None = None
profile_hue: int | None
profile_order: list[str]
title: str | None = None
title_url: str | None = None
twitter: str | None = None
website: str | None = None
class UserGroup(BaseModel):
pass
class UserSilence(BaseModel):
pass
class UserStatisticsGradeCounts(BaseModel):
a: int
s: int
sh: int
ss: int
ssh: int
class UserStatisticsLevel(BaseModel):
current: int
progress: float
class UserStatistics(BaseModel):
count_100: int
count_300: int
count_50: int
count_miss: int
country_rank: int | None = None
grade_counts: UserStatisticsGradeCounts
hit_accuracy: float
is_ranked: bool
level: UserStatisticsLevel
maximum_combo: int
play_count: int
play_time: int
pp: float
global_rank: int | None = None
global_rank_exp: int | None = None
ranked_score: int
replays_watched_by_others: int
total_hits: int
total_score: int
class WikiPage(BaseModel):
pass
class FavoriteBeatmaps(BaseModel):
beatmapset_ids: list[int]
class BeatmapPacks(BaseModel):
beatmap_packs: list[BeatmapPack]
class UserBeatmapScores(BaseModel):
scores: list['Score']
class GetUserSchema(UserExtended):
account_history: list[UserAccountHistory]
active_tournament_banners: list[ProfileBanner]
badges: list[UserBadge]
beatmap_playcounts_count: int
title: str | None
title_url: None
twitter: str | None
website: str | None
country: Country
# cover
cover: Cover
kudosu: Kudosu
account_history: list[str]
active_tournament_banner: ActiveTournamentBanner | None
active_tournament_banners: list[ActiveTournamentBanner]
badges: list[Badge]
beatmap_playcounts_count: int
comments_count: int
current_season_stats: None
daily_challenge_user_stats: DailyChallengeUserStats
favourite_beatmapset_count: int
# follow_user_mapping: list[int]
follower_count: int
graveyard_beatmapset_count: int
groups: list[UserGroup]
groups: list[str]
guest_beatmapset_count: int
is_restricted: bool | None = None
kudosu: UserKudosu
loved_beatmapset_count: int
mapping_follower_count: int
monthly_playcounts: list[UserMonthlyPlaycount]
monthly_playcounts: list[MonthlyPlaycount]
nominated_beatmapset_count: int
# page
page: Page
pending_beatmapset_count: int
previous_usernames: list[str]
rank_highest: UserRankHighest | None = None
# rank_history
rank_highest: RankHighest
ranked_beatmapset_count: int
# replays_watched_counts
replays_watched_counts: list[ReplaysWatchedCount]
scores_best_count: int
scores_first_count: int
scores_pinned_count: int
scores_recent_count: int
# session_verified: bool
statistics: UserStatistics
support_level: int
# user_achievements
team: Team | None
user_achievements: list[UserAchievement]
rank_history: RankHistory
rankHistory: RankHistory
ranked_and_approved_beatmapset_count: int
unranked_beatmapset_count: int