ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00
Code Activity

Implement osu updater

Browse source
This commit is contained in:
Steffo 2020-07-23 01:14:25 +02:00
parent 347fd489ca
commit 7afc8f81e5
4 changed files with 131 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
pyproject.toml
View file

@ -2,7 +2,7 @@
[tool.poetry]
name = "royalpack"
version = "5.13.0"
version = "5.13.1"
description = "A Royalnet command pack for the Royal Games community"
authors = ["Stefano Pigozzi <ste.pigozzi@gmail.com>"]
license = "AGPL-3.0+"
@ -23,6 +23,8 @@
riotwatcher = "^3.0.0"
royalspells = "^3.2"
steam = "*"
sqlalchemy = "^1.3.18"
bcrypt = "^3.1.7"
[tool.poetry.dependencies.royalnet]
version = "~5.10.4"

92
royalpack/commands/osu.py
View file

@ -1,11 +1,14 @@
from typing import *
import itsdangerous
import aiohttp
from royalnet.backpack import tables as rbt
import royalnet.commands as rc
import royalnet.utils as ru
from .abstract.linker import LinkerCommand
from ..types import Updatable
from ..tables import Osu
from ..stars.api_auth_login_osu import ApiAuthLoginOsuStar
@ -30,42 +33,93 @@ class OsuCommand(LinkerCommand):
def secret_key(self):
return self.config['secret_key']
async def get_updatables_of_user(self, session, user: rbt.User) -> List[Updatable]:
return []
async def get_updatables_of_user(self, session, user: rbt.User) -> List[Osu]:
return user.osu
async def get_updatables(self, session) -> List[Updatable]:
return []
async def get_updatables(self, session) -> List[Osu]:
return await ru.asyncify(session.query(self.alchemy.get(Osu)).all)
async def create(self,
session,
user: rbt.User,
args: rc.CommandArgs,
data: Optional[rc.CommandData] = None) -> Optional[Updatable]:
data: Optional[rc.CommandData] = None) -> Optional[Osu]:
serializer = itsdangerous.URLSafeSerializer(self.secret_key, salt="osu")
# TODO: Ensure the chat the link is being sent in is secure!!!
await data.reply("🔑 [b]Login necessario[/b]\n"
f"[url=https://osu.ppy.sh/oauth/authorize"
f"?client_id={self.client_id}"
f"&redirect_uri={self.base_url}{ApiAuthLoginOsuStar.path}"
f"&response_type=code"
f"&state={serializer.dumps(user.uid)}]"
f"Connetti osu! a Royalnet"
f"Connetti osu! all'account {user.username}"
f"[/url]")
return None
async def update(self, session, obj, change: Callable[[str, Any], Awaitable[None]]):
async def update(self, session, obj: Osu, change: Callable[[str, Any], Awaitable[None]]):
await obj.refresh_if_expired(client_id=self.client_id,
client_secret=self.client_secret,
base_url=self.base_url,
path=ApiAuthLoginOsuStar.path)
async with aiohttp.ClientSession(headers={"Authorization": f"Bearer {obj.access_token}"}) as session:
async with session.get("https://osu.ppy.sh/api/v2/me/osu") as response:
m = await response.json()
obj.avatar_url = m["avatar_url"]
obj.username = m["username"]
if "statistics" in m:
await change("standard_pp", m["statistics"].get("pp"))
async with session.get("https://osu.ppy.sh/api/v2/me/taiko") as response:
m = await response.json()
if "statistics" in m:
await change("taiko_pp", m["statistics"].get("pp"))
async with session.get("https://osu.ppy.sh/api/v2/me/fruits") as response:
m = await response.json()
if "statistics" in m:
await change("catch_pp", m["statistics"].get("pp"))
async with session.get("https://osu.ppy.sh/api/v2/me/mania") as response:
m = await response.json()
if "statistics" in m:
await change("mania_pp", m["statistics"].get("pp"))
async def on_increase(self, session, obj: Osu, attribute: str, old: Any, new: Any) -> None:
if attribute == "standard_pp":
await self.notify(f"📈 [b]{obj.user}[/b] è salito a [b]{new:.0f}pp[/b] su [i]osu![/i]! Congratulazioni!")
elif attribute == "taiko_pp":
await self.notify(f"📈 [b]{obj.user}[/b] è salito a [b]{new:.0f}pp[/b] su [i]osu!taiko[/i]! Congratulazioni!")
elif attribute == "catch_pp":
await self.notify(f"📈 [b]{obj.user}[/b] è salito a [b]{new:.0f}pp[/b] su [i]osu!catch[/i]! Congratulazioni!")
elif attribute == "mania_pp":
await self.notify(f"📈 [b]{obj.user}[/b] è salito a [b]{new:.0f}pp[/b] su [i]osu!mania[/i]! Congratulazioni!")
async def on_unchanged(self, session, obj: Osu, attribute: str, old: Any, new: Any) -> None:
pass
async def on_increase(self, session, obj: Updatable, attribute: str, old: Any, new: Any) -> None:
pass
async def on_decrease(self, session, obj: Osu, attribute: str, old: Any, new: Any) -> None:
if attribute == "standard_pp":
await self.notify(f"📉 [b]{obj.user}[/b] è sceso a [b]{new:.0f}pp[/b] su [i]osu![/i].")
elif attribute == "taiko_pp":
await self.notify(f"📉 [b]{obj.user}[/b] è sceso a [b]{new:.0f}pp[/b] su [i]osu!taiko[/i].")
elif attribute == "catch_pp":
await self.notify(f"📉 [b]{obj.user}[/b] è sceso a [b]{new:.0f}pp[/b] su [i]osu!catch[/i].")
elif attribute == "mania_pp":
await self.notify(f"📉 [b]{obj.user}[/b] è sceso a [b]{new:.0f}pp[/b] su [i]osu!mania[/i].")
async def on_unchanged(self, session, obj: Updatable, attribute: str, old: Any, new: Any) -> None:
pass
async def on_first(self, session, obj: Osu, attribute: str, old: None, new: Any) -> None:
if attribute == "standard_pp":
await self.notify(f"⭐️ [b]{obj.user}[/b] ha guadagnato i suoi primi [b]{new:.0f}pp[/b] su [i]osu![/i]!")
elif attribute == "taiko_pp":
await self.notify(f"⭐️ [b]{obj.user}[/b] ha guadagnato i suoi primi [b]{new:.0f}pp[/b] su [i]osu!taiko[/i]!")
elif attribute == "catch_pp":
await self.notify(f"⭐️ [b]{obj.user}[/b] ha guadagnato i suoi primi [b]{new:.0f}pp[/b] su [i]osu!catch[/i]!")
elif attribute == "mania_pp":
await self.notify(f"⭐️ [b]{obj.user}[/b] ha guadagnato i suoi primi [b]{new:.0f}pp[/b] su [i]osu!mania[/i]!")
async def on_decrease(self, session, obj: Updatable, attribute: str, old: Any, new: Any) -> None:
pass
async def on_first(self, session, obj: Updatable, attribute: str, old: None, new: Any) -> None:
pass
async def on_reset(self, session, obj: Updatable, attribute: str, old: Any, new: None) -> None:
pass
async def on_reset(self, session, obj: Osu, attribute: str, old: Any, new: None) -> None:
if attribute == "standard_pp":
await self.notify(f"⬜️ [b]{obj.user}[/b] non è più classificato su [i]osu![/i].")
elif attribute == "taiko_pp":
await self.notify(f" ⬜️[b]{obj.user}[/b] non è più classificato su [i]osu!taiko[/i].")
elif attribute == "catch_pp":
await self.notify(f"⬜️ [b]{obj.user}[/b] non è più classificato su [i]osu!catch[/i].")
elif attribute == "mania_pp":
await self.notify(f"⬜️ [b]{obj.user}[/b] non è più classificato su [i]osu!mania[/i].")

17
royalpack/stars/api_auth_login_osu.py
View file

@ -1,6 +1,7 @@
import royalnet.utils as ru
import royalnet.backpack.tables as rbt
import royalnet.constellation.api as rca
import royalnet.constellation.api.apierrors as rcae
import itsdangerous
import aiohttp
import aiohttp.client_exceptions
@ -46,6 +47,7 @@ class ApiAuthLoginOsuStar(rca.ApiStar):
async def get(self, data: rca.ApiData) -> ru.JSON:
"""Login to Royalnet with your osu! account."""
OsuT = self.alchemy.get(Osu)
TokenT = self.alchemy.get(rbt.Token)
code = data.str("code")
state = data.str("state", optional=True)
@ -81,6 +83,17 @@ class ApiAuthLoginOsuStar(rca.ApiStar):
)
data.session.add(osu)
await data.session_commit()
else:
osu = await ru.asyncify(
data.session.query(OsuT).filter_by(osu_id=m["id"]).all
)
if osu is None:
raise rcae.ForbiddenError("Unknown osu! account")
user = osu.user
raise rca.MethodNotImplementedError()
token: rbt.Token = TokenT.generate(alchemy=self.alchemy, user=user, expiration_delta=datetime.timedelta(days=7))
data.session.add(token)
await data.session_commit()
return token.json()

41
royalpack/tables/osu.py
View file

@ -38,7 +38,27 @@ class Osu(Updatable):
@declared_attr
def username(self):
return Column(String)
return Column(String, nullable=False)
@declared_attr
def avatar_url(self):
return Column(String, nullable=False)
@declared_attr
def standard_pp(self):
return Column(Float)
@declared_attr
def taiko_pp(self):
return Column(Float)
@declared_attr
def catch_pp(self):
return Column(Float)
@declared_attr
def mania_pp(self):
return Column(Float)
async def refresh(self, *, client_id, client_secret, base_url, path):
j = await oauth_refresh(url="https://osu.ppy.sh/oauth/token",
@ -53,3 +73,22 @@ class Osu(Updatable):
async def refresh_if_expired(self, *, client_id, client_secret, base_url, path):
if datetime.datetime.now() >= self.expiration_date:
await self.refresh(client_id=client_id, client_secret=client_secret, base_url=base_url, path=path)
def json(self) -> dict:
return {
"osu_id": self.osu_id,
"username": self.username,
"avatar_url": self.avatar_url,
"standard": {
"pp": self.standard_pp,
},
"taiko": {
"pp": self.taiko_pp,
},
"catch": {
"pp": self.catch_pp,
},
"mania": {
"pp": self.mania_pp,
},
}