2.2.0

backend/app/config/setting.py

@@ -43,6 +43,7 @@ class Settings(BaseSettings):
     SUMMARY: str = "接口汇总"  # 文档概述
     DOCS_URL: str = "/docs"  # Swagger UI路径
     REDOC_URL: str = "/redoc"  # ReDoc路径
+    LJDOC_URL: str = "/ljdoc"   # LangJin UI路径
     ROOT_PATH: str = "/api/v1"  # API路由前缀
 
     # ================================================= #
@@ -166,6 +167,8 @@ class Settings(BaseSettings):
     SWAGGER_CSS_URL: str = "static/swagger/swagger-ui/swagger-ui.css"
     SWAGGER_JS_URL: str = "static/swagger/swagger-ui/swagger-ui-bundle.js"
     REDOC_JS_URL: str = "static/swagger/redoc/bundles/redoc.standalone.js"
+    CUSTOM_CSS_URL: str = "static/swagger/custom-ui/styles.css"
+    CUSTOM_JS_URL: str = "static/swagger/custom-ui/scripts.js"
     FAVICON_URL: str = "static/swagger/favicon.png"
 
     # ================================================= #

backend/app/core/dependencies.py

@@ -1,7 +1,7 @@
 import json
 from collections.abc import AsyncGenerator
 
-from fastapi import Depends, Request
+from fastapi import Depends, Query, Request
 from redis.asyncio.client import Redis
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
@@ -129,6 +129,110 @@ async def get_current_user(
     return auth
 
 
+async def get_current_user_ws(
+    token: str = Query(..., description="认证token"),
+    db: AsyncSession = Depends(db_getter),
+    redis: Redis = Depends(redis_getter),
+) -> AuthSchema:
+    """获取当前用户（WebSocket专用，从查询参数获取token）
+
+    参数:
+    - token (str): 认证token
+    - db (AsyncSession): 数据库会话
+    - redis (Redis): Redis连接
+
+    返回:
+    - AuthSchema: 认证信息模型
+    """
+    return await _verify_token(token, db, redis)
+
+
+async def _verify_token(
+    token: str,
+    db: AsyncSession,
+    redis: Redis,
+) -> AuthSchema:
+    """验证token并返回用户信息
+
+    参数:
+    - token (str): 认证token
+    - db (AsyncSession): 数据库会话
+    - redis (Redis): Redis连接
+
+    返回:
+    - AuthSchema: 认证信息模型
+    """
+    if not token:
+        raise CustomException(msg="认证已失效", code=10401, status_code=401)
+
+    # 处理Bearer token（如果通过查询参数传递时包含Bearer前缀）
+    if token.startswith("Bearer"):
+        token = token.split(" ")[1]
+
+    payload = decode_access_token(token)
+    if not payload or not hasattr(payload, "is_refresh") or payload.is_refresh:
+        raise CustomException(msg="非法凭证", code=10401, status_code=401)
+
+    online_user_info = payload.sub
+    # 从Redis中获取用户信息
+    user_info = json.loads(online_user_info)  # 确保是字典类型
+
+    session_id = user_info.get("session_id")
+    if not session_id:
+        raise CustomException(msg="认证已失效", code=10401, status_code=401)
+
+    # 检查用户是否在线
+    online_ok = await RedisCURD(redis).exists(
+        key=f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:{session_id}"
+    )
+    if not online_ok:
+        raise CustomException(msg="认证已失效", code=10401, status_code=401)
+
+    # 如果启用了滑动过期，自动续期token
+    if settings.TOKEN_SLIDING_EXPIRE:
+        await RedisCURD(redis).expire(
+            key=f"{RedisInitKeyConfig.ACCESS_TOKEN.key}:{session_id}",
+            expire=settings.ACCESS_TOKEN_EXPIRE_MINUTES,
+        )
+        await RedisCURD(redis).expire(
+            key=f"{RedisInitKeyConfig.REFRESH_TOKEN.key}:{session_id}",
+            expire=settings.REFRESH_TOKEN_EXPIRE_MINUTES,
+        )
+
+    # 关闭数据权限过滤，避免当前用户查询被拦截
+    auth = AuthSchema(db=db, check_data_scope=False)
+    username = user_info.get("user_name")
+    if not username:
+        raise CustomException(msg="认证已失效", code=10401, status_code=401)
+    # 获取用户信息，使用深层预加载确保RoleModel.creator被正确加载
+    user = await UserCRUD(auth).get_by_username_crud(
+        username=username,
+        preload=[
+            "dept",
+            selectinload(UserModel.roles),
+            "positions",
+            "created_by",
+        ],
+    )
+    if not user:
+        raise CustomException(msg="用户不存在", code=10401, status_code=401)
+    if user.status == "1":
+        raise CustomException(msg="用户已被停用", code=10401, status_code=401)
+
+    # 设置请求上下文
+    # request.scope["user_id"] = user.id
+    # request.scope["user_username"] = user.username
+
+    # 过滤可用的角色和职位
+    if hasattr(user, "roles"):
+        user.roles = [role for role in user.roles if role and role.status]
+    if hasattr(user, "positions"):
+        user.positions = [pos for pos in user.positions if pos and pos.status]
+
+    auth.user = user
+    return auth
+
+
 class AuthPermission:
     """权限验证类"""
 

backend/app/core/docs.py

@@ -0,0 +1,86 @@
+from typing import Annotated
+
+from starlette.responses import HTMLResponse
+from typing_extensions import Doc
+
+
+def get_custom_ui_html(
+    *,
+    openapi_url: Annotated[
+        str,
+        Doc(
+            """
+            The OpenAPI URL that Swagger UI should load and use.
+
+            This is normally done automatically by FastAPI using the default URL
+            `/openapi.json`.
+            """
+        ),
+    ],
+    title: Annotated[
+        str,
+        Doc(
+            """
+            The HTML `<title>` content, normally shown in the browser tab.
+            """
+        ),
+    ],
+    swagger_js_url: Annotated[
+        str,
+        Doc(
+            """
+            The URL to use to load the Swagger UI JavaScript.
+
+            It is normally set to a CDN URL.
+            """
+        ),
+    ] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js",
+    swagger_css_url: Annotated[
+        str,
+        Doc(
+            """
+            The URL to use to load the Swagger UI CSS.
+
+            It is normally set to a CDN URL.
+            """
+        ),
+    ] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css",
+    swagger_favicon_url: Annotated[
+        str,
+        Doc(
+            """
+            The URL of the favicon to use. It is normally shown in the browser tab.
+            """
+        ),
+    ] = "https://fastapi.tiangolo.com/img/favicon.png",
+) -> HTMLResponse:
+    """
+    Generate and return the HTML  that loads Swagger UI for the interactive
+    API docs (normally served at `/docs`).
+
+    You would only call this function yourself if you needed to override some parts,
+    for example the URLs to use to load Swagger UI's JavaScript and CSS.
+
+    Read more about it in the
+    [FastAPI docs for Configure Swagger UI](https://fastapi.tiangolo.com/how-to/configure-swagger-ui/)
+    and the [FastAPI docs for Custom Docs UI Static Assets (Self-Hosting)](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/).
+    """
+    html = f"""
+    <!DOCTYPE html>
+    <html>
+    <head>
+    <link type="text/css" rel="stylesheet" href="{swagger_css_url}">
+    <link rel="shortcut icon" href="{swagger_favicon_url}">
+    <title>{title}</title>
+    </head>
+    <body>
+    <div id="swagger-ui"></div>
+    <script src="{swagger_js_url}"></script>
+    <script>
+        ui.configure({{url: '{openapi_url}'}});
+        ui.initialize();
+    </script>
+    </body>
+    </html>
+    """
+    return HTMLResponse(html)

backend/app/plugin/init_app.py

@@ -14,6 +14,7 @@
 from fastapi_limiter.depends import RateLimiter, WebSocketRateLimiter
 
 from app.config.setting import settings
+from app.core.docs import get_custom_ui_html
 from app.core.exceptions import handle_exception
 from app.core.http_limit import http_limit_callback, ws_limit_callback
 from app.core.logger import log
@@ -212,3 +213,13 @@ async def custom_redoc_html():
             redoc_js_url=settings.REDOC_JS_URL,
             redoc_favicon_url=settings.FAVICON_URL,
         )
+
+    @app.get(settings.LJDOC_URL, include_in_schema=False)
+    async def custom_ui_html():
+        return get_custom_ui_html(
+            openapi_url=str(app.root_path) + str(app.openapi_url),
+            title=app.title + " - LangJin UI",
+            swagger_js_url=settings.CUSTOM_JS_URL,
+            swagger_css_url=settings.CUSTOM_CSS_URL,
+            swagger_favicon_url=settings.FAVICON_URL
+        )