Gnm merge 2025-11-04

.github/workflows/ci.yml

@@ -15,7 +15,7 @@ jobs:
       pull-requests: write
     services:
       elasticsearch:
-        image: docker.elastic.co/elasticsearch/elasticsearch:7.16.2
+        image: docker.elastic.co/elasticsearch/elasticsearch:7.17.29
         # Wait for elasticsearch to report healthy before continuing.
         # see https://github.com/actions/example-services/blob/master/.github/workflows/postgres-service.yml#L28
         options: -e "discovery.type=single-node" --expose 9200 --health-cmd "curl localhost:9200/_cluster/health" --health-interval 10s --health-timeout 5s --health-retries 10

build.sbt

@@ -63,7 +63,7 @@ Global / concurrentRestrictions := Seq(
 )
 
 val awsSdkVersion = "1.12.470"
-val awsSdkV2Version = "2.31.12"
+val awsSdkV2Version = "2.32.33"
 val elastic4sVersion = "8.18.2"
 val okHttpVersion = "3.12.1"
 
@@ -107,6 +107,9 @@ lazy val commonLib = project("common-lib").settings(
     "org.scanamo" %% "scanamo" % "2.0.0",
     // declare explicit dependency on desired version of aws sdk v2 dynamo
     "software.amazon.awssdk" % "dynamodb" % awsSdkV2Version,
+    // declare explicit dependency on desired version of aws sdk v2 bedrock runtime
+    "software.amazon.awssdk" % "bedrockruntime" % awsSdkV2Version,
+    "software.amazon.awssdk" % "s3vectors" % awsSdkV2Version,
     ws,
     "org.testcontainers" % "elasticsearch" % "1.19.2" % Test
   ),

common-lib/src/main/scala/com/gu/mediaservice/lib/aws/Bedrock.scala

@@ -0,0 +1,95 @@
+package com.gu.mediaservice.lib.aws
+
+import software.amazon.awssdk.services.bedrockruntime.model._
+import software.amazon.awssdk.services.bedrockruntime._
+import com.gu.mediaservice.lib.config.CommonConfig
+import play.api.libs.json.Json
+import software.amazon.awssdk.core.SdkBytes
+
+import java.net.URI
+import com.gu.mediaservice.lib.logging.LogMarker
+import play.api.libs.json.OFormat.oFormatFromReadsAndOWrites
+import play.api.libs.json._
+
+object Bedrock {
+  private case class BedrockRequest(
+   input_type: String,
+   embedding_types: List[String],
+   images: List[String]
+ )
+  private implicit val bedrockRequestFormat: OFormat[BedrockRequest] = Json.format[BedrockRequest]
+}
+
+import scala.concurrent.{ExecutionContext, Future}
+
+class Bedrock(config: CommonConfig)
+  extends AwsClientV2BuilderUtils {
+
+  // TODO: figure out what the more usual pattern for turning off localstack behaviour is
+  override def awsLocalEndpointUri: Option[URI] = None
+
+  override def isDev: Boolean = config.isDev
+
+  val client: BedrockRuntimeClient = {
+    withAWSCredentialsV2(BedrockRuntimeClient.builder())
+      .build()
+  }
+
+  private def createRequestBody(base64EncodedImage: String, fileType: CohereCompatibleMimeType): InvokeModelRequest = {
+    val images = fileType match {
+        case CohereJpeg =>  List(s"data:image/jpg;base64,$base64EncodedImage")
+        case CoherePng => List(s"data:image/png;base64,$base64EncodedImage")
+    }
+
+    val body = Bedrock.BedrockRequest(
+      input_type = "image",
+      embedding_types = List("float"),
+      images = images
+    )
+    val jsonBody = Json.toJson(body).toString()
+
+    val request: InvokeModelRequest = {
+      InvokeModelRequest
+        .builder()
+        .accept("*/*")
+        .body(SdkBytes.fromUtf8String(jsonBody))
+        .contentType("application/json")
+        .modelId("cohere.embed-english-v3")
+        .build()
+    }
+    request
+  }
+
+  private def sendBedrockEmbeddingRequest(base64EncodedImage: String, fileType: CohereCompatibleMimeType)(
+    implicit logMarker: LogMarker
+  ): InvokeModelResponse = {
+    try {
+      val response = client.invokeModel(createRequestBody(base64EncodedImage, fileType))
+      logger.info(
+        logMarker,
+        s"Bedrock API call to create image embedding completed with status: ${response.sdkHttpResponse().statusCode()}"
+      )
+      response
+    }
+    catch {
+      case e: Exception =>
+        logger.error(logMarker, "Exception during Bedrock API call to create image embedding", e)
+        throw e
+    }
+  }
+
+  def createImageEmbedding(base64EncodedImage: String, fileType: CohereCompatibleMimeType)(implicit ec: ExecutionContext, logMarker: LogMarker): Future[List[Float]] = {
+    val bedrockFuture = Future { sendBedrockEmbeddingRequest(base64EncodedImage, fileType) }
+    bedrockFuture.map { response =>
+      val responseBody = response.body().asUtf8String()
+      val json = Json.parse(responseBody)
+      // Extract the embedding array (first element since it's an array of arrays)
+      val embedding = (json \ "embeddings" \ "float")(0).as[List[Float]]
+      logger.info(
+        logMarker,
+        s"Successfully extracted image embedding. Vector size: ${embedding.size}"
+      )
+      embedding
+    }
+  }
+}

common-lib/src/main/scala/com/gu/mediaservice/lib/aws/Embedder.scala

@@ -0,0 +1,44 @@
+package com.gu.mediaservice.lib.aws
+import com.gu.mediaservice.lib.logging.{GridLogging, LogMarker}
+import com.gu.mediaservice.model.{Jpeg, MimeType, Png, Tiff}
+import software.amazon.awssdk.services.s3vectors.model.PutVectorsResponse
+
+import java.nio.file.{Files, Path}
+import java.util.Base64
+import scala.concurrent.{ExecutionContext, Future}
+
+sealed trait CohereCompatibleMimeType
+case object CohereJpeg extends CohereCompatibleMimeType
+case object CoherePng extends CohereCompatibleMimeType
+
+class Embedder(s3vectors: S3Vectors, bedrock: Bedrock) extends GridLogging {
+  // https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-embed-v3.html#:~:text=The%20image%20must%20be%20in%20either%20image/jpeg%20or%20image/png%20format%20and%20has%20a%20maximum%20size%20of%205MB
+  def meetsCohereRequirements(fileType: MimeType, imageFilePath: Path)(implicit logMarker: LogMarker): Either[String, CohereCompatibleMimeType]= {
+    val fileSize = Files.size(imageFilePath)
+    val fiveMB = 5_000_000
+
+    fileType match {
+      case _ if fileSize > fiveMB => Left(s"Image file is >5MB. File size: $fileSize")
+      case Jpeg => Right(CohereJpeg)
+      case Png => Right(CoherePng)
+      case Tiff => Left("Image file type is not supported. File type: Tiff")
+    }
+  }
+
+  def createEmbeddingAndStore(fileType: MimeType, imageFilePath: Path, imageId: String)(implicit ec: ExecutionContext, logMarker: LogMarker
+  ): Future[Option[PutVectorsResponse]] = {
+    meetsCohereRequirements(fileType, imageFilePath)(logMarker) match {
+      case Left(error) => {
+        logger.info(logMarker, s"Skipping image embedding for $imageId as it does not meet the requirements: $error")
+        Future.successful(None)
+      }
+      case Right(imageType) => {
+        val base64EncodedString: String = Base64.getEncoder().encodeToString(Files.readAllBytes(imageFilePath))
+        val embeddingFuture = bedrock.createImageEmbedding(base64EncodedString, imageType)
+        embeddingFuture.map { embedding =>
+          Some(s3vectors.storeEmbeddingInS3VectorStore(embedding, imageId))
+        }
+      }
+    }
+  }
+}

common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3Vectors.scala

@@ -0,0 +1,69 @@
+package com.gu.mediaservice.lib.aws
+import com.gu.mediaservice.lib.config.CommonConfig
+import com.gu.mediaservice.lib.logging.LogMarker
+
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.s3vectors._
+import software.amazon.awssdk.services.s3vectors.model.{PutInputVector, PutVectorsRequest, PutVectorsResponse, VectorData}
+
+import java.net.URI
+import scala.concurrent.{ExecutionContext, Future}
+import scala.jdk.CollectionConverters._
+
+class S3Vectors(config: CommonConfig)
+  extends AwsClientV2BuilderUtils {
+
+  // TODO: figure out what the more usual pattern for turning off localstack behaviour is
+  override def awsLocalEndpointUri: Option[URI] = None
+
+  override def isDev: Boolean = config.isDev
+
+  // The S3 Vector Store is not yet available in eu-west-1, so we are using eu-central-1 because it's closest to us.
+  override def awsRegionV2: Region = Region.EU_CENTRAL_1
+
+  val client: S3VectorsClient = {
+    withAWSCredentialsV2(S3VectorsClient.builder())
+      .build()
+  }
+
+  private def createRequestBody(embedding: List[Float], imageId: String): PutVectorsRequest = {
+    val vectorData: VectorData = VectorData
+      .builder()
+      .float32(embedding.map(float2Float).asJava)
+      .build()
+
+    val inputVector: PutInputVector = PutInputVector
+      .builder()
+      .data(vectorData)
+      .key(imageId)
+      .build()
+
+    val request: PutVectorsRequest = PutVectorsRequest
+      .builder()
+      .indexName("cohere-embed-english-v3")
+      .vectorBucketName(s"image-embeddings-${config.stage.toLowerCase}")
+      .vectors(inputVector)
+      .build()
+
+    request
+  }
+
+  def storeEmbeddingInS3VectorStore(bedrockEmbedding: List[Float], imageId: String)(implicit logMarker: LogMarker
+  ): PutVectorsResponse = {
+    try {
+      val request = createRequestBody(bedrockEmbedding, imageId)
+      val response = client.putVectors(request)
+      logger.info(
+        logMarker,
+        s"S3 Vector Store API call to store image embedding completed with status: ${response.sdkHttpResponse().statusCode()}"
+      )
+      response
+    }
+   catch {
+    case e: Exception =>
+      logger.error(logMarker, s"Exception during S3 Vector Store API call to store image embedding for $imageId: ", e)
+      throw e
+    }
+  }
+
+}

common-lib/src/main/scala/com/gu/mediaservice/lib/cleanup/PhotographerRenamer.scala

@@ -234,6 +234,7 @@ object PhotographerRenamer extends MetadataCleaner {
     "Huseyin Demirci" -> "Hüseyin Demirci",
     "Huseyin Yildiz" -> "Hüseyin Yıldız",
     "Ian Macnicol" -> "Ian MacNicol",
+    "Igor Pavicevic" -> "Igor Pavićević",
     "Inti Ocon" -> "Inti Ocón",
     "Ints Kalnins" -> "Ints Kalniņš",
     "Irek Dorozanski" -> "Irek Dorożański",
@@ -274,6 +275,7 @@ object PhotographerRenamer extends MetadataCleaner {
     "Jerome Prebois" -> "Jérôme Prébois",
     "Jerome Prevost" -> "Jérôme Prévost",
     "Jerome Sessini" -> "Jérôme Sessini",
+    "Jerzy Muszynski" -> "Jerzy Muszyński",
     "Jesus Bustamante" -> "Jesús Bustamante",
     "Jesus Diges" -> "Jesús Diges",
     "Jesus Merida" -> "Jesús Mérida",
@@ -338,6 +340,7 @@ object PhotographerRenamer extends MetadataCleaner {
     "Klebher Vasquez" -> "Klebher Vásquez",
     "Koca Sulejmanovic" -> "Koca Sulejmanović",
     "Krisztian Elek" -> "Krisztián Elek",
+    "Krzysztof Cwik" -> "Krzysztof Ćwik",
     "Krzysztof Swiderski" -> "Krzysztof Świderski",
     "Kuba Stezycki" -> "Kuba Stężycki",
     "Laszlo Balogh" -> "László Balogh",
@@ -378,6 +381,7 @@ object PhotographerRenamer extends MetadataCleaner {
     "Manu Fernandez" -> "Manu Fernández",
     "Manuel Vazquez" -> "Manuel Vázquez",
     "Manuel Velazquez" -> "Manuel Velázquez",
+    "Marek Antoni Iwanczuk" -> "Marek Antoni Iwańczuk",
     "Marc Mccormack" -> "Marc McCormack",
     "Marcelo Del Pozo" -> "Marcelo del Pozo",
     "Marcelo Hernandez" -> "Marcelo Hernández",
@@ -419,6 +423,7 @@ object PhotographerRenamer extends MetadataCleaner {
     "Milos Bicanski" -> "Miloš Bičanski",
     "Milos Vujovic" -> "Miloš Vujović",
     "Miro Kuzmanovic" -> "Miro Kuzmanović",
+    "Mitar Mitrovic" -> "Mitar Mitrović",
     "Moises Castillo" -> "Moisés Castillo",
     "Morne de Klerk" -> "Morné de Klerk",
     "Murat Ozgur Guvendik" -> "Murat Özgür Güvendik",
@@ -482,6 +487,7 @@ object PhotographerRenamer extends MetadataCleaner {
     "Radoslaw Jozwiak" -> "Radosław Jóźwiak",
     "Rafal Gaglewski" -> "Rafał Gąglewski",
     "Rafal Guz" -> "Rafał Guz",
+    "Raimonda Kulikauskiene" -> "Raimonda Kulikauskienė",
     "Ramon Buxo Martinez" -> "Ramon Buxó Martínez",
     "Ramon Costa" -> "Ramón Costa",
     "Ramon de la Rocha" -> "Ramón de la Rocha",

common-lib/src/main/scala/com/gu/mediaservice/lib/cleanup/RedundantTokenRemover.scala

@@ -21,6 +21,8 @@ object RedundantTokenRemover extends MetadataCleaner {
     "Stringer",
     "Stringer .",
     "STR",
+    "STR New",
+    "-STR",
     "supplied",
     "Supplied",
     "SUPPLIED",

dev/script/generate-config/service-config.js

@@ -74,6 +74,7 @@ function getImageLoaderConfig(config) {
         |metrics.request.enabled=false
         |transcoded.mime.types="image/tiff"
         |upload.quarantine.enabled=false
+        |s3.vectors.shouldEmbed=false
         |`;
 }
 

dev/script/get-s3-vector-store-records.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+# Check if environment argument is provided
+if [ $# -ne 1 ]; then
+    echo "Usage: $0 <environment>"
+    echo "Environment options: dev, test, prod"
+    exit 1
+fi
+
+# Validate environment argument
+env=$1
+if [[ ! "$env" =~ ^(dev|test|prod)$ ]]; then
+    echo "Invalid environment. Please use dev, test, or prod"
+    exit 1
+fi
+
+aws s3vectors list-vectors \
+  --vector-bucket-name "image-embeddings-$env" \
+  --index-name cohere-embed-english-v3 \
+  --profile media-service \
+  --region eu-central-1

image-loader/app/ImageLoaderComponents.scala

@@ -1,5 +1,5 @@
 import com.gu.mediaservice.GridClient
-import com.gu.mediaservice.lib.aws.SimpleSqsMessageConsumer
+import com.gu.mediaservice.lib.aws.{Bedrock, S3Vectors, SimpleSqsMessageConsumer, Embedder}
 import com.gu.mediaservice.lib.config.Services
 import com.gu.mediaservice.lib.imaging.ImageOperations
 import com.gu.mediaservice.lib.logging.GridLogging
@@ -27,8 +27,11 @@ class ImageLoaderComponents(context: Context) extends GridComponents(context, ne
   val imageOperations = new ImageOperations(context.environment.rootPath.getAbsolutePath)
   val notifications = new Notifications(config)
   val downloader = new Downloader()(ec,wsClient)
-  val uploader = new Uploader(store, config, imageOperations, notifications, imageProcessor)
-  val projector = Projector(config, imageOperations, imageProcessor, auth)
+
+  val maybeEmbedder: Option[Embedder] = if (config.shouldEmbed) Some(new Embedder(new S3Vectors(config), new Bedrock(config))) else None
+
+  val uploader = new Uploader(store, config, imageOperations, notifications, maybeEmbedder, imageProcessor)
+  val projector = Projector(config, imageOperations, imageProcessor, auth, maybeEmbedder)
   val quarantineUploader: Option[QuarantineUploader] = (config.uploadToQuarantineEnabled, config.quarantineBucket) match {
     case (true, Some(bucketName)) =>{
       val quarantineStore = new QuarantineStore(config)

image-loader/app/lib/ImageLoaderConfig.scala

@@ -33,6 +33,8 @@ class ImageLoaderConfig(resources: GridConfigResources) extends CommonConfig(res
   val uploadStatusTable: String = string("dynamo.table.upload.status")
   val uploadStatusExpiry: FiniteDuration = configuration.get[FiniteDuration]("uploadStatus.recordExpiry")
 
+  val shouldEmbed: Boolean = boolean("s3.vectors.shouldEmbed")
+
   /**
     * Load in the chain of image processors from config. This can be a list of
     * companion objects, class names, both with and without config.