From 55a0b3b90563004a1c7d52f922fecd581097719e Mon Sep 17 00:00:00 2001 From: ngfchl Date: Fri, 3 Apr 2026 14:04:39 +0800 Subject: [PATCH] =?UTF-8?q?update.=20=E4=BD=BF=E7=94=A8django-ninja-jwt?= =?UTF-8?q?=E4=BD=9C=E4=B8=BAToken=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/settings.py | 4 +- server/urls.py | 96 +++++++++++++--------------------------------- 2 files changed, 29 insertions(+), 71 deletions(-) diff --git a/server/settings.py b/server/settings.py index 85fc5e1..b323f81 100644 --- a/server/settings.py +++ b/server/settings.py @@ -11,8 +11,9 @@ https://docs.djangoproject.com/en/6.0/ref/settings/ """ import os from pathlib import Path -from loguru import logger + import environ +from loguru import logger env = environ.Env() @@ -76,6 +77,7 @@ INSTALLED_APPS = [ 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', + 'ninja_extra', 'import_export', 'django_ckeditor_5', 'base.apps.BaseConfig', diff --git a/server/urls.py b/server/urls.py index 0862e00..e2fb74a 100644 --- a/server/urls.py +++ b/server/urls.py @@ -14,83 +14,40 @@ Including another URLconf 1. Import the include() function: from django.urls import include, path 2. Add a URL to urlpatterns: path('blog/', include('blog.urls')) """ +import importlib +import pkgutil +import traceback + +from django.conf import settings from django.conf.urls.static import static from django.contrib import admin from django.contrib.staticfiles.urls import staticfiles_urlpatterns -from django.urls import path, include, re_path - -import logging -import os -import pkgutil -import traceback -import importlib - -import jwt -from django.contrib.auth.models import User from django.http import JsonResponse -from jwt import exceptions -from ninja import NinjaAPI +from django.urls import path, include +from loguru import logger from ninja.errors import ValidationError -from ninja.security import HttpBearer +from ninja_extra import NinjaExtraAPI +from ninja_jwt.controller import NinjaJWTDefaultController from base.views import custom_upload_file from common.common_response import CommonResponse -from .settings import SECRET_KEY - -logger = logging.getLogger('ptools') -class GlobalAuth(HttpBearer): - def authenticate(self, request, token): - try: - logger.debug('用户认证中') - logger.debug(f'解析 Auth Token：{token}') - res = jwt.decode(token, SECRET_KEY, algorithms=["HS256"]) - user = User.objects.filter(id=res.get('id'), username=res.get('username')).first() - logger.debug(user) - if not user: - raise exceptions.PyJWKError('用户不存在') - request.user = user - return token - except exceptions.PyJWKError: - msg = 'Token 认证失败，请重新登陆！' - logger.error(msg) - raise - except exceptions.ExpiredSignatureError: - msg = '认证令牌已过期，请重新登陆！' - logger.error(msg) - raise - except Exception as e: - msg = '认证失败，请检查用户是否存在' - logger.error(msg) - raise - - -# 2. 获取当前文件所在的目录路径 (假设此文件与各个 app 文件夹同级，或者在特定配置目录下) -# 如果你的此文件在项目根目录，而 app 文件夹也在根目录： -from django.conf import settings - -project_root = settings.BASE_DIR - - -# 如果 app 文件夹是 Django 的已安装应用，可能需要通过 settings.INSTALLED_APPS 获取， -# 但通常直接扫描物理目录更灵活。这里假设扫描当前目录下的所有子文件夹。 - -# 3. 自动发现并注册路由 -def auto_load_routers(base_path, package_name_prefix=None): +def auto_load_routers(package_name_prefix=None): """ base_path: 物理路径，例如 /path/to/project package_name_prefix: 包名前缀，例如 'myproject.' (如果这些文件夹是包的一部分) """ + base_path = settings.BASE_DIR # 获取该路径下所有子模块/包 # 注意：这需要这些文件夹是合法的 Python 包 (即包含 __init__.py)，或者在 Python 路径中 - for importer, modname, ispkg in pkgutil.iter_modules([base_path]): + for importer, modname, is_pkg in pkgutil.iter_modules([base_path]): # 过滤条件： # 1. 必须是包 (文件夹) # 2. 不以 '_' 开头 (排除 __pycache__, _private 等) # 3. 排除当前文件所在的脚本名 (防止递归或错误) - if ispkg and not modname.startswith('_'): + if is_pkg and not modname.startswith('_'): # 构造完整的模块路径：例如 "authorize.views" # 如果这些文件夹直接位于根路径且不是大包的一部分，package_name_prefix 可能为空 @@ -126,8 +83,19 @@ def auto_load_routers(base_path, package_name_prefix=None): print(f"❌ 注册失败 {modname}: {e}") -api_v1 = NinjaAPI(version='1.0.0', auth=GlobalAuth()) -auto_load_routers(project_root) +class UnifiedNinjaAPI(NinjaExtraAPI): + def create_response(self, request, data, *, status=None, headers=None, temporal_response=None): + if not isinstance(data, CommonResponse): + if isinstance(data, tuple) and len(data) == 2: + data, status = data + data = CommonResponse.success(data=data) + return super().create_response(request, data, status=status, temporal_response=temporal_response) + + +api_v1 = UnifiedNinjaAPI(version='1.0.0') +api_v1.register_controllers(NinjaJWTDefaultController) + +auto_load_routers() @api_v1.exception_handler(ValidationError) @@ -138,18 +106,6 @@ def validation_errors(request, exc): return JsonResponse(data=CommonResponse.error(msg=f"输入参数验证失败！").to_dict(), status=422, safe=False) -@api_v1.exception_handler(exceptions.PyJWKError) # 捕获 HttpBearer 认证异常 -def custom_http_bearer_exception_handler(request, exc): - logger.error(exc) - return JsonResponse(data=CommonResponse.error(msg=f"用户Token认证失败：{exc}").to_dict(), status=401) - - -@api_v1.exception_handler(exceptions.ExpiredSignatureError) # 捕获 HttpBearer 认证异常 -def custom_http_bearer_exception_handler(request, exc): - logger.error(exc) - return JsonResponse(data=CommonResponse.error(msg=f"用户登录信息已过期，请重新登录！！").to_dict(), status=401) - - urlpatterns = [ path('', include("front.urls")), path('admin/', admin.site.urls),