手撸一个能跑的人脸设备 WebSocket 服务端三阶段状态机实战讲了一堆协议不来点能跑的代码总觉得悬。这篇我给一个最小可运行的 Python WebSocket 服务端骨架把连接 → 注册 → 登录 → 消息循环整个状态机跑通。先说清楚这是示例实现生产要补wss/TLS、token 持久化与轮换、并发隔离、限流、消息校验、异常处理。别直接上生产但拿它理解协议流程够了。状态机长这样CloudId 非法CloudId 合法token 不符token 相符连接建立 CONNECTED收到 Register?拒绝并断开生成 token / 存库 → REGISTERED收到 Login?FailUnknownToken 断开LOGGED_IN 消息循环TimeLog_v2 → 回 OK 落库KeepAlive → 回 OK其它命令 → 回 OK代码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_textxml_text.replace( ,).replace( ,)returnET.fromstring(xml_text)defbuild(tag,name,trans_id,result,**extra):parts[f{tag}{name}/{tag}]iftrans_id:parts.append(fTransID{trans_id}/TransID)parts.append(fResult{result}/Result)fork,vinextra.items():parts.append(f{k}{v}/{k})return?xml version1.0?Message.join(parts)/Messageasyncdefhandler(ws):stateCONNECTEDserialNonetry:asyncforrawinws:msgparse(raw)req(msg.findtext(Request)ormsg.findtext(Event)ormsg.findtext(Response))trans_idmsg.findtext(TransID)# ---- Register 阶段校验 CloudId ----ifreqRegister:ifmsg.findtext(CloudId)notinVALID_CLOUD_IDS:awaitws.send(build(Response,Register,trans_id,Fail))returntokensecrets.token_hex(16)# 服务端生成安全 tokenserialmsg.findtext(DeviceSerialNo)DEVICES[serial]token stateREGISTEREDawaitws.send(build(Response,Register,trans_id,OK,Tokentoken))continue# ---- Login 阶段校验 token ----ifreqLogin:ifDEVICES.get(serial)!msg.findtext(Token):awaitws.send(build(Response,Login,trans_id,FailUnknownToken))returnstateLOGGED_INawaitws.send(build(Response,Login,trans_id,OK))continue# ---- 未登录就发业务消息拒绝 ----ifstate!LOGGED_IN:awaitws.send(build(Response,reqorUnknown,trans_id,Fail))continue# ---- 已登录消息循环 ----ifreqTimeLog_v2:# 真实项目里这里落库 推业务系统考勤大屏/微信推送awaitws.send(build(Response,TimeLog_v2,trans_id,OK))continueifreqKeepAlive:awaitws.send(build(Response,KeepAlive,trans_id,OK))continue# 其它命令示例里原样回 OKawaitws.send(build(Response,reqorUnknown,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)一行搞定但别忘了编码方向。上生产前你还得补的换成wss://TLS并开启证书校验别开AllowNoCertificate。token 持久化到数据库支持轮换和吊销。多设备并发状态要按连接隔离别用全局变量串台。服务器侧限流 鉴权设备会每 10s 重试别被刷爆。业务命令的 handler 一个个补上人员同步、生物模板、设备控制……前面每篇的字段细节直接能用。有了这骨架把各命令的 handler 填上就是个能用的接入服务。配合前面几篇的字段细节对接基本没有盲区了。下一篇给一份 46 命令全景速查地图方便你当字典翻。