mirror of
https://github.com/1Panel-dev/MaxKB.git
synced 2025-12-26 10:02:46 +00:00
124 lines
6.8 KiB
Python
124 lines
6.8 KiB
Python
# coding=utf-8
|
||
"""
|
||
@project: MaxKB
|
||
@Author：虎虎
|
||
@file： application_access_token.py
|
||
@date：2025/6/9 17:49
|
||
@desc:
|
||
"""
|
||
import hashlib
|
||
|
||
import uuid_utils.compat as uuid
|
||
from django.core.cache import cache
|
||
from django.db.models import QuerySet
|
||
from django.utils.translation import gettext_lazy as _
|
||
from rest_framework import serializers
|
||
|
||
from application.models import ApplicationAccessToken, Application
|
||
from common.constants.cache_version import Cache_Version
|
||
from common.database_model_manage.database_model_manage import DatabaseModelManage
|
||
from common.exception.app_exception import AppApiException
|
||
|
||
|
||
class AccessTokenEditSerializer(serializers.Serializer):
|
||
access_token_reset = serializers.BooleanField(required=False,
|
||
label=_("Reset Token"))
|
||
is_active = serializers.BooleanField(required=False, label=_("Is it enabled"))
|
||
access_num = serializers.IntegerField(required=False, max_value=10000000,
|
||
min_value=0,
|
||
label=_("Number of visits"))
|
||
white_active = serializers.BooleanField(required=False,
|
||
label=_("Whether to enable whitelist"))
|
||
white_list = serializers.ListSerializer(required=False, child=serializers.CharField(required=True,
|
||
label=_("Whitelist")),
|
||
label=_("Whitelist")),
|
||
show_source = serializers.BooleanField(required=False,
|
||
label=_("Whether to display knowledge sources"))
|
||
show_exec = serializers.BooleanField(required=False,
|
||
label=_("Display execution details"))
|
||
language = serializers.CharField(required=False, allow_blank=True, allow_null=True,
|
||
label=_("language"))
|
||
authentication = serializers.BooleanField(default=False, label="Do you need authentication")
|
||
|
||
authentication_value = serializers.JSONField(required=False, label="Certified value", default=dict)
|
||
|
||
|
||
class AccessTokenSerializer(serializers.Serializer):
|
||
application_id = serializers.UUIDField(required=True, label=_("Application ID"))
|
||
workspace_id = serializers.CharField(required=False, allow_null=True, allow_blank=True, label=_("Workspace ID"))
|
||
|
||
def is_valid(self, *, raise_exception=False):
|
||
super().is_valid(raise_exception=True)
|
||
workspace_id = self.data.get('workspace_id')
|
||
query_set = QuerySet(Application).filter(id=self.data.get('application_id'))
|
||
if workspace_id:
|
||
query_set = query_set.filter(workspace_id=workspace_id)
|
||
if not query_set.exists():
|
||
raise AppApiException(500, _('Application id does not exist'))
|
||
|
||
def edit(self, instance):
|
||
self.is_valid(raise_exception=True)
|
||
AccessTokenEditSerializer(data=instance).is_valid(raise_exception=True)
|
||
application_access_token = QuerySet(ApplicationAccessToken).get(
|
||
application_id=self.data.get('application_id'))
|
||
if 'is_active' in instance:
|
||
application_access_token.is_active = instance.get("is_active")
|
||
if 'access_token_reset' in instance and instance.get('access_token_reset'):
|
||
application_access_token.access_token = hashlib.md5(str(uuid.uuid7()).encode()).hexdigest()[8:24]
|
||
if 'access_num' in instance and instance.get('access_num') is not None:
|
||
application_access_token.access_num = instance.get("access_num")
|
||
if 'white_active' in instance and instance.get('white_active') is not None:
|
||
application_access_token.white_active = instance.get("white_active")
|
||
if 'white_list' in instance and instance.get('white_list') is not None:
|
||
application_access_token.white_list = instance.get('white_list')
|
||
if 'show_source' in instance and instance.get('show_source') is not None:
|
||
application_access_token.show_source = instance.get('show_source')
|
||
if 'show_exec' in instance and instance.get('show_exec') is not None:
|
||
application_access_token.show_exec = instance.get('show_exec')
|
||
if 'language' in instance and instance.get('language') is not None:
|
||
application_access_token.language = instance.get('language')
|
||
if 'language' not in instance or instance.get('language') is None:
|
||
application_access_token.language = None
|
||
|
||
application_access_token.save()
|
||
license_is_valid = cache.get(Cache_Version.SYSTEM.get_key(key='license_is_valid'),
|
||
version=Cache_Version.SYSTEM.get_version())
|
||
if license_is_valid:
|
||
if instance.get('authentication') is not None and instance.get(
|
||
'authentication_value') is not None:
|
||
application_access_token.authentication = instance.get('authentication')
|
||
application_access_token.authentication_value = instance.get('authentication_value')
|
||
application_access_token.save()
|
||
return self.one(with_valid=False)
|
||
|
||
def one(self, with_valid=True):
|
||
if with_valid:
|
||
self.is_valid(raise_exception=True)
|
||
application_id = self.data.get("application_id")
|
||
application_access_token = QuerySet(ApplicationAccessToken).filter(
|
||
application_id=application_id).first()
|
||
if application_access_token is None:
|
||
application_access_token = ApplicationAccessToken(application_id=application_id,
|
||
access_token=hashlib.md5(
|
||
str(uuid.uuid7()).encode()).hexdigest()[
|
||
8:24], is_active=True)
|
||
application_access_token.save()
|
||
other = {}
|
||
license_is_valid = cache.get(Cache_Version.SYSTEM.get_key(key='license_is_valid'),
|
||
version=Cache_Version.SYSTEM.get_version())
|
||
if license_is_valid:
|
||
other = {'authentication': application_access_token.authentication,
|
||
'authentication_value': application_access_token.authentication_value}
|
||
|
||
return {'application_id': application_access_token.application_id,
|
||
'access_token': application_access_token.access_token,
|
||
"is_active": application_access_token.is_active,
|
||
'access_num': application_access_token.access_num,
|
||
'white_active': application_access_token.white_active,
|
||
'white_list': application_access_token.white_list,
|
||
'show_source': application_access_token.show_source,
|
||
'show_exec': application_access_token.show_exec,
|
||
'language': application_access_token.language,
|
||
**other,
|
||
}
|