Live show ranking.

npps4/game/ranking.py

@@ -4,7 +4,9 @@
 from .. import idol
 from .. import util
 from ..system import common
+from ..system import ranking
 from ..system import reward
+from ..system import unit
 from ..system import user
 
 
@@ -46,12 +48,59 @@ class RankingResponse(common.TimestampMixin, PageableMixin):
 @idol.register("ranking", "live")
 async def ranking_live(context: idol.SchoolIdolUserParams, request: RankingLiveRequest) -> RankingResponse:
     current_user = await user.get_current(context)
-    util.stub("ranking", "live", request)
+    total_cnt, player_scores = await ranking.get_live_ranking(context, request.live_difficulty_id, request.page)
+
+    rank_player_scores: list[RankingData] = []
+    for i, (user_id, score) in enumerate(player_scores, 1):
+        # TODO: Deduplicate with partyInfo code
+        target_user = await user.get(context, user_id)
+        if target_user is None:
+            continue
+
+        # Get unit center info
+        unit_center = await unit.get_unit_center(context, target_user)
+        if unit_center is None:
+            continue
+
+        unit_data = await unit.get_unit(context, unit_center)
+        unit.validate_unit(target_user, unit_data)
+        unit_info = await unit.get_unit_info(context, unit_data.unit_id)
+        if unit_info is None:
+            continue
+
+        removable_skills = await unit.get_unit_removable_skills(context, unit_data)
+        unit_full_data, unit_stats = await unit.get_unit_data_full_info(context, unit_data)
+
+        rank_player_scores.append(
+            RankingData(
+                rank=i,
+                score=score,
+                user_data=models.UserData(user_id=user_id, name=target_user.name, level=target_user.level),
+                center_unit_info=models.CenterUnitInfo(
+                    unit_id=unit_data.unit_id,
+                    level=unit_full_data.level,
+                    rank=unit_data.rank,
+                    love=unit_data.love,
+                    display_rank=unit_data.display_rank,
+                    unit_skill_exp=unit_data.skill_exp,
+                    unit_removable_skill_capacity=unit_data.unit_removable_skill_capacity,
+                    smile=unit_stats.smile,
+                    cute=unit_stats.pure,
+                    cool=unit_stats.cool,
+                    is_love_max=unit_full_data.is_love_max,
+                    is_level_max=unit_full_data.is_level_max,
+                    is_rank_max=unit_full_data.is_rank_max,
+                    removable_skill_ids=removable_skills,
+                ),
+                setting_award_id=target_user.active_award,
+            )
+        )
+
     return RankingResponse(
         page=request.page,
         rank=None,
-        items=[],
-        total_cnt=0,
+        items=rank_player_scores,
+        total_cnt=total_cnt,
         present_cnt=await reward.count_presentbox(context, current_user),
     )
 

npps4/system/ranking.py

@@ -0,0 +1,31 @@
+import sqlalchemy
+
+from .. import idol
+from ..db import main
+
+QUERY_PER_PAGE = 20
+
+
+async def get_live_ranking(context: idol.BasicSchoolIdolContext, live_difficulty_id: int, page: int):
+    q = (
+        sqlalchemy.select(sqlalchemy.func.count())
+        .select_from(main.LiveClear)
+        .where(main.LiveClear.live_difficulty_id == live_difficulty_id, main.LiveClear.clear_cnt > 0)
+    )
+    result = await context.db.main.execute(q)
+    total_cnt = result.scalar() or 0
+    player_scores: list[tuple[int, int]] = []
+
+    if total_cnt > 0:
+        q = (
+            sqlalchemy.select(main.LiveClear)
+            .where(main.LiveClear.live_difficulty_id == live_difficulty_id, main.LiveClear.clear_cnt > 0)
+            .order_by(main.LiveClear.hi_score.desc())
+            .limit(QUERY_PER_PAGE)
+            .offset(page * QUERY_PER_PAGE)
+        )
+        result = await context.db.main.execute(q)
+        for row in result.scalars():
+            player_scores.append((row.user_id, row.hi_score))
+
+    return total_cnt, player_scores