初始化

backend/middleware/user_context.py

@@ -0,0 +1,94 @@
+"""Mokee Gateway 用户上下文中间件
+
+网关转发请求时注入以下请求头，本中间件读取并存入 request.state.user：
+  X-User-Id, X-User-Name, X-User-Account, X-User-Email,
+  X-User-Post, X-User-Roles, X-System-Id, X-System-Code, X-System-Name
+
+本地开发时（无网关），使用 DEV_MODE=true 环境变量或 X-Dev-Mock-User 请求头
+启用默认 admin 用户，无需启动网关即可调试。
+"""
+
+import os
+from starlette.middleware.base import BaseHTTPMiddleware
+from fastapi import Request
+
+# 网关注入的所有请求头
+GATEWAY_HEADERS = [
+    "X-User-Id",
+    "X-User-Name",
+    "X-User-Account",
+    "X-User-Email",
+    "X-User-Post",
+    "X-User-Roles",
+    "X-System-Id",
+    "X-System-Code",
+    "X-System-Name",
+]
+
+# 本地开发默认用户（模拟网关注入）
+DEV_DEFAULT_USER = {
+    "X-User-Id": "admin",
+    "X-User-Name": "管理员",
+    "X-User-Account": "admin",
+    "X-User-Email": "admin@zgitm.com",
+    "X-User-Post": "",
+    "X-User-Roles": '[{"roleId":"1","roleName":"管理员","roleCode":"admin"}]',
+    "X-System-Id": "1",
+    "X-System-Code": "mokeetext-edit",
+    "X-System-Name": "mokee编辑器",
+}
+
+
+class UserContext:
+    """用户上下文 — 提供快捷访问方法"""
+
+    @staticmethod
+    def from_request(request: Request) -> dict:
+        """从 request.state 获取用户信息"""
+        return getattr(request.state, "user", {})
+
+    @staticmethod
+    def get_user_id(request: Request) -> str:
+        """获取当前用户账号名（用于 created_by 显示），未登录返回 'admin'"""
+        user = UserContext.from_request(request)
+        # 优先用登录账号名（如 admin），其次用用户名，最后用 ID
+        return user.get("X-User-Account") or user.get("X-User-Name") or user.get("X-User-Id", "admin")
+
+    @staticmethod
+    def get_user_name(request: Request) -> str:
+        return UserContext.from_request(request).get("X-User-Name", "")
+
+    @staticmethod
+    def get_user_account(request: Request) -> str:
+        return UserContext.from_request(request).get("X-User-Account", "")
+
+    @staticmethod
+    def get_system_code(request: Request) -> str:
+        return UserContext.from_request(request).get("X-System-Code", "")
+
+
+class UserContextMiddleware(BaseHTTPMiddleware):
+    """将网关注入的用户信息存入 request.state.user
+
+    生产环境：读取网关转发的 X-User-* 请求头
+    本地开发：DEV_MODE=true 时使用 DEV_DEFAULT_USER
+    """
+
+    async def dispatch(self, request: Request, call_next):
+        user_info = {}
+
+        # 读取网关注入的请求头
+        for header in GATEWAY_HEADERS:
+            value = request.headers.get(header)
+            if value:
+                user_info[header] = value
+
+        # 本地开发：无网关时使用默认用户
+        if not user_info.get("X-User-Id"):
+            dev_mode = os.getenv("DEV_MODE", "false").lower() == "true"
+            if dev_mode or request.headers.get("X-Dev-Mock-User"):
+                user_info = DEV_DEFAULT_USER.copy()
+
+        request.state.user = user_info
+        response = await call_next(request)
+        return response