ongeki: proper handling of music ranking list

core/data/alembic/versions/9c42e54a27fe_remove_ongeki_static_music_ranking_list.py

@@ -0,0 +1,40 @@
+"""remove ongeki_static_music_ranking_list
+
+Revision ID: 9c42e54a27fe
+Revises: 41f77ef50588
+Create Date: 2025-01-06 18:24:16.306748
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision = '9c42e54a27fe'
+down_revision = '41f77ef50588'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index('ongeki_static_music_ranking_uk', table_name='ongeki_static_music_ranking_list')
+    op.drop_table('ongeki_static_music_ranking_list')
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('ongeki_static_music_ranking_list',
+    sa.Column('id', mysql.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('version', mysql.INTEGER(), autoincrement=False, nullable=False),
+    sa.Column('musicId', mysql.INTEGER(), autoincrement=False, nullable=False),
+    sa.Column('point', mysql.INTEGER(), autoincrement=False, nullable=False),
+    sa.Column('userName', mysql.VARCHAR(length=255), nullable=True),
+    sa.PrimaryKeyConstraint('id'),
+    mysql_collate='utf8mb4_0900_ai_ci',
+    mysql_default_charset='utf8mb4',
+    mysql_engine='InnoDB'
+    )
+    op.create_index('ongeki_static_music_ranking_uk', 'ongeki_static_music_ranking_list', ['version', 'musicId'], unique=True)
+    # ### end Alembic commands ###

titles/ongeki/base.py

@@ -157,19 +157,27 @@ class OngekiBase:
         return {"type": data["type"], "length": 0, "gameIdlistList": []}
 
     async def handle_get_game_ranking_api_request(self, data: Dict) -> Dict:
-        game_ranking_list = await self.data.static.get_ranking_list(self.version)
-        
-        ranking_list = []
-        for music in game_ranking_list:
-            tmp = music._asdict()
-            ranking_list.append(tmp)
+        try:
+            date = datetime.now(pytz.timezone('Asia/Tokyo')) - timedelta(days=1,hours=7)
 
-        if ranking_list is None:
-            return {"length": 0, "gameRankingList": []}
-        return {
-            "type": data["type"],
-            "gameRankingList": ranking_list,
-        }
+            # type 1 - current ranking; type 2 - previous ranking
+            if data["type"] == 2:
+                date = date - timedelta(1)
+
+            rankings = await self.data.score.get_rankings(date)
+
+            if not rankings or (data["type"] == 1 and len(rankings) < 10):
+                return {"type": data["type"], "gameRankingList": []}
+
+            ranking_list = []
+            for count, music_id in rankings:
+                ranking_list.append({"id": music_id, "point": count, "userName": ""})
+
+            return {"type": data["type"], "gameRankingList": ranking_list}
+
+        except Exception as e:
+            self.logger.error(f"Error while getting game ranking: {e}")
+            return {"type": data["type"], "gameRankingList": []}
 
     async def handle_get_game_point_api_request(self, data: Dict) -> Dict:
         get_game_point = await self.data.static.get_static_game_point()

titles/ongeki/schema/score.py

@@ -4,11 +4,13 @@ from sqlalchemy import Column, Table, UniqueConstraint
 from sqlalchemy.dialects.mysql import insert
 from sqlalchemy.engine import Row
 from sqlalchemy.schema import ForeignKey
-from sqlalchemy.sql import select
+from sqlalchemy.sql import select, func
 from sqlalchemy.types import TIMESTAMP, Boolean, Float, Integer, String
 
 from core.data.schema import BaseData, metadata
 
+from datetime import datetime, timedelta
+
 score_best: Table = Table(
     "ongeki_score_best",
     metadata,
@@ -209,6 +211,18 @@ class OngekiScoreData(BaseData):
             return None
         return result.lastrowid
 
+    async def get_rankings(self, date: datetime) -> Optional[List[Row]]:
+        sql = (
+            select([func.count(playlog.c.id), playlog.c.musicId])
+                .where(playlog.c.playDate == date.date())
+                .group_by(playlog.c.musicId)
+                .order_by(func.count(playlog.c.id).desc())
+                .limit(10)
+        )
+        result = await self.execute(sql)
+        if result:
+            return result.fetchall()
+
     async def put_playlog(self, aime_id: int, playlog_data: Dict) -> Optional[int]:
         playlog_data["user"] = aime_id
 

titles/ongeki/schema/static.py

@@ -98,18 +98,6 @@ cards = Table(
     mysql_charset="utf8mb4",
 )
 
-music_ranking = Table(
-    "ongeki_static_music_ranking_list",
-    metadata,
-    Column("id", Integer, primary_key=True, nullable=False),
-    Column("version", Integer, nullable=False),
-    Column("musicId", Integer, nullable=False),
-    Column("point", Integer, nullable=False),
-    Column("userName", String(255)),
-    UniqueConstraint("version", "musicId", name="ongeki_static_music_ranking_uk"),
-    mysql_charset="utf8mb4",
-)
-
 rewards = Table(
     "ongeki_static_rewards",
     metadata,
@@ -425,13 +413,6 @@ class OngekiStaticData(BaseData):
             return None
         return result.fetchone()
 
-    async def get_ranking_list(self, version: int) -> Optional[List[Dict]]:
-        sql = select(music_ranking.c.musicId.label('id'), music_ranking.c.point, music_ranking.c.userName).where(music_ranking.c.version == version)
-        result = await self.execute(sql)
-        if result is None:
-            return None
-        return result.fetchall()
-
     async def put_reward(self, version: int, rewardId: int, rewardname: str, itemKind: int, itemId: int) -> Optional[int]:
         sql = insert(rewards).values(
                 version=version,