feat/refactor: clean-up !with parameter parsing & add more param support

app/api/domains/osu.py

@@ -68,6 +68,7 @@
 from app.repositories import scores as scores_repo
 from app.repositories import stats as stats_repo
 from app.repositories import users as users_repo
+from app.repositories import first_place_scores as first_place_scores_repo
 from app.repositories.achievements import Achievement
 from app.usecases import achievements as achievements_usecases
 from app.usecases import user_achievements as user_achievements_usecases
@@ -812,10 +813,13 @@ async def osuSubmitModularSelector(
                 "client_flags": score.client_flags,
                 "user_id": score.player.id,
                 "perfect": score.perfect,
-                "checksum": score.client_checksum,
+                "checksum": score.client_checksum
             },
         )
 
+    #if score.rank == 1 and not player.restricted:
+    await first_place_scores_repo.create_or_update(bmap.md5, score.mode, score.id)
+
     if score.passed:
         replay_data = await replay_file.read()
 

app/commands.py

@@ -414,50 +414,36 @@ async def top(ctx: Context) -> str | None:
 class ParsingError(str): ...
 
 
-def parse__with__command_args(
+def parse__with__args(
     mode: int,
     args: Sequence[str],
-) -> Mapping[str, Any] | ParsingError:
+) -> ScoreParams | ParsingError:
     """Parse arguments for the !with command."""
-
-    if not args or len(args) > 4:
-        return ParsingError("Invalid syntax: !with <acc/nmiss/combo/mods ...>")
-
-    # !with 95% 1m 429x hddt
-    acc = mods = combo = nmiss = None
-
-    # parse acc, misses, combo and mods from arguments.
-    # tried to balance complexity vs correctness here
-    for arg in (str.lower(arg) for arg in args):
-        # mandatory suffix, combo & nmiss
-        if combo is None and arg.endswith("x") and arg[:-1].isdecimal():
-            combo = int(arg[:-1])
-            # if combo > bmap.max_combo:
-            #    return "Invalid combo."
-        elif nmiss is None and arg.endswith("m") and arg[:-1].isdecimal():
-            nmiss = int(arg[:-1])
-            # TODO: store nobjects?
-            # if nmiss > bmap.combo:
-            #    return "Invalid misscount."
+    score_args = ScoreParams(mode=mode)
+
+    for arg in args:
+        if arg.endswith("%") and arg[:-1].replace(".", "", 1).isdecimal():
+            score_args.acc = float(arg[:-1])
+            if not 0 <= score_args.acc <= 100:
+                return ParsingError("Invalid accuracy.")
+        elif arg.endswith("x") and arg[:-1].isdecimal(): # ignore mods like "dtrx"
+            score_args.combo = int(arg[:-1])
+        elif arg.endswith("m"):
+            score_args.nmiss = int(arg[:-1])
+        elif arg.endswith("x100"):
+            score_args.n100 = int(arg[:-4])
+        elif arg.endswith("x50"):
+            score_args.n50 = int(arg[:-3])
+        elif arg.endswith("xkatu"):
+            score_args.nkatu = int(arg[:-5])
+        elif arg.endswith("xgeki"):
+            score_args.ngeki = int(arg[:-5])
+        elif len(mods_str := arg.removeprefix("+")) % 2 == 0:
+            score_args.mods = Mods.from_modstr(mods_str).filter_invalid_combos(mode)
         else:
-            # optional prefix/suffix, mods & accuracy
-            arg_stripped = arg.removeprefix("+").removesuffix("%")
-            if mods is None and arg_stripped.isalpha() and len(arg_stripped) % 2 == 0:
-                mods = Mods.from_modstr(arg_stripped)
-                mods = mods.filter_invalid_combos(mode)
-            elif acc is None and arg_stripped.replace(".", "", 1).isdecimal():
-                acc = float(arg_stripped)
-                if not 0 <= acc <= 100:
-                    return ParsingError("Invalid accuracy.")
-            else:
-                return ParsingError(f"Unknown argument: {arg}")
-
-    return {
-        "acc": acc,
-        "mods": mods,
-        "combo": combo,
-        "nmiss": nmiss,
-    }
+            return ParsingError(f"Invalid parameter '{arg}'.")
+
+    return score_args
 
 
 @command(Privileges.UNRESTRICTED, aliases=["w"], hidden=True)
@@ -480,41 +466,16 @@ async def _with(ctx: Context) -> str | None:
 
     mode_vn = ctx.player.last_np["mode_vn"]
 
-    command_args = parse__with__command_args(mode_vn, ctx.args)
-    if isinstance(command_args, ParsingError):
-        return str(command_args)
-
-    msg_fields = []
-
-    score_args = ScoreParams(mode=mode_vn)
-
-    mods = command_args["mods"]
-    if mods is not None:
-        score_args.mods = mods
-        msg_fields.append(f"{mods!r}")
-
-    nmiss = command_args["nmiss"]
-    if nmiss:
-        score_args.nmiss = nmiss
-        msg_fields.append(f"{nmiss}m")
-
-    combo = command_args["combo"]
-    if combo is not None:
-        score_args.combo = combo
-        msg_fields.append(f"{combo}x")
-
-    acc = command_args["acc"]
-    if acc is not None:
-        score_args.acc = acc
-        msg_fields.append(f"{acc:.2f}%")
+    score_args = parse__with__args(mode_vn, ctx.args)
+    if isinstance(score_args, ParsingError):
+        return str(score_args)
 
     result = app.usecases.performance.calculate_performances(
         osu_file_path=str(BEATMAPS_PATH / f"{bmap.id}.osu"),
         scores=[score_args],  # calculate one score
     )
 
-    return "{msg}: {pp:.2f}pp ({stars:.2f}*)".format(
-        msg=" ".join(msg_fields),
+    return "{pp:.2f}pp ({stars:.2f}*)".format(
         pp=result[0]["performance"]["pp"],
         stars=result[0]["difficulty"]["stars"],  # (first score result)
     )

app/objects/score.py

@@ -325,7 +325,6 @@ def calculate_performance(self, beatmap_id: int) -> tuple[float, float]:
             mods=int(self.mods),
             combo=self.max_combo,
             ngeki=self.ngeki,
-            n300=self.n300,
             nkatu=self.nkatu,
             n100=self.n100,
             n50=self.n50,

app/repositories/first_place_scores.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from typing import TypedDict
+from typing import cast
+
+from sqlalchemy import Column
+from sqlalchemy import Index
+from sqlalchemy import BigInteger
+from sqlalchemy import SmallInteger
+from sqlalchemy import String
+from sqlalchemy.dialects.mysql import insert as mysql_insert
+from sqlalchemy import select
+from sqlalchemy import and_
+from sqlalchemy import PrimaryKeyConstraint
+
+import app.state.services
+from app.repositories import Base
+from app.constants.gamemodes import GameMode
+
+
+class FirstPlaceScoresTable(Base):
+    __tablename__ = "first_place_scores"
+
+    map_md5 = Column("map_md5", String(32), nullable=False)
+    mode = Column("mode", SmallInteger, nullable=False)
+    score_id = Column("score_id", BigInteger, nullable=False)
+
+    __table_args__ = (
+        Index("first_place_scores_map_md5_mode_index", map_md5, mode),
+        PrimaryKeyConstraint(map_md5, mode)
+    )
+
+
+READ_PARAMS = (
+    FirstPlaceScoresTable.map_md5,
+    FirstPlaceScoresTable.mode,
+    FirstPlaceScoresTable.score_id
+)
+
+
+class FirstPlaceScore(TypedDict):
+    map_md5: str
+    mode: int
+    score_id: int
+
+
+async def create_or_update(
+    map_md5: str,
+    mode: int,
+    score_id: int
+) -> None:
+    insert_stmt = mysql_insert(FirstPlaceScoresTable).values(
+        map_md5=map_md5,
+        mode=mode,
+        score_id=score_id
+    ).on_duplicate_key_update(
+        score_id=score_id
+    )
+    print(insert_stmt)
+    await app.state.services.database.execute(insert_stmt)
+
+
+async def fetch_one(map_md5: str, mode: GameMode) -> FirstPlaceScore | None:
+    select_stmt = select(*READ_PARAMS).where(and_(FirstPlaceScoresTable.map_md5 == map_md5, FirstPlaceScoresTable.mode == mode))
+    score = await app.state.services.database.fetch_one(select_stmt)
+    return cast(FirstPlaceScore | None, score)

app/repositories/stats.py

@@ -225,6 +225,8 @@ async def partial_update(
 
     await app.state.services.database.execute(update_stmt)
 
+    print(update_stmt)
+
     select_stmt = (
         select(*READ_PARAMS)
         .where(StatsTable.id == player_id)

app/usecases/performance.py

@@ -17,11 +17,9 @@ class ScoreParams:
     mods: int | None = None
     combo: int | None = None
 
-    # caller may pass either acc OR 300/100/50/geki/katu/miss
-    # passing both will result in a value error being raised
+    # caller may pass either acc OR 100/50/geki/katu/miss
     acc: float | None = None
 
-    n300: int | None = None
     n100: int | None = None
     n50: int | None = None
     ngeki: int | None = None
@@ -76,10 +74,10 @@ def calculate_performances(
 
     for score in scores:
         if score.acc and (
-            score.n300 or score.n100 or score.n50 or score.ngeki or score.nkatu
+            score.n100 or score.n50 or score.ngeki or score.nkatu
         ):
             raise ValueError(
-                "Must not specify accuracy AND 300/100/50/geki/katu. Only one or the other.",
+                "Must not specify accuracy AND 100/50/geki/katu. Only one or the other.",
             )
 
         # rosupp ignores NC and requires DT
@@ -92,7 +90,6 @@ def calculate_performances(
             mods=score.mods or 0,
             combo=score.combo,
             acc=score.acc,
-            n300=score.n300,
             n100=score.n100,
             n50=score.n50,
             n_geki=score.ngeki,

migrations/base.sql

@@ -210,6 +210,16 @@ create table ratings
 	primary key (userid, map_md5)
 );
 
+create table first_place_scores
+(
+    map_md5 char(32) not null,
+    mode tinyint not null,
+    score_id bigint not null,
+    primary key(map_md5, mode)
+);
+create index first_place_scores_map_md5_mode_index
+	on first_place_scores (map_md5, mode);
+
 create table scores
 (
 	id bigint unsigned auto_increment

migrations/migrations.sql

@@ -475,3 +475,14 @@ create index users_country_index
 # v5.2.2
 create index scores_fetch_leaderboard_generic_index
 	on scores (map_md5, status, mode);
+
+# v5.2.3
+create table first_place_scores
+(
+    map_md5 char(32) not null,
+    mode tinyint not null,
+    score_id bigint not null,
+    primary key(map_md5, mode)
+);
+create index first_place_scores_map_md5_mode_index
+	on first_place_scores (map_md5, mode);

pyproject.toml

@@ -34,7 +34,7 @@ profile = "black"
 
 [tool.poetry]
 name = "bancho-py"
-version = "5.2.2"
+version = "5.2.3"
 description = "An osu! server implementation optimized for maintainability in modern python"
 authors = ["Akatsuki Team"]
 license = "MIT"

tests/unit/commands_test.py

@@ -0,0 +1,28 @@
+import pytest
+
+import app.commands
+from app.usecases.performance import ScoreParams
+
+@pytest.mark.parametrize(
+    ("test_input", "expected"),
+    [
+        (
+            # covers all parameters
+            {"mode": 0, "args": "+hddtezfl 600x 99.37% 5x100 4x50 3xgeki 1xkatu 7m "},
+            ScoreParams(mode=0, mods=4206, combo=600, acc=99.37, n100=5, n50=4, ngeki=3, nkatu=1, nmiss=7)
+        ),
+
+        (
+            # specifically covers different mode & mods without "+" prefix
+            {"mode": 1, "args": "hdhr"},
+            ScoreParams(mode=1, mods=30)
+        ),
+        (
+            # accuracy out of range
+            {"mode": 0, "args": "100.0001%"},
+            app.commands.ParsingError("Invalid accuracy.")
+        )
+    ]
+)
+def test_parse__with__args(test_input, expected):
+    assert app.commands.parse__with__args(**test_input) == expected