[PAN-2508] ShipBob <> Gorgias Sample Integration in Python

SHIPBOB_TO_GORGIAS/Python/.gitignore

@@ -0,0 +1,6 @@
+# Byte-compiled and cached files
+__pycache__/
+*.py[cod]
+
+# Local virtual environments
+.venv/

SHIPBOB_TO_GORGIAS/Python/PANDIUM.yaml

@@ -0,0 +1,46 @@
+version: 1.0
+base: python:3.14 # node, ruby, java, php, .net, go, kotlin, rust
+build: pipenv install # modify for language
+run: pipenv run python -m sb2gorgias # modify for language
+configs:
+  schema:
+    name: pandium_configs
+    type: object
+    properties:
+      # Fallback cursor used on a tenant's first run, before any metadata exists.
+      order_start_date:
+        type: string
+        format: date
+      # Display newest order at the top of the customer sidebar list.
+      newest_order_first:
+        type: boolean
+        default: false
+  uischema:
+    type: VerticalLayout
+    elements:
+      - label: Order Cutoff Date
+        scope: '#/properties/order_start_date'
+        type: Control
+      - label: Display Newest Order at Top
+        scope: '#/properties/newest_order_first'
+        type: Control
+# Shared across both flows. The platform validates a run's stdout against this
+# schema and shallow-merges it (top-level) into tenant metadata.
+metadata_schema:
+  schema:
+    name: metadata_schema
+    type: object
+    properties:
+      # Flow A (cron) — the sync cursor: high-water marks for new / updated orders.
+      new_order_start_date:
+        type: string
+        format: date-time
+      updated_order_start_date:
+        type: string
+        format: date-time
+      # Flow B (webhook) — dedupe map of shipment_id -> ISO time the ticket was created.
+      processed_shipments:
+        type: object
+        additionalProperties:
+          type: string
+          format: date-time

SHIPBOB_TO_GORGIAS/Python/Pipfile

@@ -0,0 +1,15 @@
+[[source]]
+url = "https://pypi.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+coloredlogs = "*"
+requests = "*"
+
+[dev-packages]
+ipdb = "*"
+pytest = "*"
+
+[requires]
+python_version = "3.14"

SHIPBOB_TO_GORGIAS/Python/pytest.ini

@@ -0,0 +1,5 @@
+[pytest]
+# Put the project root on sys.path so tests can `import sb2gorgias`.
+pythonpath = .
+testpaths = tests
+addopts = -q

SHIPBOB_TO_GORGIAS/Python/sb2gorgias/__main__.py

@@ -0,0 +1,51 @@
+import json
+import logging
+import sys
+from typing import Any
+
+import coloredlogs
+
+from . import flow_a, flow_b
+from .lib import Pandium
+
+# Logs go to stderr; stdout is reserved for the JSON metadata Pandium reads back.
+# Install on the root logger so every sb2gorgias module's logs surface, then
+# quiet urllib3's per-request connection chatter.
+coloredlogs.install(level='DEBUG', stream=sys.stderr)
+logging.getLogger('urllib3').setLevel(logging.WARNING)
+logger = logging.getLogger(__name__)
+
+
+def run(mode: str, pandium: Pandium) -> dict[str, Any]:
+    """Dispatch on run mode. Each branch returns the metadata dict to print to
+    stdout, which the platform validates against ``metadata_schema`` and
+    shallow-merges into tenant metadata."""
+    match mode:
+        case 'init':
+            # No dynamic config to resolve; just report which secrets are wired up.
+            logger.info('Available secrets: %s', ', '.join(pandium.secrets) or '(none)')
+            return {}
+
+        case 'webhook':
+            # Flow B — ShipBob shipment_delivered -> Gorgias ticket.
+            return flow_b.run(pandium)
+
+        case _:
+            # Flow A — scheduled ShipBob orders -> Gorgias customer sidebar.
+            return flow_a.run(pandium)
+
+
+def main() -> None:
+    pandium = Pandium.from_env()
+    mode = pandium.run_mode or ''
+
+    logger.info('sb2gorgias starting — run mode: %s', mode or '(cron)')
+
+    std_out = run(mode, pandium)
+    # Flow A's timeout handler prints and exits on its own; on every normal path
+    # the metadata is printed here as the last line of stdout.
+    print(json.dumps(std_out))
+
+
+if __name__ == '__main__':
+    main()

SHIPBOB_TO_GORGIAS/Python/sb2gorgias/flow_a.py

@@ -0,0 +1,162 @@
+"""Flow A — CRON: ShipBob orders -> Gorgias customer sidebar.
+
+Keeps each Gorgias customer's ``data.pandium.shipbob_orders`` in sync with that
+customer's recent ShipBob orders. Runs on a schedule and resumes where the last
+run left off, using tenant metadata as the cursor.
+
+The run is bounded at ~10 minutes by Pandium. To stay resumable, the loop keeps a
+single in-memory *timeout record* (the cursor) current as each order is
+processed, and a SIGALRM handler flushes that record to stdout before the hard
+kill. Exiting 0 on timeout means the partial cursor is merged into metadata and
+the next run picks up from there.
+"""
+
+import json
+import logging
+import signal
+import sys
+from datetime import datetime, timedelta
+
+from .gorgias import GorgiasAPI, _get
+from .shipbob import ShipBobAPI
+
+logger = logging.getLogger(__name__)
+
+ALARM_SECONDS = 540  # self-imposed 9-min alarm, ahead of Pandium's ~10-min kill
+ONE_MONTH = timedelta(days=30)
+MAX_ORDERS_TO_SYNC = 10  # most recent N orders kept on each customer
+
+
+def clamp(value, now: datetime) -> datetime:
+    """Keep a cursor within [now - 1 month, now]. Unparseable/missing values fall
+    back to one month ago (the oldest window we ever fetch)."""
+    floor = now - ONE_MONTH
+    try:
+        dt = value if isinstance(value, datetime) else datetime.fromisoformat(str(value))
+    except (TypeError, ValueError):
+        return floor
+    if dt.tzinfo is not None:
+        dt = dt.replace(tzinfo=None)  # compare naive-to-naive
+    return min(max(dt, floor), now)
+
+
+def _upsert(orders: list, order_payload: dict, newest_first: bool) -> list:
+    """Merge ``order_payload`` into a customer's order list (replace by id, else
+    append), then sort and trim to the most recent ``MAX_ORDERS_TO_SYNC``."""
+    for i, existing in enumerate(orders):
+        if existing.get('id') == order_payload['id']:
+            orders[i] = order_payload
+            return orders  # in-place replace; no re-sort/trim needed
+
+    orders.append(order_payload)
+    orders.sort(key=lambda o: o.get('id', 0), reverse=newest_first)
+    if len(orders) > MAX_ORDERS_TO_SYNC:
+        orders = orders[:MAX_ORDERS_TO_SYNC] if newest_first else orders[-MAX_ORDERS_TO_SYNC:]
+    return orders
+
+
+def process_order(sb_order: dict, gorgias: GorgiasAPI, cache: dict, newest_first: bool) -> None:
+    """Find-or-create the order's Gorgias customer, then PUT/POST its updated
+    ``data.pandium.shipbob_orders``. ``cache`` accumulates customer payloads
+    within a run so multiple orders for one customer batch onto the same record.
+    """
+    key = gorgias.customer_key(sb_order)
+    email = gorgias.valid_email(_get(sb_order, 'recipient.email', ''))
+
+    if key not in cache:
+        try:
+            existing = gorgias.find_customer(
+                email=email or None, external_id=None if email else key
+            )
+        except Exception as err:
+            logger.error('Skipping order %s — cannot fetch customer %s: %s',
+                         _get(sb_order, 'id', ''), key, err)
+            return
+
+        if existing:
+            data = existing.get('data') or {}
+            data.setdefault('pandium', {})
+            if not isinstance(data['pandium'].get('shipbob_orders'), list):
+                data['pandium']['shipbob_orders'] = []
+            cache[key] = {'id': existing['id'], 'data': data}
+        else:
+            cache[key] = gorgias.new_customer_payload(sb_order, key)
+
+    customer = cache[key]
+    customer['data']['pandium']['shipbob_orders'] = _upsert(
+        customer['data']['pandium']['shipbob_orders'],
+        gorgias.order_data_payload(sb_order),
+        newest_first,
+    )
+
+    try:
+        if 'id' in customer:
+            gorgias.update_customer(customer['id'], customer)
+        else:
+            customer['id'] = gorgias.create_customer(customer)
+    except Exception as err:
+        logger.error('Failed to upsert Gorgias customer %s: %s', key, err)
+
+
+def run(pandium) -> dict:
+    now = datetime.now()
+    metadata = pandium.metadata or {}
+    fallback = pandium.config.get('order_start_date')
+
+    new_cursor = clamp(metadata.get('new_order_start_date') or fallback, now)
+    updated_cursor = clamp(metadata.get('updated_order_start_date') or fallback, now)
+
+    # The timeout record: the cursor we flush on either outcome. Values are ISO
+    # strings advanced as orders are processed.
+    record = {
+        'new_order_start_date': new_cursor.isoformat(),
+        'updated_order_start_date': updated_cursor.isoformat(),
+    }
+
+    def flush():
+        print(json.dumps(record))
+
+    def on_alarm(signum, frame):
+        logger.warning('Approaching the run-time limit — flushing cursor for the next run.')
+        flush()
+        sys.exit(0)  # timed-out run still counts as successful → partial cursor merged
+
+    signal.signal(signal.SIGALRM, on_alarm)
+    signal.alarm(ALARM_SECONDS)
+
+    shipbob = ShipBobAPI(pandium)
+    gorgias = GorgiasAPI(pandium)
+    newest_first = str(pandium.config.get('newest_order_first', '')).lower() == 'true'
+    cache: dict = {}
+
+    # New orders: SortOrder=Oldest, so created_date advances forward monotonically.
+    logger.info('Syncing new ShipBob orders since %s', record['new_order_start_date'])
+    page = 1
+    while True:
+        orders = shipbob.get_new_orders_page(new_cursor, page)
+        if not orders:
+            break
+        for order in orders:
+            process_order(order, gorgias, cache, newest_first)
+            created = order.get('created_date')
+            if created:
+                # created_date is YYYY-MM-DDThh:mm:ss.sssssss+00:00; trim to 26
+                # chars for a valid (naive, microsecond) date-time.
+                record['new_order_start_date'] = created[:26]
+        page += 1
+
+    # Updated orders: keyed off shipment last_update_at (see get_updated_orders_page).
+    logger.info('Syncing updated ShipBob orders since %s', record['updated_order_start_date'])
+    page = 1
+    while True:
+        orders = shipbob.get_updated_orders_page(updated_cursor, page)
+        if not orders:
+            break
+        for order in orders:
+            process_order(order, gorgias, cache, newest_first)
+            # last_update_at is YYYY-MM-DDThh:mm:ss.sss+00:00; trim to 23 chars.
+            record['updated_order_start_date'] = shipbob.get_update_date(order, updated_cursor)[:23]
+        page += 1
+
+    signal.alarm(0)  # made it — cancel the alarm; __main__ prints the record on return
+    return record