Video S3 Flow

The VideoS3 Flow is designed to fetch and process video 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 videos 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.

• 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.mp4 to products which have a base_code value as foo and color attribute equal to blue and video 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 videos for products that have the same values.

• Separator: Used for separating video order index value from product values, every video file should have video 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 videos. The 'fetch missing' functionality is a cron job operating every 4 hours for all enabled flows. It queries omnitron for products with missing video 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 “VideoS3 Bucket Settings” button at the top of configuration screen

Script Card (Optional)​

The output data structure for the VideoS3 scripting step is expected to be the same as input.

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 ERP is placed in the input field, and the result of the script is viewed in the result field.

Script Step Example Input:​

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

[ {
  "is_pre_product" : false,
  "videos" : [ {
    "object_key" : "mpd-omnitron-ab87ed4f-dd41-4152-9fbe-31411c9e7de1-upload/AKN-058-BLUZ-XL_1.mp4",
    "url" : "https://s3.eu-central-1.amazonaws.com/dev-entegrator-upload-bucket/mpd-omnitron-ab87ed4f-dd41-4152-9fbe-31411c9e7de1-upload/AKN-058-BLUZ-XL_1.mp4",
    "order" : "1"
  } ],
  "product_sku" : "AKN-058-BLUZ-XL",
  "product_pk" : 106
} ]

POST Write Data To Omnitron​

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

Path: /api/v1/product_video/

Request body is a multipart form request with product ID, video order and binary video data.

After the new videos are installed, the old videos of the updated products are automatically deleted, ensuring that only the newly installed videos remain.