mirror of
https://github.com/1Panel-dev/MaxKB.git
synced 2025-12-26 18:32:48 +00:00
258 lines
10 KiB
Python
258 lines
10 KiB
Python
# 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)
|