From 6988c1a6a778a2b22dc408faff46b28d9130206a Mon Sep 17 00:00:00 2001 From: Miku AuahDark Date: Wed, 22 Jan 2025 14:37:03 +0800 Subject: [PATCH] Remove old achievement checker code. --- npps4/system/achievement.py | 451 ------------------------------------ 1 file changed, 451 deletions(-) diff --git a/npps4/system/achievement.py b/npps4/system/achievement.py index c99a5bb..5a3769f 100644 --- a/npps4/system/achievement.py +++ b/npps4/system/achievement.py @@ -1376,457 +1376,6 @@ async def check(context: idol.BasicSchoolIdolContext, target_user: main.User, /, return container + type_53_container -async def test_params(ach_info: achievement.Achievement, args: collections.abc.Sequence[int | None]): - if args: - for i in range(min(len(args), 11)): - if args[i] is not None: - if getattr(ach_info, f"params{i + 1}", None) != args[i]: - return False - - return True - - -async def check_type_countable( - context: idol.BasicSchoolIdolContext, - user: main.User, - achievement_type: int, - count: int, - pindex: int = 1, - *args: int | None, - test: Callable[ - [achievement.Achievement, collections.abc.Sequence[int | None]], collections.abc.Awaitable[bool] - ] = test_params, -): - time = util.time() - await update_resettable_achievement(context, user, time) - - q = sqlalchemy.select(main.Achievement).where( - main.Achievement.user_id == user.id, - main.Achievement.achievement_type == achievement_type, - main.Achievement.is_accomplished == False, - ) - result = await context.db.main.execute(q) - - achieved: list[main.Achievement] = [] - new: list[main.Achievement] = [] - - for ach in result.scalars(): - ach_info = await get_achievement_info(context, ach.achievement_id) - if ach_info is None: - raise ValueError("achievement info is none, database is corrupted?") - - if await test(ach_info, args): - target_amount = int(getattr(ach_info, f"params{pindex}", None) or 1) - if count >= target_amount: - # Achieved. - ach.count = min(count, target_amount) - ach.is_accomplished = True - achieved.append(ach) - - # Update reset value - match ach.reset_type: - case 1: - ach.reset_value = util.get_days_since_unix(time) - case 2: - ach.reset_value = util.get_weeks_since_unix(time) - case 3: - ach.reset_value = util.get_months_since_unix(time) - - # New achievement - for next_ach_id in await get_next_achievement_ids(context, ach.achievement_id): - new_ach_info = await get_achievement_info(context, next_ach_id) - if new_ach_info is not None and not await has_achievement( - context, user, new_ach_info.achievement_id - ): - new_ach = await add_achievement(context, user, new_ach_info, time) - # Append to new achievement - new.append(new_ach) - else: - ach.count = count - - await context.db.main.flush() - - await context.db.main.flush() - return AchievementContext(accomplished=achieved, new=new) - - -async def check_type_increment( - context: idol.BasicSchoolIdolContext, - user: main.User, - achievement_type: int, - increment: bool, - pindex: int = 1, - *args: int | None, - test: Callable[ - [achievement.Achievement, collections.abc.Sequence[int | None]], collections.abc.Awaitable[bool] - ] = test_params, -): - q = sqlalchemy.select(main.Achievement).where( - main.Achievement.user_id == user.id, - main.Achievement.achievement_type == achievement_type, - main.Achievement.is_accomplished == False, - ) - result = await context.db.main.execute(q) - - time = util.time() - achieved: list[main.Achievement] = [] - new: list[main.Achievement] = [] - - for ach in result.scalars(): - ach_info = await get_achievement_info(context, ach.achievement_id) - if ach_info is None: - raise ValueError("achievement info is none, database is corrupted?") - - if await test(ach_info, args): - target_amount = int(getattr(ach_info, f"params{pindex}", None) or 1) - count = ach.count + increment - if count >= target_amount: - # Achieved. - ach.count = min(count, target_amount) - ach.is_accomplished = True - achieved.append(ach) - - # Update reset value - match ach.reset_type: - case 1: - ach.reset_value = util.get_days_since_unix(time) - case 2: - ach.reset_value = util.get_weeks_since_unix(time) - - # New achievement - for next_ach_id in await get_next_achievement_ids(context, ach.achievement_id): - new_ach_info = await get_achievement_info(context, next_ach_id) - if new_ach_info is not None and not await has_achievement( - context, user, new_ach_info.achievement_id - ): - new_ach = await add_achievement(context, user, new_ach_info, time) - # Append to new achievement - new.append(new_ach) - else: - ach.count = count - - await context.db.main.flush() - return AchievementContext(accomplished=achieved, new=new) - - -class RecursiveAchievementCall[**P](Protocol): - def __call__( - self, context: idol.BasicSchoolIdolContext, user: main.User, /, *args: P.args, **kwargs: P.kwargs - ) -> collections.abc.Awaitable[AchievementContext]: ... - - -def recursive_achievement(achievement_type: int, /): - def wrapper0[**P](func: RecursiveAchievementCall[P]): - async def wrapper1( - context: idol.BasicSchoolIdolContext, user: main.User, /, *args: P.args, **kwargs: P.kwargs - ) -> AchievementContext: - nonlocal achievement_type - result = AchievementContext() - - while True: - keep_traversing = False - ach_result = await func(context, user, *args, **kwargs) - - if ach_result.has_achievement(): - result = result + ach_result - - for ach in ach_result.new: - if ach.achievement_type == achievement_type: - keep_traversing = True - break - - if not keep_traversing: - break - - return result - - return wrapper1 - - return wrapper0 - - -async def check_type_1(context: idol.BasicSchoolIdolContext, user: main.User, /, increment: bool): - """ - Check live show clear achievements. - """ - return await check_type_increment(context, user, 1, increment) - - -async def check_type_2(context: idol.BasicSchoolIdolContext, user: main.User, /, difficulty: int, increment: bool): - """ - Check live show clear achievements by specified difficulty. - """ - return await check_type_increment(context, user, 2, increment, 2, difficulty) - - -async def check_type_3(context: idol.BasicSchoolIdolContext, user: main.User, /, rank: int, increment: bool): - """ - Check live show clear achievements by specified score rank. - """ - return await check_type_increment(context, user, 3, increment, 2, rank) - - -async def check_type_4(context: idol.BasicSchoolIdolContext, user: main.User, /, rank: int, increment: bool): - """ - Check live show clear achievements by specified combo rank. - """ - return await check_type_increment(context, user, 4, increment, 2, rank) - - -async def check_type_7(context: idol.BasicSchoolIdolContext, user: main.User, /, unit_type_id: int, increment: bool): - """ - Check live show clear achievements by specified combo rank. - """ - return await check_type_increment(context, user, 4, increment, 2, unit_type_id) - - -@recursive_achievement(18) -async def check_type_18(context: idol.BasicSchoolIdolContext, user: main.User, /, club_members: int): - """ - Check amount of club members collected. - """ - return await check_type_countable(context, user, 18, club_members) - - -@recursive_achievement(19) -async def check_type_19(context: idol.BasicSchoolIdolContext, user: main.User, /, idolized: int): - """ - Check amount of idolized club members. - """ - return await check_type_countable(context, user, 19, idolized) - - -@recursive_achievement(20) -async def check_type_20(context: idol.BasicSchoolIdolContext, user: main.User, /, max_love: int): - """ - Check amount of max bonded idolized club members. - """ - return await check_type_countable(context, user, 20, max_love) - - -@recursive_achievement(21) -async def check_type_21(context: idol.BasicSchoolIdolContext, user: main.User, /, max_level: int): - """ - Check amount of max leveled idolized club members. - """ - return await check_type_countable(context, user, 21, max_level) - - -async def check_type_23(context: idol.BasicSchoolIdolContext, user: main.User, /, scenario_id: int): - """ - Check main story clear. - """ - - # Note, there's no "count" in scenario but the database requires it, so specify unused parameter index (e.g. 2) - return await check_type_countable(context, user, 23, 1, 2, scenario_id) - - -async def check_type_27(context: idol.BasicSchoolIdolContext, user: main.User, /, nlogins: int): - """ - Check login bonus count - """ - return await check_type_countable(context, user, 27, nlogins) - - -async def check_type_29(context: idol.BasicSchoolIdolContext, user: main.User, /): - """ - Check one-time login bonus. - """ - return await check_type_countable(context, user, 29, 1) - - -@recursive_achievement(30) -async def check_type_30(context: idol.BasicSchoolIdolContext, user: main.User, /, rank: int | None = None): - """ - Check player rank. - """ - return await check_type_countable(context, user, 30, user.level if rank is None else rank) - - -@recursive_achievement(32) -async def check_type_32(context: idol.BasicSchoolIdolContext, user: main.User, /, live_track_id: int): - """ - Check live show clear of specific `live_track_id` - """ - - # Note, there's no "count" in here but the database requires it, so specify unused parameter index (e.g. 2) - return await check_type_countable(context, user, 32, 1, 2, live_track_id) - - -async def check_type_37(context: idol.BasicSchoolIdolContext, user: main.User, /, live_track_id: int, increment: bool): - """ - Check live show clear of specific `live_track_id` - """ - - # Note, there's no "count" in here but the database requires it, so specify unused parameter index (e.g. 2) - return await check_type_increment(context, user, 37, increment, 2, live_track_id) - - -class _Type50Checker: - def __init__(self, context: idol.BasicSchoolIdolContext, user: main.User, unit_deck: list[int], /): - self.context = context - self.user = user - self.unit_deck = unit_deck - self.achievement_groups: list[set[int]] = [] - - async def init(self): - if len(self.achievement_groups) == 0: - previous_unit_set: dict[int, set[int]] = {} - - for unit_id in self.unit_deck: - if unit_id in previous_unit_set: - self.achievement_groups.append(previous_unit_set[unit_id]) - else: - # Perform query on unit type - unit_info = await unit.get_unit_info(self.context, unit_id) - result: set[int] = set() - if unit_info is not None: - q = sqlalchemy.select(achievement.UnitTypeGroup.achievement_unit_type_group_id).where( - achievement.UnitTypeGroup.unit_type_id == unit_info.unit_type_id - ) - r = await self.context.db.achievement.execute(q) - result.update(r.scalars()) - - previous_unit_set[unit_id] = result - self.achievement_groups.append(result) - - async def all(self, achievement_unit_type_group_id: int): - await self.init() - for group_id in self.achievement_groups: - if achievement_unit_type_group_id not in group_id: - return False - - return True - - async def any(self, achievement_unit_type_group_id: int): - await self.init() - for group_id in self.achievement_groups: - if achievement_unit_type_group_id in group_id: - return True - - return False - - async def false(self, achievement_unit_type_group_id: int): - return False - - async def test_member_type(self, params8: int | None, params9: int | None): - if params8 is None: - return True - return await _Type50Checker.comparer.get(params9, _Type50Checker.false)(self, params8) - - async def __call__(self, ach_info: achievement.Achievement, args: collections.abc.Sequence[int | None]): - # args: - # 0. live_track_id - # 1. difficulty - # 2. attribute_id - # 3. score_rank - # 4. combo_rank - - args = util.ensure_no_none(args) - - return ( - (ach_info.params1 is None or ach_info.params1 == args[0]) - and (ach_info.params2 is None or ach_info.params2 == args[1]) - and (ach_info.params3 is None or ach_info.params3 == args[2]) - and ach_info.params4 is None # TODO - and (ach_info.params5 is None or ach_info.params5 >= args[3]) - and (ach_info.params6 is None or ach_info.params6 >= args[4]) - and (await self.test_member_type(ach_info.params8, ach_info.params9)) - ) - - comparer = {None: false, 1: all, 2: any} - - -async def check_type_50( - context: idol.BasicSchoolIdolContext, - user: main.User, - /, - live_track_id: int, - difficulty: int, - attribute: int, - score_rank: int, - combo_rank: int, - unit_deck: list[int], - increment: bool, -): - """ - Check live show clear where all or one of the deck are in specific group. - """ - - return await check_type_increment( - context, - user, - 50, - increment, - 10, - live_track_id, - difficulty, - attribute, - score_rank, - combo_rank, - test=_Type50Checker(context, user, unit_deck), - ) - - -@recursive_achievement(53) -async def check_type_53( - context: idol.BasicSchoolIdolContext, user: main.User, /, achievement_category_id: int, count: int -): - """ - Check amount of achievement cleared with specific category - """ - return await check_type_countable(context, user, 53, count, 2, achievement_category_id) - - -async def check_type_53_recursive(context: idol.BasicSchoolIdolContext, user: main.User, /): - """ - Check amount of achievement cleared on all available categories - """ - result_complete = AchievementContext() - - while True: - keep_going = False - all_accomplished_by_category = await count_accomplished_achievement_by_category(context, user) - result = AchievementContext() - for achievement_category_id, amount in all_accomplished_by_category: - result.extend(await check_type_53(context, user, achievement_category_id, amount)) - - result_complete.extend(result) - # Should we keep recurse? - for ach in result.new: - if ach.achievement_type == 53: - keep_going = True - break - - if not keep_going: - break - - return result_complete - - -async def check_type_57(context: idol.BasicSchoolIdolContext, user: main.User, /, completed_scenario_count: int): - """ - Check amount of cleared main stories. - """ - - return await check_type_countable(context, user, 57, completed_scenario_count) - - -async def check_type_58(context: idol.BasicSchoolIdolContext, user: main.User, /, increment: bool): - """ - Check live show clear achievements. - """ - return await check_type_increment(context, user, 58, increment) - - -async def check_type_59(context: idol.BasicSchoolIdolContext, user: main.User, /, unlocked_scenario_count: int): - """ - Check amount of unlocked main stories. - """ - - return await check_type_countable(context, user, 59, unlocked_scenario_count) - - async def get_achievement_filter_ids(context: idol.BasicSchoolIdolContext): q = sqlalchemy.select(achievement.FilterCategory) result = await context.db.achievement.execute(q)