woocommerce_get_order / woocommerce_find_orders

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

def execute_action(context):
    order = woocommerce_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 re

import requests

# Never returned: the customer's IP and browser, the order key and payment URL
# (they open the order and its payment page), and marketing attribution and tracking cookies.
_WOOCOMMERCE_ORDER_HIDDEN = {
    "customer_ip_address", "customer_user_agent", "order_key", "cart_hash",
    "payment_url", "_links",
}
_WOOCOMMERCE_META_HIDDEN = re.compile(r"attribution|ip_address|user_agent|^pys_|cookie|stripe|paypal|ppcp", re.I)


def _woocommerce_clean_order(order):
    order = {k: v for k, v in order.items() if k not in _WOOCOMMERCE_ORDER_HIDDEN}
    order["meta_data"] = [
        meta for meta in order.get("meta_data") or []
        if not _WOOCOMMERCE_META_HIDDEN.search(str(meta.get("key") or ""))
    ]
    return order


def _woocommerce_store(context):
    """The current business's WooCommerce connection: {"domain", "consumerKey", "consumerSecret"}."""
    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("woocommerce")
    if not store:
        raise ValueError("WooCommerce is not connected for this business")
    return store


def _woocommerce_get(store, path, params=None):
    response = requests.get(
        f"https://{store['domain']}/wp-json/wc/v3/{path}",
        auth=(store["consumerKey"], store["consumerSecret"]),
        headers={"User-Agent": "octocom"},
        params=params or {},
        timeout=20,
    )
    if response.status_code == 404:
        return None
    response.raise_for_status()
    return response.json()


def _woocommerce_with_notes(store, order):
    # Shipping plugins often write the tracking number into an order note.
    order["notes"] = _woocommerce_get(store, f"orders/{order['id']}/notes") or []
    return _woocommerce_clean_order(order)


def _woocommerce_indexed_order_id(context, number):
    """The order ID Octocom's order sync indexed under this number, or None."""
    business = context.get("business") or {}
    slug = business.get("slug") or (context.get("conversation") or {}).get("businessSlug")
    response = requests.get(
        f"{OCTOCOM_API_BASE_URL}/rest/private/integrations/woocommerce/order-id",
        params={"businessSlug": slug, "orderNumber": number},
        headers={"X-API-Key": OCTOCOM_ORG_API_KEY},
        timeout=20,
    )
    response.raise_for_status()
    return response.json().get("orderId")


def woocommerce_get_order(context, order_number):
    """The WooCommerce order with this number (plus its notes), or None."""
    store = _woocommerce_store(context)
    number = re.sub(r"\s+", "", str(order_number)).lstrip("#")
    if not number:
        return None

    # Custom order numbers and marketplace references, as the order sync saw them.
    indexed_id = _woocommerce_indexed_order_id(context, number)
    if indexed_id:
        order = _woocommerce_get(store, f"orders/{indexed_id}")
        if order:
            return _woocommerce_with_notes(store, order)

    if number.isdigit():
        order = _woocommerce_get(store, f"orders/{number}")
        if order and str(order.get("number")) == number:
            return _woocommerce_with_notes(store, order)

    # WooCommerce's search also covers order meta, so a custom order number or
    # a marketplace (Amazon, eBay) reference stored on the order is found too.
    hits = _woocommerce_get(store, "orders", {"search": number}) or []
    needle = number.lower()
    for hit in hits:
        if str(hit.get("number")).lower() == needle:
            return _woocommerce_with_notes(store, hit)
    for hit in hits:
        for meta in hit.get("meta_data") or []:
            if isinstance(meta.get("value"), str) and re.sub(r"\s+", "", meta["value"]).lower() == needle:
                return _woocommerce_with_notes(store, hit)
    # A lone hit for something that looks like an order number is the order the
    # search matched it on (e.g. an ERP reference); a name or postcode is not.
    if len(hits) == 1 and len(needle) >= 6 and any(ch.isdigit() for ch in needle):
        return _woocommerce_with_notes(store, hits[0])
    return None


def woocommerce_find_orders(context, email, limit=10):
    """The customer's WooCommerce orders, newest first: their account's orders and guest orders with this billing email."""
    store = _woocommerce_store(context)
    email = email.strip().lower()
    orders = {}
    for customer in _woocommerce_get(store, "customers", {"email": email, "role": "all"}) or []:
        for order in _woocommerce_get(store, "orders", {"customer": customer["id"], "per_page": limit}) or []:
            orders[order["id"]] = order
    for order in _woocommerce_get(store, "orders", {"search": email, "per_page": max(limit, 20)}) or []:
        if ((order.get("billing") or {}).get("email") or "").strip().lower() == email:
            orders[order["id"]] = order
    newest = sorted(orders.values(), key=lambda o: o.get("date_created_gmt") or "", reverse=True)
    return [_woocommerce_with_notes(store, order) for order in newest[:limit]]