"""模块:Talk 用户级 Socket.IO 命名空间(/talk) 面向企业聊天(Talk)的终端用户实时通道。与同文件体系中的 /msync 命名空间不同: /msync 面向“服务”,鉴权是服务密钥 HMAC,没有终端用户身份;/talk 面向“用户”, 鉴权是用户 JWT,身份态在连接建立时解析并绑定到 sid。 事件协议(字段全部 PascalCase): 客户端 → 服务端: connect (握手,从 Cookie 或 auth.Token 尝试认证) talk:auth {Token} 用户令牌认证(Cookie 不可用时的兜底) talk:send {ConversationUUID, MsgType, Content, AttachmentSHA1, AttachmentName, AttachmentSize, AttachmentMime} 发送消息(需已认证) talk:read {ConversationUUID, LastReadMessageUUID} 标记已读(需已认证) talk:typing {ConversationUUID} 正在输入(需已认证) talk:ping 应用层心跳(需已认证) 服务端 → 客户端(推送/响应): talk:ready _ok({UserUUID, CompanyUUID}) 认证成功(连接或 talk:auth 后) talk:message _ok(消息 dict) 会话内新消息广播 talk:read _ok({ConversationUUID, UserUUID, LastReadMessageUUID}) 已读回执广播 talk:typing _ok({ConversationUUID, UserUUID}) 输入中广播(跳过发起者) talk:pong _ok() 心跳响应 talk:error _err(...) 处理异常时单独推给触发的 sid 鉴权链路(与 HTTP 侧完全一致,不另写一套): Cookie AccessToken(JWT) --VerifyUserToken--> UserUUID/CompanyUUID/TenantUUID/RoleUUID Cookie CurrentRoleUUID --ResolveRoleContext--> current_role / current_company_uuid / current_tenant_uuid 其中角色/公司/租户上下文复用 gvsdsdk.auth.middleware.ResolveRoleContext(唯一实现), 通过一个轻量 shim 对象代替 Django HttpRequest 传入。 令牌用 SECRET_KEY 作为 HMAC 密钥,算法 HS256(见 IssueUserToken),有效期 12 小时。 单进程限制(重要): python-socketio 运行在 async_mode='gevent' 的单进程内,且未配置 Redis / 消息队列 adapter。因此: * 本模块的 _sid_user_map、会话房间(room)均只存在于当前进程内存; * BroadcastToConversation / BroadcastToUser 只能触达“连接到本进程”的 socket, 无法跨 worker 广播; * 若未来以多 worker/多进程部署(如 gunicorn -w N、gevent 多实例),必须引入 socketio.RedisManager(或其它 adapter)才能实现跨进程房间与广播; 同时 _sid_user_map 也应迁移到共享存储(Redis)以免身份态不一致。 """ import datetime import logging import threading import jwt from django.conf import settings from gvsdsdk.auth.middleware import ResolveRoleContext from gvsdsdk.model_utils import UUIDObj from gvsdsdk.msync.sio import sio logger = logging.getLogger(__name__) #: 用户级命名空间 TALK_NS = '/talk' #: 用户令牌有效期(小时) TALK_TOKEN_TTL_HOURS = 12 #: JWT 算法(HS256,密钥为 settings.SECRET_KEY) TALK_TOKEN_ALGORITHM = 'HS256' #: 与 gvsds/auth.py 保持一致的 Cookie 名(此处不直接 import gvsds.auth, #: 以避免 gvsdsdk 对 gvsds 业务包的强依赖;如需改名请同步两处) JWT_ACCESS_COOKIE = 'AccessToken' #: sid → 用户上下文字典(进程内,单进程限制见模块 docstring) _sid_user_map = {} _map_lock = threading.Lock() # ── 基础助手(与 gvsdsdk.msync.sio 的响应约定保持一致) ── def _ok(data=None, msg='success'): return {'code': 200, 'msg': msg, 'data': data} def _err(msg, code=400): return {'code': code, 'msg': msg, 'data': None} def _d(data, key, default=None): """安全获取字典字段""" if not isinstance(data, dict): return default return data.get(key, default) def _UuidBytes(value): """任意 UUID 表示(str/bytes/uuid.UUID/UUIDObj)→ 16 字节;非法或空返回 None""" if value is None or value == '': return None try: return UUIDObj(value).bytes except (ValueError, TypeError, AttributeError): return None def _UuidStr(value): """任意 UUID 表示 → 字符串;非法或空返回 None""" if value is None or value == '': return None try: return UUIDObj(value).str except (ValueError, TypeError, AttributeError): return None def _ConvRoom(ConversationUUID): """会话房间名(字符串形式,str/bytes 入参均可处理);非法返回 None""" conversation_str = _UuidStr(ConversationUUID) if not conversation_str: return None return f'conv:{conversation_str}' def _ParseCookies(RawCookie): """解析 WSGI environ['HTTP_COOKIE'] 为 {name: value}""" cookies = {} if not RawCookie: return cookies from http.cookies import SimpleCookie jar = SimpleCookie() try: jar.load(RawCookie) except Exception: logger.exception('[TALK] Cookie 解析失败') return cookies for key, morsel in jar.items(): cookies[key] = morsel.value return cookies class _RequestShim: """轻量请求 shim——让 ResolveRoleContext 可在 WSGI environ 场景复用 仅提供 ResolveRoleContext 所需的属性:COOKIES / user, 以及会被其原地赋值的 current_role / current_company_uuid / current_tenant_uuid。 """ def __init__(self, Cookies, User): self.COOKIES = Cookies or {} self.user = User self.current_role = None self.current_company_uuid = None self.current_tenant_uuid = None # ── 令牌签发 / 校验 ── def IssueUserToken(UserUUID, CompanyUUID, TenantUUID, RoleUUID): """签发用户级令牌 使用 settings.SECRET_KEY 作为 HMAC 密钥,算法 HS256(复用 simplejwt 已依赖的 PyJWT, 不新增第三方依赖)。令牌内 UUID 以字符串形式承载(JSON 安全),校验后统一转为字节。 有效期 TALK_TOKEN_TTL_HOURS(12)小时。 Args: UserUUID: 用户 UUID(str/bytes/uuid.UUID/UUIDObj) CompanyUUID: 当前公司 UUID,可为 None TenantUUID: 当前租户 UUID,可为 None RoleUUID: 当前角色 UUID,可为 None Returns: str: 编码后的 JWT """ now = datetime.datetime.now(datetime.timezone.utc) payload = { 'UserUUID': _UuidStr(UserUUID), 'CompanyUUID': _UuidStr(CompanyUUID), 'TenantUUID': _UuidStr(TenantUUID), 'RoleUUID': _UuidStr(RoleUUID), 'iat': now, 'exp': now + datetime.timedelta(hours=TALK_TOKEN_TTL_HOURS), } return jwt.encode(payload, settings.SECRET_KEY, algorithm=TALK_TOKEN_ALGORITHM) def VerifyUserToken(Token): """校验用户级令牌 Args: Token: JWT 字符串 Returns: dict | None: 合法时返回 {'UserUUID','CompanyUUID','TenantUUID','RoleUUID'}(值均为 16 字节 或 None);令牌缺失/非法/过期,或缺少 UserUUID 时返回 None。 """ if not Token: return None try: claims = jwt.decode(Token, settings.SECRET_KEY, algorithms=[TALK_TOKEN_ALGORITHM]) except jwt.InvalidTokenError: # InvalidTokenError 是 ExpiredSignatureError 等所有解码异常的基类 return None user_uuid = _UuidBytes(claims.get('UserUUID')) if not user_uuid: return None return { 'UserUUID': user_uuid, 'CompanyUUID': _UuidBytes(claims.get('CompanyUUID')), 'TenantUUID': _UuidBytes(claims.get('TenantUUID')), 'RoleUUID': _UuidBytes(claims.get('RoleUUID')), } def ResolveUserFromEnviron(Environ): """从 WSGI environ 解析当前用户上下文 流程:environ['HTTP_COOKIE'] → AccessToken → VerifyUserToken 得到用户身份; 再用 CurrentRoleUUID 经 ResolveRoleContext(唯一角色解析实现)解析当前 公司/租户/角色,并以解析结果为准(解析不到时回退到令牌内的声明)。 Args: Environ: WSGI environ 字典 Returns: dict | None: 认证成功返回 {'UserUUID','CompanyUUID','TenantUUID','RoleUUID'}(值均为 16 字节或 None); 未认证返回 None。 """ try: cookies = _ParseCookies((Environ or {}).get('HTTP_COOKIE', '')) claims = VerifyUserToken(cookies.get(JWT_ACCESS_COOKIE, '')) if claims is None: return None user_uuid = claims['UserUUID'] user = _GetUser(user_uuid) # 复用唯一角色解析实现(内部为同步 ORM;Main.py 已 monkey.patch_all(), # gevent 下会正确让出协程,无需额外处理) shim = _RequestShim(cookies, user) ResolveRoleContext(shim, user) company_uuid = _UuidBytes(shim.current_company_uuid) or claims.get('CompanyUUID') tenant_uuid = _UuidBytes(shim.current_tenant_uuid) or claims.get('TenantUUID') role_uuid = _UuidBytes(getattr(shim.current_role, 'RoleUUID', None)) or claims.get('RoleUUID') return { 'UserUUID': user_uuid, 'CompanyUUID': company_uuid, 'TenantUUID': tenant_uuid, 'RoleUUID': role_uuid, } except Exception: logger.exception('[TALK] ResolveUserFromEnviron 解析失败') return None def _GetUser(UserUUID): """按 UserUUID(16 字节)取用户实例;取不到返回 None""" try: from gvsds.apps.core.models import User except ImportError: logger.warning('[TALK] gvsds.apps.core.models.User 不可用') return None return User.objects.filter(UserUUID=UserUUID).first() # ── 广播 ── def BroadcastToConversation(ConversationUUID, Event, Data, SkipSid=None): """向会话房间广播事件 房间名固定为 f'conv:{ConversationUUID}'(ConversationUUID 传 str/bytes 均可)。 Args: ConversationUUID: 会话 UUID Event: 事件名 Data: 事件负载 SkipSid: 可选,跳过该 sid(如输入中事件不回发给发起者) 注意:单进程限制——只能触达本进程内的连接(见模块 docstring)。 """ room = _ConvRoom(ConversationUUID) if not room: logger.warning('[TALK] BroadcastToConversation 收到非法 ConversationUUID: %r', ConversationUUID) return sio.emit(Event, Data, namespace=TALK_NS, room=room, skip_sid=SkipSid) def BroadcastToUser(UserUUID, Event, Data): """向某用户的全部连接推送(用于未读角标等) 注意:单进程限制——仅本进程内该用户的连接可见(见模块 docstring)。 """ target = _UuidBytes(UserUUID) if not target: logger.warning('[TALK] BroadcastToUser 收到非法 UserUUID: %r', UserUUID) return with _map_lock: sids = [sid for sid, ctx in _sid_user_map.items() if ctx.get('UserUUID') == target] for sid in sids: try: sio.emit(Event, Data, namespace=TALK_NS, to=sid) except Exception: logger.exception('[TALK] BroadcastToUser 推送失败 sid=%s', sid) # ── sid 身份态助手 ── def _SetSidUser(sid, Context): with _map_lock: _sid_user_map[sid] = Context def _GetSidUser(sid): with _map_lock: return _sid_user_map.get(sid) def _PopSidUser(sid): with _map_lock: return _sid_user_map.pop(sid, None) def _ReadyPayload(Context): return _ok({ 'UserUUID': _UuidStr(Context.get('UserUUID')), 'CompanyUUID': _UuidStr(Context.get('CompanyUUID')), }) def _Fail(sid, Error, Code=500): """统一异常处理:记录堆栈、向该 sid 推 talk:error、返回 _err""" logger.exception('[TALK] 处理失败 sid=%s code=%s', sid, Code) payload = _err(str(Error), Code) try: sio.emit('talk:error', payload, namespace=TALK_NS, to=sid) except Exception: logger.exception('[TALK] 推送 talk:error 失败 sid=%s', sid) return payload def _JoinUserConversations(sid, Context): """把 sid 加入该用户所在的所有会话房间 会话列表由 gvsds 侧 Talk 模块提供(gvsds.apps.talk.manager.ListConversationUUIDsForUser)。 该模块可能尚未启用,故惰性导入并容错:导入失败时仅绑定 sid→user,不加入任何房间。 """ try: from gvsds.apps.talk.manager import ListConversationUUIDsForUser except ImportError: logger.warning('[TALK] gvsds.apps.talk.manager 不可用,跳过会话房间加入') return try: try: conversation_uuids = ListConversationUUIDsForUser( Context.get('UserUUID'), Context.get('CompanyUUID'), ) except TypeError: # 兼容仅接收 UserUUID 的实现 conversation_uuids = ListConversationUUIDsForUser(Context.get('UserUUID')) except Exception: logger.exception('[TALK] 获取用户会话列表失败 sid=%s', sid) return for conversation_uuid in (conversation_uuids or []): room = _ConvRoom(conversation_uuid) if not room: continue try: sio.enter_room(sid, room, namespace=TALK_NS) except Exception: logger.exception('[TALK] 加入会话房间失败 sid=%s room=%s', sid, room) # ── 事件处理器 ── @sio.on('connect', namespace=TALK_NS) def on_talk_connect(sid, environ, auth=None): """连接握手:尝试从 Cookie 或 auth.Token 认证 认证失败“不拒绝连接”(否则前端只看到握手失败、拿不到原因), 仅记录未认证状态并等待 talk:auth;所有业务事件都会再次校验认证态。 """ try: context = ResolveUserFromEnviron(environ) if context is None: claims = VerifyUserToken(_d(auth, 'Token', '')) if claims: context = dict(claims) if not context: logger.warning('[TALK] connect 未认证 sid=%s,等待 talk:auth', sid) return _SetSidUser(sid, context) _JoinUserConversations(sid, context) sio.emit('talk:ready', _ReadyPayload(context), namespace=TALK_NS, to=sid) logger.info('[TALK] connect 已认证 sid=%s user=%s', sid, _UuidStr(context.get('UserUUID'))) except Exception as e: _Fail(sid, e) @sio.on('disconnect', namespace=TALK_NS) def on_talk_disconnect(sid, reason=None): """断开:清理身份态并离开所有房间""" try: _PopSidUser(sid) try: rooms = sio.rooms(sid, namespace=TALK_NS) or [] except Exception: rooms = [] for room in rooms: if room == sid: continue try: sio.leave_room(sid, room, namespace=TALK_NS) except Exception: logger.exception('[TALK] 离开房间失败 sid=%s room=%s', sid, room) logger.info('[TALK] disconnect sid=%s reason=%s', sid, reason) except Exception: logger.exception('[TALK] disconnect 清理失败 sid=%s', sid) @sio.on('talk:auth', namespace=TALK_NS) def on_talk_auth(sid, data): """用户令牌认证(Cookie 不可用时的兜底),成功后绑定 sid→user 并加入会话房间""" try: claims = VerifyUserToken(_d(data, 'Token', '')) if not claims: return _err('认证失败', 401) _SetSidUser(sid, claims) _JoinUserConversations(sid, claims) payload = _ReadyPayload(claims) sio.emit('talk:ready', payload, namespace=TALK_NS, to=sid) return payload except Exception as e: return _Fail(sid, e) @sio.on('talk:send', namespace=TALK_NS) def on_talk_send(sid, data): """发送消息:落库后向会话房间广播 talk:message""" try: context = _GetSidUser(sid) if not context or not context.get('UserUUID'): return _err('未认证', 401) conversation_uuid = _UuidBytes(_d(data, 'ConversationUUID')) if not conversation_uuid: return _err('缺少 ConversationUUID', 400) try: from gvsds.apps.talk.manager import SendMessageByUser except ImportError: return _err('Talk 服务未启用', 501) try: message = SendMessageByUser( context['UserUUID'], context.get('CompanyUUID'), conversation_uuid, MsgType=_d(data, 'MsgType', 0), Content=_d(data, 'Content', ''), AttachmentSHA1=_d(data, 'AttachmentSHA1', ''), AttachmentName=_d(data, 'AttachmentName', ''), AttachmentSize=_d(data, 'AttachmentSize', 0), AttachmentMime=_d(data, 'AttachmentMime', ''), SkipSid=sid, ) except Exception as e: # 业务异常:向发起者单独推送 talk:error return _Fail(sid, e) # 注意:talk:message 广播已由 TalkManager.SendMessage 内部完成 # (REST 与 socket 共用同一广播出口),此处不得重复广播 return _ok(message) except Exception as e: return _Fail(sid, e) @sio.on('talk:read', namespace=TALK_NS) def on_talk_read(sid, data): """标记已读:落库后向会话房间广播 talk:read""" try: context = _GetSidUser(sid) if not context or not context.get('UserUUID'): return _err('未认证', 401) conversation_uuid = _UuidBytes(_d(data, 'ConversationUUID')) if not conversation_uuid: return _err('缺少 ConversationUUID', 400) last_read_uuid = _UuidBytes(_d(data, 'LastReadMessageUUID')) try: from gvsds.apps.gerp.apps.talk.manager import MarkReadByUser except ImportError: return _err('Talk 服务未启用', 501) try: success = MarkReadByUser(context['UserUUID'], conversation_uuid, last_read_uuid) except Exception as e: return _Fail(sid, e) if not success: return _err('标记已读失败', 400) # 注意:talk:read 广播已由 TalkManager.MarkRead 内部完成(BroadcastRead), # 此处不得重复广播 return _ok(msg='已标记已读') except Exception as e: return _Fail(sid, e) @sio.on('talk:typing', namespace=TALK_NS) def on_talk_typing(sid, data): """正在输入:向会话房间广播(跳过发起者自身)""" try: context = _GetSidUser(sid) if not context or not context.get('UserUUID'): return _err('未认证', 401) conversation_uuid = _UuidBytes(_d(data, 'ConversationUUID')) if not conversation_uuid: return _err('缺少 ConversationUUID', 400) BroadcastToConversation(conversation_uuid, 'talk:typing', { 'ConversationUUID': _UuidStr(conversation_uuid), 'UserUUID': _UuidStr(context['UserUUID']), }, SkipSid=sid) return _ok() except Exception as e: return _Fail(sid, e) @sio.on('talk:ping', namespace=TALK_NS) def on_talk_ping(sid, data=None): """应用层心跳""" try: payload = _ok() sio.emit('talk:pong', payload, namespace=TALK_NS, to=sid) return payload except Exception as e: return _Fail(sid, e)