UnisKB/apps/application/serializers/application_access_token.py

# 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,
                }