UnisKB/apps/chat/views/chat.py

# coding=utf-8
"""
    @project: MaxKB
    @Author：虎虎
    @file： chat.py
    @date：2025/6/6 11:18
    @desc:
"""
import requests
from django.http import HttpResponse, StreamingHttpResponse
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
from rest_framework.parsers import MultiPartParser
from rest_framework.request import Request
from rest_framework.views import APIView

from application.api.application_api import SpeechToTextAPI, TextToSpeechAPI
from chat.api.chat_api import ChatAPI
from chat.api.chat_authentication_api import ChatAuthenticationAPI, ChatAuthenticationProfileAPI, ChatOpenAPI, OpenAIAPI
from chat.serializers.chat import OpenChatSerializers, ChatSerializers, SpeechToTextSerializers, \
    TextToSpeechSerializers, OpenAIChatSerializer
from chat.serializers.chat_authentication import AnonymousAuthenticationSerializer, ApplicationProfileSerializer, \
    AuthProfileSerializer
from common.auth import TokenAuth
from common.constants.permission_constants import ChatAuth
from common.exception.app_exception import AppAuthenticationFailed
from common.result import result
from knowledge.models import FileSourceType
from oss.serializers.file import FileSerializer
from users.api import CaptchaAPI
from users.serializers.login import CaptchaSerializer


def stream_image(response):
    """生成器函数，用于流式传输图片数据"""
    for chunk in response.iter_content(chunk_size=4096):
        if chunk:  # 过滤掉保持连接的空块
            yield chunk


class ResourceProxy(APIView):
    def get(self, request: Request):
        image_url = request.query_params.get("url")
        if not image_url:
            return result.error("Missing 'url' parameter")
        try:

            # 发送GET请求，流式获取图片内容
            response = requests.get(
                image_url,
                stream=True,  # 启用流式响应
                allow_redirects=True,
                timeout=10
            )
            content_type = response.headers.get('Content-Type', '').split(';')[0]
            # 创建Django流式响应
            django_response = StreamingHttpResponse(
                stream_image(response),  # 使用生成器
                content_type=content_type
            )

            return django_response
        except Exception as e:
            return result.error(f"Image request failed: {str(e)}")


class OpenAIView(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['POST'],
        description=_('OpenAI Interface Dialogue'),
        summary=_('OpenAI Interface Dialogue'),
        operation_id=_('OpenAI Interface Dialogue'),  # type: ignore
        request=OpenAIAPI.get_request(),
        responses=None,
        tags=[_('Chat')]  # type: ignore
    )
    def post(self, request: Request, application_id: str):
        return OpenAIChatSerializer(data={'application_id': application_id, 'chat_user_id': request.auth.chat_user_id,
                                          'chat_user_type': request.auth.chat_user_type}).chat(request.data)


class AnonymousAuthentication(APIView):
    def options(self, request, *args, **kwargs):
        return HttpResponse(
            headers={"Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true",
                     "Access-Control-Allow-Methods": "POST",
                     "Access-Control-Allow-Headers": "Origin,Content-Type,Cookie,Accept,Token"}, )

    @extend_schema(
        methods=['POST'],
        description=_('Application Anonymous Certification'),
        summary=_('Application Anonymous Certification'),
        operation_id=_('Application Anonymous Certification'),  # type: ignore
        request=ChatAuthenticationAPI.get_request(),
        responses=None,
        tags=[_('Chat')]  # type: ignore
    )
    def post(self, request: Request):
        return result.success(
            AnonymousAuthenticationSerializer(data={'access_token': request.data.get("access_token")}).auth(
                request),
            headers={"Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": "true",
                     "Access-Control-Allow-Methods": "POST",
                     "Access-Control-Allow-Headers": "Origin,Content-Type,Cookie,Accept,Token"}
        )


class ApplicationProfile(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['GET'],
        description=_("Get application related information"),
        summary=_("Get application related information"),
        operation_id=_("Get application related information"),  # type: ignore
        request=None,
        responses=None,
        tags=[_('Chat')]  # type: ignore
    )
    def get(self, request: Request):
        if isinstance(request.auth, ChatAuth):
            return result.success(ApplicationProfileSerializer(
                data={'application_id': request.auth.application_id}).profile())
        raise AppAuthenticationFailed(401, "身份异常")


class AuthProfile(APIView):
    @extend_schema(
        methods=['GET'],
        description=_("Get application authentication information"),
        summary=_("Get application authentication information"),
        operation_id=_("Get application authentication information"),  # type: ignore
        parameters=ChatAuthenticationProfileAPI.get_parameters(),
        responses=None,
        tags=[_('Chat')]  # type: ignore
    )
    def get(self, request: Request):
        return result.success(
            AuthProfileSerializer(data={'access_token': request.query_params.get("access_token")}).profile())


class ChatView(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['POST'],
        description=_("dialogue"),
        summary=_("dialogue"),
        operation_id=_("dialogue"),  # type: ignore
        request=ChatAPI.get_request(),
        parameters=ChatAPI.get_parameters(),
        responses=None,
        tags=[_('Chat')]  # type: ignore
    )
    def post(self, request: Request, chat_id: str):
        return ChatSerializers(data={'chat_id': chat_id,
                                     'chat_user_id': request.auth.chat_user_id,
                                     'chat_user_type': request.auth.chat_user_type,
                                     'application_id': request.auth.application_id,
                                     'debug': False}
                               ).chat(request.data)


class OpenView(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['GET'],
        description=_("Get the session id according to the application id"),
        summary=_("Get the session id according to the application id"),
        operation_id=_("Get the session id according to the application id"),  # type: ignore
        parameters=ChatOpenAPI.get_parameters(),
        responses=None,
        tags=[_('Chat')]  # type: ignore
    )
    def get(self, request: Request):
        return result.success(OpenChatSerializers(
            data={'application_id': request.auth.application_id,
                  'chat_user_id': request.auth.chat_user_id, 'chat_user_type': request.auth.chat_user_type,
                  'debug': False}).open())


class CaptchaView(APIView):
    @extend_schema(methods=['GET'],
                   summary=_("Get Chat captcha"),
                   description=_("Get Chat captcha"),
                   operation_id=_("Get Chat captcha"),  # type: ignore
                   tags=[_("Chat")],  # type: ignore
                   responses=CaptchaAPI.get_response())
    def get(self, request: Request):
        username = request.query_params.get('username', None)
        accessToken = request.query_params.get('accessToken', None)
        return result.success(CaptchaSerializer().chat_generate(username, 'chat', accessToken))


class SpeechToText(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['POST'],
        description=_("speech to text"),
        summary=_("speech to text"),
        operation_id=_("speech to text"),  # type: ignore
        request=SpeechToTextAPI.get_request(),
        responses=SpeechToTextAPI.get_response(),
        tags=[_('Chat')]  # type: ignore
    )
    def post(self, request: Request):
        return result.success(
            SpeechToTextSerializers(
                data={'application_id': request.auth.application_id})
            .speech_to_text({'file': request.FILES.get('file')}))


class TextToSpeech(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['POST'],
        description=_("text to speech"),
        summary=_("text to speech"),
        operation_id=_("text to speech"),  # type: ignore
        request=TextToSpeechAPI.get_request(),
        responses=TextToSpeechAPI.get_response(),
        tags=[_('Chat')]  # type: ignore
    )
    def post(self, request: Request):
        byte_data = TextToSpeechSerializers(
            data={'application_id': request.auth.application_id}).text_to_speech(request.data)
        return HttpResponse(byte_data, status=200, headers={'Content-Type': 'audio/mp3',
                                                            'Content-Disposition': 'attachment; filename="abc.mp3"'})


class UploadFile(APIView):
    authentication_classes = [TokenAuth]
    parser_classes = [MultiPartParser]

    @extend_schema(
        methods=['POST'],
        description=_("Upload files"),
        summary=_("Upload files"),
        operation_id=_("Upload files"),  # type: ignore
        request=TextToSpeechAPI.get_request(),
        responses=TextToSpeechAPI.get_response(),
        tags=[_('Application')]  # type: ignore
    )
    def post(self, request: Request, chat_id: str):
        files = request.FILES.getlist('file')
        file_ids = []
        meta = {}
        for file in files:
            file_url = FileSerializer(
                data={'file': file, 'meta': meta, 'source_id': chat_id, 'source_type': FileSourceType.CHAT, }).upload()
            file_ids.append({'name': file.name, 'url': file_url, 'file_id': file_url.split('/')[-1]})
        return result.success(file_ids)