Spaces:
Build error
Build error
File size: 6,837 Bytes
8e5d22a 9b4e099 | 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 | """
Normalisation d'images : dérivé JPEG 1500px max + thumbnail pour le pipeline IA.
"""
# 1. stdlib
import io
import logging
from pathlib import Path
# 2. third-party
from PIL import Image
# 3. local
from app.schemas.image import ImageDerivativeInfo
from app.services.ingest.iiif_fetcher import fetch_iiif_image
logger = logging.getLogger(__name__)
# Constantes de normalisation
_MAX_DERIVATIVE_PX = 1500 # grand côté max du dérivé envoyé à l'IA
_MAX_THUMBNAIL_PX = 256 # grand côté max du thumbnail
_DERIVATIVE_QUALITY = 90 # qualité JPEG dérivé
_THUMBNAIL_QUALITY = 75 # qualité JPEG thumbnail
def _resize_to_max(image: Image.Image, max_size: int) -> Image.Image:
"""Redimensionne l'image pour que son grand côté vaille max_size.
Si l'image est déjà plus petite ou égale, retourne une copie sans upscaling.
Le ratio d'aspect est préservé. Utilise LANCZOS pour la qualité.
"""
w, h = image.size
if max(w, h) <= max_size:
return image.copy()
if w >= h:
new_w = max_size
new_h = max(1, round(h * max_size / w))
else:
new_h = max_size
new_w = max(1, round(w * max_size / h))
return image.resize((new_w, new_h), Image.Resampling.LANCZOS)
def create_derivatives(
source_bytes: bytes,
original_url: str,
corpus_slug: str,
folio_label: str,
base_data_dir: Path = Path("data"),
) -> ImageDerivativeInfo:
"""Produit un dérivé JPEG (1500px max) et un thumbnail depuis des bytes image.
Structure de sortie (CLAUDE.md §3) :
{base_data_dir}/corpora/{corpus_slug}/derivatives/{folio_label}.jpg
{base_data_dir}/corpora/{corpus_slug}/derivatives/{folio_label}_thumb.jpg
Args:
source_bytes: contenu brut de l'image source (JPEG, PNG, TIFF, etc.).
original_url: URL d'origine (conservée dans le schéma de sortie).
corpus_slug: identifiant du corpus (ex. "beatus-lat8878").
folio_label: identifiant du folio (ex. "0013r").
base_data_dir: racine du dossier data (défaut : Path("data")).
Returns:
ImageDerivativeInfo avec dimensions et chemins des fichiers produits.
Raises:
PIL.UnidentifiedImageError: si les bytes ne sont pas une image valide.
OSError: si l'écriture sur disque échoue.
"""
image = Image.open(io.BytesIO(source_bytes))
# Convertir en RGB pour garantir un JPEG valide (PNG RGBA, palette, etc.)
if image.mode != "RGB":
image = image.convert("RGB")
original_width, original_height = image.size
logger.info(
"Image ouverte",
extra={
"corpus": corpus_slug,
"folio": folio_label,
"original_size": f"{original_width}x{original_height}",
},
)
# Dossier de sortie
derivatives_dir = base_data_dir / "corpora" / corpus_slug / "derivatives"
derivatives_dir.mkdir(parents=True, exist_ok=True)
# Dérivé IA : grand côté <= 1500px
deriv_image = _resize_to_max(image, _MAX_DERIVATIVE_PX)
derivative_width, derivative_height = deriv_image.size
derivative_path = derivatives_dir / f"{folio_label}.jpg"
deriv_image.save(derivative_path, format="JPEG", quality=_DERIVATIVE_QUALITY)
# Thumbnail : grand côté <= 256px
thumb_image = _resize_to_max(image, _MAX_THUMBNAIL_PX)
thumbnail_width, thumbnail_height = thumb_image.size
thumbnail_path = derivatives_dir / f"{folio_label}_thumb.jpg"
thumb_image.save(thumbnail_path, format="JPEG", quality=_THUMBNAIL_QUALITY)
logger.info(
"Dérivés produits",
extra={
"corpus": corpus_slug,
"folio": folio_label,
"derivative": f"{derivative_width}x{derivative_height}",
"thumbnail": f"{thumbnail_width}x{thumbnail_height}",
},
)
return ImageDerivativeInfo(
original_url=original_url,
original_width=original_width,
original_height=original_height,
derivative_path=str(derivative_path),
derivative_width=derivative_width,
derivative_height=derivative_height,
thumbnail_path=str(thumbnail_path),
thumbnail_width=thumbnail_width,
thumbnail_height=thumbnail_height,
)
def fetch_and_normalize(
url: str,
corpus_slug: str,
folio_label: str,
base_data_dir: Path = Path("data"),
) -> ImageDerivativeInfo:
"""Point d'entrée principal : télécharge depuis une URL IIIF et produit les dérivés.
Chaîne fetch_iiif_image() → create_derivatives().
Args:
url: URL complète de l'image IIIF.
corpus_slug: identifiant du corpus.
folio_label: identifiant du folio.
base_data_dir: racine du dossier data.
Returns:
ImageDerivativeInfo rempli.
"""
source_bytes = fetch_iiif_image(url)
return create_derivatives(source_bytes, url, corpus_slug, folio_label, base_data_dir)
# ── Mode IIIF natif : images en mémoire, jamais sur disque ───────────────────
def fetch_ai_derivative_bytes(
iiif_service_url: str | None,
fallback_url: str | None,
) -> tuple[bytes, int, int]:
"""Retourne (jpeg_bytes, width, height) pour l'IA — jamais sauvé sur disque.
- Si iiif_service_url est fourni : utilise l'IIIF Image API pour demander
au serveur un dérivé 1500px directement redimensionné côté serveur.
- Sinon (fallback_url) : télécharge l'image complète et redimensionne
en mémoire.
Returns:
Tuple (jpeg_bytes, derivative_width, derivative_height).
Raises:
ValueError: si aucune source n'est fournie.
httpx.HTTPStatusError: si le serveur retourne une erreur.
"""
from app.services.ingest.iiif_fetcher import fetch_iiif_derivative, fetch_iiif_image
if iiif_service_url:
raw_bytes = fetch_iiif_derivative(iiif_service_url, max_px=_MAX_DERIVATIVE_PX)
elif fallback_url:
raw_bytes = fetch_iiif_image(fallback_url)
else:
raise ValueError("Aucune source image fournie (ni iiif_service_url ni fallback_url)")
# Ouvrir en mémoire pour obtenir les dimensions (et redimensionner si fallback)
image = Image.open(io.BytesIO(raw_bytes))
if image.mode != "RGB":
image = image.convert("RGB")
if not iiif_service_url:
# Fallback : le serveur n'a pas redimensionné, on le fait en mémoire
image = _resize_to_max(image, _MAX_DERIVATIVE_PX)
w, h = image.size
# Encoder en JPEG en mémoire
buf = io.BytesIO()
image.save(buf, format="JPEG", quality=_DERIVATIVE_QUALITY)
jpeg_bytes = buf.getvalue()
logger.info(
"Dérivé IA en mémoire",
extra={"iiif": bool(iiif_service_url), "size": f"{w}x{h}", "bytes": len(jpeg_bytes)},
)
return jpeg_bytes, w, h
|