feat(apilot): wire layer with verified frame codec

apilot/wire.py

@@ -0,0 +1,75 @@
+"""协议线层: 24B头(9×i32)+体, 全帧 Encode6BitBuf。提炼自 protocol_bot/cq2_pure_enter.py(实测验证)。"""
+from __future__ import annotations
+import socket
+import struct
+import sys
+import time
+from dataclasses import dataclass
+
+sys.path.insert(0, "protocol_bot")
+from encode6bitbuf_final import encode6, decode6  # type: ignore
+
+HDR = struct.Struct("<9i")  # recog,msg,p1,p2,p3,zero?,len,tail → 对齐 cq2_pure_enter.send()
+
+
+@dataclass
+class Frame:
+    recog: int
+    msg_id: int
+    p1: int
+    p2: int
+    p3: int
+    tail: int
+    body: bytes = b""
+
+
+def pack_frame(f: Frame) -> bytes:
+    head = HDR.pack(0, f.msg_id, f.recog, f.p1, f.p2, f.p3, 0, len(f.body), f.tail)
+    return encode6(head) + f.body
+
+
+def unpack_stream(buf: bytes) -> tuple[list[Frame], bytes]:
+    out, pos = [], 0
+    while pos + 48 <= len(buf):
+        try:
+            m = HDR.unpack(decode6(buf[pos:pos + 48]))
+        except Exception:
+            break
+        body_len = m[7]
+        if pos + 48 + body_len > len(buf):
+            break
+        out.append(Frame(recog=m[2], msg_id=m[1], p1=m[3], p2=m[4], p3=m[5],
+                         tail=m[8], body=buf[pos + 48:pos + 48 + body_len]))
+        pos += 48 + body_len
+    return out, buf[pos:]
+
+
+class Wire:
+    """同步 TCP 线客户端。"""
+
+    def __init__(self, host: str, port: int, timeout: float = 10) -> None:
+        self.sock = socket.create_connection((host, port), timeout=timeout)
+
+    def send(self, f: Frame) -> None:
+        self.sock.sendall(pack_frame(f))
+
+    def recv(self, deadline_s: float) -> list[Frame]:
+        deadline = time.time() + deadline_s
+        buf = b""
+        while time.time() < deadline:
+            self.sock.settimeout(max(0.2, deadline - time.time()))
+            try:
+                d = self.sock.recv(65536)
+                if not d:
+                    break
+                buf += d
+            except (socket.timeout, ConnectionResetError, OSError):
+                break
+        frames, _ = unpack_stream(buf)
+        return frames
+
+    def close(self) -> None:
+        try:
+            self.sock.close()
+        except OSError:
+            pass

tests/test_wire.py

@@ -0,0 +1,31 @@
+# tests/test_wire.py
+import pytest
+from apilot.wire import Frame, pack_frame, unpack_stream
+
+def test_frame_roundtrip():
+    f = Frame(recog=900000, msg_id=20, p1=123, p2=14481607, p3=8043,
+              tail=7, body=b"*#acct/role/12345/9/1/0/110585/0/1/1/1/1")
+    packed = pack_frame(f)
+    frames, rest = unpack_stream(packed)
+    assert rest == b""
+    assert len(frames) == 1
+    g = frames[0]
+    assert (g.recog, g.msg_id, g.p1, g.p2, g.p3, g.tail, g.body) == \
+           (f.recog, f.msg_id, f.p1, f.p2, f.p3, f.tail, f.body)
+
+def test_unpack_partial_packet():
+    f = Frame(recog=0, msg_id=529, p1=0, p2=0, p3=0, tail=1, body=b"key32abc")
+    packed = pack_frame(f)
+    frames, rest = unpack_stream(packed[:-10])   # 掐掉10字节
+    assert frames == [] and len(rest) == len(packed) - 10
+
+def test_pack_matches_reference_impl():
+    """与 cq2_pure_enter.send() 的原始拼包逐字节一致(防字段序回归)。"""
+    import struct, random
+    sys_path_hack = __import__("sys"); sys_path_hack.path.insert(0, "protocol_bot")
+    from encode6bitbuf_final import encode6
+    random.seed(42)
+    body = b"*#1186115037/role/1/2/3/1/0/110585/0/1/1/1/1"
+    ref = encode6(struct.pack("<9i", 0, 20, 900000, 111, 14481607, 8043, 0, len(body), 9)) + body
+    mine = pack_frame(Frame(recog=900000, msg_id=20, p1=111, p2=14481607, p3=8043, tail=9, body=body))
+    assert mine == ref