""" El generador de imágenes por la API, para la landing. La landing presenta el mismo recorrido que la pantalla de Tools con otro diseño: pide el catálogo, sube la foto, espera el análisis, pide la generación y la espera. Todo pasa por el backend de la landing, que tiene la clave; el navegador del visitante nunca habla con Tools. Estas vistas no deciden nada: traducen entre HTTP y los mismos servicios que usa la pantalla (`analysis.request_analysis`, `services.request_website_generation`). Lo que cambia para la landing —el prompt con la nota que manda, el tope por visitante— vive en esos servicios, no acá. Sólo se ve lo que entró por la landing: un análisis o un pedido hecho desde la pantalla de Tools contesta 404 por la API. Los endpoints de la v1 (`image-generations`) siguen contestando `410 Gone` (decisión 2 de `ai-image-generator-v2`): quien los llame recibe un mensaje que explica qué pasó y no un 404. """ import json from http import HTTPStatus from django.conf import settings from django.core.exceptions import ValidationError as DjangoValidationError from drf_spectacular.utils import OpenApiParameter, extend_schema from rest_framework import serializers, status from rest_framework.response import Response from apps.ai import analysis as photo_analysis from apps.ai import quota, services, website from apps.ai.models import GenerationOrigin, ImageGeneration, PhotoAnalysis from apps.ai.prompt import AnswerInvalid from apps.core.api import BaseApiView, ErrorSerializer from apps.core.errors import ApiError, Deprecated, ErrorCode, NotFound, ValidationFailed DEPRECATION_MESSAGE = ( 'The image generator v1 API was replaced by the v2 generator and is no longer served. ' 'Use /api/v1/ai/analyses and /api/v1/ai/generations.' ) #: Las cabeceras con que el backend de la landing dice quién es el visitante. A #: Tools le llega la IP del servidor de la landing, no la del visitante, y sin #: esto el tope contaría a todos los visitantes como uno solo. VISITOR_IP_HEADER = 'X-Visitor-Ip' VISITOR_USER_AGENT_HEADER = 'X-Visitor-User-Agent' VISITOR_MISSING = 'VISITOR_MISSING' class VisitorLimitError(ApiError): """El 429 del tope por visitante, con la ventana agotada y `Retry-After`.""" def __init__(self, error: quota.VisitorLimitReached): super().__init__( ErrorCode.RATE_LIMITED, f'This visitor reached the limit of {error.limit} generations per {error.window}.', HTTPStatus.TOO_MANY_REQUESTS, { 'reason': error.code, 'window': error.window, 'limit': error.limit, 'retry_after': error.retry_after, }, ) self.retry_after = error.retry_after def as_response(self): response = super().as_response() response['Retry-After'] = str(self.retry_after) return response # --- Documentación ---------------------------------------------------------- class CatalogSerializer(serializers.Serializer): version = serializers.CharField() catalog = serializers.DictField() texts = serializers.DictField(child=serializers.CharField()) class AnalysisSerializer(serializers.Serializer): id = serializers.UUIDField() state = serializers.CharField() error_code = serializers.CharField(allow_null=True) areas = serializers.ListField(child=serializers.CharField()) pool_present = serializers.BooleanField() confidence = serializers.DictField() points = serializers.DictField() created_at = serializers.DateTimeField() finished_at = serializers.DateTimeField(allow_null=True) class AnalysisCreateSerializer(serializers.Serializer): image = serializers.ImageField() class GenerationSerializer(serializers.Serializer): id = serializers.UUIDField() analysis_id = serializers.UUIDField() state = serializers.CharField() error_code = serializers.CharField(allow_null=True) areas = serializers.ListField(child=serializers.DictField()) note = serializers.CharField() has_markup = serializers.BooleanField() result_url = serializers.URLField(allow_null=True) download_url = serializers.URLField(allow_null=True) created_at = serializers.DateTimeField() finished_at = serializers.DateTimeField(allow_null=True) class GenerationCreateSerializer(serializers.Serializer): analysis_id = serializers.UUIDField() areas = serializers.CharField( help_text='JSON list of containers: [{"area", "service", "answers", "whole_area"}].' ) note = serializers.CharField(required=False, help_text='Up to 600 characters.') markup = serializers.ImageField( required=False, help_text='Transparent PNG with the marks, the size of the analyzed photo. Needs a note.', ) VISITOR_PARAMETERS = [ OpenApiParameter( VISITOR_IP_HEADER, str, OpenApiParameter.HEADER, required=True, description='Visitor IP.' ), OpenApiParameter( VISITOR_USER_AGENT_HEADER, str, OpenApiParameter.HEADER, required=True, description='Visitor user agent.', ), ] # --- Catálogo --------------------------------------------------------------- class CatalogView(BaseApiView): @extend_schema(responses={200: CatalogSerializer, 304: None}, tags=['ai']) def get(self, request): """ The questions of the generator, their texts in English and their version. Send the stored version in `If-None-Match` to get a `304` when nothing changed. Image URLs carry `?v=` of their content: a new hash means a new file to download. """ version = website.catalog_version() etag = f'"{version}"' if _etag_matches(request.headers.get('If-None-Match', ''), version): response = Response(status=status.HTTP_304_NOT_MODIFIED) else: static_url = request.build_absolute_uri(settings.STATIC_URL) response = Response({'version': version, **website.catalog_payload(static_url)}) response['ETag'] = etag return response def _etag_matches(header: str, version: str) -> bool: candidates = [candidate.strip() for candidate in header.split(',') if candidate.strip()] return any( candidate == '*' or candidate.removeprefix('W/').strip('"') == version for candidate in candidates ) # --- Análisis --------------------------------------------------------------- class AnalysisListView(BaseApiView): @extend_schema( request={'multipart/form-data': AnalysisCreateSerializer}, responses={202: AnalysisSerializer, 422: ErrorSerializer}, tags=['ai'], ) def post(self, request): """ Uploads the visitor's photo and queues its analysis. `202`: the analysis runs in the background; poll its detail until `state` is `COMPLETED` or `FAILED`. The photo is compressed on arrival (up to 25 MB in). """ replay = self.replay(request) if replay is not None: return replay upload = request.FILES.get('image') if upload is None: raise ValidationFailed('The photo is missing.', field='image', reason='IMAGE_MISSING') try: analysis = photo_analysis.request_analysis( None, upload, origin=GenerationOrigin.WEBSITE ) except (photo_analysis.SourceImageTooLarge, services.SourceImageInvalid) as error: raise ValidationFailed( 'The photo is too large or is not an image.', field='image', reason=error.code ) from None response = Response(website.analysis_props(analysis), status=status.HTTP_202_ACCEPTED) return self.store(request, response) class AnalysisDetailView(BaseApiView): @extend_schema(responses={200: AnalysisSerializer, 404: ErrorSerializer}, tags=['ai']) def get(self, request, analysis_id): """An analysis requested by the website, to poll while it runs.""" return Response(website.analysis_props(_website_analysis(analysis_id))) def _website_analysis(analysis_id) -> PhotoAnalysis: """El análisis de la landing, o 404. Un identificador mal formado tampoco existe.""" try: analysis = PhotoAnalysis.objects.filter( id=analysis_id, origin=GenerationOrigin.WEBSITE ).first() except (ValueError, TypeError, DjangoValidationError): analysis = None if analysis is None: raise NotFound('That analysis does not exist.') return analysis # --- Generaciones ----------------------------------------------------------- class GenerationListView(BaseApiView): @extend_schema( request={'multipart/form-data': GenerationCreateSerializer}, parameters=VISITOR_PARAMETERS, responses={ 202: GenerationSerializer, 404: ErrorSerializer, 409: ErrorSerializer, 422: ErrorSerializer, 429: ErrorSerializer, }, tags=['ai'], ) def post(self, request): """ Queues the generation of a design over a completed analysis. The visitor's note leads: the answers are the base, and what the note asks for is generated over them, as long as it is work of the company's services (or outdoor decor around it). With marks, the note says what each mark is. Validation errors are `422` with `details.field` (`areas[0].color`, `markup`, `notes`...) and `details.reason`. Over the visitor limit it is `429` with `details.window` and `Retry-After`. """ replay = self.replay(request) if replay is not None: return replay fingerprint = _visitor_fingerprint(request) analysis = _website_analysis(request.data.get('analysis_id') or '') areas = _areas_of(request.data.get('areas')) try: generation = services.request_website_generation( analysis, areas, note=request.data.get('note') or '', markup_layer=request.FILES.get('markup'), fingerprint=fingerprint, ) except services.AnalysisNotReady as error: raise ApiError( ErrorCode.CONFLICT, 'The analysis has not completed.', HTTPStatus.CONFLICT, {'field': 'analysis_id', 'reason': error.code}, ) from None except AnswerInvalid as error: raise ValidationFailed( 'The request does not match the catalog.', field=error.key, reason=error.code ) from None except quota.VisitorLimitReached as error: raise VisitorLimitError(error) from None props = website.generation_props(generation, request.build_absolute_uri) return self.store(request, Response(props, status=status.HTTP_202_ACCEPTED)) class GenerationDetailView(BaseApiView): @extend_schema(responses={200: GenerationSerializer, 404: ErrorSerializer}, tags=['ai']) def get(self, request, generation_id): """A generation requested by the website, to poll until it has its image.""" generation = ImageGeneration.objects.filter( id=generation_id, origin=GenerationOrigin.WEBSITE ).first() if generation is None: raise NotFound('That generation does not exist.') return Response(website.generation_props(generation, request.build_absolute_uri)) def _visitor_fingerprint(request) -> str: ip = request.headers.get(VISITOR_IP_HEADER, '').strip() user_agent = request.headers.get(VISITOR_USER_AGENT_HEADER, '').strip() if not ip or not user_agent: raise ValidationFailed( f'The {VISITOR_IP_HEADER} and {VISITOR_USER_AGENT_HEADER} headers are required.', field=VISITOR_IP_HEADER if not ip else VISITOR_USER_AGENT_HEADER, reason=VISITOR_MISSING, ) return quota.visitor_fingerprint(ip, user_agent) def _areas_of(value): """ Los contenedores, como lista. En multipart llegan como texto JSON —la única forma de mandar una lista de objetos junto a un archivo—; en JSON llegan ya como lista. """ if isinstance(value, str): try: return json.loads(value) except json.JSONDecodeError: raise ValidationFailed( 'The areas are not valid JSON.', field='areas', reason='ANSWER_INVALID' ) from None return value if value is not None else [] # --- v1, deprecada ---------------------------------------------------------- class ImageGenerationListView(BaseApiView): @extend_schema(deprecated=True, responses={410: ErrorSerializer}, tags=['ai']) def post(self, request): """ Deprecated. Always answers 410 Gone. The v1 image request (kind plus flat answers) no longer exists: use `/api/v1/ai/analyses` and `/api/v1/ai/generations`. """ raise Deprecated(DEPRECATION_MESSAGE) class ImageGenerationDetailView(BaseApiView): @extend_schema(deprecated=True, responses={410: ErrorSerializer}, tags=['ai']) def get(self, request, generation_id): """Deprecated. Always answers 410 Gone, whatever the identifier.""" raise Deprecated(DEPRECATION_MESSAGE)