Implement "First clear" loveca bonus.

Fixes #36.

Author: MikuAuahDark

npps4/game/live.py

@@ -1,4 +1,5 @@
 import copy
+import itertools
 import json
 import math
 
@@ -13,6 +14,7 @@
 from ..system import class_system as class_system_module
 from ..system import common
 from ..system import effort
+from ..system import item
 from ..system import live
 from ..system import live_model
 from ..system import museum
@@ -588,6 +590,30 @@ async def live_reward(context: idol.SchoolIdolUserParams, request: LiveRewardReq
     live_clear_data.hi_combo_cnt = max(live_clear_data.hi_combo_cnt, request.max_combo)
     live_clear_data.clear_cnt = live_clear_data.clear_cnt + 1
 
+    special_reward: list[common.AnyItem] = []
+    if live_clear_data.clear_cnt == 1:
+        # First clear. Give loveca only if:
+        # * User cleared Easy, Normal, and Hard
+        # * User cleared Expert
+        # * User cleared Master
+        if live_setting.difficulty > 3:
+            # User cleared Expert or higher. Give loveca directly.
+            special_reward.append(item.loveca(1))
+        else:
+            # User cleared Easy, Normal, or Hard. Only give loveca if all is cleared.
+            enh_list = (await live.get_enh_live_difficulty_ids(context, request.live_difficulty_id)).copy()
+            enh_list[live_setting.difficulty] = request.live_difficulty_id
+            cleared = True
+
+            for i in range(1, 4):
+                live_clear_data_adjacent = await live.get_live_clear_data(context, current_user, enh_list[i])
+                if live_clear_data_adjacent is None or live_clear_data_adjacent.clear_cnt == 0:
+                    cleared = False
+                    break
+
+            if cleared:
+                special_reward.append(item.loveca(1))
+
     # Get accomplished live goals
     old_live_goals = set(await live.get_achieved_goal_id_list(context, old_live_clear_data))
     new_live_goals = set(await live.get_achieved_goal_id_list(context, live_clear_data))
@@ -598,7 +624,7 @@ async def live_reward(context: idol.SchoolIdolUserParams, request: LiveRewardReq
     )
 
     # Give live goal rewards
-    for reward_data in live_goal_rewards:
+    for reward_data in itertools.chain(live_goal_rewards, special_reward):
         await advanced.add_item(context, current_user, reward_data)
 
     # This is the intended EXP and G drop
@@ -815,7 +841,7 @@ async def live_reward(context: idol.SchoolIdolUserParams, request: LiveRewardReq
         goal_accomp_info=LiveRewardGoalAccomplishedInfo(
             achieved_ids=accomplished_live_goals, rewards=live_goal_rewards
         ),
-        special_reward_info=[],  # TODO: Give 1 loveca on clearing this track for the first time.
+        special_reward_info=special_reward,
         accomplished_achievement_list=await achievement.to_game_representation(
             context, accomplished_achievement.accomplished, accomplished_achievement_rewards
         ),

npps4/game/profile.py

@@ -69,7 +69,7 @@ async def profile_livecount(context: idol.SchoolIdolUserParams, request: Profile
     if target_user is None:
         raise idol.error.by_code(idol.error.ERROR_CODE_USER_NOT_EXIST)
 
-    cleared = await live.get_cleard_live_count(context, target_user)
+    cleared = await live.get_cleared_live_count(context, target_user)
     return ProfileLiveCountResponse(
         [ProfileLiveCount(difficulty=i, clear_cnt=cleared.get(i, 0)) for i in (1, 2, 3, 4, 6)]
     )

npps4/system/live.py

@@ -44,10 +44,6 @@ async def init(context: idol.BasicSchoolIdolContext, user: main.User):
     for normallive in result.scalars():
         setting_data = await context.db.live.get(live.LiveSetting, normallive.live_setting_id)
         assert setting_data is not None
-        # live_clear = main.LiveClear(
-        #     user_id=user.id, live_difficulty_id=normallive.live_difficulty_id, difficulty=setting_data.difficulty
-        # )
-        # context.db.main.add(live_clear)
         if setting_data.live_track_id not in unlocked:
             if await unlock_normal_live(context, user, setting_data.live_track_id):
                 unlocked.add(setting_data.live_track_id)
@@ -618,9 +614,41 @@ async def pull_precise_score_with_beatmap(
     return json.loads(gzip.decompress(replay.precise_log)), notes_list, replay.timestamp
 
 
-async def get_cleard_live_count(context: idol.BasicSchoolIdolContext, /, user: main.User) -> dict[int, int]:
+async def get_cleared_live_count(context: idol.BasicSchoolIdolContext, /, user: main.User) -> dict[int, int]:
     q = sqlalchemy.select(main.LiveClear.difficulty, sqlalchemy.func.count(main.LiveClear.live_difficulty_id)).group_by(
         main.LiveClear.difficulty
     )
     result = await context.db.main.execute(q)
     return {r[0]: r[1] for r in result}
+
+
+@common.context_cacheable("adjacent_live_difficulty_id")
+async def get_enh_live_difficulty_ids(context: idol.BasicSchoolIdolContext, /, live_difficulty_id: int):
+    output: dict[int, int] = {}
+
+    live_info_base = await get_live_info_table(context, live_difficulty_id)
+    if live_info_base is None:
+        raise ValueError(f"invalid live_difficulty_id {live_difficulty_id}")
+
+    live_setting = await get_live_setting(context, live_info_base.live_setting_id)
+    if live_setting is None:
+        raise ValueError(f"invalid live_setting_id {live_info_base.live_setting_id}")
+
+    q = sqlalchemy.select(live.LiveSetting).where(
+        live.LiveSetting.live_track_id == live_setting.live_track_id, live.LiveSetting.difficulty <= 3
+    )
+    live_settings = (await context.db.live.execute(q)).scalars().all()
+    live_setting_map = {l.live_setting_id: l for l in live_settings}
+
+    live_info_type = type(live_info_base)
+    q = sqlalchemy.select(live_info_type).where(
+        live_info_type.live_setting_id.in_([l.live_setting_id for l in live_settings])
+    )
+    live_infos = (await context.db.live.execute(q)).scalars().all()
+
+    for live_info in live_infos:
+        output[live_setting_map[live_info.live_setting_id].difficulty] = live_info.live_difficulty_id
+        # Also set cache for the retrieved live difficulty ids
+        context.set_cache("adjacent_live_difficulty_id", live_info.live_difficulty_id, output)
+
+    return output

scripts/migrations/7_give_loveca_to_cleared_songs.py

@@ -0,0 +1,61 @@
+import npps4.script_dummy  # isort:skip
+
+import sqlalchemy
+
+import npps4.idol
+import npps4.db.main
+import npps4.system.item
+import npps4.system.live
+import npps4.system.reward
+
+revision = "7_give_loveca_to_cleared_songs"
+prev_revision = "6_send_subunit_rewards_take2"
+
+FIXES_ACHIEVEMENT_IDS = [10090010, *range(10090012, 10090019)]
+
+
+async def main(context: npps4.idol.BasicSchoolIdolContext):
+    q = sqlalchemy.select(npps4.db.main.User)
+
+    async for target_user in await context.db.main.stream_scalars(q):
+        live_difficulty_id_checked: set[int] = set()
+        loveca = 0
+
+        q = sqlalchemy.select(npps4.db.main.LiveClear).where(npps4.db.main.LiveClear.user_id == target_user.id)
+        async for live_clear in await context.db.main.stream_scalars(q):
+            if live_clear.live_difficulty_id in live_difficulty_id_checked:
+                continue
+
+            live_setting = await npps4.system.live.get_live_setting_from_difficulty_id(
+                context, live_clear.live_difficulty_id
+            )
+            if live_setting is None:
+                raise ValueError(f"invalid live_difficulty_id {live_clear.live_difficulty_id}")
+
+            if live_setting.difficulty > 3:
+                # User cleared Expert or higher. Give loveca directly.
+                loveca = loveca + 1
+            else:
+                # User cleared Easy, Normal, or Hard. Only give loveca if all is cleared.
+                enh_list = (
+                    await npps4.system.live.get_enh_live_difficulty_ids(context, live_clear.live_difficulty_id)
+                ).copy()
+                enh_list[live_setting.difficulty] = live_clear.live_difficulty_id
+                cleared = True
+
+                for i in range(1, 4):
+                    live_clear_data_adjacent = await npps4.system.live.get_live_clear_data(
+                        context, target_user, enh_list[i]
+                    )
+                    if live_clear_data_adjacent is None or live_clear_data_adjacent.clear_cnt == 0:
+                        cleared = False
+                        break
+
+                if cleared:
+                    loveca = loveca + 1
+                    live_difficulty_id_checked.update(enh_list.values())
+
+        if loveca > 0:
+            print(f"Given {loveca} loveca to user {target_user.id} ({target_user.invite_code})")
+            item = npps4.system.item.loveca(loveca)
+            await npps4.system.reward.add_item(context, target_user, item, "First Live Clear Bonuses")