""" Una imagen del proveedor, a pedido, desde la terminal. Es una herramienta de trabajo, no una pieza del producto: sirve para conseguir una imagen cuando hace falta una —la foto de muestra de una opción nueva, una prueba de encuadre, lo que sea— usando la clave y el proveedor que el proyecto ya tiene configurados. **No sabe para qué se la pide.** No conoce el catálogo, ni las tarjetas, ni los esquemas, y por eso no se entera cuando alguno cambia. Hace una cosa: recibe un prompt, opcionalmente unas imágenes sobre las que editar y una máscara que acota dónde, llama al proveedor, guarda lo que vuelve en `var/ai-images/` —que está fuera de git— e imprime la ruta. Dónde termina ese archivo es decisión de quien lo pidió, en otro paso y a mano. Cuesta dinero real: una llamada por corrida. No reintenta y no toca nada del proyecto. **Lo que imprime va en inglés y no pasa por el catálogo.** Es la salida de una herramienta de operación, no interfaz. """ import base64 import re from contextlib import ExitStack from datetime import UTC, datetime from pathlib import Path from django.conf import settings from django.core.management.base import BaseCommand, CommandError from openai import APIConnectionError, APIStatusError, APITimeoutError, RateLimitError from PIL import Image, UnidentifiedImageError from apps.ai import provider from apps.ai.models import AiSettings #: Dónde caen las imágenes. Bajo `var/`, que el `.gitignore` ya excluye: lo que #: sale de acá no está versionado hasta que alguien lo mueva a propósito. OUTPUT_DIR = 'ai-images' #: Cuántas imágenes admite `images.edit` en una llamada, según el SDK. MAX_INPUT_IMAGES = 16 #: Para el nombre del archivo: lo que quede del prompt después de sacarle todo #: lo que no sea una letra o un número. SLUG_MAX_LENGTH = 40 def output_dir() -> Path: return Path(settings.BASE_DIR) / 'var' / OUTPUT_DIR def slug_of(prompt: str) -> str: """Las primeras palabras del prompt, para reconocer el archivo de un vistazo.""" cleaned = re.sub(r'[^a-z0-9]+', '-', prompt.lower()).strip('-') return cleaned[:SLUG_MAX_LENGTH].rstrip('-') or 'image' class Command(BaseCommand): help = 'Ask the image provider for one image and save it under var/ai-images (costs money).' def add_arguments(self, parser): parser.add_argument('prompt', help='What to generate, in English.') parser.add_argument( '--image', action='append', default=[], metavar='PATH', help=( 'Edit over this image instead of generating from text. ' f'Repeatable, up to {MAX_INPUT_IMAGES}; the first one is the base.' ), ) parser.add_argument( '--mask', metavar='PATH', help=( 'A PNG with alpha, the size of the first image: its fully transparent areas ' 'are where the first image may be edited. Needs --image.' ), ) parser.add_argument('--model', help='Override the model. Defaults to the configured one.') parser.add_argument( '--quality', help='Override the quality. Defaults to the configured one.' ) parser.add_argument('--size', help='Ask for this size, e.g. 1536x864.') parser.add_argument('--out', metavar='PATH', help='Write here instead of var/ai-images.') def handle(self, *args, **options): if not provider.is_configured(): raise CommandError('No OPENAI_API_KEY in the environment.') sources = [Path(path) for path in options['image']] if len(sources) > MAX_INPUT_IMAGES: raise CommandError(f'At most {MAX_INPUT_IMAGES} images per call, got {len(sources)}.') for source in sources: if not source.is_file(): raise CommandError(f'{source} is not a file.') mask = self._checked_mask(options['mask'], sources) # La configuración del proyecto como punto de partida: pedir a mano una # imagen con un modelo que la pantalla no ofrece esconde la diferencia # cuando después se compara con lo que sale del producto. ai_settings = AiSettings.current() editing = bool(sources) model = options['model'] or ( ai_settings.image_model if editing else provider.GENERATE_MODEL ) quality = options['quality'] or ai_settings.image_quality response = self._call(options['prompt'], sources, mask, model, quality, options['size']) encoded = self._first_image(response) if not encoded: raise CommandError('The provider answered without an image.') target = self._target(options['out'], options['prompt']) target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(base64.b64decode(encoded)) usage = getattr(response, 'usage', None) self.stdout.write( f'{"Edited" if editing else "Generated"} with {model}, quality {quality}' + (f', usage {usage}' if usage else '') + '.' ) # La ruta sola y en la última línea: es lo que se lee para mover el # archivo, y se lee mejor sin nada alrededor. self.stdout.write(str(target.resolve())) def _checked_mask(self, raw: str | None, sources: list[Path]) -> Path | None: """ La máscara, comprobada antes de gastar. El proveedor la aplica sobre la primera imagen y exige que tenga su mismo tamaño y un canal alfa: sin eso contesta un 400, y es mejor decirlo acá con el motivo que cobrar el viaje para enterarse. """ if raw is None: return None mask = Path(raw) if not sources: raise CommandError('--mask needs at least one --image to apply to.') if not mask.is_file(): raise CommandError(f'{mask} is not a file.') try: with Image.open(mask) as opened_mask, Image.open(sources[0]) as base: if opened_mask.format != 'PNG' or 'A' not in opened_mask.getbands(): raise CommandError(f'{mask} must be a PNG with an alpha channel.') if opened_mask.size != base.size: raise CommandError( f'{mask} is {opened_mask.size[0]}x{opened_mask.size[1]} but the first ' f'image is {base.size[0]}x{base.size[1]}; they must match.' ) except (UnidentifiedImageError, OSError) as error: raise CommandError(f'Could not open the mask or the first image: {error}') from error return mask def _call( self, prompt: str, sources: list[Path], mask: Path | None, model: str, quality: str, size: str | None, ): api = provider.client() extra = {'size': size} if size else {} try: if not sources: return api.images.generate( model=model, prompt=prompt, quality=quality, n=1, **extra ) with ExitStack() as stack: handles = [stack.enter_context(source.open('rb')) for source in sources] if mask is not None: extra['mask'] = stack.enter_context(mask.open('rb')) return api.images.edit( model=model, image=handles if len(handles) > 1 else handles[0], prompt=prompt, quality=quality, n=1, **extra, ) except (RateLimitError, APITimeoutError, APIConnectionError, APIStatusError) as error: # Sin reintento: es una corrida a mano, y quien la pidió está # mirando la terminal y puede volver a tipearla. raise CommandError(f'The provider failed: {error}') from error def _first_image(self, response) -> str | None: data = getattr(response, 'data', None) or [] return getattr(data[0], 'b64_json', None) if data else None def _target(self, out: str | None, prompt: str) -> Path: if out: return Path(out) stamp = datetime.now(UTC).strftime('%Y%m%d-%H%M%S') return output_dir() / f'{stamp}-{slug_of(prompt)}.png'