SAP S/4HANA & Business One

import com.sap.gateway.ip.core.customdev.util.Message
import groovy.json.JsonSlurper
import groovy.json.JsonOutput

def Message processData(Message message) {
    def body = message.getBody(String)
    def slurper = new JsonSlurper()
    def sapEvent = slurper.parseText(body)

    // Extract material document header fields
    def matDoc = sapEvent.d ?: sapEvent
    def materialDocument = matDoc.MaterialDocument ?: matDoc.MaterialDocumentNumber
    def materialDocumentYear = matDoc.MaterialDocumentYear ?: new Date().format("yyyy")
    def postingDate = matDoc.PostingDate ?: matDoc.DocumentDate
    def headerText = matDoc.MaterialDocumentHeaderText ?: ""

    // Extract line items
    def items = matDoc.to_MaterialDocumentItem?.results ?: matDoc.Items ?: []
    def recordLines = items.collect { item ->
        [
            itemNumber       : item.MaterialDocumentItem ?: item.LineNumber,
            material         : item.Material ?: item.ItemCode,
            plant            : item.Plant ?: item.Warehouse,
            storageLocation  : item.StorageLocation ?: "",
            batch            : item.Batch ?: "",
            serialNumber     : item.SerialNumber ?: "",
            quantity         : item.QuantityInEntryUnit ?: item.Quantity,
            entryUnit        : item.EntryUnit ?: item.UnitOfMeasure,
            movementType     : item.GoodsMovementType ?: item.MovementType,
            purchaseOrder    : item.PurchaseOrder ?: "",
            purchaseOrderItem: item.PurchaseOrderItem ?: ""
        ]
    }

    // Build Certyo record payload
    def tenantId = message.getProperty("CertyoTenantId")
    def certyoRecord = [
        tenantId       : tenantId,
        database       : "sap-s4hana",
        collection     : "MaterialDocuments",
        recordId       : "${materialDocument}-${materialDocumentYear}",
        recordVersion  : "1",
        operationType  : "insert",
        sourceTimestamp : postingDate,
        idempotencyKey : "${materialDocument}-${materialDocumentYear}-${postingDate?.take(10) ?: 'unknown'}",
        recordPayload  : [
            materialDocument     : materialDocument,
            materialDocumentYear : materialDocumentYear,
            postingDate          : postingDate,
            headerText           : headerText,
            documentType         : matDoc.MaterialDocumentType ?: "WE",
            referenceDocument    : matDoc.ReferenceDocument ?: "",
            items                : recordLines
        ]
    ]

    message.setBody(JsonOutput.toJson(certyoRecord))
    message.setHeader("Content-Type", "application/json")
    message.setHeader("X-API-Key", message.getProperty("CertyoApiKey"))

    return message
}

const cds = require("@sap/cds");

class CertyoIntegrationService extends cds.ApplicationService {
  async init() {
    const s4 = await cds.connect.to("API_MATERIAL_DOCUMENT_SRV");
    const { MaterialDocumentHeaders } = s4.entities;

    // React to new material documents from S/4HANA
    this.on("MaterialDocumentCreated", async (req) => {
      const { materialDocument, materialDocumentYear } = req.data;

      // Fetch full document with line items from S/4HANA
      const doc = await s4.run(
        SELECT.one
          .from(MaterialDocumentHeaders)
          .where({
            MaterialDocument: materialDocument,
            MaterialDocumentYear: materialDocumentYear,
          })
          .columns((h) => {
            h("*");
            h.to_MaterialDocumentItem((i) => i("*"));
          })
      );

      if (!doc) {
        console.warn(
          `Material document ${materialDocument}/${materialDocumentYear} not found`
        );
        return;
      }

      // Transform to Certyo record format
      const certyoRecord = {
        tenantId: process.env.CERTYO_TENANT_ID,
        database: "sap-s4hana",
        collection: "MaterialDocuments",
        recordId: `${doc.MaterialDocument}-${doc.MaterialDocumentYear}`,
        recordVersion: "1",
        operationType: "insert",
        sourceTimestamp: doc.PostingDate,
        idempotencyKey: `${doc.MaterialDocument}-${doc.MaterialDocumentYear}-${doc.PostingDate}`,
        recordPayload: {
          materialDocument: doc.MaterialDocument,
          materialDocumentYear: doc.MaterialDocumentYear,
          postingDate: doc.PostingDate,
          documentDate: doc.DocumentDate,
          headerText: doc.MaterialDocumentHeaderText,
          referenceDocument: doc.ReferenceDocument,
          items: (doc.to_MaterialDocumentItem || []).map((item) => ({
            itemNumber: item.MaterialDocumentItem,
            material: item.Material,
            plant: item.Plant,
            storageLocation: item.StorageLocation,
            batch: item.Batch,
            serialNumber: item.SerialNumber,
            quantity: parseFloat(item.QuantityInEntryUnit),
            entryUnit: item.EntryUnit,
            movementType: item.GoodsMovementType,
            purchaseOrder: item.PurchaseOrder,
            purchaseOrderItem: item.PurchaseOrderItem,
          })),
        },
      };

      // Call Certyo API
      const response = await fetch("https://www.certyos.com/api/v1/records", {
        method: "POST",
        headers: {
          "X-API-Key": process.env.CERTYO_API_KEY,
          "Content-Type": "application/json",
        },
        body: JSON.stringify(certyoRecord),
      });

      if (!response.ok) {
        const errorBody = await response.text();
        console.error(
          `Certyo ingestion failed for ${materialDocument}: ${response.status} ${errorBody}`
        );
        throw new Error(`Certyo returned ${response.status}`);
      }

      const result = await response.json();
      console.log(
        `Ingested ${materialDocument} -> recordHash: ${result.recordHash}`
      );

      return result;
    });

    await super.init();
  }
}

module.exports = CertyoIntegrationService;

"""
SAP Material Document Poller -> Certyo Ingestion
Polls SAP OData API for new material documents and ingests them into Certyo.
Supports S/4HANA Cloud, S/4HANA On-Premise (via Cloud Connector), and Business One.
"""

import os
import sys
import time
import logging
from datetime import datetime, timedelta, timezone
from dataclasses import dataclass, field

import requests
from requests.auth import HTTPBasicAuth

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

# --- Configuration ---
SAP_BASE_URL = os.environ["SAP_BASE_URL"]           # e.g. https://my-s4.example.com/sap/opu/odata/sap
SAP_USERNAME = os.environ["SAP_USERNAME"]
SAP_PASSWORD = os.environ["SAP_PASSWORD"]
CERTYO_API_KEY = os.environ["CERTYO_API_KEY"]
CERTYO_TENANT_ID = os.environ["CERTYO_TENANT_ID"]
CERTYO_BASE_URL = os.environ.get("CERTYO_BASE_URL", "https://www.certyos.com")
POLL_INTERVAL_SECONDS = int(os.environ.get("POLL_INTERVAL_SECONDS", "30"))

# S/4HANA Material Document OData endpoint
MATDOC_ENDPOINT = "/API_MATERIAL_DOCUMENT_SRV/A_MaterialDocumentHeader"


@dataclass
class MaterialDocument:
    material_document: str
    material_document_year: str
    posting_date: str
    document_date: str
    header_text: str = ""
    reference_document: str = ""
    items: list = field(default_factory=list)


def fetch_new_material_documents(since: datetime) -> list[MaterialDocument]:
    """Fetch material documents created after the given timestamp."""
    since_str = since.strftime("%Y-%m-%dT%H:%M:%S")
    url = f"{SAP_BASE_URL}{MATDOC_ENDPOINT}"
    params = {
        "$filter": f"CreationDate gt datetime'{since_str}'",
        "$expand": "to_MaterialDocumentItem",
        "$orderby": "CreationDate asc",
        "$top": 100,
        "$format": "json",
    }

    response = requests.get(
        url,
        params=params,
        auth=HTTPBasicAuth(SAP_USERNAME, SAP_PASSWORD),
        headers={"Accept": "application/json"},
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()
    results = data.get("d", {}).get("results", [])

    documents = []
    for doc in results:
        items = []
        for item in doc.get("to_MaterialDocumentItem", {}).get("results", []):
            items.append({
                "itemNumber": item.get("MaterialDocumentItem", ""),
                "material": item.get("Material", ""),
                "plant": item.get("Plant", ""),
                "storageLocation": item.get("StorageLocation", ""),
                "batch": item.get("Batch", ""),
                "serialNumber": item.get("SerialNumber", ""),
                "quantity": float(item.get("QuantityInEntryUnit", 0)),
                "entryUnit": item.get("EntryUnit", ""),
                "movementType": item.get("GoodsMovementType", ""),
                "purchaseOrder": item.get("PurchaseOrder", ""),
                "purchaseOrderItem": item.get("PurchaseOrderItem", ""),
            })

        documents.append(MaterialDocument(
            material_document=doc["MaterialDocument"],
            material_document_year=doc["MaterialDocumentYear"],
            posting_date=doc.get("PostingDate", ""),
            document_date=doc.get("DocumentDate", ""),
            header_text=doc.get("MaterialDocumentHeaderText", ""),
            reference_document=doc.get("ReferenceDocument", ""),
            items=items,
        ))

    return documents


def ingest_to_certyo(doc: MaterialDocument) -> dict:
    """Send a single material document to Certyo for blockchain anchoring."""
    record_id = f"{doc.material_document}-{doc.material_document_year}"
    posting_date_short = doc.posting_date[:10] if doc.posting_date else "unknown"

    payload = {
        "tenantId": CERTYO_TENANT_ID,
        "database": "sap-s4hana",
        "collection": "MaterialDocuments",
        "recordId": record_id,
        "recordVersion": "1",
        "operationType": "insert",
        "sourceTimestamp": doc.posting_date,
        "idempotencyKey": f"{record_id}-{posting_date_short}",
        "recordPayload": {
            "materialDocument": doc.material_document,
            "materialDocumentYear": doc.material_document_year,
            "postingDate": doc.posting_date,
            "documentDate": doc.document_date,
            "headerText": doc.header_text,
            "referenceDocument": doc.reference_document,
            "items": doc.items,
        },
    }

    response = requests.post(
        f"{CERTYO_BASE_URL}/api/v1/records",
        headers={
            "X-API-Key": CERTYO_API_KEY,
            "Content-Type": "application/json",
        },
        json=payload,
        timeout=15,
    )

    if response.status_code == 202:
        result = response.json()
        logger.info("Ingested %s -> recordHash: %s", record_id, result["recordHash"])
        return result
    else:
        logger.error(
            "Certyo rejected %s: %s %s", record_id, response.status_code, response.text
        )
        response.raise_for_status()
        return {}


def bulk_ingest_to_certyo(documents: list[MaterialDocument]) -> dict:
    """Send up to 1000 material documents in a single bulk request."""
    records = []
    for doc in documents:
        record_id = f"{doc.material_document}-{doc.material_document_year}"
        posting_date_short = doc.posting_date[:10] if doc.posting_date else "unknown"
        records.append({
            "database": "sap-s4hana",
            "collection": "MaterialDocuments",
            "recordId": record_id,
            "recordVersion": "1",
            "operationType": "insert",
            "sourceTimestamp": doc.posting_date,
            "idempotencyKey": f"{record_id}-{posting_date_short}",
            "recordPayload": {
                "materialDocument": doc.material_document,
                "materialDocumentYear": doc.material_document_year,
                "postingDate": doc.posting_date,
                "documentDate": doc.document_date,
                "headerText": doc.header_text,
                "referenceDocument": doc.reference_document,
                "items": doc.items,
            },
        })

    response = requests.post(
        f"{CERTYO_BASE_URL}/api/v1/records/bulk",
        headers={
            "X-API-Key": CERTYO_API_KEY,
            "Content-Type": "application/json",
        },
        json={
            "tenantId": CERTYO_TENANT_ID,
            "records": records,
        },
        timeout=30,
    )

    if response.status_code == 202:
        result = response.json()
        logger.info("Bulk ingested %d records", len(records))
        return result
    else:
        logger.error("Bulk ingest failed: %s %s", response.status_code, response.text)
        response.raise_for_status()
        return {}


def run_poll_loop():
    """Main polling loop. Tracks high-water mark to avoid reprocessing."""
    high_water_mark = datetime.now(timezone.utc) - timedelta(hours=1)
    logger.info("Starting SAP -> Certyo poller (interval: %ds)", POLL_INTERVAL_SECONDS)

    while True:
        try:
            documents = fetch_new_material_documents(high_water_mark)
            if documents:
                logger.info("Found %d new material documents", len(documents))
                if len(documents) > 10:
                    bulk_ingest_to_certyo(documents)
                else:
                    for doc in documents:
                        ingest_to_certyo(doc)
                # Advance high-water mark
                high_water_mark = datetime.now(timezone.utc)
            else:
                logger.debug("No new documents")
        except requests.RequestException as e:
            logger.error("Poll cycle failed: %s", e)

        time.sleep(POLL_INTERVAL_SECONDS)


if __name__ == "__main__":
    run_poll_loop()