Image S3 Flow

The Image S3 Flow is designed to fetch and process image files from an Amazon S3 bucket. Every execution of the flow retrieves new files located in a dedicated S3 folder configured for the flow. These files are matched to products based on their file names and the file name template defined in the flow settings.

The flow exclusively ingests data from the S3 bucket and does not support outbound settings or further outbound operations. Its primary purpose is to ensure that product-related images are fetched and associated correctly based on the naming conventions configured in the flow.

Flow Type​

• Inbound Flow: Indicates that the flow will read data from the S3 bucket. A cron schedule is set up to ensure it runs, for example, every 5 minutes.

Flow Steps​

• Login Step: Handles logging into Omnitron. If an error occurs during this step, the details are logged.

• ReadDataFromS3: Fetches new files from the specified S3 folder for processing.

• Pre Script: Executes a custom script before any transformation or mapping steps, this is useful if filename structures are not conventionally parseable like, “{sku}_{image_order}.jpg”.

• Mapping: Performs data transformations on the input using the JOLT library to group files based on products and extracting product descriptors (sku, base_code, etc.) for further processing.

• Filename Template Task: Matches files to products based on the configured filename template rules.

• Script: Executes additional transformations or operations on the data using custom scripts.

• Write Data To Omnitron: Writes the processed and matched data back to the Omnitron system.

Detailed Flow Designer Settings​

Configuration Card​

• Domain URL: Omnitron’s domain URL, Example: https://demo.omnitron.akinon.net
• Timezone Settings: Timezone used for date-based queries from ERP, this value is used with inbound task requests to modify modified_date__gt value for a specific timezone. The date time format is following: yyyy-MM-dd'T'HH:mm:ss.SSSSSS, ex: 2024-12-31T13:30:59.000001

• Filename template: File naming template, used for matching parts of filename to omnitron products. It can be used to match attribute values, sku, base_code etc. to product values. Default: {sku}. "{attributes__color}_{base_code}" will try to match a file like blue_foo_1.jpg to products which have a base_code value as foo and color attribute equal to blue and image order will be 1. This value will also be used in reverse while triggering the flow using an SKU value, for a product with SKU “bar”, base_code “foo” and color “yellow” when the flow is triggered with the value “bar” Integrator will automatically create the search string “yellow_foo” and look for files with matching names if finds any it will update all the images for products that have the same values.

• Separator: Used for separating image order index value from product values, every image file should have image order value at the last part of the filename (before file extension) separated by this value. By default: “_”

• Productization Settings:

  • Enabled: Fetch missing setting, when enabled, allows fetching products from ERP that have been newly productized but lack images. The 'fetch missing' functionality is a cron job operating every 4 hours for all enabled flows. It queries Omnitron for products with missing image values.

  • Fetch items after productization date limit: Determines how many days within which productized products will be queried from Omnitron.

• S3 Bucket

  • Access Key: S3 API access key, by default this is initialized by Integrator with the access key of a specially created bucket for the flow.

  • Secret Key: S3 API secret key, by default this is initialized by Integrator with the secret key of a specially created bucket for the flow.

  • Bucket Name: S3 bucket name, by default this is initialized by Integrator with the bucket name of a specially created bucket for the flow.

  • Prefix: S3 bucket path prefix, when entered Integrator will search the files in the given path. By default this value is initialized by Integrator with the patch of the specially created bucket for the flow.

  

Access Key, Secret Key, Bucket Name values are only changeable using the “ImageS3 Bucket Settings” button at the top of the configuration screen.

Pre Script and Script Cards (Optional)​

The output data structures for ImageS3 scripting steps are expected to be the same as inputs.

This is a basic script, and the process steps within it can be updated to activate the use of each script step.

Scripts can read incoming data from the inputStream, where the variable input_text contains a JSON string. This string should be parsed using Python's json library. Any outgoing data is written using the outputStream.write() method after converting the relevant Python object back into a JSON string.

Additionally, the script allows for the use of attributes, which provide supplementary information tied to the current execution process. These attributes can be freely accessed or modified throughout the script. For example, the get_attribute() function is used to read attribute values, while the session.putAttribute() method is used to write new string attribute values. Each attribute consists of a key-value pair, where the key uniquely identifies the attribute, and the value can be referenced in subsequent steps.

Attributes with keys starting with the prefix log.will be automatically logged at the end of the execution if a log file is generated, ensuring that important information is captured and available for later review.

Example Script:

import json
import traceback
from java.nio.charset import StandardCharsets
from org.apache.commons.io import IOUtils
from org.apache.nifi.processor.io import StreamCallback

def get_attribute(flow_file, attr_name):
   all_var = flow_file.getAttribute("allVar")
   if all_var:
       all_attributes = json.loads(all_var)
       return all_attributes.get(attr_name, None)
   return flow_file.getAttribute(attr_name)

class TransformCallback(StreamCallback):
   def __init__(self, flowFile):
       self.flowFile = flowFile
       self.omnitronToken = get_attribute(flowFile, "token.omnitron")
       self.erpToken = get_attribute(flowFile, "token.erp")


   def process(self, inputStream, outputStream):
       input_text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
       input_obj = json.loads(input_text)

       # Transform content
       output_obj = self.transform(input_obj)

       # Write output content
       outputStream.write(bytearray(json.dumps(output_obj, indent=4).encode('utf-8')))
      
   def transform(self, output_obj):
       # Transform content
       return output_obj

flowFile = session.get()
if flowFile != None:
   try:
       flowFile = session.write(flowFile, TransformCallback(flowFile))
       # Finish by transferring the FlowFile to an output relationship
       session.transfer(flowFile, REL_SUCCESS)
   except:
       var = traceback.format_exc()
       session.putAttribute(flowFile, 'log.error.script_error', str(var))
       session.transfer(flowFile, REL_FAILURE)
   session.commit()

Script Testing: The response from S3 is placed in the input field, and the result of the script is viewed in the result field.

Pre Script Step Example Input:​

The "Key" values are parsed and matched with the corresponding products after the Pre Script step. If the filenames are not conventionally parseable by the integrator, the Key values can be adjusted to ensure a successful match. However, since these values are used to generate image URLs in subsequent steps, any modifications to the Key values will also affect the resulting URLs. These changes should be addressed and corrected in the script step.

[ {
  "Size" : 912150,
  "StorageClass" : "STANDARD",
  "LastModified" : "2024-12-16T12:35:35.000Z",
  "ETag" : "\"...\"",
  "Key" : "yellow_foo_1.jpg"
}, {
  "Size" : 417604,
  "StorageClass" : "STANDARD",
  "LastModified" : "2024-12-16T12:35:35.000Z",
  "ETag" : "\"...\"",
  "Key" : "yellow_foo_2.jpg"
}, {
  "Size" : 613717,
  "StorageClass" : "STANDARD",
  "LastModified" : "2024-12-16T12:53:51.000Z",
  "ETag" : "\"...\"",
  "Key" : "blue_foo_1.jpg"
}, {
  "Size" : 738279,
  "StorageClass" : "STANDARD",
  "LastModified" : "2024-12-16T12:53:51.000Z",
  "ETag" : "\"...\"",
  "Key" : "blue_foo_2.jpg"
}]

Script Step Example Input​

The "url" fields in this input are generated based on the "Key" values from the previous step. If the "Key" values are modified in the pre-script step, the corresponding "url" fields should be updated in this script step to ensure the URLs are correct.

[ {
  "images" : [ {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/yellow_foo_1.jpg",
    "order" : "1"
  }, {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/yellow_foo_2.jpg",
    "order" : "2"
  } ],
  "product_sku" : "bar"
},{
  "images" : [ {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/yellow_foo_1.jpg",
    "order" : "1"
  }, {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/yellow_foo_2.jpg",
    "order" : "2"
  } ],
  "product_sku" : "bar_xl"
}, {
  "images" : [ {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/blue_foo_1.jpg",
    "order" : "1"
  }, {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/blue_foo_2.jpg",
    "order" : "2"
  } ],
  "product_sku" : "barbar"
}]

POST Write Data To Omnitron​

Integrator automatically posts the results of the previous steps after dividing the output into individual products.

Path: /api/i1/downloadable_image/

Example Body:

{
  "images" : [ {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/yellow_foo_1.jpg",
    "order" : "1",
    "image_type": "swatch"
  }, {
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/yellow_foo_2.jpg",
    "order" : "2",
    "image_type": "default"
  } ],
  "product_sku" : "bar"
}