shopify_get_order / shopify_find_orders

shopify_get_order(context: dict, order_number: str, *, fields: str = SHOPIFY_ORDER_FIELDS) -> dict | None
shopify_find_orders(context: dict, email: str, limit: int = 10, *, fields: str = SHOPIFY_ORDER_FIELDS) -> list[dict]
shopify_order_stores(context: dict) -> list[str]
shopify_order_admin_url(order: dict) -> str | None

def execute_action(context):
    order = shopify_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]}

order = shopify_get_order(
    context,
    context["args"]["orderId"],
    fields=SHOPIFY_ORDER_FIELDS + " customAttributes { key value } metafield(namespace: \"custom\", key: \"gift\") { value }",
)

import re

SHOPIFY_ORDERS_API_VERSION = "2026-10"

# The order fields shopify_get_order and shopify_find_orders return, in
# Shopify's Admin GraphQL shape. Pass your own selection as fields= to read more
# or less: https://shopify.dev/docs/api/admin-graphql/2026-10/objects/Order
#
# Left out on purpose, because a support bot should never see them: the
# customer's IP and browser, checkout and cart tokens, the order status page
# URL (its key opens the order), marketing attribution and payment details.
SHOPIFY_ORDER_FIELDS = """
id
legacyResourceId
name
email
phone
createdAt
processedAt
cancelledAt
cancelReason
closedAt
displayFinancialStatus
displayFulfillmentStatus
returnStatus
note
tags
discountCodes
currencyCode
presentmentCurrencyCode
currentTotalPriceSet { shopMoney { amount currencyCode } presentmentMoney { amount currencyCode } }
currentSubtotalPriceSet { shopMoney { amount currencyCode } presentmentMoney { amount currencyCode } }
totalShippingPriceSet { shopMoney { amount currencyCode } presentmentMoney { amount currencyCode } }
currentTotalDiscountsSet { shopMoney { amount currencyCode } presentmentMoney { amount currencyCode } }
totalRefundedSet { shopMoney { amount currencyCode } presentmentMoney { amount currencyCode } }
customer { id displayName defaultEmailAddress { emailAddress } }
shippingAddress { name company address1 address2 city province zip country countryCodeV2 phone }
shippingLines(first: 5) { nodes { title } }
lineItems(first: 50) {
  nodes { id name title variantTitle sku quantity currentQuantity unfulfilledQuantity }
}
fulfillments(first: 10) {
  id name status displayStatus createdAt inTransitAt estimatedDeliveryAt deliveredAt
  trackingInfo(first: 5) { company number url }
  fulfillmentLineItems(first: 20) { nodes { quantity lineItem { id } } }
}
refunds(first: 10) {
  createdAt note
  totalRefundedSet { shopMoney { amount currencyCode } presentmentMoney { amount currencyCode } }
}
"""


def shopify_order_stores(context):
    """
    The myshopify.com domains of the current business's Shopify stores, oldest first.

    Raises:
        ValueError: If the business is not in the context or has no Shopify store.
    """
    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")
    shopify = (get_integration_credentials(context).get(slug) or {}).get("shopify")
    if not shopify:
        raise ValueError("Shopify is not connected for this business")
    return [store["domain"] for store in shopify.get("stores") or [shopify]]


def _shopify_orders_query(context, shop, query, variables):
    body = shopify_graphql(context, query, variables, api_version=SHOPIFY_ORDERS_API_VERSION, shop=shop)
    if body.get("errors"):
        raise ValueError(f"Shopify order lookup failed on {shop}: {body['errors']}")
    return body.get("data") or {}


def _shopify_search_value(value):
    """Quote a value for Shopify's search syntax, e.g. email:"jane@example.com"."""
    return '"' + str(value).replace("\\", "\\\\").replace('"', '\\"') + '"'


def shopify_get_order(context, order_number, *, fields=SHOPIFY_ORDER_FIELDS):
    """
    Look up a Shopify order by its order number, e.g. "#1001" or "1001".

    Uses Shopify's order search, so a number typed without the store's name
    prefix still matches ("1001" finds "#HP1001"). Searches every Shopify store
    connected to the current business, in the order they were connected, and
    returns the first match.

    Args:
        context: The context object.
        order_number: The order number as the customer gave it. Spaces,
            hyphens and a leading "#" are ignored; letters are upper-cased.
        fields: The Order fields to return, as a GraphQL selection.
            Defaults to SHOPIFY_ORDER_FIELDS; pass your own string to read
            other fields.

    Returns:
        The order in Shopify's Admin GraphQL shape (camelCase fields,
        connections as {"nodes": [...]}), plus "shop": the myshopify.com
        domain of the store it came from. None if no store has the order.
        Field reference:
        https://shopify.dev/docs/api/admin-graphql/2026-10/objects/Order

    Raises:
        ValueError: If Shopify is not connected for the business, or Shopify
            reports a GraphQL error.
        requests.HTTPError: If Shopify rejects the request.
    """
    number = re.sub(r"[\s-]+", "", str(order_number)).lstrip("#").upper()
    if not number:
        return None
    query = "query($q: String!) { orders(first: 1, query: $q) { nodes { %s } } }" % fields
    for shop in shopify_order_stores(context):
        data = _shopify_orders_query(context, shop, query, {"q": f"name:#{number}"})
        nodes = (data.get("orders") or {}).get("nodes") or []
        if nodes:
            return {**nodes[0], "shop": shop}
    return None


def shopify_find_orders(context, email, limit=10, *, fields=SHOPIFY_ORDER_FIELDS):
    """
    Find the most recent Shopify orders placed with an email address.

    Searches every Shopify store connected to the current business and
    returns only orders whose email, or whose customer's email, is exactly
    this address (ignoring case).

    Args:
        context: The context object.
        email: The customer's email address.
        limit: Maximum number of orders to return (newest first). Default 10.
        fields: The Order fields to return, as a GraphQL selection.
            Defaults to SHOPIFY_ORDER_FIELDS.

    Returns:
        A list of orders shaped like shopify_get_order's, newest first, each
        with "shop". Empty if there are none.

    Raises:
        ValueError: If Shopify is not connected for the business, or Shopify
            reports a GraphQL error.
        requests.HTTPError: If Shopify rejects the request.
    """
    email = str(email or "").strip()
    limit = int(limit)
    if not email or limit < 1:
        return []
    # Shopify's search can match more loosely than the address, so keep exact
    # matches only. The aliased fields carry what that needs and are removed,
    # whatever `fields` selects.
    query = (
        "query($q: String!, $first: Int!) { orders(first: $first, query: $q, sortKey: CREATED_AT, reverse: true) "
        "{ nodes { %s _email: email _createdAt: createdAt _customer: customer { defaultEmailAddress { emailAddress } } } } }"
    ) % fields
    found = []
    for shop in shopify_order_stores(context):
        data = _shopify_orders_query(context, shop, query, {"q": f"email:{_shopify_search_value(email)}", "first": limit})
        for node in (data.get("orders") or {}).get("nodes") or []:
            order_email = node.pop("_email", None) or ""
            created_at = node.pop("_createdAt", None) or ""
            customer = node.pop("_customer", None) or {}
            customer_email = (customer.get("defaultEmailAddress") or {}).get("emailAddress") or ""
            if email.lower() in {order_email.lower(), customer_email.lower()}:
                found.append((created_at, {**node, "shop": shop}))
    found.sort(key=lambda match: match[0], reverse=True)
    return [order for _, order in found[:limit]]


def shopify_order_admin_url(order):
    """The order's page in the Shopify admin, or None without "shop" and "legacyResourceId"."""
    shop, order_id = order.get("shop"), order.get("legacyResourceId")
    if not shop or not order_id:
        return None
    return f"https://admin.shopify.com/store/{shop.split('.')[0]}/orders/{order_id}"


def _shopify_display_status(value):
    return str(value).replace("_", " ").lower() if value else None


def _shopify_widget_order(order):
    money = (order.get("currentTotalPriceSet") or {}).get("shopMoney") or {}
    return {
        "id": order.get("id"), "number": order.get("name"), "url": shopify_order_admin_url(order),
        "status": "cancelled" if order.get("cancelledAt") else _shopify_display_status(
            order.get("displayFulfillmentStatus") or order.get("displayFinancialStatus")),
        "email": order.get("email"), "total": money.get("amount"),
        "currency": money.get("currencyCode"), "placed_at": order.get("createdAt"),
        "products": [{"name": p.get("name") or p.get("title"), "sku": p.get("sku"),
                      "quantity": p["currentQuantity"] if p.get("currentQuantity") is not None else p.get("quantity")}
                     for p in (order.get("lineItems") or {}).get("nodes") or []],
        "shipments": [{"status": _shopify_display_status(f.get("displayStatus") or f.get("status")),
                       "carrier": ", ".join(t["company"] for t in f.get("trackingInfo") or [] if t.get("company")) or None,
                       "tracking_number": ", ".join(t["number"] for t in f.get("trackingInfo") or [] if t.get("number")) or None,
                       "tracking_url": ", ".join(t["url"] for t in f.get("trackingInfo") or [] if t.get("url")) or None}
                      for f in order.get("fulfillments") or []],
    }


def shopify_order_widget(context):
    """Read-only sidebar orders from the saved order ID and customer's email, linked to the Shopify admin."""
    return order_widget(context, get_order=shopify_get_order, find_orders=shopify_find_orders, map_order=_shopify_widget_order)