erpnew/flower/urls.py

"""
URL configuration for flower project.

The `urlpatterns` list routes URLs to views. For more information please see:
    https://docs.djangoproject.com/en/5.2/topics/http/urls/
Examples:
Function views
    1. Add an import:  from my_app import views
    2. Add a URL to urlpatterns:  path('', views.home, name='home')
Class-based views
    1. Add an import:  from other_app.views import Home
    2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')
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'))
"""
from django.contrib import admin
from django.contrib.auth import get_user_model
from django.contrib.auth.models import update_last_login
from django.core.exceptions import ObjectDoesNotExist
from django.urls import path, include
from rest_framework import status
from rest_framework.response import Response
from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView
from sse.views import create_sse_event, push_test_event, get_sse_status, shutdown_sse
from rest_framework_simplejwt.settings import api_settings
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
from rest_framework_simplejwt.views import (
    TokenObtainPairView,
    TokenRefreshView,
)

from basic_info.models import EmployeeStatusEnum
from flower.app_version import AppVersionView
from flower.error_code import AuthErrorCode
from flower.error_code_views import ErrorCodeListView

# 自定义后台站点标题（simpleui 也会读取）
admin.site.site_header = "宇问科技"
admin.site.site_title = "宇问科技"
admin.site.index_title = "管理后台"

def login_error_response(error_code: AuthErrorCode, code: str, message: str, http_status: int):
    return Response(
        {
            'error_code': int(error_code),
            'code': code,
            'message': message,
            'detail': message,
        },
        status=http_status,
    )


class CustomTokenObtainPairView(TokenObtainPairView):
    """自定义登录视图"""

    def post(self, request, *args, **kwargs):
        data = request.data or {}
        username = (data.get('username') or '').strip()
        password = data.get('password')

        if not username and not password:
            return login_error_response(
                AuthErrorCode.MISSING_CREDENTIALS,
                'missing_credentials',
                '请输入用户名和密码',
                status.HTTP_400_BAD_REQUEST,
            )
        if not username:
            return login_error_response(
                AuthErrorCode.MISSING_USERNAME,
                'missing_username',
                '请输入用户名',
                status.HTTP_400_BAD_REQUEST,
            )
        if not password:
            return login_error_response(
                AuthErrorCode.MISSING_PASSWORD,
                'missing_password',
                '请输入密码',
                status.HTTP_400_BAD_REQUEST,
            )

        UserModel = get_user_model()
        try:
            user = UserModel._default_manager.get_by_natural_key(username)
        except UserModel.DoesNotExist:
            return login_error_response(
                AuthErrorCode.USER_NOT_FOUND,
                'user_not_found',
                '用户不存在',
                status.HTTP_401_UNAUTHORIZED,
            )

        if not user.check_password(password):
            return login_error_response(
                AuthErrorCode.INVALID_PASSWORD,
                'invalid_password',
                '密码错误',
                status.HTTP_401_UNAUTHORIZED,
            )

        if not user.is_active:
            return login_error_response(
                AuthErrorCode.USER_INACTIVE,
                'user_inactive',
                '该用户已被禁用',
                status.HTTP_403_FORBIDDEN,
            )

        try:
            employee = user.employee
        except ObjectDoesNotExist:
            return login_error_response(
                AuthErrorCode.EMPLOYEE_NOT_BOUND,
                'employee_not_bound',
                '该用户未绑定员工身份',
                status.HTTP_403_FORBIDDEN,
            )

        if employee.status != EmployeeStatusEnum.ACTIVE:
            return login_error_response(
                AuthErrorCode.EMPLOYEE_INACTIVE,
                'employee_inactive',
                '该员工已离职或停用',
                status.HTTP_403_FORBIDDEN,
            )

        refresh = TokenObtainPairSerializer.get_token(user)
        if api_settings.UPDATE_LAST_LOGIN:
            update_last_login(None, user)

        return Response(
            {
                'refresh': str(refresh),
                'access': str(refresh.access_token),
            },
            status=status.HTTP_200_OK,
        )


urlpatterns = [
    # JWT 登录
    path('api/auth/login/', CustomTokenObtainPairView.as_view(), name='token_obtain_pair'),
    path('api/auth/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
    path('api/error-codes/', ErrorCodeListView.as_view(), name='error_code_list'),
    path('api/app-version/', AppVersionView.as_view(), name='app_version'),

    # API 文档
    path('api/schema/', SpectacularAPIView.as_view(), name='schema'),
    path('api/docs/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'),

    path('admin/', admin.site.urls),
    path('api/v1/', include('api_v1.urls')),
    path('api/v2/', include('api_v2.urls')),
    path('api/core/', include('api_core.urls')),
    path('api/backend/', include('api_man.urls')),

    # sse 相关端点
    path('sse/', create_sse_event, name='sse_event'),
    path('sse/push/', push_test_event, name='push_sse_event'),
    path('sse/status/', get_sse_status, name='sse_status'),
    path('sse/shutdown/', shutdown_sse, name='shutdown_sse'),
]