fe0dcb78c3
All checks were successful
Build and Deploy Backend / build-and-deploy (push) Successful in 7m15s
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""
|
||
电子吧唧 - 视图（App端）
|
||
"""
|
||
from django.http import StreamingHttpResponse
|
||
from rest_framework import viewsets
|
||
from rest_framework.decorators import action
|
||
from rest_framework.permissions import IsAuthenticated
|
||
from drf_spectacular.utils import extend_schema
|
||
|
||
from utils.response import success, error
|
||
from apps.admins.authentication import AppJWTAuthentication
|
||
from .models import BadgeImage
|
||
from .serializers import BadgeImageSerializer, Text2ImageSerializer, Image2ImageSerializer
|
||
|
||
|
||
@extend_schema(tags=['电子吧唧'])
|
||
class BadgeViewSet(viewsets.ViewSet):
|
||
"""电子吧唧视图集（App端）"""
|
||
|
||
authentication_classes = [AppJWTAuthentication]
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
@action(detail=False, methods=['get'], url_path='history')
|
||
def history(self, request):
|
||
"""
|
||
获取生成历史
|
||
GET /api/v1/badge/history/
|
||
"""
|
||
images = BadgeImage.objects.filter(user=request.user).order_by('-created_at')[:50]
|
||
serializer = BadgeImageSerializer(images, many=True)
|
||
return success(data={'images': serializer.data})
|
||
|
||
@action(detail=False, methods=['post'], url_path='generate/t2i')
|
||
def generate_t2i(self, request):
|
||
"""
|
||
文生图 (SSE 流式)
|
||
POST /api/v1/badge/generate/t2i/
|
||
"""
|
||
serializer = Text2ImageSerializer(data=request.data)
|
||
if not serializer.is_valid():
|
||
return error(message=str(serializer.errors))
|
||
|
||
data = serializer.validated_data
|
||
user = request.user
|
||
|
||
badge_image = BadgeImage.objects.create(
|
||
user=user,
|
||
prompt=data['prompt'],
|
||
style=data.get('style', ''),
|
||
source='t2i',
|
||
generation_status='generating',
|
||
)
|
||
|
||
from .services.badge_image_service import generate_t2i_stream
|
||
|
||
response = StreamingHttpResponse(
|
||
generate_t2i_stream(
|
||
user=user,
|
||
badge_image=badge_image,
|
||
prompt=data['prompt'],
|
||
style=data.get('style'),
|
||
),
|
||
content_type='text/event-stream',
|
||
)
|
||
response['Cache-Control'] = 'no-cache'
|
||
response['X-Accel-Buffering'] = 'no'
|
||
return response
|
||
|
||
@action(detail=False, methods=['post'], url_path='generate/i2i')
|
||
def generate_i2i(self, request):
|
||
"""
|
||
图生图 (SSE 流式)
|
||
POST /api/v1/badge/generate/i2i/
|
||
"""
|
||
serializer = Image2ImageSerializer(data=request.data, context={'request': request})
|
||
if not serializer.is_valid():
|
||
return error(message=str(serializer.errors))
|
||
|
||
data = serializer.validated_data
|
||
user = request.user
|
||
image_file = data['image']
|
||
image_bytes = image_file.read()
|
||
|
||
badge_image = BadgeImage.objects.create(
|
||
user=user,
|
||
prompt=data.get('prompt', ''),
|
||
style=data.get('style', ''),
|
||
source='i2i',
|
||
strength=data.get('strength', 0.7),
|
||
generation_status='generating',
|
||
)
|
||
|
||
from .services.badge_image_service import generate_i2i_stream
|
||
|
||
response = StreamingHttpResponse(
|
||
generate_i2i_stream(
|
||
user=user,
|
||
badge_image=badge_image,
|
||
image_bytes=image_bytes,
|
||
prompt=data.get('prompt', ''),
|
||
style=data.get('style'),
|
||
strength=data.get('strength', 0.7),
|
||
),
|
||
content_type='text/event-stream',
|
||
)
|
||
response['Cache-Control'] = 'no-cache'
|
||
response['X-Accel-Buffering'] = 'no'
|
||
return response
|