Refactor osu!direct handlers

.env.example

@@ -14,7 +14,7 @@ READ_DB_PORT=3306
 READ_DB_USER=
 READ_DB_PASS=
 READ_DB_NAME=
-DIRECT_URL=https://us.kitsu.moe/api
+DIRECT_URL=https://catboy.best/api
 API_KEYS_POOL=
 ALLOW_CUSTOM_CLIENTS=False
 SRV_URL=https://akatsuki.gg

app/api/direct.py

@@ -4,6 +4,7 @@
 from typing import Any
 from typing import Optional
 from urllib.parse import unquote_plus
+from urllib.parse import urlparse
 
 from fastapi import Depends
 from fastapi import Path
@@ -16,24 +17,12 @@
 import config
 from app.adapters import amplitude
 from app.constants.ranked_status import RankedStatus
+from app.models.cheesegull import CheesegullBeatmapset
 from app.models.user import User
+from app.usecases.cheesegull import format_beatmapset_to_direct
+from app.usecases.cheesegull import format_beatmapset_to_direct_card
 from app.usecases.user import authenticate_user
 
-USING_CHIMU = "https://api.chimu.moe/v1" == config.DIRECT_URL
-USING_KITSU = "https://us.kitsu.moe/api" == config.DIRECT_URL
-CHIMU_SET_ID_SPELLING = "SetId" if USING_CHIMU else "SetID"
-
-DIRECT_SET_INFO_FMTSTR = (
-    "{{{chimu_set_id_spelling}}}.osz|{{Artist}}|{{Title}}|{{Creator}}|"
-    "{{RankedStatus}}|10.0|{{LastUpdate}}|{{{chimu_set_id_spelling}}}|"
-    "0|{{HasVideo}}|0|0|0|{{diffs}}"
-).format(chimu_set_id_spelling="SetId" if USING_CHIMU else "SetID")
-
-DIRECT_MAP_INFO_FMTSTR = (
-    "[{DifficultyRating:.2f}⭐] {DiffName} "
-    "{{cs: {CS} / od: {OD} / ar: {AR} / hp: {HP}}}@{Mode}"
-)
-
 
 async def osu_direct(
     user: User = Depends(authenticate_user(Query, "u", "h")),
@@ -46,9 +35,6 @@ async def osu_direct(
 
     params: dict[str, Any] = {"amount": 101, "offset": page_num}
 
-    if "akatsuki.gg" in config.DIRECT_URL or "akatest.space" in config.DIRECT_URL:
-        params["osu_direct"] = True
-
     if unquote_plus(query) not in ("Newest", "Top Rated", "Most Played"):
         params["query"] = query
 
@@ -67,36 +53,18 @@ async def osu_direct(
         if response.status_code != status.HTTP_200_OK:
             return b"-1\nFailed to retrieve data from the beatmap mirror."
 
-        result = response.json()
-
-        # if USING_KITSU: # kitsu is kinda annoying here and sends status in body
-        #    if result["code"] != 200:
-        #        return b"-1\nFailed to retrieve data from the beatmap mirror."
-
+        beatmapsets = [
+            CheesegullBeatmapset.model_validate(beatmapset)
+            for beatmapset in response.json()
+        ]
     except asyncio.exceptions.TimeoutError:
         return b"-1\n3rd party beatmap mirror we depend on timed out. Their server is likely down."
 
-    result_len = len(result)
-    ret = [f"{'101' if result_len == 100 else result_len}"]
-
-    for bmap in result:
-        if not bmap["ChildrenBeatmaps"]:
-            continue
+    beatmapset_count = len(beatmapsets)
 
-        diff_sorted_maps = sorted(
-            bmap["ChildrenBeatmaps"],
-            key=lambda x: x["DifficultyRating"],
-        )
-
-        diffs_str = ",".join(
-            DIRECT_MAP_INFO_FMTSTR.format(**bm) for bm in diff_sorted_maps
-        )
-        ret.append(
-            DIRECT_SET_INFO_FMTSTR.format(
-                **bmap,
-                diffs=diffs_str,
-            ),
-        )
+    osu_direct_response = [
+        format_beatmapset_to_direct(beatmapset) for beatmapset in beatmapsets
+    ]
 
     if config.AMPLITUDE_API_KEY:
         asyncio.create_task(
@@ -115,7 +83,12 @@ async def osu_direct(
             ),
         )
 
-    return "\n".join(ret).encode()
+    # direct will only ever ask for 100 beatmaps
+    # but if it *should* fetch the next page then it wants a hint there's more beatmaps
+    direct_beatmapset_count = 101 if beatmapset_count >= 100 else beatmapset_count
+    beatmapset_count_line = f"{direct_beatmapset_count}\n"
+
+    return (beatmapset_count_line + "\n".join(osu_direct_response)).encode()
 
 
 async def beatmap_card(
@@ -130,14 +103,12 @@ async def beatmap_card(
 
         map_set_id = bmap.set_id
 
-    url = f"{config.DIRECT_URL}/{'set' if USING_CHIMU else 's'}/{map_set_id}"
+    url = f"{config.DIRECT_URL}/s/{map_set_id}"
     response = await app.state.services.http_client.get(url, timeout=5)
     if response.status_code != 200:
         return
 
-    result = response.json()
-
-    json_data = result["data"] if USING_CHIMU else result
+    beatmapset = CheesegullBeatmapset.model_validate(response.json())
 
     if config.AMPLITUDE_API_KEY:
         asyncio.create_task(
@@ -152,17 +123,13 @@ async def beatmap_card(
             ),
         )
 
-    return (
-        "{chimu_spell}.osz|{Artist}|{Title}|{Creator}|"
-        "{RankedStatus}|10.0|{LastUpdate}|{chimu_spell}|"
-        "0|0|0|0|0".format(**json_data, chimu_spell=json_data[CHIMU_SET_ID_SPELLING])
-    ).encode()
+    return format_beatmapset_to_direct_card(beatmapset).encode()
 
 
 async def download_map(set_id: str = Path(...)):
-    domain = config.DIRECT_URL.split("/")[2]
+    parsed_url = urlparse(config.DIRECT_URL)
 
     return RedirectResponse(
-        url=f"https://{domain}/d/{set_id}",
+        url=f"https://{parsed_url.netloc}/d/{set_id}",
         status_code=status.HTTP_301_MOVED_PERMANENTLY,
     )

app/models/___init__.py

@@ -2,6 +2,7 @@
 
 from . import achievement
 from . import beatmap
+from . import cheesegull
 from . import score
 from . import stats
 from . import user

app/models/cheesegull.py

@@ -0,0 +1,40 @@
+from pydantic import BaseModel
+from pydantic import Field
+
+from datetime import datetime
+
+class CheesegullBeatmap(BaseModel):
+    id: int = Field(..., alias="BeatmapID")
+    beatmapset_id: int = Field(..., alias="ParentSetID")
+    version: str = Field(..., alias="DiffName")
+    checksum: str = Field(..., alias="FileMD5")
+    mode: int = Field(..., alias="Mode")
+    bpm: float = Field(..., alias="BPM")
+    approach_rate: float = Field(..., alias="AR")
+    overall_difficulty: float = Field(..., alias="OD")
+    circle_size: float = Field(..., alias="CS")
+    health_points: float = Field(..., alias="HP")
+    total_length: int = Field(..., alias="TotalLength")
+    hit_length: int = Field(..., alias="HitLength")
+    play_count: int = Field(..., alias="Playcount")
+    pass_count: int = Field(..., alias="Passcount")
+    max_combo: int = Field(..., alias="MaxCombo")
+    difficulty_rating: float = Field(..., alias="DifficultyRating")
+
+
+class CheesegullBeatmapset(BaseModel):
+    id: int = Field(..., alias="SetID")
+    beatmaps: list[CheesegullBeatmap] = Field(..., alias="ChildrenBeatmaps")
+    ranked_status: int = Field(..., alias="RankedStatus")
+    approved_date: datetime = Field(..., alias="ApprovedDate")
+    last_update: datetime = Field(..., alias="LastUpdate")
+    last_checked: datetime = Field(..., alias="LastChecked")
+    artist: str = Field(..., alias="Artist")
+    title: str = Field(..., alias="Title")
+    creator: str = Field(..., alias="Creator")
+    source: str = Field(..., alias="Source")
+    tags: str = Field(..., alias="Tags")
+    has_video: bool = Field(..., alias="HasVideo")
+    genre: int | None = Field(None, alias="Genre")
+    language: int | None = Field(None, alias="Language")
+    favourites: int = Field(..., alias="Favourites")

app/usecases/__init__.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from . import beatmap
+from . import cheesegull
 from . import countries
 from . import discord
 from . import leaderboards

app/usecases/cheesegull.py

@@ -0,0 +1,32 @@
+from app.models.cheesegull import CheesegullBeatmap
+from app.models.cheesegull import CheesegullBeatmapset
+
+def format_beatmapset_to_direct(beatmapset: CheesegullBeatmapset) -> str:
+    # TODO: replace some of the placeholder values
+
+    difficulty_sorted_beatmapsets = sorted(
+        beatmapset.beatmaps,
+        key=lambda x: x.difficulty_rating,
+    )
+    formatted_beatmaps = ",".join(format_beatmap_to_direct(beatmap) for beatmap in difficulty_sorted_beatmapsets)
+
+    return (
+        f"{beatmapset.id}.osz|{beatmapset.artist}|{beatmapset.title}|{beatmapset.creator}|"
+        f"{beatmapset.ranked_status}|10.0|{beatmapset.last_update}|{beatmapset.id}|"
+        f"0|{beatmapset.has_video}|0|0|0|{formatted_beatmaps}"
+    )
+
+def format_beatmap_to_direct(beatmap: CheesegullBeatmap) -> str:
+    return (
+        f"[{beatmap.difficulty_rating:.2f}⭐] {beatmap.version} "
+        f"{{cs: {beatmap.circle_size} / od: {beatmap.overall_difficulty} / ar: {beatmap.approach_rate} / hp: {beatmap.health_points}}}@{beatmap.mode}"
+    )
+
+def format_beatmapset_to_direct_card(beatmapset: CheesegullBeatmapset) -> str:
+    # TODO: replace some of the placeholder values
+
+    return (
+        f"{beatmapset.id}.osz|{beatmapset.artist}|{beatmapset.title}|{beatmapset.creator}|"
+        f"{beatmapset.ranked_status}|10.0|{beatmapset.last_update}|{beatmapset.id}|"
+        "0|0|0|0|0"
+    )