MaxKB/apps/chat/views/chat_record.py at 8305bc68ca7f5352999856ce3dbb14f148ef845e · 1Panel-dev/MaxKB · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
# coding=utf-8
"""
    @project: MaxKB
    @Author：虎虎
    @file： chat_record.py
    @date：2025/6/23 10:42
    @desc:
"""
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
from rest_framework.request import Request
from rest_framework.views import APIView

from application.serializers.application_chat_record import ChatRecordOperateSerializer
from chat.api.chat_api import HistoricalConversationAPI, PageHistoricalConversationAPI, \
    PageHistoricalConversationRecordAPI, HistoricalConversationRecordAPI, HistoricalConversationOperateAPI
from chat.api.vote_api import VoteAPI
from chat.serializers.chat_record import VoteSerializer, HistoricalConversationSerializer, \
    HistoricalConversationRecordSerializer, HistoricalConversationOperateSerializer
from common import result
from common.auth import ChatTokenAuth


class VoteView(APIView):
    authentication_classes = [ChatTokenAuth]

    @extend_schema(
        methods=['PUT'],
        description=_("Like, Dislike"),
        summary=_("Like, Dislike"),
        operation_id=_("Like, Dislike"),  # type: ignore
        parameters=VoteAPI.get_parameters(),
        request=VoteAPI.get_request(),
        responses=VoteAPI.get_response(),
        tags=[_('Chat')]  # type: ignore
    )
    def put(self, request: Request, chat_id: str, chat_record_id: str):
        return result.success(VoteSerializer(
            data={'chat_id': chat_id,
                  'chat_record_id': chat_record_id
                  }).vote(request.data))


class HistoricalConversationView(APIView):
    authentication_classes = [ChatTokenAuth]

    @extend_schema(
        methods=['GET'],
        description=_("Get historical conversation"),
        summary=_("Get historical conversation"),
        operation_id=_("Get historical conversation"),  # type: ignore
        parameters=HistoricalConversationAPI.get_parameters(),
        responses=HistoricalConversationAPI.get_response(),
        tags=[_('Chat')]  # type: ignore
    )
    def get(self, request: Request):
        return result.success(HistoricalConversationSerializer(
            data={
                'application_id': request.auth.application_id,
                'chat_user_id': request.auth.chat_user_id,
            }).list())

    class Operate(APIView):
        authentication_classes = [ChatTokenAuth]

        @extend_schema(
            methods=['PUT'],
            description=_("Modify conversation about"),
            summary=_("Modify conversation about"),
            operation_id=_("Modify conversation about"),  # type: ignore
            parameters=HistoricalConversationOperateAPI.get_parameters(),
            request=HistoricalConversationOperateAPI.get_request(),
            responses=HistoricalConversationOperateAPI.get_response(),
            tags=[_('Chat')]  # type: ignore
        )
        def put(self, request: Request, chat_id: str):
            return result.success(HistoricalConversationOperateSerializer(
                data={
                    'application_id': request.auth.application_id,
                    'chat_user_id': request.auth.chat_user_id,
                    'chat_id': chat_id,
                }).edit_abstract(request.data)
                                  )

        @extend_schema(
            methods=['DELETE'],
            description=_("Delete history conversation"),
            summary=_("Delete history conversation"),
            operation_id=_("Delete history conversation"),  # type: ignore
            parameters=HistoricalConversationOperateAPI.get_parameters(),
            responses=HistoricalConversationOperateAPI.get_response(),
            tags=[_('Chat')]  # type: ignore
        )
        def delete(self, request: Request, chat_id: str):
            return result.success(HistoricalConversationOperateSerializer(
                data={
                    'application_id': request.auth.application_id,
                    'chat_user_id': request.auth.chat_user_id,
                    'chat_id': chat_id,
                }).logic_delete())

    class BatchDelete(APIView):
        authentication_classes = [ChatTokenAuth]

        @extend_schema(
            methods=['DELETE'],
            description=_("Batch delete history conversation"),
            summary=_("Batch delete history conversation"),
            operation_id=_("Batch delete history conversation"),  # type: ignore
            parameters=HistoricalConversationOperateAPI.get_parameters(),
            responses=HistoricalConversationOperateAPI.get_response(),
            tags=[_('Chat')]  # type: ignore
        )
        def delete(self, request: Request):
            return result.success(HistoricalConversationOperateSerializer.Clear(data={
                'application_id': request.auth.application_id,
                'chat_user_id': request.auth.chat_user_id,
            }).batch_logic_delete())

    class PageView(APIView):
        authentication_classes = [ChatTokenAuth]

        @extend_schema(
            methods=['GET'],
            description=_("Get historical conversation by page"),
            summary=_("Get historical conversation by page"),
            operation_id=_("Get historical conversation by page"),  # type: ignore
            parameters=PageHistoricalConversationAPI.get_parameters(),
            responses=PageHistoricalConversationAPI.get_response(),
            tags=[_('Chat')]  # type: ignore
        )
        def get(self, request: Request, current_page: int, page_size: int):
            return result.success(HistoricalConversationSerializer(
                data={
                    'application_id': request.auth.application_id,
                    'chat_user_id': request.auth.chat_user_id,
                }).page(current_page, page_size))


class HistoricalConversationRecordView(APIView):
    authentication_classes = [ChatTokenAuth]

    @extend_schema(
        methods=['GET'],
        description=_("Get historical conversation records"),
        summary=_("Get historical conversation records"),
        operation_id=_("Get historical conversation records"),  # type: ignore
        parameters=HistoricalConversationRecordAPI.get_parameters(),
        responses=HistoricalConversationRecordAPI.get_response(),
        tags=[_('Chat')]  # type: ignore
    )
    def get(self, request: Request, chat_id: str):
        return result.success(HistoricalConversationRecordSerializer(
            data={
                'chat_id': chat_id,
                'application_id': request.auth.application_id,
                'chat_user_id': request.auth.chat_user_id,
            }).list())

    class PageView(APIView):
        authentication_classes = [ChatTokenAuth]

        @extend_schema(
            methods=['GET'],
            description=_("Get historical conversation records by page "),
            summary=_("Get historical conversation records by page"),
            operation_id=_("Get historical conversation records by page"),  # type: ignore
            parameters=PageHistoricalConversationRecordAPI.get_parameters(),
            responses=PageHistoricalConversationRecordAPI.get_response(),
            tags=[_('Chat')]  # type: ignore
        )
        def get(self, request: Request, chat_id: str, current_page: int, page_size: int):
            return result.success(HistoricalConversationRecordSerializer(
                data={
                    'chat_id': chat_id,
                    'application_id': request.auth.application_id,
                    'chat_user_id': request.auth.chat_user_id,
                }).page(current_page, page_size))


class ChatRecordView(APIView):
    authentication_classes = [ChatTokenAuth]

    @extend_schema(
        methods=['GET'],
        description=_("Get conversation details"),
        summary=_("Get conversation details"),
        operation_id=_("Get conversation details"),  # type: ignore
        parameters=PageHistoricalConversationRecordAPI.get_parameters(),
        responses=PageHistoricalConversationRecordAPI.get_response(),
        tags=[_('Chat')]  # type: ignore
    )
    def get(self, request: Request, chat_id: str, chat_record_id: str):
        return result.success(ChatRecordOperateSerializer(
            data={
                'chat_id': chat_id,
                'chat_record_id': chat_record_id,
                'application_id': request.auth.application_id,
                'chat_user_id': request.auth.chat_user_id,
            }).one(False))