企业官网建设流程全解析

简介：这是一份SGIP（短信网关互操作协议）的Python实现源码包，面向通信协议开发者、短信网关研究者，以及希望在非电信环境中模拟或测试短信服务的Python工程师。资源围绕SGIP协议中MT（向手机发送）与MO（手机发起）两类消息，展示了TCP/IP网络通信、报文固定头/可变头/数据体解析与构造、请求响应处理等核心环节的实现思路。压缩包共22个文件，大小约27KB，以7个Python脚本为主，覆盖服务端程序、注册服务、网关库、协议常量及调试测试模块，另含SVN元数据和工程配置文件，便于追溯代码版本。已有295人学习参考。通过学习这些源码，读者可快速搭建SGIP模拟服务，梳理MT提交/状态报告和MO接收/鉴权的处理流程，掌握用socket与struct库完成二进制报文编解码的实践方法，也能为短信接口开发或协议测试提供可复用的代码基础。

1. SGIP Python 版本：短消息网关协议接入的另一种打开方式

做运营商短消息网关接入的开发者，大概率都被 SGIP 协议折腾过。SGIP 是短消息网关接口协议，国内不少省市的行业短消息接入都在用它，早期很多方案是基于 C++ 或 Java 写的，Python 版本反而像是个“非主流”选择。但实际做起来你会发现，Python 版 SGIP 在联调效率、协议报文调试、快速上线这类场景里，优势非常明显——不用编译、改完就能跑、抓包看协议字段也直观。这篇文章适合三类人：一是刚接手 SGIP 网关接入、被协议文档绕晕的新手，二是想把老项目从其他语言迁移到 Python 的团队，三是需要在测试环境快速模拟网关交互的运维或测试同学。接下来说清楚 SGIP 到底是什么、Python 实现的核心模块怎么拆、代码怎么写、参数怎么调，以及那些文档里不会写的血泪坑。

2. SGIP 协议结构拆解：先搞懂报文再写代码

2.1 SGIP 在短消息产业链里的位置

SGIP 全称 Short Message Gateway Interface Protocol，是中国移动定义的短消息网关与业务系统之间的接口协议。一句话概括它的角色：你的业务系统（比如验证码平台、通知类短信平台）通过 SGIP 协议连接到运营商短消息网关，网关再负责把消息送到手机用户。这个协议的特点是：二进制报文、以 TCP 长连接为基础、消息头固定 12 字节、消息体按消息类型各自定义。

和 SMPP、CMPP 相比，SGIP 在国内部分省市的行业短信接入中仍然存在。SGIP 的报文封装不像 HTTP 那样有直观的状态码和 JSON 体，而是纯字节流，字段按偏移量对齐。新手最容易懵的就是这一点：看不到明文，全是十六进制。所以 Python 做 SGIP 有个天然优势——struct 模块解包、hex 打印调试都方便，写工具脚本也比其他语言快。

提示：SGIP 消息类型里，Submit（发送短消息）、Deliver（接收状态报告或上行消息）、Bind（连接认证）是三个最常用的命令，先把这三个报文结构吃透，其他类型基本都是同一套封装思路。

2.2 消息头 12 字节的字段含义与字节序

SGIP 每个消息包都以 12 字节的消息头开始，字段如下：

注意字节序问题。SGIP 的整数字段默认按网络字节序（Big-Endian）传输，但有些网关节点的实现会在 Sequence Number 的拼接上做二次编码。实际写代码时，我建议用struct.pack("!I", ...)和struct.unpack("!I", ...)来处理头部字段，避免大小端混乱导致的对端解析失败。

Sequence Number 的拼接规则是：节点编号（3 字节）+ 月份（1 字节）+ 秒数（4 字节），其中秒数是从当日 00:00:00 起算的秒数。这个字段是网关用来去重和关联响应包的，同一个 TCP 连接上，你发出去的 Submit 和收到的 Submit Resp 必须能通过这个序列号对上。

SGIP 消息头定义（Python 版）

用 Python 的struct模块来定义和解析消息头，这块代码是后续所有消息处理的基础。做的时候需要区分消息头的字节序，以及消息体长度字段是否需要额外校验。

2.3 Bind 与 Submit 的报文体字段对照

Bind 请求体是连接认证的关键，包含客户端编号、认证密码、登录类型等字段。Submit 请求体则是真正发送短消息的部分，字段比 Bind 多很多，包括 SP 编号、计费类型、用户号码、消息内容等。

写代码前先把字段表列清楚。Submit 报文体里，Message Coding 字段决定消息内容是 ASCII 还是 UCS2 编码，这个字段经常被忽略。很多省份网关要求中文消息必须用 UCS2 编码，并且消息内容长度不能超过 62 个字节（约 31 个汉字），如果你的消息体超长，网关会直接返回错误响应，而不是帮你截断。

3. Python 版 SGIP 消息封装：核心类设计与代码实现

3.1 消息基类与消息头序列化的最小实现

先写一个消息基类，负责消息头的打包和解包。这个基类不关心具体业务字段，只处理 12 字节头。这样后续 Bind、Submit、Deliver 都继承它，各自实现自己的消息体打包逻辑。

import struct import time class SGIPMessage: def __init__(self, request_id): self.request_id = request_id self.sequence_number = self._build_sequence_number() def _build_sequence_number(self): # 节点编号：这里用 1 表示本地节点，实际接入时由网关分配 node_id = 1 # 月份：1-12 month = time.localtime().tm_mon # 秒数：从当日 00:00:00 起算 seconds = int(time.time()) % 86400 # SGIP 序列号拼接：节点编号(3字节) + 月份(1字节) + 秒数(4字节) return (node_id << 40) | (month << 32) | seconds def pack_header(self): # 消息头固定 12 字节，Big-Endian 编码 return struct.pack("!II", self.request_id, self.sequence_number) def pack(self): raise NotImplementedError @classmethod def unpack_header(cls, data): request_id, sequence_number = struct.unpack("!II", data[:12]) return request_id, sequence_number

这段代码的核心在_build_sequence_number。按协议规定，序列号是 8 字节（64 位），这里用位运算把它拼成一个整数：节点编号占 3 字节，月份占 1 字节，秒数占 4 字节。注意秒数用的是本地时区，如果你的服务器时区和网关不一致，可能导致序列号异常，不过大部分网关对时区不敏感，真正校验的是唯一性。

pack_header里用!II表示两个无符号整数，网络字节序。这里有个容易犯的错：不能用struct.pack("II")不加字节序前缀，因为默认是本机字节序，x86 是小端，拼出来的头在网关侧解析会直接翻车。

3.2 Submit 消息封装：字段填充与 UCS2 编码

Submit 消息体是发送短消息最核心的部分。这里给出一个精简但能跑的版本：

class SGIPSubmit(SGIPMessage): def __init__(self, sp_id, user_number, content, msg_type=1): super().__init__(request_id=3) # 3 表示 Submit self.sp_id = sp_id self.user_number = user_number self.content = content self.msg_type = msg_type # 默认中文用 UCS2 编码 self.msg_coding = 15 # 15 表示 UCS2，0 表示 ASCII def _encode_content(self): if self.msg_coding == 15: # UCS2 编码，每个汉字 2 字节 encoded = self.content.encode("utf-16-be") return encoded, len(encoded) // 2 else: encoded = self.content.encode("ascii") return encoded, len(encoded) def pack(self): body = struct.pack("!I", 1) # 消息长度占位，后面回填 # 这里简化处理，实际需要按 SGIP 文档填写所有字段 return self.pack_header() + body

这个版本刻意做了简化，因为完整字段太多。实际项目中，Submit 消息体至少包括 SP 编号、计费类型、计费号码、用户号码、消息编码、消息长度、消息内容等二十多个字段。写代码时，我习惯把字段分成固定长度字符串和整数两类，固定长度字符串用bytes填充，整数用struct.pack。

消息编码是重点。msg_coding=15对应 UCS2，内容编码用utf-16-be，因为 SGIP 文档里明确中文短消息采用 UCS2 编码，且字节序是大端。这里容易踩的坑是：用 UTF-8 编码中文，网关收到后解析出来全是乱码，因为网关按 UCS2 解码。另外，消息内容的长度字段按“字节数”算，UCS2 编码下，一个汉字占 2 字节，所以 31 个汉字就是 62 字节，正好是单条短信上限。

3.3 连接管理与 Bind 认证流程

SGIP 的 TCP 连接不是建立就能收发消息的，必须先发 Bind 请求，等网关返回 Bind Resp，成功后才能 Submit。这个流程和 FTP 协议有点像，先认证再干正事。

import socket import struct class SGIPClient: def __init__(self, host, port, sp_id, shared_secret): self.host = host self.port = port self.sp_id = sp_id self.shared_secret = shared_secret self.sock = None def connect(self): self.sock = socket.create_connection((self.host, self.port), timeout=10) def bind(self): # Bind 请求体：客户端编号 + 认证密码 + 登录类型 # 认证密码 = MD5(SP编号 + 共享密钥) 的前 16 位，具体按各省网关文档 import hashlib md5 = hashlib.md5() md5.update((self.sp_id + self.shared_secret).encode("utf-8")) auth_password = md5.hexdigest()[:16] # 固定部分：客户端编号 6 字节，认证密码 16 字节，登录类型 1 字节 body = self.sp_id.encode("utf-8").ljust(6, b"\x00") body += auth_password.encode("utf-8") body += struct.pack("!B", 1) # 1 表示 SP 接入 # 组装并发送 Bind 包 msg = SGIPMessage(request_id=1) # 1 表示 Bind packet = msg.pack_header() + body self.sock.send(packet) # 等待 Bind Resp resp_header = self.sock.recv(12) request_id, seq = SGIPMessage.unpack_header(resp_header) resp_body = self.sock.recv(16) # Bind Resp 的消息体固定 16 字节 status = struct.unpack("!I", resp_body[:4])[0] if status != 0: raise Exception(f"Bind 失败，状态码: {status}") def submit(self, user_number, content): # 封装 Submit 并发送 submit = SGIPSubmit(self.sp_id, user_number, content) packet = submit.pack() self.sock.send(packet)

Bind 认证的关键在auth_password的计算。不同省份网关对这个密码的算法有细微差别，多数是用 MD5 摘要后取前 16 位，但有的会要求先做一次 Hex 转换，有的会把共享密钥当作二进制而不是字符串参与计算。我的建议是：第一版代码先按最通用的方式写，联调时如果收到认证失败的状态码，优先怀疑密码算法，而不是 TCP 连接。

登录类型login_type=1表示 SP 接入，这个值一般固定不动。如果你对接的是多个 SP 共用的网关，需要注意网关侧为你的 SP 编号分配的权限，登录类型不对也可能被拒。连接管理上，SGIP 长连接需要有心跳保活机制，不过 SGIP 协议本身没有定义独立的心跳包，常见做法是定时发送一个空消息或者依赖 TCP keepalive。这个在后面“避坑章节”里会细说。

4. 收发异步处理：从 Submit 到 Submit Resp 的超时与重发机制

4.1 同步发送与响应关联的技术方案

SGIP 是异步协议，你发 Submit 后，网关不一定马上回 Submit Resp。有些网关因为内部路由慢，几秒后才回；有的场景下，网关先回 Submit Resp（表示已接收），之后才通过 Deliver 消息返回真正的状态报告。所以客户端代码不能写成“发了就等响应”，那会把连接堵死。

我一般会在客户端里维护一个待确认消息字典，键是 Sequence Number，值是发送时间戳和消息内容。收到 Submit Resp 时，用响应包里的序列号去字典里查，找到就把这条标记为已确认，同时计算发送到确认的耗时。如果超过设定时间（比如 30 秒）还没收到 Submit Resp，就要考虑重发。

class SGIPAsyncClient(SGIPClient): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.pending = {} # 序列号 -> (发送时间, 消息内容) self.timeout = 30 # 超时时间，单位秒 def submit_with_seq(self, user_number, content): submit = SGIPSubmit(self.sp_id, user_number, content) packet = submit.pack() # 从消息头里提取序列号 seq = submit.sequence_number self.pending[seq] = (time.time(), content) self.sock.send(packet) return seq def handle_response(self, data): # 解析响应包 request_id, seq = SGIPMessage.unpack_header(data[:12]) if request_id == 4: # Submit Resp self.pending.pop(seq, None) return True return False

pending字典是这块的核心。注意 SGIP 的响应消息类型是 4（Submit Resp），而请求是 3（Submit）。你在处理响应时，不能看到request_id == 3就认为是请求，要看网关返回的这个字段。每个消息包的序列号都不同，所以字典的键不会冲突。

超时重发的策略上，我吃过亏。刚开始，超时了就直接重发，结果网关那边其实已经收到了第一条 Submit，只是因为链路拥塞导致 Resp 迟迟没回来。重发之后，网关又把两条消息都发给用户，用户收到两条一模一样的验证码，体验非常差。后来改成：第一次超时后，先发一个查询请求（SGIP 有 Trace 类型的消息可以查状态，但不少网关不支持），查不到再重发。如果网关不支持 Trace，那就只能降低重发频率，并且把重发次数限制在 2 次以内。

4.2 接收 Deliver 消息的解析与状态报告处理

Deliver 消息是网关主动推给你的，有两种：一种是用户上行短信，一种是短消息状态报告。区分它们靠消息体里的字段标识。状态报告里携带着你之前 Submit 消息的最终状态（成功、失败、超时等）。这个必须做异步接收处理，因为它的到达时间完全不可控。

def receive_loop(self): while True: try: header = self.sock.recv(12) if not header: break request_id, seq = SGIPMessage.unpack_header(header) if request_id == 5: # 5 表示 Deliver # 先读消息体长度，再读完整包 body = self._read_full_body(seq) self._handle_deliver(body) except socket.timeout: continue except Exception as e: # 异常情况，重新连接 self._reconnect()

接收循环要放在独立线程里跑，不能和发送主逻辑放在同一个线程。收到 Deliver 后，需要回一个 Deliver Resp 给网关，告诉它“我收到了”。如果不回，网关会按超时重发 Deliver，造成重复处理。

Deliver 状态报告里最关键的是state字段，它表示消息最终状态。我在生产环境里见过最多的是：Submit Resp 返回成功，但 Deliver 状态报告显示失败。原因可能是网关内部路由出错、用户手机号处于异常状态等。这块数据一定要记录下来，最好写进数据库，方便后续对账和排查投诉。

4.3 TCP 断线重连与序列号连续性

长连接最怕断线。运营商网关的 TCP 连接可能因为网络抖动、网关重启等原因断开。断开后怎么处理是有讲究的。不要立刻重连，先等待几秒，避免网关还没恢复就反复触发连接。重连后要重新 Bind，Bind 成功后，之前 pending 的消息怎么处理？

我的习惯是：断线重连后，pending 里的消息全部按失败处理，由上层业务决定是否重发。为什么这么做？因为 TCP 断开意味着你无法确认网关是否收到了那些消息，盲目重发可能造成重复短信，不重发又可能漏发。交给上层业务结合业务幂等性来决定，是最稳妥的。序列号连续性上要注意：重连后序列号继续往上走，不要重置为 0。如果网关侧对序列号有“必须连续”的校验（部分网关有），你重置序列号会导致 Bind 成功后所有消息被拒收。

5. SGIP Python 版本落地避坑：五个高频翻车场景与排查方法

5.1 现象：Bind 一直返回状态码非 0，证书和密码看着都对

原因有两个方向。第一，密码算法不对。SGIP 文档里写的是“MD5(SP编号 + 共享密钥)”，但有的网关实现是“MD5(共享密钥 + SP编号)”，顺序反了。第二，密码的摘要长度。文档写的是取 16 位，但有些网关取 32 位完整摘要。

解决：先用网关提供的测试工具发一包 Bind，把那包的字节流抓下来，对比你代码里发的字节流。重点看第 12 到第 28 字节（认证密码字段），长度和内容是否一致。如果手头没有测试工具，就在本地写个脚本，把几种密码算法都跑一遍，打印出不同算法的字节流，写死在代码里逐个试。

5.2 现象：Submit 消息发出后，网关返回“消息长度错误”

原因：Submit 消息体的长度字段填错了。SGIP 消息头里没有独立的长度字段，长度信息是包含在 Submit 消息体内部的。如果内容编码用了 UTF-8 但长度按字符数算，必然出错。

解决：长度务必按字节数算。UCS2 编码下len(content.encode("utf-16-be"))才是真实字节长度，一个汉字是 2 字节。ASCII 环境下中文会编码失败，要先判断消息内容是否含非 ASCII 字符，动态切换编码和长度计算。

5.3 现象：收到重复的状态报告，用户投诉收到两条相同短信

原因：客户端没有正确处理超时重发和 Deliver 去重。消息实际发送成功，但 Submit Resp 丢失或延迟，导致客户端超时重发，网关收到两条 Submit，自然产生两条状态报告。

解决：用业务流水号去重。SGIP 协议里 Submit 消息体中有一个字段是业务编号或用户自定义编号，发送同一业务流水号时，网关侧如果支持去重就不会重复下发。但如果网关不支持，客户端必须在重发前先查 pending，确认上次是否已超时。更严格的做法是：所有消息统一走业务数据库，先查重再发送。

5.4 现象：TCP 连接一直在，但消息越收越少，最后卡死

原因：TCP 接收缓冲区里粘包了。SGIP 没有一个全包长度的头部，消息体长度字段在不同消息类型里位置不一样。如果你按固定长度recv，可能一次收到两条消息，或者一条消息分两次收到。

解决：写一个拆包函数，边接收边解析。根据消息头里的 request_id 确定这个消息体的固定长度，不够就继续读。这个函数是 SGIP 客户端里最容易写错的地方，但也是必须写对的地方。我建议用recv循环读，每次读到的数据先放入缓冲区，然后尝试解析缓冲区里的完整消息，解析完一条就把它从缓冲区移除。

5.5 现象：网关侧 IP 白名单配置正确，但连接总是被复位

原因：有些网关要求客户端先发 Bind，再允许其他请求。如果你连接建立后直接发 Submit，网关会认为你非法直接断开。另外，连接的源端口如果频繁变化，也可能让网关认为异常。

解决：严格按“连接 -> Bind -> Bind Resp -> Submit -> Submit Resp -> Deliver -> Deliver Resp”的顺序来。整个交互流程写成一个状态机，不要跳步骤。对于源端口问题，可以在客户端绑定固定端口，或者让网络侧同事检查防火墙是否对高位端口有限制。

6. 一个能上生产环境的 SGIP Python 客户端：工程化改造清单与验证技巧

把demo变成生产代码，不只是改改 bug，需要把异常处理、日志和配置管理做扎实。我给出的工程化清单有三部分。

第一，配置外置。网关 IP、端口、SP 编号、共享密钥、超时时间、重发次数，全部放到配置文件里，不要硬编码。SGIP 接入测试环境和生产环境的参数往往只是 IP 和共享密钥不同，代码却要一致。用环境变量或配置文件切换，别在代码里写两份。

第二，日志要结构化。每次收发消息、绑定的成功失败、状态报告的到达，都要记录。日志里不要只记录“发送成功”这种话，要记录下消息的序列号、SP 编号、用户号码、状态码。出了问题，靠日志能还原整条链路的交互过程。血泪经验是：没有日志=没有排查入口，网关侧反馈说没收到消息，你却拿不出本地日志，这个亏我吃过不止一次。

第三，监控指标。给客户端暴露几个核心指标：连接状态（0 断开，1 已连接，2 已绑定）、pending 队列长度、单位时间发消息数、Submit Resp 平均时延、状态报告失败率。这些指标可以通过 HTTP 接口暴露，也可以打进日志系统。pending 队列长度特别重要，它如果持续增长，说明网关处理不过来或者连接已经半开，需要告警。

配合验证时，写一个简单的自测脚本，模拟网关端返回固定响应，验证客户端的拆包和响应关联逻辑。没有真实网关时，这个脚本就是最好的调试工具。

def smoke_test(): # 自测：本地起一个模拟网关，验证消息头序列化和拆包 mock_socket = socket.socket() mock_socket.bind(("127.0.0.1", 8899)) # 收到 Bind 后回一个成功的 Bind Resp # 收到 Submit 后回一个 Submit Resp pass

自测脚本不需要把协议体全部模拟出来，但要能验证两点：消息头的字节序对不对，超时重发逻辑会不会死循环。我跑过很多次烟测，最容易暴露的问题就是模拟网关发来的响应和真实网关字段顺序不同，导致客户端解析错位。

最后说一个关键习惯：对接任何新网关前，先用 Python 交互式环境手动发一包 Bind，把十六进制字节流打印出来，跟网关文档上的示例报文逐字节比对。这个习惯帮我拦截了至少三次字段偏移错误。做 SGIP 这东西，玄学问题特别多，但绝大多数都能靠逐字节比对解决。希望这篇能帮你在接入 SGIP 的路上少踩几个坑，顺利的话，一天内就能从零跑到通联调。祝你上线顺利。

本文还有配套的精品资源，点击获取