Merge pull request #261 from cc004/dev

Dev

Author: Lanly109

autopcr/core/datamgr.py

@@ -596,6 +596,76 @@ def is_quest_cleared(self, quest: int) -> bool:
     def is_quest_sweepable(self, quest: int) -> bool:
         return quest in self.quest_dict and self.quest_dict[quest].clear_flg == 3
 
+    def get_talent_quest_single(self, talent_id: int) -> str:
+        max_quest_id = self.cleared_talent_quest_ids.get(talent_id, 0)
+        quest = "0-0" if max_quest_id == 0 else f"{db.quest_name[max_quest_id][4:]}"
+        return quest
+
+    def get_talent_quest_info(self) -> str:
+        msg = []
+        for talent_id in sorted([area.talent_id for area in db.talent_quest_area_data.values()]):
+            talent_name = db.talents[talent_id].talent_name 
+            msg.append(f"{talent_name}{self.get_talent_quest_single(talent_id)}")
+        return "/".join(msg)
+
+    def get_talent_level_single(self, talent_info: TalentLevelInfo) -> str:
+        material = db.talent_level_material[talent_info.talent_id]
+        cur_point = talent_info.total_point
+        up_point = talent_info.total_point + self.get_inventory((material.reward_type, material.item_id)) * material.point
+        info = f"{db.get_talent_level(cur_point)}"
+        if True and cur_point != up_point: # always show for pretty format
+            info += f"→{db.get_talent_level(up_point)}"
+        return info
+
+    def get_talent_level_info(self) -> str:
+        msg = []
+        for talent_info in self.princess_knight_info.talent_level_info_list:
+            msg.append(f"{db.talents[talent_info.talent_id].talent_name}{self.get_talent_level_single(talent_info)}")
+        return "/".join(msg)
+
+    def get_talent_skill_info(self) -> str:
+        page = 0 if not self.princess_knight_info.talent_skill_last_enhanced_page_node_list else db.talent_skill_node[self.princess_knight_info.talent_skill_last_enhanced_page_node_list[0].node_id].page_num
+        joined_nodes = flow(self.princess_knight_info.talent_skill_last_enhanced_page_node_list) \
+                        .where(lambda x: db.talent_skill_node[x.node_id].is_joined_node()) \
+                        .to_list()
+        max_joined_node = max(joined_nodes, key=lambda x: x.node_id, default=TalentSkillNodeInfo(node_id=1))
+        joined_node_after = flow(self.princess_knight_info.talent_skill_last_enhanced_page_node_list) \
+                            .where(lambda x: x.node_id > max_joined_node.node_id) \
+                            .group_by(lambda x: db.talent_skill_node[x.node_id].pos_x) \
+                            .to_dict(lambda g: g.key, lambda g: g.to_list())
+        joined_node_after_max = {pos: max(nodes, key=lambda x: x.node_id) for pos, nodes in joined_node_after.items()}
+
+        prefix = f"第{page}页 第{len(joined_nodes)}合流"
+        msg = []
+        if not joined_node_after and joined_nodes:
+            prefix += f"[{max_joined_node.enhance_level}级]"
+        elif joined_node_after:
+            for pos, nodes in joined_node_after.items():
+                msg.append(f"{db.talent_skill_node[nodes[0].node_id].pos()}{len(nodes)}[{joined_node_after_max[pos].enhance_level}级]")
+
+        info = prefix + " " + "/".join(msg)
+        info += f"(余{self.get_inventory(db.xinyou)})"
+        return info
+
+    def get_master_skill_info(self) -> str:
+        cur_node_id = self.princess_knight_info.team_skill_latest_node.node_id
+        up_node_id = cur_node_id
+        num = self.get_inventory(db.master_fragment) + self.get_inventory(db.master_ffragment) // 100
+        next_id = sorted([k for k in db.team_skill_node if k > cur_node_id])
+        for node_id in next_id:
+            tot = 0
+            for enhance_level_id in db.team_skill_node[node_id].enhance_level_ids():
+                tot += db.team_skill_enhance_level[enhance_level_id].consume_num
+            if num >= tot:
+                num -= tot
+                up_node_id = node_id
+
+        info = f"MP{cur_node_id}"
+        if True and cur_node_id != up_node_id: # always show for pretty format
+            info += f"→{up_node_id}"
+        info += f"(余{num})"
+        return info
+
     async def request(self, request: Request[TResponse], next: RequestHandler) -> TResponse:
         resp = await next.request(request)
         if resp:

autopcr/core/pcrclient.py

@@ -1281,59 +1281,62 @@ async def serlize_reward(self, reward_list: List[InventoryInfo], target: Union[I
             result.append(f"{db.get_inventory_name_san(target)}x0({self.data.get_inventory(target)})")
         return '\n'.join(result) if result else "无"
 
-    async def serialize_unit_info(self, unit_data: Union[UnitData, UnitDataLight]) -> Tuple[bool, str]:
-        info = []
-        ok = True
-        def add_info(prefix, cur, expect = None):
-            if expect:
-                nonlocal ok
-                info.append(f'{prefix}:{cur}/{expect}')
-                ok &= (cur == expect)
-            else:
-                info.append(f'{prefix}:{cur}')
-        unit_id = unit_data.id
-        add_info("等级", unit_data.unit_level, db.team_max_level)
-        if unit_data.battle_rarity:
-            add_info("星级", f"{unit_data.battle_rarity}-{unit_data.unit_rarity}")
-        else:
-            add_info("星级", f"{unit_data.unit_rarity}")
-        add_info("品级", unit_data.promotion_level, db.equip_max_rank)
-        for id, union_burst in enumerate(unit_data.union_burst):
-            if union_burst.skill_level:
-                add_info(f"ub{id}", union_burst.skill_level, unit_data.unit_level)
-        for id, skill in enumerate(unit_data.main_skill):
-            if skill.skill_level:
-                add_info(f"skill{id}", skill.skill_level, unit_data.unit_level)
-        for id, skill in enumerate(unit_data.ex_skill):
-            if skill.skill_level:
-                add_info(f"ex{id}", skill.skill_level, unit_data.unit_level)
-        equip_info = []
-        for id, equip in enumerate(unit_data.equip_slot):
-            equip_id = getattr(db.unit_promotion[unit_id][unit_data.promotion_level], f'equip_slot_{id + 1}')
-            if not equip.is_slot:
-                if equip_id != 999999:
-                    equip_info.append('-')
-                    ok = False
-                else:
-                    equip_info.append('*')
-            else:
-                star = db.get_equip_star_from_pt(equip_id, equip.enhancement_pt)
-                ok &= (star == 5)
-                equip_info.append(str(star))
-        equip_info = '/'.join(equip_info)
-        add_info("装备", equip_info)
-
-        for id, equip in enumerate(unit_data.unique_equip_slot):
-            equip_slot = id + 1
-            have_unique = (equip_slot in db.unit_unique_equip and unit_id in db.unit_unique_equip[equip_slot])
-            max_level = 0 if not have_unique else db.unique_equipment_max_level[equip_slot]
-            if have_unique:
-                if not equip.is_slot:
-                    add_info(f"专武{id}", '-', max_level)
-                else:
-                    add_info(f"专武{id}", db.get_unique_equip_level_from_pt(equip_slot, equip.enhancement_pt), max_level)
-        
-        return ok, ' '.join(info)
+    def unit_info_to_dict(self, unit_id: int) -> Dict:
+        unit_data = {
+            "星级": "无",
+            "等级": "无",
+            "品级": "无",
+            "装备": "无",
+            "UB": 0,
+            "S1": 0,
+            "S2": 0,
+            "EX": 0,
+            "剧情": "无",
+            "专武1": "无",
+            "专武2": "无",
+            "普碎数": "无",
+            "金碎数": "无",
+        }
+
+        read_story = set(self.data.read_story_ids)
+        if unit_id in self.data.unit:
+            unitinfo = self.data.unit[unit_id]
+            rank = f"R{unitinfo.promotion_level}"
+            equip = ''.join('-' if not solt.is_slot else str(solt.enhancement_level) for solt in unitinfo.equip_slot)
+            kizuna_unit = set()
+            for story in db.chara2story[unit_id]:
+                if story.story_id in db.story_detail:
+                    kizuna_unit.add(db.story_detail[story.story_id].requirement_id)
+
+            love = []
+            for other in kizuna_unit:
+                if other not in db.unlock_unit_condition: continue
+                unit_name = db.get_unit_name(other)
+                other_id = other // 100
+                love_level = self.data.unit_love_data[other_id].love_level if other_id in self.data.unit_love_data else 0
+                unit_story = [story.story_id for story in db.unit_story if story.story_group_id == other_id]
+                total_storys = len(unit_story)
+                read_storys = len([story for story in unit_story if story in read_story])
+                love.append(f"{unit_name}好感{love_level}({read_storys}/{total_storys})")
+            love_str = '\n'.join(love) if love else "无"
+
+            unit_data.update({
+                "星级": f"{unitinfo.unit_rarity}★",
+                "等级": unitinfo.unit_level,
+                "品级": rank,
+                "装备": equip,
+                "UB": unitinfo.union_burst[0].skill_level if unitinfo.union_burst else "无",
+                "S1": unitinfo.main_skill[0].skill_level if unitinfo.main_skill else "无",
+                "S2": unitinfo.main_skill[1].skill_level if len(unitinfo.main_skill) > 1 else "无",
+                "EX": unitinfo.ex_skill[0].skill_level if unitinfo.ex_skill else "无",
+                "剧情": love_str,
+                "专武1": "未实装" if not unitinfo.unique_equip_slot else "无" if not unitinfo.unique_equip_slot[0].is_slot else unitinfo.unique_equip_slot[0].enhancement_level,
+                "专武2": "未实装" if len(unitinfo.unique_equip_slot) < 2 else "无" if not unitinfo.unique_equip_slot[1].is_slot else unitinfo.unique_equip_slot[1].enhancement_level,
+                "普碎数": self.data.get_inventory((eInventoryType.Item, db.unit_to_memory[unit_id])) if unit_id in db.unit_to_memory else "无",
+                "金碎数": self.data.get_inventory((eInventoryType.Item, db.unit_to_pure_memory[unit_id])) if unit_id in db.unit_to_pure_memory else "无",
+            })
+
+        return unit_data
 
     async def recover_challenge(self, quest: int):
         req = QuestRecoverChallengeRequest()

autopcr/db/database.py

@@ -764,6 +764,14 @@ def event_story_detail(self) -> List[EventStoryDetail]:
                 .to_list()
             )
 
+    @lazy_property
+    def story_detail(self) -> Dict[int, StoryDetail]:
+        with self.dbmgr.session() as db:
+            return (
+                StoryDetail.query(db)
+                .to_dict(lambda x: x.story_id, lambda x: x)
+            )
+
     @lazy_property
     def unit_story(self) -> List[StoryDetail]:
         with self.dbmgr.session() as db:
@@ -1554,17 +1562,50 @@ def experience_knight_rank(self) -> Dict[int, ExperienceKnightRank]:
     @lazy_property
     def talents(self) -> Dict[int, Talent]:
         with self.dbmgr.session() as db:
-            return Talent.query(db).to_dict(lambda x: x.talent_id, lambda x: x)
+            return (
+                Talent.query(db)
+                .to_dict(lambda x: x.talent_id, lambda x: x)
+            )
+
+    @lazy_property
+    def talent_level_material(self) -> Dict[int, TalentLevelMaterial]:
+        with self.dbmgr.session() as db:
+            return (
+                TalentLevelMaterial.query(db)
+                .to_dict(lambda x: x.talent_id, lambda x: x)
+            )
 
     @lazy_property
     def talent_skill_node(self) -> Dict[int, TalentSkillNode]:
         with self.dbmgr.session() as db:
-            return TalentSkillNode.query(db).to_dict(lambda x: x.node_id, lambda x: x)
+            return (
+                TalentSkillNode.query(db)
+                .to_dict(lambda x: x.node_id, lambda x: x)
+            )
+
+    @lazy_property
+    def team_skill_enhance_level(self) -> Dict[int, TeamSkillEnhanceLevel]:
+        with self.dbmgr.session() as db:
+            return (
+                TeamSkillEnhanceLevel.query(db)
+                .to_dict(lambda x: x.enhance_level_id, lambda x: x)
+            )
+
+    @lazy_property
+    def team_skill_node(self) -> Dict[int, TeamSkillNode]:
+        with self.dbmgr.session() as db:
+            return (
+                TeamSkillNode.query(db)
+                .to_dict(lambda x: x.node_id, lambda x: x)
+            )
 
     @lazy_property
     def experience_talent_level(self) -> Dict[int, ExperienceTalentLevel]:
         with self.dbmgr.session() as db:
-            return ExperienceTalentLevel.query(db).to_dict(lambda x: x.talent_level, lambda x: x)
+            return (
+                ExperienceTalentLevel.query(db)
+                .to_dict(lambda x: x.talent_level, lambda x: x)
+            )
 
     def get_ex_equip_star_from_pt(self, id: int, pt: int) -> int:
         rarity = self.get_ex_equip_rarity(id)
@@ -2209,7 +2250,7 @@ def unit_unique_equip_level_candidate(self, equip_slot: int) -> List[int]:
         return list(range(st, self.unique_equipment_max_level[equip_slot] + 1))
 
     def last_normal_quest(self) -> List[int]:
-        quest_ids = sorted(self.normal_quest_data.keys(), reverse=True)
+        quest_ids = sorted([k for k, v in self.normal_quest_data.items() if self.parse_time(v.start_time) <= apiclient.datetime] , reverse=True)
         return quest_ids[:5]
         last_start_time = flow(self.normal_quest_data.values()) \
                 .where(lambda x: db.parse_time(x.start_time) <= apiclient.datetime) \

autopcr/db/methods.py

@@ -195,3 +195,17 @@ def pos(self) -> str:
         elif self.pos_x == 8:
             return "右"
         return "未知"
+
+@method
+class TeamSkillNode(models.TeamSkillNode):
+    def enhance_level_ids(self) -> Iterator[int]:
+        if self.enhance_level_id_1 != 0:
+            yield self.enhance_level_id_1
+        if self.enhance_level_id_2 != 0:
+            yield self.enhance_level_id_2
+        if self.enhance_level_id_3 != 0:
+            yield self.enhance_level_id_3
+        if self.enhance_level_id_4 != 0:
+            yield self.enhance_level_id_4
+        if self.enhance_level_id_5 != 0:
+            yield self.enhance_level_id_5

autopcr/http_server/validator.py

@@ -90,8 +90,8 @@ def enable_manual_validator():
     def usermgr_load(qid: str, readonly=False):
         result = usermgr_load_legacy(qid, readonly=readonly)
         accountmgr_load_legacy = result.load
-        def accountmgr_load(account: str = "", readonly=False):
-            result = accountmgr_load_legacy(account=account, readonly=readonly)
+        def accountmgr_load(account: str = "", readonly=False, force_use_all=False):
+            result = accountmgr_load_legacy(account=account, readonly=readonly, force_use_all=force_use_all)
             async def post_login():
                 validate_dict[qid].append(ValidateInfo(status="ok"))
             account_aenter_legacy = result._do_aenter

autopcr/module/accountmgr.py

@@ -231,9 +231,12 @@ def is_clan_battle_forbidden(self):
         return self._parent._parent.is_clan_battle_forbidden(username)
 
 class AccountBatch(Account):
-    def __init__(self, parent: 'AccountManager', qid: str, accounts: str = BATCHINFO, readonly: bool = False):
+    def __init__(self, parent: 'AccountManager', qid: str, accounts: str = BATCHINFO, readonly: bool = False, force_use_all: bool = False):
         super().__init__(parent, qid, accounts, readonly)
-        self.enable_account = sorted(set([x for x in self.data.batch_accounts]) & set(self._parent.accounts()))
+        if force_use_all:
+            self.enable_account = sorted(list(self._parent.accounts()))
+        else:
+            self.enable_account = sorted(set([x for x in self.data.batch_accounts]) & set(self._parent.accounts()))
 
     def generate_info(self):
         accounts = list(self.enable_account)
@@ -264,22 +267,26 @@ async def do_from_key_pre(alias: str):
             ret_list = [x.get_result() for x in resps]
             ret = copy(ret_list[0])
             ret.log = '\n'.join(f"==={name}===\n{x.log}" for name, x in zip(alias, ret_list) if x.log)
+            ret.status = eResultStatus.ERROR if any(x.status == eResultStatus.ERROR for x in ret_list) else eResultStatus.WARNING if any(x.status == eResultStatus.WARNING for x in ret_list) else eResultStatus.SUCCESS
             if any(r.table.header for r in ret_list):
                 ret.table.header = next(r.table.header for r in ret_list)
-                ret.table.header = ["昵称"] + ret.table.header
+                ret.table.header = ["昵称", '状态'] + ret.table.header
                 ret.table.data = [
-                    {**d, '昵称': name}
+                    {**d, '昵称': name, '状态': f"#{x.status.value}"}
                     for name, x in zip(alias, ret_list)
                     if x.table
                     for d in x.table.data
                 ]
             else:
-                ret.table.header = ["昵称", "结果"]
+                ret.table.header = ["昵称", "状态", "结果"]
                 ret.table.data = [
-                    {'昵称': name, '结果': x.log}
+                    {
+                     '昵称': name, 
+                     '状态': f"#{x.status.value}",
+                     '结果': x.log
+                    }
                     for name, x in zip(alias, ret_list)
                 ]
-            ret.status = eResultStatus.ERROR if any(x.status == eResultStatus.ERROR for x in ret_list) else eResultStatus.WARNING if any(x.status == eResultStatus.WARNING for x in ret_list) else eResultStatus.SUCCESS
             all = await self.save_single_result(key, ret)
             resps = [all] + resps
         self.data.single_result[key] = resps
@@ -342,11 +349,11 @@ def set_password(self, password: str):
     def validate_password(self, password: str) -> bool:
         return self.secret.password == password
 
-    def load(self, account: str = "", readonly = False) -> Account:
+    def load(self, account: str = "", readonly = False, force_use_all: bool = False) -> Account:
         if not AccountManager.pathsyntax.fullmatch(account):
             raise AccountException(f'非法账户名{account}')
         if account == BATCHINFO:
-            return AccountBatch(self, self.qid, account, readonly = readonly)
+            return AccountBatch(self, self.qid, account, readonly = readonly, force_use_all = force_use_all)
         if not account:
             account = self.secret.default_account
         if not account and len(list(self.accounts())) == 1: