MAGIC_REQUEST  = 0x80
MAGIC_RESPONSE = 0x81

def classify_status(status: int) -> str:
    if status == 0x0000:
        return "SUCCESS"
    elif status <= 0x0009:
        return "CLIENT_ERROR"  # 业务错误
    elif status <= 0x0080:
        return "RESERVED"
    else:
        return "SERVER_ERROR"  # 服务端错误

#!/usr/bin/env python3
"""binary_protocol.py — Memcached 二进制协议编解码器"""

import struct
from dataclasses import dataclass
from typing import Optional

# 常量
MAGIC_REQUEST  = 0x80
MAGIC_RESPONSE = 0x81

HEADER_FORMAT = "!BBHBBHIIQ"  # 24 字节头部
HEADER_SIZE = 24

# 操作码
OP_GET      = 0x00
OP_SET      = 0x01
OP_ADD      = 0x02
OP_REPLACE  = 0x03
OP_DELETE   = 0x04
OP_INCR     = 0x05
OP_DECR     = 0x06
OP_QUIT     = 0x07
OP_FLUSH    = 0x08
OP_VERSION  = 0x0B
OP_APPEND   = 0x0E
OP_PREPEND  = 0x0F
OP_STAT     = 0x10

# 状态码
STATUS_SUCCESS      = 0x0000
STATUS_KEY_NOT_FOUND = 0x0001
STATUS_KEY_EXISTS   = 0x0002
STATUS_VALUE_TOO_LARGE = 0x0003
STATUS_INVALID_ARGS = 0x0004
STATUS_NOT_STORED   = 0x0005
STATUS_NON_NUMERIC  = 0x0006
STATUS_UNKNOWN_CMD  = 0x0081
STATUS_OUT_OF_MEM   = 0x0082
STATUS_INTERNAL_ERR = 0x0084

@dataclass
class BinaryRequest:
    opcode: int
    key: bytes = b""
    value: bytes = b""
    extras: bytes = b""
    cas: int = 0
    opaque: int = 0

    def encode(self) -> bytes:
        key_length = len(self.key)
        extras_length = len(self.extras)
        body_length = extras_length + key_length + len(self.value)

        header = struct.pack(
            HEADER_FORMAT,
            MAGIC_REQUEST,   # magic
            self.opcode,     # opcode
            key_length,      # key length
            extras_length,   # extras length
            0x00,            # data type
            0x0000,          # reserved
            body_length,     # total body length
            self.opaque,     # opaque
            self.cas         # cas
        )

        return header + self.extras + self.key + self.value

@dataclass
class BinaryResponse:
    opcode: int
    status: int
    key: bytes
    value: bytes
    extras: bytes
    cas: int
    opaque: int

    @classmethod
    def decode(cls, data: bytes) -> 'BinaryResponse':
        if len(data) < HEADER_SIZE:
            raise ValueError("数据不足")

        (magic, opcode, key_length, extras_length,
         data_type, status, body_length,
         opaque, cas) = struct.unpack(HEADER_FORMAT, data[:HEADER_SIZE])

        if magic != MAGIC_RESPONSE:
            raise ValueError(f"无效的响应魔数: 0x{magic:02x}")

        body = data[HEADER_SIZE:HEADER_SIZE + body_length]
        extras = body[:extras_length]
        key = body[extras_length:extras_length + key_length]
        value = body[extras_length + key_length:]

        return cls(
            opcode=opcode,
            status=status,
            key=key,
            value=value,
            extras=extras,
            cas=cas,
            opaque=opaque
        )

    @property
    def success(self) -> bool:
        return self.status == STATUS_SUCCESS


# 编码示例
req = BinaryRequest(
    opcode=OP_SET,
    key=b"user:1001",
    value=b'{"name":"Bob"}',
    extras=struct.pack(">II", 0, 3600)  # flags + exptime
)
encoded = req.encode()
print(f"请求帧大小: {len(encoded)} 字节")
print(f"Hex: {encoded.hex()}")

def send_request_with_opaque(sock, request: BinaryRequest, opaque: int):
    request.opaque = opaque
    sock.sendall(request.encode())

def recv_response(sock) -> BinaryResponse:
    header = b""
    while len(header) < HEADER_SIZE:
        chunk = sock.recv(HEADER_SIZE - len(header))
        if not chunk:
            raise ConnectionError("连接关闭")
        header += chunk

    (_, _, key_length, extras_length, _, _, body_length, _, _) = \
        struct.unpack(HEADER_FORMAT, header)

    body = b""
    while len(body) < body_length:
        chunk = sock.recv(body_length - len(body))
        if not chunk:
            raise ConnectionError("连接关闭")
        body += chunk

    return BinaryResponse.decode(header + body)

import struct
import socket
from concurrent.futures import ThreadPoolExecutor

class BinaryMemcachedClient:
    def __init__(self, host='127.0.0.1', port=11211):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((host, port))
        self._opaque_counter = 0

    def _next_opaque(self) -> int:
        self._opaque_counter += 1
        return self._opaque_counter

    def set(self, key: str, value: bytes, flags: int = 0,
            exptime: int = 0, cas: int = 0) -> bool:
        extras = struct.pack(">II", flags, exptime)
        req = BinaryRequest(
            opcode=OP_SET,
            key=key.encode(),
            value=value,
            extras=extras,
            cas=cas,
            opaque=self._next_opaque()
        )
        self.sock.sendall(req.encode())
        resp = recv_response(self.sock)
        return resp.success

    def get(self, key: str) -> tuple[bytes, int, int] | None:
        req = BinaryRequest(
            opcode=OP_GET,
            key=key.encode(),
            opaque=self._next_opaque()
        )
        self.sock.sendall(req.encode())
        resp = recv_response(self.sock)

        if resp.status == STATUS_KEY_NOT_FOUND:
            return None

        flags = struct.unpack(">I", resp.extras[:4])[0] if resp.extras else 0
        return resp.value, flags, resp.cas

    def delete(self, key: str) -> bool:
        req = BinaryRequest(
            opcode=OP_DELETE,
            key=key.encode(),
            opaque=self._next_opaque()
        )
        self.sock.sendall(req.encode())
        resp = recv_response(self.sock)
        return resp.status == STATUS_SUCCESS

    def version(self) -> str:
        req = BinaryRequest(opcode=OP_VERSION, opaque=self._next_opaque())
        self.sock.sendall(req.encode())
        resp = recv_response(self.sock)
        return resp.value.decode()

    def close(self):
        req = BinaryRequest(opcode=OP_QUIT, opaque=self._next_opaque())
        self.sock.sendall(req.encode())
        self.sock.close()


# 使用
client = BinaryMemcachedClient()
print(f"Version: {client.version()}")
client.set("test", b"hello", flags=0, exptime=300)
result = client.get("test")
if result:
    print(f"Value: {result[0].decode()}")
client.close()