magento2_get_order / magento2_find_orders

magento2_get_order(context: dict, order_number: str) -> dict | None
magento2_find_orders(context: dict, email: str, limit: int = 10) -> list[dict]

def execute_action(context):
    order = magento2_get_order(context, context["args"]["orderId"])
    if order is None:
        return {"success": False, "message": "The order could not be found."}
    return {"success": True, "orders": [order]}

import base64
import hashlib
import hmac
import re
import secrets
import time
from urllib.parse import quote

import requests

# Never returned: the customer's IP address and card details on the payment.
_MAGENTO2_ORDER_HIDDEN = {"remote_ip", "x_forwarded_for"}
_MAGENTO2_PAYMENT_HIDDEN = {
    "cc_last4", "cc_exp_month", "cc_exp_year", "cc_owner", "cc_type",
    "cc_trans_id", "last_trans_id", "additional_information",
}


def _magento2_clean_order(order):
    order = {k: v for k, v in order.items() if k not in _MAGENTO2_ORDER_HIDDEN}
    if isinstance(order.get("payment"), dict):
        order["payment"] = {k: v for k, v in order["payment"].items() if k not in _MAGENTO2_PAYMENT_HIDDEN}
    return order


def _magento2_store(context):
    """The current business's Magento connection, as get_integration_credentials returns it."""
    business = context.get("business") or {}
    slug = business.get("slug") or (context.get("conversation") or {}).get("businessSlug")
    if not slug:
        raise ValueError("The business is not available in the context")
    store = (get_integration_credentials(context).get(slug) or {}).get("magento2")
    if not store:
        raise ValueError("Magento is not connected for this business")
    return store


def _magento2_get(store, resource, filters, extra_params=None):
    """GET /rest/V1/<resource> with AND-ed searchCriteria filters, signed like the connector."""
    url = store["storeUrl"].rstrip("/") + "/rest/V1/" + resource
    if store.get("storeId") is not None:
        filters = [("store_id", store["storeId"])] + filters
    params = {}
    for index, (field, value) in enumerate(filters):
        prefix = f"searchCriteria[filterGroups][{index}][filters][0]"
        params[prefix + "[field]"] = field
        params[prefix + "[value]"] = str(value)
        params[prefix + "[conditionType]"] = "eq"
    params.update(extra_params or {})
    headers = {"Accept": "application/json", "User-Agent": "mozilla/5.0"}
    if store.get("useAccessTokenAsBearerToken"):
        headers["Authorization"] = "Bearer " + store["accessToken"]
    else:
        oauth = {
            "oauth_consumer_key": store["consumerKey"],
            "oauth_token": store["accessToken"],
            "oauth_nonce": secrets.token_hex(16),
            "oauth_timestamp": str(int(time.time())),
            "oauth_signature_method": "HMAC-SHA256",
            "oauth_version": "1.0",
        }
        encode = lambda value: quote(str(value), safe="~")
        pairs = sorted((encode(k), encode(v)) for k, v in list(params.items()) + list(oauth.items()))
        base = "GET&" + encode(url) + "&" + encode("&".join(k + "=" + v for k, v in pairs))
        key = encode(store["consumerSecret"]) + "&" + encode(store["accessTokenSecret"])
        oauth["oauth_signature"] = base64.b64encode(
            hmac.new(key.encode(), base.encode(), hashlib.sha256).digest()
        ).decode()
        headers["Authorization"] = "OAuth " + ", ".join(
            encode(k) + '="' + encode(v) + '"' for k, v in sorted(oauth.items())
        )
    # Most stores allowlist Octocom's proxy IP; skipProxy stores are called directly.
    send = requests.request if store.get("skipProxy") else proxy_request
    response = send("GET", url, params=params, headers=headers, timeout=20, allow_redirects=False)
    response.raise_for_status()
    return response.json().get("items") or []


def _magento2_with_shipments(store, order):
    # Tracking numbers live on the order's shipments, not on the order.
    order["shipments"] = _magento2_get(store, "shipments", [("order_id", order["entity_id"])])
    return _magento2_clean_order(order)


def magento2_get_order(context, order_number):
    """The Magento order with this increment ID (plus its shipments), or None."""
    store = _magento2_store(context)
    number = re.sub(r"\s+", "", str(order_number)).lstrip("#")
    if not number:
        return None
    candidates = [number]
    # Customers often drop the leading zeros of Magento's 9-digit numbers.
    if number.isdigit() and len(number) < 9:
        candidates.append(number.zfill(9))
    for candidate in candidates:
        orders = _magento2_get(store, "orders", [("increment_id", candidate)])
        if orders:
            return _magento2_with_shipments(store, orders[0])
    return None


def magento2_find_orders(context, email, limit=10):
    """The customer's Magento orders by email, newest first."""
    store = _magento2_store(context)
    orders = _magento2_get(store, "orders", [("customer_email", email.strip())], {
        "searchCriteria[pageSize]": str(limit),
        "searchCriteria[currentPage]": "1",
        "searchCriteria[sortOrders][0][field]": "created_at",
        "searchCriteria[sortOrders][0][direction]": "DESC",
    })
    return [_magento2_with_shipments(store, order) for order in orders[:limit]]


def _magento2_widget_order(order):
    return {
        "id": order.get("entity_id"), "number": order.get("increment_id"), "status": order.get("status"),
        "email": order.get("customer_email"), "total": order.get("grand_total"),
        "currency": order.get("order_currency_code"), "placed_at": order.get("created_at"),
        "products": [{"name": p.get("name"), "quantity": p.get("qty_ordered"), "sku": p.get("sku"), "total": p.get("row_total_incl_tax")}
                     for p in order.get("items") or [] if not p.get("parent_item_id")],
        "shipments": [{"carrier": t.get("title") or t.get("carrier_code"), "tracking_number": t.get("track_number")}
                      for shipment in order.get("shipments") or [] for t in shipment.get("tracks") or []],
    }


def magento2_order_widget(context):
    """Read-only sidebar orders from the saved order ID and customer's email."""
    return order_widget(context, get_order=magento2_get_order, find_orders=magento2_find_orders, map_order=_magento2_widget_order)