From cd789dd2dd4dbfad48d1484f63357ee7733fe1be Mon Sep 17 00:00:00 2001
From: TermiNexus <wyt18222152539wyt@163.com>
Date: Sat, 26 Sep 2026 23:46:21 +0800
Subject: [PATCH] =?UTF-8?q?feat(msync):=20=E6=96=B0=E5=A2=9E=E7=94=A8?=
 =?UTF-8?q?=E6=88=B7=E7=BA=A7=20Socket.IO=20=E5=91=BD=E5=90=8D=E7=A9=BA?=
 =?UTF-8?q?=E9=97=B4=20/talk?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- talk_socket.py：Talk 用户令牌鉴权（Cookie AccessToken 或 talk:auth）+ 复用 ResolveRoleContext
  解析公司/租户/角色
- 事件 talk:auth/send/read/typing/ping ↔ 推送 talk:ready/message/read/typing/pong/error，
  会话房间统一 conv:<UUID>
- create_sio_app 惰性导入注册命名空间；单进程 gevent 无 Redis adapter（多 worker 需补）
---
 msync/sio.py   |   5 +
 talk_socket.py | 541 +++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 546 insertions(+)
 create mode 100644 talk_socket.py

diff --git a/msync/sio.py b/msync/sio.py
index 7e1409a..6f6d3e9 100644
--- a/msync/sio.py
+++ b/msync/sio.py
@@ -590,4 +590,9 @@ def on_app_manifest_report(sid, data):
 
 
 def create_sio_app(django_app):
+    # 注册用户级命名空间 /talk（导入即完成事件注册）
+    try:
+        import gvsdsdk.talk_socket  # noqa: F401
+    except Exception:
+        logger.exception('[SIO] talk 命名空间注册失败')
     return socketio.WSGIApp(sio, django_app)
diff --git a/talk_socket.py b/talk_socket.py
new file mode 100644
index 0000000..8db247e
--- /dev/null
+++ b/talk_socket.py
@@ -0,0 +1,541 @@
+"""模块：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)