import msgpack

# V1 消息格式
v1_message = {
    "version": 1,
    "type": "user_update",
    "data": {
        "id": 123,
        "name": "Alice",
    }
}

# V2 消息格式（新增字段）
v2_message = {
    "version": 2,
    "type": "user_update",
    "data": {
        "id": 123,
        "name": "Alice",
        "email": "[email protected]",  # 新增
    }
}

# 安全的反序列化
def decode_message(raw: bytes) -> dict:
    msg = msgpack.unpackb(raw, raw=False)
    version = msg.get("version", 1)
    
    if version == 1:
        # V1 → V2 升级
        msg["data"]["email"] = None  # 填充默认值
        msg["version"] = 2
    
    return msg

import msgpack

# 扩展类型定义
TYPE_V1 = 10
TYPE_V2 = 11

def encode_message(data, version=2):
    if version == 1:
        return msgpack.packb(data, use_bin_type=True)
    else:
        body = msgpack.packb(data, use_bin_type=True)
        return msgpack.packb(msgpack.ExtType(TYPE_V2, body), use_bin_type=True)

def decode_message(raw):
    try:
        obj = msgpack.unpackb(raw, raw=False)
        if isinstance(obj, msgpack.ExtType) and obj.code == TYPE_V2:
            return msgpack.unpackb(obj.data, raw=False)
    except:
        pass
    # 回退到 V1
    return msgpack.unpackb(raw, raw=False)

# 客户端-服务端类型协商
NEGOTIATE_REQUEST = {
    "type": "negotiate",
    "supported_formats": ["msgpack", "json"],
    "preferred_format": "msgpack",
    "schema_version": 2,
}

NEGOTIATE_RESPONSE = {
    "type": "negotiate_ack",
    "selected_format": "msgpack",
    "schema_version": 2,
}

import msgpack

# ❌ 长键名
long_keys = {
    "user_identifier": 123,
    "user_display_name": "Alice",
    "user_email_address": "[email protected]",
    "account_creation_timestamp": 1700000000,
}
# 编码: ~90 bytes

# ✅ 短键名（用于内部通信）
short_keys = {
    "u": 123,           # user_identifier
    "n": "Alice",       # user_display_name
    "e": "[email protected]",  # user_email_address
    "t": 1700000000,    # account_creation_timestamp
}
# 编码: ~50 bytes (节省 44%)

# 映射表
KEY_MAP = {
    "u": "user_identifier",
    "n": "user_display_name",
    "e": "user_email_address",
    "t": "account_creation_timestamp",
}

def expand_keys(data: dict) -> dict:
    return {KEY_MAP.get(k, k): v for k, v in data.items()}

import msgpack
import lz4.frame
import zlib
import snappy

def compress_msgpack(data: dict, method: str = "lz4") -> bytes:
    """压缩 MessagePack 数据"""
    packed = msgpack.packb(data, use_bin_type=True)
    
    if method == "lz4":
        return lz4.frame.compress(packed)
    elif method == "zlib":
        return zlib.compress(packed, level=6)
    elif method == "snappy":
        return snappy.compress(packed)
    return packed

def decompress_msgpack(data: bytes, method: str = "lz4") -> dict:
    """解压并解码"""
    if method == "lz4":
        decompressed = lz4.frame.decompress(data)
    elif method == "zlib":
        decompressed = zlib.decompress(data)
    elif method == "snappy":
        decompressed = snappy.decompress(data)
    else:
        decompressed = data
    
    return msgpack.unpackb(decompressed, raw=False)

# 对比
data = {"items": [{"id": i, "name": f"Item {i}", "value": i * 1.5} for i in range(100)]}

original = msgpack.packb(data, use_bin_type=True)
lz4_compressed = compress_msgpack(data, "lz4")
zlib_compressed = compress_msgpack(data, "zlib")

print(f"原始:      {len(original):6d} bytes")
print(f"LZ4 压缩:  {len(lz4_compressed):6d} bytes ({len(lz4_compressed)/len(original)*100:.1f}%)")
print(f"Zlib 压缩: {len(zlib_compressed):6d} bytes ({len(zlib_compressed)/len(original)*100:.1f}%)")

import msgpack
import io

# ❌ 不好: 逐条序列化
def bad_batch(items):
    results = []
    for item in items:
        packed = msgpack.packb(item)
        results.append(packed)
    return results

# ✅ 好: 批量序列化
def good_batch(items):
    buf = io.BytesIO()
    for item in items:
        msgpack.pack(item, buf)
    return buf.getvalue()

# ✅ 更好: 一次序列化整个数组
def best_batch(items):
    return msgpack.packb(items, use_bin_type=True)

import msgpack
from functools import lru_cache

# 缓存编解码器配置
@lru_cache(maxsize=1)
def get_encoder():
    return msgpack.Packer(use_bin_type=True)

# 复用 Unpacker
class MsgPackPool:
    def __init__(self, size=10):
        self._pool = []
        self._size = size
    
    def get_unpacker(self):
        if self._pool:
            return self._pool.pop()
        return msgpack.Unpacker(raw=False)
    
    def return_unpacker(self, unpacker):
        if len(self._pool) < self._size:
            unpacker.feed(b"")  # 重置
            self._pool.append(unpacker)

import msgpack

# ❌ 不好: 直接信任外部数据
def unsafe_handler(data: bytes):
    obj = msgpack.unpackb(data, raw=False)
    execute_command(obj["cmd"])  # 危险!

# ✅ 好: 验证输入
def safe_handler(data: bytes):
    obj = msgpack.unpackb(data, raw=False)
    
    # 类型检查
    if not isinstance(obj, dict):
        raise ValueError("期望字典类型")
    
    # 白名单验证
    allowed_commands = {"ping", "status", "query"}
    cmd = obj.get("cmd")
    if cmd not in allowed_commands:
        raise ValueError(f"不允许的命令: {cmd}")
    
    # 参数验证
    params = obj.get("params", {})
    if not isinstance(params, dict):
        raise ValueError("参数必须为字典类型")
    
    execute_command(cmd, params)

import msgpack

# 恶意数据: 极深嵌套
def create_deep_nested(depth=10000):
    data = {"value": "leaf"}
    for _ in range(depth):
        data = {"child": data}
    return data

# ❌ 默认行为: 可能栈溢出
try:
    malicious = msgpack.packb(create_deep_nested(10000))
    msgpack.unpackb(malicious)  # 可能崩溃
except Exception as e:
    print(f"错误: {e}")

# ✅ 使用 max_buffer_size 限制
try:
    msgpack.unpackb(malicious, max_buffer_size=1024*1024)
except msgpack.exceptions.BufferFull:
    print("缓冲区溢出: 数据过大")

import msgpack

# 限制允许的类型
class SafeUnpacker:
    ALLOWED_TYPES = {int, float, str, bool, type(None), list, dict, bytes}
    
    @staticmethod
    def unpack(data: bytes) -> any:
        obj = msgpack.unpackb(data, raw=False)
        SafeUnpacker._validate(obj)
        return obj
    
    @staticmethod
    def _validate(obj, depth=0):
        if depth > 100:
            raise ValueError("嵌套过深")
        
        if type(obj) not in SafeUnpacker.ALLOWED_TYPES:
            raise ValueError(f"不允许的类型: {type(obj)}")
        
        if isinstance(obj, dict):
            for key, value in obj.items():
                SafeUnpacker._validate(key, depth + 1)
                SafeUnpacker._validate(value, depth + 1)
        elif isinstance(obj, list):
            for item in obj:
                SafeUnpacker._validate(item, depth + 1)

import msgpack

MAX_MESSAGE_SIZE = 10 * 1024 * 1024  # 10MB
MAX_STRING_LENGTH = 1024 * 1024       # 1MB
MAX_ARRAY_LENGTH = 100000             # 10万元素
MAX_MAP_LENGTH = 10000                # 1万键值对

def safe_unpack(data: bytes) -> any:
    if len(data) > MAX_MESSAGE_SIZE:
        raise ValueError(f"消息过大: {len(data)} bytes")
    
    return msgpack.unpackb(
        data,
        raw=False,
        max_buffer_size=MAX_MESSAGE_SIZE,
        max_str_len=MAX_STRING_LENGTH,
        max_bin_len=MAX_STRING_LENGTH,
        max_array_len=MAX_ARRAY_LENGTH,
        max_map_len=MAX_MAP_LENGTH,
        ext_hook=_safe_ext_handler,
    )

def _safe_ext_handler(code, data):
    # 只允许已知的扩展类型
    ALLOWED_EXT_TYPES = {-1}  # timestamp
    if code not in ALLOWED_EXT_TYPES:
        raise ValueError(f"不允许的扩展类型: {code}")
    return msgpack.ExtType(code, data)

# ⚠️ 不要序列化包含敏感信息的对象
import msgpack

# ❌ 不好: 序列化整个用户对象
user = get_user_from_db(123)  # 包含 password_hash
packed = msgpack.packb(user.__dict__)

# ✅ 好: 只序列化安全字段
def safe_user_dict(user):
    return {
        "id": user.id,
        "name": user.name,
        "email": user.email,
        # 不包含 password_hash, api_key 等
    }

packed = msgpack.packb(safe_user_dict(user))

# 使用 xxd 查看二进制
python3 -c "import msgpack; print(msgpack.packb({'hello': 'world'}).hex())"
# 输出: 82a568656c6c6f a5776f726c64

# 格式化输出
python3 -c "
import msgpack
data = msgpack.packb({'id': 1, 'name': 'Alice', 'tags': ['admin']})
hex_str = data.hex()
for i in range(0, len(hex_str), 2):
    print(hex_str[i:i+2], end=' ')
print()
"
# 输出: 82 a2 69 64 01 a4 6e 61 6d 65 a5 41 6c 69 63 65 a4 74 61 67 73 91 a5 61 64 6d 69 6e

import msgpack

def inspect_msgpack(data: bytes):
    """详细解析 MessagePack 二进制"""
    offset = 0
    while offset < len(data):
        byte = data[offset]
        
        if byte <= 0x7f:
            print(f"  [{offset:4d}] positive fixint: {byte}")
            offset += 1
        elif byte >= 0xe0:
            print(f"  [{offset:4d}] negative fixint: {byte - 256}")
            offset += 1
        elif byte == 0xc0:
            print(f"  [{offset:4d}] nil")
            offset += 1
        elif byte == 0xc2:
            print(f"  [{offset:4d}] false")
            offset += 1
        elif byte == 0xc3:
            print(f"  [{offset:4d}] true")
            offset += 1
        elif 0xa0 <= byte <= 0xbf:
            length = byte - 0xa0
            string = data[offset+1:offset+1+length].decode("utf-8")
            print(f"  [{offset:4d}] fixstr({length}): \"{string}\"")
            offset += 1 + length
        elif 0x80 <= byte <= 0x8f:
            count = byte - 0x80
            print(f"  [{offset:4d}] fixmap({count})")
            offset += 1
        elif 0x90 <= byte <= 0x9f:
            count = byte - 0x90
            print(f"  [{offset:4d}] fixarray({count})")
            offset += 1
        else:
            print(f"  [{offset:4d}] 0x{byte:02x}")
            offset += 1

# 使用
data = msgpack.packb({"name": "Alice", "active": True, "score": 95})
print("MessagePack 结构:")
inspect_msgpack(data)

import msgpack
import logging
import json

logger = logging.getLogger(__name__)

class MsgPackFormatter(logging.Formatter):
    """MessagePack 日志格式化器"""
    
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }
        
        # 附加异常信息
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        
        return json.dumps(log_data, ensure_ascii=False)

# 请求/响应日志
def log_request_response(request_data, response_data, duration_ms):
    logger.info(
        "RPC 调用",
        extra={
            "request_size": len(request_data),
            "response_size": len(response_data),
            "duration_ms": duration_ms,
            # 不要记录完整数据，可能包含敏感信息
            "request_type": type(request_data).__name__,
        }
    )

import time
import msgpack
from contextlib import contextmanager

@contextmanager
def measure_time(operation: str):
    start = time.perf_counter_ns()
    yield
    elapsed = time.perf_counter_ns() - start
    print(f"{operation}: {elapsed/1000:.1f} μs")

# 使用
data = {"users": [{"id": i, "name": f"User{i}"} for i in range(1000)]}

with measure_time("序列化 1000 用户"):
    packed = msgpack.packb(data, use_bin_type=True)

with measure_time("反序列化 1000 用户"):
    unpacked = msgpack.unpackb(packed, raw=False)

print(f"数据大小: {len(packed)} bytes")

msgpack.unpackb(data, raw=False)