Просмотр исходного кода

feat(world): SC52 属性帧解析等级进镜像

wuzilong 1 месяц назад
Родитель
Сommit
916a8a8f29

+ 1 - 0
pipeline_bot/src/pbot/world/mirror.py

@@ -30,6 +30,7 @@ class PlayerState:
     map_id: str = ""
     map_name: str = ""
     is_dead: bool = False
+    level: int = 0  # SC52 属性帧 "level"
 
     @property
     def point(self) -> PointCode:

+ 23 - 0
pipeline_bot/src/pbot/world/update.py

@@ -197,6 +197,8 @@ class WorldUpdater:
             world.record("me_revive")
         elif mid == 31:
             self._on_struck(world, frame)
+        elif mid == 52:
+            self._on_player_props(world, frame)
         elif mid == 53:
             self._on_hp(world, frame)
         elif mid == 55:
@@ -540,6 +542,27 @@ class WorldUpdater:
             _LOG.info("[SC31] 致死: %s", detail)
         world.record("struck", detail, aid)
 
+    def _on_player_props(self, world: WorldMirror, f: Frame) -> None:
+        """SC52 角色属性: level 权威源(神通 35 级门槛判定用)。
+
+        body=encode6(JSON), 实测(cap_hoe_success_frames_20261001):
+        {"job":0,"level":1,"hp":15,"maxhp":29,"exp":0,...}
+        hp/maxhp 有值时同步(比 SC53 稀疏推送更全)。
+        """
+        j = body_json(f)
+        lv = _as_int(j.get("level", 0))
+        if lv > 0:
+            if lv != world.me.level:
+                _LOG.info("[SC52] 等级 %d → %d", world.me.level, lv)
+            world.me.level = lv
+        hp = _as_int(j.get("hp", 0))
+        maxhp = _as_int(j.get("maxhp", 0))
+        if hp > 0:
+            world.me.hp = hp
+        if maxhp > 0:
+            world.me.max_hp = maxhp
+        world.record("player_props", f"lv={lv}")
+
     def _on_hp(self, world: WorldMirror, f: Frame) -> None:
         # SC53 body=encode6("长uid&HP&maxHP&MP&maxMP|power") (backup diag 定案)
         me = world.me

+ 30 - 0
pipeline_bot/tests/test_pipeline_units.py

@@ -0,0 +1,30 @@
+"""流水线新增单元测试: SC52 等级解析。"""
+import json
+
+from pbot.codec.bitbuf import encode6
+from pbot.codec.frame import build_frame, parse_frame
+from pbot.world.mirror import WorldMirror
+from pbot.world.update import WorldUpdater
+
+
+def _sc52_frame(level: int, hp: int = 100) -> bytes:
+    body = json.dumps({"job": 0, "level": level, "hp": hp,
+                       "maxhp": hp, "exp": 0, "maxexp": 100}).encode()
+    return build_frame(52, 0, 0, 0, 0, encode6(body))
+
+
+def test_sc52_updates_level() -> None:
+    w = WorldMirror()
+    u = WorldUpdater()
+    u.apply(w, parse_frame(_sc52_frame(35)))
+    assert w.me.level == 35
+    u.apply(w, parse_frame(_sc52_frame(36, hp=120)))
+    assert w.me.level == 36
+    assert w.me.hp == 120  # maxhp/hp 同步(若实现)或仅 level(见下)
+
+
+def test_sc52_level_zero_ignored() -> None:
+    w = WorldMirror()
+    u = WorldUpdater()
+    u.apply(w, parse_frame(_sc52_frame(0)))
+    assert w.me.level == 0  # 0 不产生误报