最小WebSocket服务端骨架

手撸一个能跑的人脸设备 WebSocket 服务端：三阶段状态机实战

讲了一堆协议，不来点能跑的代码总觉得悬。这篇我给一个最小可运行的 Python WebSocket 服务端骨架，把"连接 → 注册 → 登录 → 消息循环"整个状态机跑通。

先说清楚：这是示例实现，生产要补wss/TLS、token 持久化与轮换、并发隔离、限流、消息校验、异常处理。别直接上生产，但拿它理解协议流程够了。

状态机长这样

代码（Python + websockets）

依赖：pip install websockets

""" 最小 M50 人脸终端 WebSocket 接入服务端（示例实现） 仅用于理解三阶段状态机；生产环境需补充 wss/TLS、token 持久化与轮换、 并发隔离、限流、消息校验、异常处理。 """importasyncioimportbase64importsecretsimportxml.etree.ElementTreeasETimportwebsockets# 模拟数据库VALID_CLOUD_IDS={"cloud-demo-001"}DEVICES={}# serial -> tokendefdecode_name(b64data:str)->str:"""协议里的 Name/部门名是 base64 UTF-16 LE 编码，这里做解码示例"""returnbase64.b64decode(b64data).decode("utf-16-le")defparse(xml_text:str):# 协议样例标签常带不规则空格如 < Subnet >，先规整再解析xml_text=xml_text.replace(" < ","<").replace(" > ",">")returnET.fromstring(xml_text)defbuild(tag,name,trans_id,result,**extra):parts=[f"<{tag}>{name}</{tag}>"]iftrans_id:parts.append(f"<TransID>{trans_id}</TransID>")parts.append(f"<Result>{result}</Result>")fork,vinextra.items():parts.append(f"<{k}>{v}</{k}>")return'<?xml version="1.0"?><Message>'+"".join(parts)+"</Message>"asyncdefhandler(ws):state="CONNECTED"serial=Nonetry:asyncforrawinws:msg=parse(raw)req=(msg.findtext("Request")ormsg.findtext("Event")ormsg.findtext("Response"))trans_id=msg.findtext("TransID")# ---- Register 阶段：校验 CloudId ----ifreq=="Register":ifmsg.findtext("CloudId")notinVALID_CLOUD_IDS:awaitws.send(build("Response","Register",trans_id,"Fail"))returntoken=secrets.token_hex(16)# 服务端生成安全 tokenserial=msg.findtext("DeviceSerialNo")DEVICES[serial]=token state="REGISTERED"awaitws.send(build("Response","Register",trans_id,"OK",Token=token))continue# ---- Login 阶段：校验 token ----ifreq=="Login":ifDEVICES.get(serial)!=msg.findtext("Token"):awaitws.send(build("Response","Login",trans_id,"FailUnknownToken"))returnstate="LOGGED_IN"awaitws.send(build("Response","Login",trans_id,"OK"))continue# ---- 未登录就发业务消息：拒绝 ----ifstate!="LOGGED_IN":awaitws.send(build("Response",reqor"Unknown",trans_id,"Fail"))continue# ---- 已登录：消息循环 ----ifreq=="TimeLog_v2":# 真实项目里这里落库 + 推业务系统（考勤大屏/微信推送）awaitws.send(build("Response","TimeLog_v2",trans_id,"OK"))continueifreq=="KeepAlive":awaitws.send(build("Response","KeepAlive",trans_id,"OK"))continue# 其它命令：示例里原样回 OKawaitws.send(build("Response",reqor"Unknown",trans_id,"OK"))exceptExceptionase:print("conn closed:",e)asyncdefmain():# 注意：示例用 ws:// 方便本地跑，生产必须 wss://asyncwithwebsockets.serve(handler,"0.0.0.0",8765):print("server on :8765 (ws://)")awaitasyncio.Future()if__name__=="__main__":asyncio.run(main())

几个要点（对照前面几篇）

• TransID必须原样回传——请求和响应靠它关联，也是防重放的基础（安全设计那篇讲过）。
• CloudId白名单在 Register 阶段就卡掉非法设备，这是租户隔离的第一道闸。
• token用secrets.token_hex生成，别硬编码；真实环境存库 + 设过期。
• TimeLog_v2必须回OK，否则设备可能停止推送（日志接入那篇的坑）。
• XML 标签不规则空格：协议样例里常写成< Subnet >这种，解析前先.replace(" < ", "<")规整，不然ElementTree解析会出问题。
• 中文名解码就是base64.b64decode(...).decode("utf-16-le")，一行搞定，但别忘了编码方向。

上生产前你还得补的

1. 换成wss://（TLS），并开启证书校验（别开AllowNoCertificate）。
2. token 持久化到数据库，支持轮换和吊销。
3. 多设备并发：状态要按连接隔离，别用全局变量串台。
4. 服务器侧限流 + 鉴权，设备会每 10s 重试，别被刷爆。
5. 业务命令的 handler 一个个补上（人员同步、生物模板、设备控制……），前面每篇的字段细节直接能用。

有了这骨架，把各命令的 handler 填上就是个能用的接入服务。配合前面几篇的字段细节，对接基本没有盲区了。下一篇给一份 46 命令全景速查地图，方便你当字典翻。