如何序列化/反序列化Protobuf API响应及解决.proto文件编译异常

问题根因 #

你用request.urlretrieve直接保存的文件编译报错，核心原因是API返回的内容并非纯文本格式的合法.proto定义——可能是经过gzip压缩的内容，或是混入了不可打印的控制字符/乱码，导致protoc无法识别。

分步解决方法 #

1. 先确认API返回的内容属性 #

先发送请求查看响应头，判断内容类型和编码方式：

import requests

proto_url = "https://pxqsellerbff.piaoxingqiu.cn/prod/seat/20230605/6465d5b543231c00015cc844/647d89e739bde00001519ff4_1_1685948904931.proto?versionNo=2&src=WEB&channelId=&terminalSrc=WEB"
response = requests.get(proto_url)
print("Content-Type:", response.headers.get("Content-Type"))
print("Content-Encoding:", response.headers.get("Content-Encoding"))

2. 正确获取并清洗.proto内容 #

根据响应头处理压缩、解码内容，并过滤非法控制字符：

import requests
import gzip
from io import BytesIO

def save_valid_proto(url, save_path="valid_test.proto"):
    response = requests.get(url)
    response.raise_for_status()  # 确保请求成功
    
    # 处理gzip压缩的情况
    if response.headers.get("Content-Encoding") == "gzip":
        compressed_data = BytesIO(response.content)
        with gzip.GzipFile(fileobj=compressed_data, mode='rb') as f:
            proto_content = f.read().decode('utf-8')
    else:
        proto_content = response.content.decode('utf-8')
    
    # 过滤不可打印的控制字符，只保留合法文本
    filtered_content = ''.join([c for c in proto_content if c.isprintable() or c in '\n\r\t'])
    
    with open(save_path, 'w', encoding='utf-8') as f:
        f.write(filtered_content)
    print(f"合法.proto文件已保存至 {save_path}")

# 执行保存
save_valid_proto(proto_url)

3. 验证编译 #

用protoc编译处理后的文件：

protoc -I=. --python_out=. ./valid_test.proto

4. 实现序列化/反序列化 #

编译完成后，导入生成的Python模块即可处理Protobuf数据：

from valid_test_pb2 import TargetMessage  # 替换为实际生成的消息类名
import requests

# 反序列化示例：解析API返回的二进制Protobuf数据
data_response = requests.get("你的Protobuf数据API地址")
message = TargetMessage()
message.ParseFromString(data_response.content)

# 序列化示例：将对象转为二进制
serialized_data = message.SerializeToString()

关键提示 #

• 注意请求方法：原链接标注OPTIONS，但通常获取.proto定义用GET，需验证实际支持的请求方式。
• 字符过滤是必要步骤：部分API返回的内容会混入不可打印控制字符，必须过滤后才能被protoc正常识别。

内容的提问来源于stack exchange，提问作者gzldc