AWS and AI resolution tutorials

tutorials/dev-garage-aws-pipeline-revenue-management/files/TransformCsvToJson.groovy

@@ -0,0 +1,165 @@
+import com.sap.gateway.ip.core.customdev.util.Message
+import groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+
+// This script assumes the CSV->XML Converter and XML to JSON Converter have
+// already run, and that the "Prepare for Lookup" Content Modifier stashed
+// the resulting JSON in the "preEnrichmentJson" property (along with
+// businessDate, ingestedAt and sourceRegion) before the Cost LookUp read
+// replaced the message body with product_master.csv.
+//
+// This script's job is to compute profit margin per transaction using unit
+// costs from the reference file in S3, and to leave a run summary in the
+// runHeadline and runSummary properties, which participants read in the
+// message trace.
+
+Message processData(Message message) {
+
+    // 1. Pull back the JSON built by the XML to JSON Converter. The SAP CI
+    // Message wrapper only exposes single-arg getProperty() (no
+    // type-conversion overload like raw Camel's Message), and the property
+    // can come back as several different underlying types depending on
+    // Camel's stream caching -- a String, a raw byte[], or an
+    // InputStreamCache (a re-readable stream wrapper). Calling toString()
+    // on that last one just prints the Java object identity
+    // ("org.apache.camel.converter.stream.InputStreamCache@..."), not its
+    // content, so each case needs handling explicitly.
+    def rawPreEnrichmentJson = message.getProperty("preEnrichmentJson")
+    def preEnrichmentJson
+    switch (rawPreEnrichmentJson) {
+        case byte[]:
+            preEnrichmentJson = new String((byte[]) rawPreEnrichmentJson, "UTF-8")
+            break
+        case InputStream:
+            // The property can share one stream with the body. If a step in
+            // between (such as the Cost LookUp read) has already read that
+            // stream to the end, rewind it first, or getText() returns "".
+            try { rawPreEnrichmentJson.reset() } catch (ignored) { }
+            preEnrichmentJson = rawPreEnrichmentJson.getText("UTF-8")
+            break
+        default:
+            preEnrichmentJson = rawPreEnrichmentJson as String
+    }
+    if (!preEnrichmentJson?.trim()) {
+        throw new IllegalStateException("Property preEnrichmentJson is empty. In 'Prepare for Lookup', " +
+            "set the Data Type of preEnrichmentJson to java.lang.String so it holds a copy of the sales JSON.")
+    }
+    def parsedPayload = new JsonSlurper().parseText(preEnrichmentJson)
+
+    // 1a. The XML to JSON Converter mirrors the XML element names exactly,
+    // so the rows live at Records.Row -- there is no flat "records" key.
+    // Confirm this shape in your own Message Processing Log; if your
+    // converter's settings ever produce something different, adjust this
+    // line to match.
+    def rawRows = parsedPayload.Records.Row
+    // Guard against the single-record edge case: some XML->JSON conversions
+    // collapse a lone repeating element into a single object instead of a
+    // one-element array, which would otherwise break the .each{} below.
+    def rows = rawRows instanceof List ? rawRows : [rawRows]
+
+    // 1b. Attach the ingestion metadata from the properties set in
+    // "Prepare for Lookup". This also reshapes the payload into the flat
+    // { "source-region", "ingested-at", "records" } structure used
+    // everywhere else in this lab, instead of the nested Records.Row shape
+    // that only exists because that's how the XML happened to be structured.
+    def payload = [
+        "source-region": message.getProperty("sourceRegion"),
+        "ingested-at"  : message.getProperty("ingestedAt"),
+        "records"      : rows
+    ]
+
+    // 2. Parse the reference file read from S3 by Cost LookUp (the current
+    // body): regional-landing-bkt/reference/product_master.csv, with a
+    // header line "product_id,unit_cost" followed by one product per line.
+    // The header line is skipped, which also skips any byte-order mark a
+    // spreadsheet tool may add. Splitting on comma or semicolon covers Excel
+    // on European locales, which saves CSVs with semicolons.
+    def costLookup = [:]
+    message.getBody(java.io.Reader).readLines().drop(1).findAll { it.trim() }.each { line ->
+        def cols = line.split(/[,;]/)
+        costLookup[cols[0].trim()] = new BigDecimal(cols[1].trim())
+    }
+
+    // 3. The one genuinely custom piece of business logic left in code:
+    // derive profit margin, a number that doesn't exist anywhere in the
+    // source CSV, from cost data that only lives in the reference file.
+    payload.records.each { record ->
+        def unitCost = costLookup[record.product_id]
+        def quantity = record.quantity as BigDecimal
+        def totalAmount = record.total_amount as BigDecimal
+        def unitPrice = record.unit_price as BigDecimal
+
+        // Write the numeric casts back onto the record. The XML to JSON
+        // Converter serializes every XML text node as a JSON string, so
+        // without this quantity/unit_price/total_amount would stay quoted
+        // in the curated output.
+        record.quantity = quantity
+        record.unit_price = unitPrice
+        record.total_amount = totalAmount
+
+        if (unitCost != null) {
+            def cost = unitCost * quantity
+            def margin = totalAmount - cost
+            record.unit_cost = unitCost
+            record.gross_margin = margin.setScale(2, BigDecimal.ROUND_HALF_UP)
+            record.margin_percentage = totalAmount == 0 ? 0 :
+                (margin / totalAmount * 100).setScale(1, BigDecimal.ROUND_HALF_UP)
+        } else {
+            record.unit_cost = "unknown"
+            record.gross_margin = "unknown"
+            record.margin_percentage = "unknown"
+        }
+    }
+
+    // 4. Build a short run summary. Nothing here changes the curated JSON --
+    // it only sets two exchange properties, runHeadline and runSummary,
+    // which show up in Monitor -> the message's trace -> Properties for any
+    // step after this one. Edit the two path constants if your buckets differ.
+    // The sender keeps the source file in place (Post-Processing: Keep File
+    // and Process Again), so every poll reprocesses the same file.
+    def sourcePath  = "regional-landing-bkt/daily/ (kept for the next run)"
+    def curatedPath = "curated-corp/profit-margins/sales_" + message.getProperty("businessDate") + ".json"
+
+    def priced = payload.records.findAll { it.gross_margin instanceof Number }
+    def unpricedCount = payload.records.size() - priced.size()
+    def totalRevenue = priced.sum(BigDecimal.ZERO) { it.total_amount as BigDecimal }
+    def totalMargin  = priced.sum(BigDecimal.ZERO) { it.gross_margin as BigDecimal }
+    def marginPct = totalRevenue == 0 ? BigDecimal.ZERO :
+        (totalMargin / totalRevenue * 100).setScale(1, BigDecimal.ROUND_HALF_UP)
+
+    def marginBy = { String field ->
+        priced.groupBy { it[field] }.collectEntries { key, recs ->
+            [key, recs.sum(BigDecimal.ZERO) { it.gross_margin as BigDecimal }]
+        }
+    }
+    def topProduct = priced ? marginBy("product_name").max { it.value } : null
+    def topRegion  = priced ? marginBy("region").max { it.value } : null
+
+    // Locale.US pins the format to 1,234.56 regardless of the tenant JVM's
+    // default locale (which could otherwise render 1.234,56).
+    def money = { BigDecimal v -> "\$" + String.format(Locale.US, "%,.2f", v) }
+
+    def lines = []
+    lines << "Profit margin run complete"
+    lines << ""
+    lines << "Business date:  " + message.getProperty("businessDate")
+    lines << "Records:        " + payload.records.size() + " (" + unpricedCount + " without cost data)"
+    lines << "Revenue:        " + money(totalRevenue)
+    lines << "Gross margin:   " + money(totalMargin) + " (" + marginPct + "%)"
+    if (topProduct) {
+        lines << ""
+        lines << "Top product:    " + topProduct.key + " (" + money(topProduct.value) + " margin)"
+        lines << "Top region:     " + topRegion.key + " (" + money(topRegion.value) + " margin)"
+    }
+    lines << ""
+    lines << "Read from:      " + sourcePath
+    lines << "Published to:   " + curatedPath
+
+    message.setProperty("runHeadline",
+        "Profit margins ready: " + message.getProperty("businessDate") + " | " + marginPct + "% gross margin")
+    message.setProperty("runSummary", lines.join("\n"))
+
+    message.setBody(JsonOutput.toJson(payload))
+    message.setHeader("Content-Type", "application/json")
+    return message
+}

tutorials/dev-garage-aws-pipeline-revenue-management/files/xml_schema.xsd

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
+  <xs:element name="Records">
+    <xs:complexType>
+      <xs:sequence>
+        <xs:element name="Row" maxOccurs="unbounded">
+          <xs:complexType>
+            <xs:sequence>
+              <xs:element name="transaction_id" type="xs:string"/>
+              <xs:element name="store_id" type="xs:string"/>
+              <xs:element name="region" type="xs:string"/>
+              <xs:element name="product_id" type="xs:string"/>
+              <xs:element name="product_name" type="xs:string"/>
+              <xs:element name="category" type="xs:string"/>
+              <xs:element name="quantity" type="xs:string"/>
+              <xs:element name="unit_price" type="xs:string"/>
+              <xs:element name="total_amount" type="xs:string"/>
+              <xs:element name="sale_timestamp" type="xs:string"/>
+              <xs:element name="payment_method" type="xs:string"/>
+            </xs:sequence>
+          </xs:complexType>
+        </xs:element>
+      </xs:sequence>
+    </xs:complexType>
+  </xs:element>
+</xs:schema>