Релиз 0.1

This commit is contained in:
2025-11-25 22:30:20 +03:00
commit 337c71986b
10 changed files with 1091 additions and 0 deletions

16
.gitignore vendored Normal file
View File

@ -0,0 +1,16 @@
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
# Ruff
.ruff_cache
# uv
uv.lock

34
.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,34 @@
repos:
- repo: https://github.com/crate-ci/typos
rev: v1.36.3
hooks:
- id: typos
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.13.2
hooks:
- id: ruff
args: [ --fix ]
- id: ruff-format
- repo: https://github.com/RobertCraigie/pyright-python
rev: v1.1.405
hooks:
- id: pyright
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v6.0.0
hooks:
- id: trailing-whitespace
- id: check-docstring-first
- id: check-added-large-files
- id: check-yaml
- id: debug-statements
- id: check-merge-conflict
- id: double-quote-string-fixer
- id: end-of-file-fixer
- repo: meta
hooks:
- id: check-hooks-apply
- id: check-useless-excludes

0
README.md Normal file
View File

90
pyproject.toml Normal file
View File

@ -0,0 +1,90 @@
[project]
name = "osuclient"
version = "0.1.0"
description = "Client for osu! API"
readme = "README.md"
authors = [
{ name = "Miwory", email = "miwory.uwu@gmail.com" }
]
requires-python = ">=3.13"
dependencies = [
"aiohttpx==1.2.0",
"pydantic>=2.12.3",
]
[build-system]
requires = ["uv_build>=0.9.2,<0.10.0"]
build-backend = "uv_build"
[tool.uv.sources]
aiohttpx = { git = "https://git.miwory.dev/Miwory/AioHTTPX.git" }
[project.optional-dependencies]
dev = [
"ruff==0.14.2",
"pyright==1.1.406",
"poethepoet==0.37.0",
"pre-commit==4.3.0",
]
[tool.poe.tasks]
_git = "git add ."
_lint = "pre-commit run --all-files"
lint = ["_git", "_lint"]
check = "uv pip ls --outdated"
[tool.pyright]
venvPath = "."
venv = ".venv"
strictListInference = true
strictDictionaryInference = true
strictSetInference = true
deprecateTypingAliases = true
typeCheckingMode = "strict"
pythonPlatform = "All"
[tool.ruff]
target-version = "py313"
line-length = 79
fix = true
[tool.ruff.lint]
preview = true
select = [
"E",
"W",
"F",
"UP",
"A",
"B",
"C4",
"SIM",
"I",
"S",
"G",
"FAST",
"ASYNC",
"BLE",
"INT",
"ISC",
"ICN",
"PYI",
"INP",
"RSE",
"PIE",
"SLOT",
"TID",
"LOG",
"FBT",
"DTZ",
"EM",
"PERF",
"RUF",
]
ignore = ["RUF029", "S101", "S104"]
[tool.ruff.format]
quote-style = "single"
indent-style = "space"
docstring-code-format = true

View File

@ -0,0 +1 @@
__version__ = '0.1.0'

319
src/osuclient/api.py Normal file
View File

@ -0,0 +1,319 @@
from typing import Literal
from aiohttpx import status as st
from aiohttpx.client import AioHTTPXClient
from . import schema as s
class osuAPIClient(AioHTTPXClient):
def __init__(
self,
redis_url: str,
client_id: str,
client_secret: str,
callback_url: 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
super().__init__(
base_url=self.base_uri,
redis_url=redis_url,
key='osu',
limit=10,
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,
mode: Literal['fruits', 'mania', 'osu', 'taiko'] | None = None,
limit: int | None = None,
offset: int | None = None,
cache_time: int | None = None,
):
req = await self.get(
f'/users/{user}/scores/{score_type}',
params=self.clean_dict(
{
'legacy_only': legacy_only,
'include_fails': include_fails,
'mode': mode,
'limit': limit,
'offset': offset,
}
),
headers=self.clean_dict(
{
'Authorization': f'Bearer {access_token}',
'X-Cache-TTL': cache_time,
}
),
)
match req.status_code:
case st.OK:
return [s.Score.model_validate(score) for score in req.json()]
case st.NOT_FOUND:
raise s.Error(req.status_code, 'Not Found')
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 other endpoints
# https://osu.ppy.sh/docs/index.html#get-user-beatmaps

120
src/osuclient/auth.py Normal file
View File

@ -0,0 +1,120 @@
from typing import Literal
from urllib.parse import urlencode
from aiohttpx import status as st
from aiohttpx.client import AioHTTPXClient
from . import schema as s
class osuAuthClient(AioHTTPXClient):
def __init__(
self,
redis_url: str,
client_id: str,
client_secret: str,
redirect_uri: str,
):
self.base_uri = 'https://osu.ppy.sh/oauth'
self.client_id = client_id
self.client_secret = client_secret
self.redirect_uri = redirect_uri
super().__init__(
base_url=self.base_uri,
redis_url=redis_url,
key='osu',
limit=10,
logger='osu! Auth',
)
async def get_authorize_url(
self, scope: list[str] | str, state: str | None = None
):
if isinstance(scope, list):
scope = ' '.join(scope)
params = self.clean_dict(
{
'client_id': self.client_id,
'redirect_uri': self.redirect_uri,
'response_type': 'code',
'scope': scope,
'state': state,
}
)
return f'{self.base_uri}/authorize?{urlencode(params)}'
async def access_token(
self,
code: str,
scope: list[str] | str | None = None,
):
req = await self.token('authorization_code', code=code, scope=scope)
if not isinstance(req, s.UserToken):
raise s.Error(500, 'Internal Server Error')
return req
async def refresh_token(
self, refresh_token: str, scope: list[str] | str | None = None
):
req = await self.token(
'refresh_token', refresh_token=refresh_token, scope=scope
)
if not isinstance(req, s.UserToken):
raise s.Error(500, 'Internal Server Error')
return req
async def client_token(self, scope: list[str] | str = 'public'):
req = await self.token('client_credentials', scope=scope)
if isinstance(req, s.UserToken):
raise s.Error(500, 'Internal Server Error')
return req
async def token(
self,
grant_type: Literal[
'authorization_code', 'refresh_token', 'client_credentials'
],
code: str | None = None,
refresh_token: str | None = None,
scope: list[str] | str | None = None,
):
if isinstance(scope, list):
scope = ' '.join(scope)
req = await self.post(
'/token',
data=self.clean_dict(
{
'client_id': self.client_id,
'client_secret': self.client_secret,
'code': code,
'refresh_token': refresh_token,
'grant_type': grant_type,
'redirect_uri': self.redirect_uri,
'scope': scope,
}
),
)
match req.status_code:
case st.OK:
try:
return s.UserToken.model_validate(req.json())
except ValueError:
return s.Token.model_validate(req.json())
case st.BAD_REQUEST:
raise s.Error(req.status_code, req.json()['error'])
case _:
raise s.Error(req.status_code, 'Internal Server Error')

0
src/osuclient/py.typed Normal file
View File

501
src/osuclient/schema.py Normal file
View File

@ -0,0 +1,501 @@
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, Field
class Error(Exception):
status_code: int
error: str
def __init__(self, status_code: int, error: str) -> None:
self.status_code = status_code
self.error = error
super().__init__(f'{status_code}: {error}')
class Token(BaseModel):
access_token: str
expires_in: int
token_type: Literal['Bearer']
class UserToken(Token):
refresh_token: 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
beatmapset: 'Beatmapset | BeatmapsetExtended | None' = None
checksum: str | None = None
current_user_playcount: int
max_combo: int
owners: list['BeatmapOwner']
class BeatmapDifficultyAttributes(BaseModel):
star_rating: float
max_combo: int
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
playcount: int
ranked: int
url: str
class BeatmapOwner(BaseModel):
id: int
username: str
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
beatmapsets: list['Beatmapset'] | None = None
user_completion_data: 'UserCompletionData | None' = None
class BeatmapPlaycount(BaseModel):
beatmap_id: int
beatmap: 'Beatmap | None' = None
beatmapset: 'Beatmapset | None' = None
count: int
class BeatmapScores(BaseModel):
scores: list['Score']
userScore: 'BeatmapUserScore | None' = None
class BeatmapUserScore(BaseModel):
position: int
score: 'Score'
class Beatmapset(BaseModel):
artist: str
artist_unicode: str
covers: 'Covers'
creator: str
favourite_count: int
id: int
nsfw: bool
offset: int
play_count: int
preview_url: str
source: str
status: str
spotlight: bool
title: str
title_unicode: str
user_id: int
video: bool
beatmaps: list['Beatmap | BeatmapExtended']
# converts
current_nominations: list['Nomination']
# current_user_attributes
# description
# discussions
# events
# genre
has_favourited: bool
# language
# nominations
pack_tags: list[str]
# ratings
# recent_favourites
# related_users
# user
track_id: int
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 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 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
class BeatmapsetDiscussionVote(BaseModel):
beatmapset_discussion_id: int
created_at: datetime
id: int
score: int
updated_at: datetime
user_id: int
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
has_favourited: bool
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
changelog_entries: list['ChangelogEntry'] | None = None
versions: list['Versions'] | 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
github_user: 'GithubUser | None' = None
message: str | None = None
message_html: 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
current_user_attributes: 'CurrentUserAttributes | None' = None
last_read_id: int | None = None
last_message_id: int | None = None
recent_messages: list['ChatMessage'] | None = None
users: list[int] | 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
sender: 'User'
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 User(BaseModel):
pass
class UserExtended(User):
pass
class UserGroup(BaseModel):
pass
class UserSilence(BaseModel):
pass
class UserStatistics(BaseModel):
pass
class WikiPage(BaseModel):
pass
class FavoriteBeatmaps(BaseModel):
beatmapset_ids: list[int]
class BeatmapPacks(BaseModel):
beatmap_packs: list[BeatmapPack]
class UserBeatmapScores(BaseModel):
scores: list['Score']

10
src/osuclient/scopes.py Normal file
View File

@ -0,0 +1,10 @@
CHAT_READ = 'chat.read'
CHAT_WRITE = 'chat.write'
CHAT_WRITE_MANAGE = 'chat.write_manage'
DELEGATE = 'delegate'
FORUM_WRITE = 'forum.write'
FORUM_WRITE_MANAGE = 'forum.write_manage'
FRIENDS_READ = 'friends.read'
GROUP_PERMISSIONS = 'group_permissions'
IDENTIFY = 'identify'
PUBLIC = 'public'