init project

Author: shaohuzhang1

apps/common/auth/__init__.py

@@ -0,0 +1,9 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： __init__.py.py
+    @date：2025/4/14 10:44
+    @desc:
+"""
+from .authenticate import *

apps/common/auth/authenticate.py

@@ -0,0 +1,96 @@
+# coding=utf-8
+"""
+    @project: qabot
+    @Author：虎虎
+    @file： authenticate.py
+    @date：2023/9/4 11:16
+    @desc:  认证类
+"""
+import traceback
+from importlib import import_module
+
+from django.conf import settings
+from django.core import cache
+from django.core import signing
+from rest_framework.authentication import TokenAuthentication
+
+from common.exception.app_exception import AppAuthenticationFailed, AppEmbedIdentityFailed, AppChatNumOutOfBoundsFailed, \
+    ChatException, AppApiException
+from django.utils.translation import gettext_lazy as _
+
+token_cache = cache.caches['default']
+
+
+class AnonymousAuthentication(TokenAuthentication):
+    def authenticate(self, request):
+        return None, None
+
+
+def new_instance_by_class_path(class_path: str):
+    parts = class_path.rpartition('.')
+    package_path = parts[0]
+    class_name = parts[2]
+    module = import_module(package_path)
+    HandlerClass = getattr(module, class_name)
+    return HandlerClass()
+
+
+handles = [new_instance_by_class_path(class_path) for class_path in settings.AUTH_HANDLES]
+
+
+class TokenDetails:
+    token_details = None
+    is_load = False
+
+    def __init__(self, token: str):
+        self.token = token
+
+    def get_token_details(self):
+        if self.token_details is None and not self.is_load:
+            try:
+                self.token_details = signing.loads(self.token)
+            except Exception as e:
+                self.is_load = True
+        return self.token_details
+
+
+class OpenAIKeyAuth(TokenAuthentication):
+    def authenticate(self, request):
+        auth = request.META.get('HTTP_AUTHORIZATION')
+        auth = auth.replace('Bearer ', '')
+        # 未认证
+        if auth is None:
+            raise AppAuthenticationFailed(1003, _('Not logged in, please log in first'))
+        try:
+            token_details = TokenDetails(auth)
+            for handle in handles:
+                if handle.support(request, auth, token_details.get_token_details):
+                    return handle.handle(request, auth, token_details.get_token_details)
+            raise AppAuthenticationFailed(1002, _('Authentication information is incorrect! illegal user'))
+        except Exception as e:
+            traceback.format_exc()
+            if isinstance(e, AppEmbedIdentityFailed) or isinstance(e, AppChatNumOutOfBoundsFailed) or isinstance(e,
+                                                                                                                 AppApiException):
+                raise e
+            raise AppAuthenticationFailed(1002, _('Authentication information is incorrect! illegal user'))
+
+
+class TokenAuth(TokenAuthentication):
+    # 重新 authenticate 方法，自定义认证规则
+    def authenticate(self, request):
+        auth = request.META.get('HTTP_AUTHORIZATION')
+        # 未认证
+        if auth is None:
+            raise AppAuthenticationFailed(1003, _('Not logged in, please log in first'))
+        try:
+            token_details = TokenDetails(auth)
+            for handle in handles:
+                if handle.support(request, auth, token_details.get_token_details):
+                    return handle.handle(request, auth, token_details.get_token_details)
+            raise AppAuthenticationFailed(1002, _('Authentication information is incorrect! illegal user'))
+        except Exception as e:
+            traceback.format_exc()
+            if isinstance(e, AppEmbedIdentityFailed) or isinstance(e, AppChatNumOutOfBoundsFailed) or isinstance(e,
+                                                                                                                 AppApiException):
+                raise e
+            raise AppAuthenticationFailed(1002, _('Authentication information is incorrect! illegal user'))

apps/common/auth/handle/auth_base_handle.py

@@ -0,0 +1,19 @@
+# coding=utf-8
+"""
+    @project: qabot
+    @Author：虎虎
+    @file： authenticate.py
+    @date：2024/3/14 03:02
+    @desc:  认证处理器
+"""
+from abc import ABC, abstractmethod
+
+
+class AuthBaseHandle(ABC):
+    @abstractmethod
+    def support(self, request, token: str, get_token_details):
+        pass
+
+    @abstractmethod
+    def handle(self, request, token: str, get_token_details):
+        pass

apps/common/auth/handle/impl/user_token.py

@@ -0,0 +1,36 @@
+# coding=utf-8
+"""
+    @project: maxkb
+    @Author：虎虎
+    @file： authenticate.py
+    @date：2024/3/14 03:02
+    @desc:  用户认证
+"""
+from django.db.models import QuerySet
+from common.auth.handle.auth_base_handle import AuthBaseHandle
+from common.constants.authentication_type import AuthenticationType
+from common.constants.cache_version import Cache_Version
+from common.constants.permission_constants import Auth, RoleConstants
+from common.exception.app_exception import AppAuthenticationFailed
+from users.models import User
+from django.core.cache import cache
+from django.utils.translation import gettext_lazy as _
+
+
+class UserToken(AuthBaseHandle):
+    def support(self, request, token: str, get_token_details):
+        auth_details = get_token_details()
+        if auth_details is None:
+            return False
+        return True
+
+    def handle(self, request, token: str, get_token_details):
+        cache_token = cache.get(token, version=Cache_Version.TOKEN)
+        if cache_token is None:
+            raise AppAuthenticationFailed(1002, _('Login expired'))
+        auth_details = get_token_details()
+        user = QuerySet(User).get(id=auth_details['id'])
+        role = RoleConstants[user.role]
+        return user, Auth([], [],
+                          client_id=str(user.id),
+                          client_type=AuthenticationType.SYSTEM_USER.value, current_role=role)

apps/common/constants/authentication_type.py

@@ -0,0 +1,16 @@
+# coding=utf-8
+"""
+    @project: maxkb
+    @Author：虎虎
+    @file： authentication_type.py
+    @date：2023/11/14 20:03
+    @desc:
+"""
+from enum import Enum
+
+
+class AuthenticationType(Enum):
+    # 系统用户
+    SYSTEM_USER = "SYSTEM_USER"
+    # 对话用户
+    CHAT_USER = "CHAT_USER"

apps/common/constants/cache_version.py

@@ -0,0 +1,14 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： cache_version.py
+    @date：2025/4/14 19:09
+    @desc:
+"""
+from enum import Enum
+
+
+class Cache_Version(Enum):
+    # 系统用户
+    TOKEN = "TOKEN"

apps/common/constants/permission_constants.py

@@ -0,0 +1,117 @@
+"""
+    @project: qabot
+    @Author：虎虎
+    @file： permission_constants.py
+    @date：2023/9/13 18:23
+    @desc: 权限,角色 常量
+"""
+from enum import Enum
+from typing import List
+
+from django.utils.translation import gettext_lazy as _
+
+
+class Group(Enum):
+    """
+    权限组 一个组一般对应前端一个菜单
+    """
+    USER = "USER"
+
+
+class Operate(Enum):
+    """
+     一个权限组的操作权限
+    """
+    READ = 'READ'
+    EDIT = "EDIT"
+    CREATE = "CREATE"
+    DELETE = "DELETE"
+
+
+class RoleGroup(Enum):
+    # 系统用户
+    SYSTEM_USER = "SYSTEM_USER"
+    # 对话用户
+    CHAT_USER = "CHAT_USER"
+
+
+class Role:
+    def __init__(self, name: str, decs: str, group: RoleGroup):
+        self.name = name
+        self.decs = decs
+        self.group = group
+
+
+class RoleConstants(Enum):
+    ADMIN = Role(_("ADMIN"), _('Super administrator'), RoleGroup.SYSTEM_USER)
+
+
+class Permission:
+    """
+    权限信息
+    """
+
+    def __init__(self, group: Group, operate: Operate, roles=None, dynamic_tag=None):
+        if roles is None:
+            roles = []
+        self.group = group
+        self.operate = operate
+        self.roleList = roles
+        self.dynamic_tag = dynamic_tag
+
+    def __str__(self):
+        return self.group.value + ":" + self.operate.value + (
+            (":" + self.dynamic_tag) if self.dynamic_tag is not None else '')
+
+    def __eq__(self, other):
+        return str(self) == str(other)
+
+
+class PermissionConstants(Enum):
+    """
+     权限枚举
+    """
+    USER_READ = Permission(group=Group.USER, operate=Operate.READ, roles=[RoleConstants.ADMIN])
+    USER_EDIT = Permission(group=Group.USER, operate=Operate.EDIT, roles=[RoleConstants.ADMIN])
+    USER_DELETE = Permission(group=Group.USER, operate=Operate.DELETE, roles=[RoleConstants.ADMIN])
+
+
+def get_permission_list_by_role(role: RoleConstants):
+    """
+    根据角色 获取角色对应的权限
+    :param role: 角色
+    :return: 权限
+    """
+    return list(map(lambda k: PermissionConstants[k],
+                    list(filter(lambda k: PermissionConstants[k].value.roleList.__contains__(role),
+                                PermissionConstants.__members__))))
+
+
+class Auth:
+    """
+     用于存储当前用户的角色和权限
+    """
+
+    def __init__(self, role_list: List[RoleConstants], permission_list: List[PermissionConstants | Permission]
+                 , client_id, client_type, current_role: RoleConstants, **keywords):
+        self.role_list = role_list
+        self.permission_list = permission_list
+        self.client_id = client_id
+        self.client_type = client_type
+        self.keywords = keywords
+        self.current_role = current_role
+
+
+class CompareConstants(Enum):
+    # 或者
+    OR = "OR"
+    # 并且
+    AND = "AND"
+
+
+class ViewPermission:
+    def __init__(self, roleList: List[RoleConstants], permissionList: List[PermissionConstants | object],
+                 compare=CompareConstants.OR):
+        self.roleList = roleList
+        self.permissionList = permissionList
+        self.compare = compare

apps/common/exception/app_exception.py

@@ -0,0 +1,83 @@
+# coding=utf-8
+"""
+    @project: qabot
+    @Author：虎虎
+    @file： app_exception.py
+    @date：2023/9/4 14:04
+    @desc:
+"""
+from rest_framework import status
+
+
+class AppApiException(Exception):
+    """
+    项目内异常
+    """
+    status_code = status.HTTP_200_OK
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+
+class NotFound404(AppApiException):
+    """
+       未认证(未登录)异常
+       """
+    status_code = status.HTTP_404_NOT_FOUND
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+
+class AppAuthenticationFailed(AppApiException):
+    """
+    未认证(未登录)异常
+    """
+    status_code = status.HTTP_401_UNAUTHORIZED
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+
+class AppUnauthorizedFailed(AppApiException):
+    """
+    未授权(没有权限)异常
+    """
+    status_code = status.HTTP_403_FORBIDDEN
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+
+class AppEmbedIdentityFailed(AppApiException):
+    """
+    嵌入cookie异常
+    """
+    status_code = 460
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+
+class AppChatNumOutOfBoundsFailed(AppApiException):
+    """
+      访问次数超过今日访问量
+    """
+    status_code = 461
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+
+class ChatException(AppApiException):
+    status_code = 500
+
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message