Update 卖

titles/mai2/base.py

@@ -77,40 +77,33 @@ class Mai2Base:
         }
 
     async def handle_get_game_ranking_api_request(self, data: Dict) -> Dict:
-        conn = pymysql.connect(
-            host=self.core_config.database.host,
-            port=self.core_config.database.port,
-            user=self.core_config.database.username,
-            password=self.core_config.database.password,
-            database=self.core_config.database.name,
-            charset='utf8mb4'  
-        )
         try:
-            cursor = conn.cursor()
+            playlogs = await self.data.score.get_playlogs(user_id=None) 
+            ranking_list = []
 
-            query = """
+            if not playlogs:
-            SELECT musicid AS id, COUNT(*) AS point
+                self.logger.warning("No playlogs found.")
-            FROM mai2_playlog
+                return {"length": 0, "gameRankingList": []}
-            GROUP BY musicid
-            ORDER BY point DESC
-            LIMIT 100
-            """
-            cursor.execute(query)
 
-            results = cursor.fetchall()
+            music_count = {}
-            ranking_list = [{"id": row[0], "point": row[1], "userName": ""} for row in results]
+            for log in playlogs:
-            output = {
+                music_id = log.musicId  
+                music_count[music_id] = music_count.get(music_id, 0) + 1
+
+            sorted_music = sorted(music_count.items(), key=lambda item: item[1], reverse=True)
+
+            for music_id, count in sorted_music[:100]: 
+                ranking_list.append({"id": music_id, "point": count, "userName": ""})
+
+            return {
                 "type": 1,
                 "gameRankingList": ranking_list,
                 "gameRankingInstantList": None
             }
 
-            cursor.close()
-            conn.close()
-            return output
-
         except Exception as e:
-            return {'length': 0, 'gameRankingList': []}
+            self.logger.error(f"Error while getting game ranking: {e}")
+            return {"length": 0, "gameRankingList": []}
 
     async def handle_get_game_tournament_info_api_request(self, data: Dict) -> Dict:
         # TODO: Tournament support