Oracle NetSuite Integration

/**
 * @NApiVersion 2.1
 * @NScriptType Restlet
 * @NModuleScope SameAccount
 *
 * Certyo Integration RESTlet
 * Accepts item fulfillment data and ingests it into Certyo for
 * blockchain anchoring.
 *
 * Deploy: Setup > Scripting > Scripts > New > RESTlet
 * URL:    /app/site/hosting/restlet.nl?script=<id>&deploy=1
 */
define(['N/https', 'N/log', 'N/runtime'], function (https, log, runtime) {

  var CERTYO_API_URL = 'https://www.certyos.com/api/v1/records';

  /**
   * POST handler — accepts a fulfillment payload and sends it to Certyo.
   * @param {Object} requestBody - The fulfillment data from the caller
   * @returns {Object} Certyo API response or error details
   */
  function post(requestBody) {
    var scriptObj = runtime.getCurrentScript();
    var apiKey = scriptObj.getParameter({ name: 'custscript_certyo_api_key' });

    if (!apiKey) {
      log.error('Configuration Error', 'custscript_certyo_api_key parameter is not set');
      return { success: false, error: 'API key not configured' };
    }

    try {
      var certRecord = {
        tenantId:        requestBody.subsidiary || 'default',
        database:        'netsuite',
        collection:      'item_fulfillments',
        recordId:        requestBody.tranId,
        recordVersion:   String(requestBody.version || '1'),
        operationType:   requestBody.operationType || 'insert',
        recordPayload: {
          tranId:          requestBody.tranId,
          entity:          requestBody.entity,
          entityName:      requestBody.entityName,
          tranDate:        requestBody.tranDate,
          shipMethod:      requestBody.shipMethod,
          trackingNumbers: requestBody.trackingNumbers || [],
          items:           requestBody.items || [],
          memo:            requestBody.memo || '',
          location:        requestBody.location,
          createdFrom:     requestBody.createdFrom
        },
        sourceTimestamp:  requestBody.tranDate
          ? new Date(requestBody.tranDate).toISOString()
          : new Date().toISOString(),
        idempotencyKey:   requestBody.tranId + '-v' +
                          (requestBody.version || '1') + '-' +
                          new Date().toISOString().substring(0, 10)
      };

      log.audit('Certyo Ingestion', 'Sending record: ' + certRecord.recordId);

      var response = https.post({
        url: CERTYO_API_URL,
        headers: {
          'X-API-Key':    apiKey,
          'Content-Type': 'application/json',
          'Accept':       'application/json'
        },
        body: JSON.stringify(certRecord)
      });

      var statusCode = response.code;
      var responseBody = JSON.parse(response.body);

      if (statusCode === 202) {
        log.audit('Certyo Ingestion Success', JSON.stringify({
          recordId:   responseBody.recordId,
          recordHash: responseBody.recordHash,
          acceptedAt: responseBody.acceptedAt
        }));
        return {
          success:    true,
          recordHash: responseBody.recordHash,
          acceptedAt: responseBody.acceptedAt
        };
      } else {
        log.error('Certyo Ingestion Failed', 'HTTP ' + statusCode +
                  ': ' + response.body);
        return {
          success:    false,
          statusCode: statusCode,
          error:      responseBody.message || response.body
        };
      }

    } catch (e) {
      log.error('Certyo Ingestion Error', e.message + '\n' + e.stack);
      return { success: false, error: e.message };
    }
  }

  return { post: post };
});

/**
 * @NApiVersion 2.1
 * @NScriptType UserEventScript
 * @NModuleScope SameAccount
 *
 * Certyo User Event — Item Fulfillment
 * Triggers on afterSubmit to ingest fulfillment records into Certyo.
 *
 * Deploy: Setup > Scripting > Script Deployments
 *   Record Type: Item Fulfillment
 *   Event Type:  After Submit
 *   Status:      Released
 */
define(['N/https', 'N/record', 'N/log', 'N/runtime', 'N/search'],
  function (https, record, log, runtime, search) {

  var CERTYO_API_URL = 'https://www.certyos.com/api/v1/records';

  /**
   * afterSubmit — runs after the Item Fulfillment is committed to the DB.
   * @param {Object} context
   * @param {string} context.type - create | edit | delete | xedit
   * @param {Record} context.newRecord - The saved record
   */
  function afterSubmit(context) {
    // Only fire on create or edit
    if (context.type !== context.UserEventType.CREATE &&
        context.type !== context.UserEventType.EDIT) {
      return;
    }

    var scriptObj = runtime.getCurrentScript();
    var apiKey = scriptObj.getParameter({ name: 'custscript_certyo_api_key' });

    if (!apiKey) {
      log.error('Certyo Config', 'API key parameter not set — skipping ingestion');
      return;
    }

    try {
      var rec = context.newRecord;
      var fulfillmentId = rec.getValue({ fieldId: 'tranid' });

      // Extract subsidiary name for tenant mapping
      var subsidiaryId = rec.getValue({ fieldId: 'subsidiary' });
      var subsidiaryName = 'default';
      if (subsidiaryId) {
        var subLookup = search.lookupFields({
          type: search.Type.SUBSIDIARY,
          id: subsidiaryId,
          columns: ['name']
        });
        subsidiaryName = subLookup.name || 'default';
      }

      // Gather line items
      var lineCount = rec.getLineCount({ sublistId: 'item' });
      var items = [];
      for (var i = 0; i < lineCount; i++) {
        items.push({
          itemId:        rec.getSublistValue({ sublistId: 'item', fieldId: 'item', line: i }),
          itemName:      rec.getSublistText({ sublistId: 'item', fieldId: 'item', line: i }),
          quantity:      rec.getSublistValue({ sublistId: 'item', fieldId: 'quantity', line: i }),
          serialNumbers: rec.getSublistValue({ sublistId: 'item', fieldId: 'serialnumbers', line: i }) || '',
          lotNumbers:    rec.getSublistValue({ sublistId: 'item', fieldId: 'binitem', line: i }) || '',
          location:      rec.getSublistText({ sublistId: 'item', fieldId: 'location', line: i })
        });
      }

      var certRecord = {
        tenantId:       subsidiaryName.toLowerCase().replace(/\s+/g, '-'),
        database:       'netsuite',
        collection:     'item_fulfillments',
        recordId:       fulfillmentId,
        recordVersion:  context.type === context.UserEventType.CREATE ? '1' : '2',
        operationType:  context.type === context.UserEventType.CREATE ? 'insert' : 'update',
        recordPayload: {
          fulfillmentId: fulfillmentId,
          entity:        rec.getValue({ fieldId: 'entity' }),
          entityName:    rec.getText({ fieldId: 'entity' }),
          tranDate:      rec.getValue({ fieldId: 'trandate' }),
          status:        rec.getValue({ fieldId: 'shipstatus' }),
          shipMethod:    rec.getText({ fieldId: 'shipmethod' }),
          trackingNumbers: (rec.getValue({ fieldId: 'linkedtrackingnumbers' }) || '').split(','),
          createdFrom:   rec.getValue({ fieldId: 'createdfrom' }),
          memo:          rec.getValue({ fieldId: 'memo' }) || '',
          items:         items
        },
        sourceTimestamp: new Date(rec.getValue({ fieldId: 'trandate' })).toISOString(),
        idempotencyKey:  fulfillmentId + '-' +
                         (context.type === context.UserEventType.CREATE ? 'v1' : 'v2') + '-' +
                         new Date().toISOString().substring(0, 10)
      };

      log.audit('Certyo Ingestion', 'Sending ' + fulfillmentId + ' (' + items.length + ' lines)');

      var response = https.post({
        url: CERTYO_API_URL,
        headers: {
          'X-API-Key':    apiKey,
          'Content-Type': 'application/json',
          'Accept':       'application/json'
        },
        body: JSON.stringify(certRecord)
      });

      if (response.code === 202) {
        var body = JSON.parse(response.body);
        log.audit('Certyo Accepted', JSON.stringify({
          recordId:   body.recordId,
          recordHash: body.recordHash
        }));

        // Store the record hash on a custom field for later verification
        record.submitFields({
          type: record.Type.ITEM_FULFILLMENT,
          id:   rec.id,
          values: {
            'custbody_certyo_record_hash': body.recordHash,
            'custbody_certyo_status':      'Pending'
          },
          options: { enableSourcing: false, ignoreMandatoryFields: true }
        });
      } else {
        log.error('Certyo Error', 'HTTP ' + response.code + ': ' + response.body);
      }

    } catch (e) {
      // Log but don't throw — we don't want to block the fulfillment save
      log.error('Certyo Integration Error', e.message + '\n' + e.stack);
    }
  }

  return { afterSubmit: afterSubmit };
});

"""
NetSuite -> Certyo batch synchronization script.

Uses NetSuite SuiteQL (REST API) to query recent Item Fulfillments,
then pushes them to Certyo's bulk ingestion endpoint.

Requirements:
    pip install requests requests_oauthlib

Environment variables:
    NETSUITE_ACCOUNT_ID    - e.g. "1234567" or "1234567_SB1" for sandbox
    NETSUITE_CONSUMER_KEY
    NETSUITE_CONSUMER_SECRET
    NETSUITE_TOKEN_ID
    NETSUITE_TOKEN_SECRET
    CERTYO_API_KEY
    CERTYO_TENANT_ID
"""

import os
import json
import time
import logging
from datetime import datetime, timedelta, timezone

import requests
from requests_oauthlib import OAuth1

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

# --- NetSuite configuration ---
NS_ACCOUNT_ID = os.environ["NETSUITE_ACCOUNT_ID"]
NS_BASE_URL = f"https://{NS_ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest"

ns_auth = OAuth1(
    client_key=os.environ["NETSUITE_CONSUMER_KEY"],
    client_secret=os.environ["NETSUITE_CONSUMER_SECRET"],
    resource_owner_key=os.environ["NETSUITE_TOKEN_ID"],
    resource_owner_secret=os.environ["NETSUITE_TOKEN_SECRET"],
    realm=NS_ACCOUNT_ID,
    signature_method="HMAC-SHA256",
)

# --- Certyo configuration ---
CERTYO_API_URL = "https://www.certyos.com/api/v1/records/bulk"
CERTYO_API_KEY = os.environ["CERTYO_API_KEY"]
CERTYO_TENANT_ID = os.environ["CERTYO_TENANT_ID"]


def query_netsuite_fulfillments(since_hours: int = 24) -> list[dict]:
    """Query recent Item Fulfillments via SuiteQL."""
    since = (datetime.now(timezone.utc) - timedelta(hours=since_hours)).strftime(
        "%m/%d/%Y %H:%M:%S"
    )

    suiteql = f"""
        SELECT
            t.tranId,
            t.tranDate,
            t.entity,
            BUILTIN.DF(t.entity)        AS entityName,
            BUILTIN.DF(t.shipMethod)     AS shipMethodName,
            t.linkedTrackingNumbers,
            t.memo,
            t.lastModifiedDate,
            tl.item,
            BUILTIN.DF(tl.item)          AS itemName,
            tl.quantity,
            tl.serialNumbers
        FROM
            transaction t
            JOIN transactionLine tl ON t.id = tl.transaction
        WHERE
            t.type = 'ItemShip'
            AND t.lastModifiedDate >= TO_DATE('{since}', 'MM/DD/YYYY HH24:MI:SS')
        ORDER BY
            t.tranId, tl.lineSequenceNumber
    """

    logger.info("Querying NetSuite fulfillments since %s", since)

    response = requests.post(
        f"{NS_BASE_URL}/query/v1/suiteql",
        auth=ns_auth,
        headers={
            "Content-Type": "application/json",
            "Prefer": "transient",
        },
        json={"q": suiteql.strip()},
    )
    response.raise_for_status()

    data = response.json()
    items = data.get("items", [])
    logger.info("Retrieved %d fulfillment lines from NetSuite", len(items))
    return items


def group_by_fulfillment(lines: list[dict]) -> dict[str, dict]:
    """Group line items by fulfillment tranId."""
    fulfillments: dict[str, dict] = {}

    for line in lines:
        tran_id = line["tranId"]
        if tran_id not in fulfillments:
            fulfillments[tran_id] = {
                "tranId": tran_id,
                "tranDate": line["tranDate"],
                "entity": line["entity"],
                "entityName": line["entityName"],
                "shipMethod": line.get("shipMethodName", ""),
                "trackingNumbers": line.get("linkedTrackingNumbers", ""),
                "memo": line.get("memo", ""),
                "lastModified": line["lastModifiedDate"],
                "items": [],
            }

        fulfillments[tran_id]["items"].append(
            {
                "itemId": str(line["item"]),
                "itemName": line["itemName"],
                "quantity": float(line["quantity"]),
                "serialNumbers": line.get("serialNumbers", ""),
            }
        )

    return fulfillments


def build_certyo_records(fulfillments: dict[str, dict]) -> list[dict]:
    """Transform NetSuite fulfillments into Certyo record format."""
    records = []
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    for tran_id, ful in fulfillments.items():
        records.append(
            {
                "database": "netsuite",
                "collection": "item_fulfillments",
                "recordId": tran_id,
                "recordVersion": "1",
                "operationType": "upsert",
                "recordPayload": {
                    "tranId": ful["tranId"],
                    "entity": ful["entity"],
                    "entityName": ful["entityName"],
                    "tranDate": ful["tranDate"],
                    "shipMethod": ful["shipMethod"],
                    "trackingNumbers": ful["trackingNumbers"],
                    "memo": ful["memo"],
                    "items": ful["items"],
                },
                "sourceTimestamp": ful["lastModified"],
                "idempotencyKey": f"{tran_id}-sync-{today}",
            }
        )

    return records


def ingest_to_certyo(records: list[dict]) -> None:
    """Send records to Certyo in batches of up to 1000."""
    batch_size = 1000

    for i in range(0, len(records), batch_size):
        batch = records[i : i + batch_size]
        logger.info(
            "Sending batch %d-%d (%d records) to Certyo",
            i + 1,
            i + len(batch),
            len(batch),
        )

        response = requests.post(
            CERTYO_API_URL,
            headers={
                "X-API-Key": CERTYO_API_KEY,
                "Content-Type": "application/json",
            },
            json={
                "tenantId": CERTYO_TENANT_ID,
                "records": batch,
            },
        )

        if response.status_code == 202:
            result = response.json()
            logger.info(
                "Batch accepted: %d records ingested", len(batch)
            )
        else:
            logger.error(
                "Batch failed: HTTP %d — %s", response.status_code, response.text
            )
            response.raise_for_status()

        # Respect rate limits between batches
        if i + batch_size < len(records):
            time.sleep(1)


def main():
    lines = query_netsuite_fulfillments(since_hours=24)
    if not lines:
        logger.info("No new fulfillments found — nothing to sync")
        return

    fulfillments = group_by_fulfillment(lines)
    logger.info("Grouped into %d unique fulfillments", len(fulfillments))

    records = build_certyo_records(fulfillments)
    ingest_to_certyo(records)
    logger.info("Sync complete: %d records sent to Certyo", len(records))


if __name__ == "__main__":
    main()