mirror of
https://github.com/RYGhub/royalnet.git
synced 2024-11-27 13:34:28 +00:00
125 lines
6.2 KiB
Python
125 lines
6.2 KiB
Python
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
|
|
|
|
|
|
class OsuCommand(LinkerCommand):
|
|
name = "osu"
|
|
|
|
description = "Connetti e sincronizza il tuo account di osu!"
|
|
|
|
@property
|
|
def client_id(self):
|
|
return self.config[self.name]['client_id']
|
|
|
|
@property
|
|
def client_secret(self):
|
|
return self.config[self.name]['client_secret']
|
|
|
|
@property
|
|
def base_url(self):
|
|
return self.config['base_url']
|
|
|
|
@property
|
|
def secret_key(self):
|
|
return self.config['secret_key']
|
|
|
|
async def get_updatables_of_user(self, session, user: rbt.User) -> List[Osu]:
|
|
return user.osu
|
|
|
|
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[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 account di osu! a {user.username}"
|
|
f"[/url]")
|
|
return 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_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_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_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].")
|