Fluxa Agentic Checkout

interface:
  display_name: "FluxA Agentic Checkout"
  short_description: "Automate supported checkouts, then hand off safely"
  default_prompt: "Use FluxA Agentic Checkout to open a checkout link, autofill supported payment surfaces, and stop for human takeover when verification or unsupported flows appear."

*
!.gitignore

python-dotenv
socksio
playwright
pypinyin

#!/usr/bin/env python3
"""General checkout runner with Playwright automation and handoff signals."""

from __future__ import annotations

import argparse
import asyncio
import json
import re
import shutil
import sys
import tempfile
import time
from pathlib import Path

try:  # pragma: no cover - runtime integration
    from dotenv import load_dotenv
except Exception:  # pragma: no cover
    load_dotenv = None


SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from shopify.checkout import run_checkout_flow
from shopify.candidates import normalize_candidates
from shopify.navigation import run_navigation_playwright_fallback
from shopify.results import default_payload, normalize_checkout_state
from shopify.runtime import (
    Config,
    build_sensitive_data,
    get_env,
    json_dumps,
    load_secrets,
    normalize_text,
    parse_bool,
    resolve_path,
    resolve_proxy_settings,
)
from order_store import OrderStore


SCRIPT_VERSION = "checkout_playwright_handoff@v3"
SKILL_ROOT = SCRIPT_DIR.parent

STRIPE_HOSTS = ("checkout.stripe.com", "buy.stripe.com", "billing.stripe.com")
SHOPIFY_HOST_MARKERS = ("myshopify.com", "shopify.com")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Run the general checkout Playwright flow with structured handoff results.")
    parser.add_argument("--query", default=None)
    parser.add_argument("--entry-url", action="append", default=[])
    parser.add_argument("--entry-urls-json", default=None)
    parser.add_argument("--product-url", action="append", default=[])
    parser.add_argument("--candidate-urls-json", default=None)
    parser.add_argument("--mode", choices=["preview", "execute"], default=None)
    parser.add_argument("--headed", action="store_true")
    parser.add_argument("--headless", action="store_true")
    parser.add_argument("--secrets-path", default=None)
    parser.add_argument("--user-data-dir", default=None)
    parser.add_argument("--fresh-profile", action="store_true")
    parser.add_argument("--out-dir", default=None)
    parser.add_argument("--record-trace", action="store_true")
    parser.add_argument("--record-video", action="store_true")
    parser.add_argument("--browser-channel", default=None)
    parser.add_argument("--max-run-seconds", type=int, default=None)
    parser.add_argument("--max-total-usd", type=float, default=None)
    parser.add_argument("--keep-open-seconds", type=float, default=None)
    parser.add_argument("--action-delay-seconds", type=float, default=None)
    parser.add_argument("--manual-verification-timeout-seconds", type=int, default=None)
    parser.add_argument("--proxy-server", default=None)
    parser.add_argument("--proxy-bypass", default=None)
    parser.add_argument("--proxy-username", default=None)
    parser.add_argument("--proxy-password", default=None)
    parser.add_argument("--confirm-delivery", action="store_true")
    parser.add_argument("--confirm-legal-consent", action="store_true")
    parser.add_argument("--resident-id-number", default=None)
    parser.add_argument("--record-order-on-success", action="store_true")
    parser.add_argument("--no-record-order-on-success", dest="record_order_on_success", action="store_false")
    parser.set_defaults(record_order_on_success=None)
    parser.add_argument("--order-label", default=None)
    parser.add_argument("--order-note", default=None)
    parser.add_argument("--order-currency", default=None)
    parser.add_argument("--order-store-dir", default=None)
    return parser


def resolve_entry_urls_json(args: argparse.Namespace) -> str:
    explicit = (
        getattr(args, "entry_urls_json", None)
        or getattr(args, "candidate_urls_json", None)
        or get_env("ENTRY_URLS_JSON")
        or get_env("CANDIDATE_URLS_JSON")
    )
    if explicit:
        return explicit

    raw_urls = [value.strip() for value in getattr(args, "entry_url", []) if str(value or "").strip()]
    raw_urls.extend(value.strip() for value in getattr(args, "product_url", []) if str(value or "").strip())
    for env_name in ("ENTRY_URL", "PRODUCT_URL"):
        env_value = str(get_env(env_name, "") or "").strip()
        if env_value:
            raw_urls.append(env_value)

    if raw_urls:
        deduped = []
        seen = set()
        for url in raw_urls:
            if url in seen:
                continue
            seen.add(url)
            deduped.append(url)
        return json.dumps(deduped, ensure_ascii=False)

    raise ValueError("Provide --entry-url, ENTRY_URL, --entry-urls-json, or the legacy product/candidate URL flags.")


def normalize_entry_candidates(raw_candidates: str) -> list[dict[str, str]]:
    return normalize_candidates(raw_candidates)


def is_shopify_like_url(url: str) -> bool:
    lowered = str(url or "").lower()
    return any(marker in lowered for marker in SHOPIFY_HOST_MARKERS) or any(
        token in lowered for token in ("/products/", "/collections/", "/cart", "/checkout", "/checkouts/")
    )


def is_direct_checkout_like_url(url: str) -> bool:
    lowered = str(url or "").lower()
    return (
        any(host in lowered for host in STRIPE_HOSTS)
        or "/checkout" in lowered
        or "/checkouts/" in lowered
        or looks_like_cart_checkout_url(lowered)
    )


def looks_like_cart_checkout_url(url: str) -> bool:
    lowered = str(url or "").lower()
    return bool(re.search(r"/cart/[^/?#]+:\d", lowered))


def choose_route(entry_candidates: list[dict[str, str]]) -> str:
    first_url = entry_candidates[0]["url"] if entry_candidates else ""
    if is_shopify_like_url(first_url) and not is_direct_checkout_like_url(first_url):
        return "shopify_navigation"
    return "direct_checkout"


def build_direct_navigation_result(entry_candidate: dict[str, str]) -> dict[str, object]:
    entry_url = entry_candidate["url"]
    cart_checkout = looks_like_cart_checkout_url(entry_url)
    payload = default_payload()
    payload.update(
        {
            "entryUrl": entry_url,
            "candidateChosen": entry_url,
            "productUrl": None if cart_checkout else entry_url,
            "checkoutUrl": entry_url,
            "hint": "Direct checkout-like surface provided; attempting in-place autofill before any handoff.",
            "outcome": {"status": "checkout_reached", "hint": "Direct checkout-like surface provided."},
        }
    )
    return {
        "navigation": payload,
        "artifacts": {
            "conversation": None,
            "screenshots": [],
            "urls": [entry_url],
            "errors": [],
            "rankedCandidates": [entry_candidate],
            "method": "direct_checkout",
        },
    }


def resolve_config(args: argparse.Namespace) -> Config:
    query = args.query or get_env("QUERY", "checkout preview")
    candidate_urls_json = resolve_entry_urls_json(args)
    mode = args.mode or get_env("MODE", "preview")
    headed = True if args.headed else False
    if not args.headed and not args.headless:
        headed = parse_bool(get_env("HEADED"), False)
    if args.headless:
        headed = False

    secrets_raw = args.secrets_path or get_env("SECRETS_PATH")
    if not secrets_raw:
        raise ValueError("SECRETS_PATH is required.")

    user_data_dir_value = args.user_data_dir or get_env("USER_DATA_DIR")
    user_data_dir = resolve_path(user_data_dir_value) if user_data_dir_value else None
    fresh_profile = True if args.fresh_profile else parse_bool(get_env("FRESH_PROFILE"), True)
    out_dir = resolve_path(args.out_dir or get_env("OUT_DIR", "artifacts/agentic-checkout"))
    record_trace = True if args.record_trace else parse_bool(get_env("RECORD_TRACE"), False)
    record_video = True if args.record_video else parse_bool(get_env("RECORD_VIDEO"), False)
    browser_channel = args.browser_channel or get_env("BROWSER_CHANNEL")
    max_run_seconds = args.max_run_seconds or int(get_env("MAX_RUN_SECONDS", 600))
    max_total_usd = args.max_total_usd if args.max_total_usd is not None else float(get_env("MAX_TOTAL_USD", 10))
    keep_open_seconds = args.keep_open_seconds if args.keep_open_seconds is not None else float(get_env("KEEP_OPEN_SECONDS", 0))
    action_delay_seconds = (
        args.action_delay_seconds
        if args.action_delay_seconds is not None
        else float(get_env("ACTION_DELAY_SECONDS", 1.0 if headed else 0.4))
    )
    manual_verification_timeout_seconds = (
        args.manual_verification_timeout_seconds
        if args.manual_verification_timeout_seconds is not None
        else int(get_env("MANUAL_VERIFICATION_TIMEOUT_SECONDS", 300))
    )
    proxy_server, proxy_bypass, proxy_username, proxy_password = resolve_proxy_settings(args)
    confirm_delivery = True if getattr(args, "confirm_delivery", False) else parse_bool(get_env("CONFIRM_DELIVERY"), False)
    confirm_legal_consent = (
        True if getattr(args, "confirm_legal_consent", False) else parse_bool(get_env("CONFIRM_LEGAL_CONSENT"), False)
    )
    resident_id_number = normalize_text(getattr(args, "resident_id_number", None) or get_env("RESIDENT_ID_NUMBER"))
    record_order_on_success = (
        args.record_order_on_success
        if getattr(args, "record_order_on_success", None) is not None
        else parse_bool(get_env("RECORD_ORDER_ON_SUCCESS"), default=(mode == "execute"))
    )
    order_label = getattr(args, "order_label", None) or get_env("ORDER_LABEL")
    order_note = getattr(args, "order_note", None) or get_env("ORDER_NOTE")
    order_currency = getattr(args, "order_currency", None) or get_env("ORDER_CURRENCY", "USD")
    order_store_dir_value = getattr(args, "order_store_dir", None) or get_env("CHECKOUT_HANDOFF_ORDER_DIR")
    order_store_dir = resolve_path(order_store_dir_value) if order_store_dir_value else SKILL_ROOT / "data" / "paid_orders"

    return Config(
        query=query,
        candidate_urls_json=candidate_urls_json,
        mode=mode,
        secrets_path=resolve_path(secrets_raw),
        out_dir=out_dir,
        user_data_dir=user_data_dir,
        fresh_profile=fresh_profile,
        headed=headed,
        use_vision=False,
        record_trace=record_trace,
        record_video=record_video,
        browser_channel=browser_channel,
        max_run_seconds=max_run_seconds,
        max_total_usd=max_total_usd,
        keep_open_seconds=keep_open_seconds,
        action_delay_seconds=action_delay_seconds,
        manual_verification_timeout_seconds=manual_verification_timeout_seconds,
        llm_provider="playwright_only",
        browser_use_model="",
        openai_model="",
        max_steps=0,
        proxy_server=proxy_server,
        proxy_bypass=proxy_bypass,
        proxy_username=proxy_username,
        proxy_password=proxy_password,
        confirm_delivery=confirm_delivery,
        confirm_legal_consent=confirm_legal_consent,
        resident_id_number=resident_id_number or None,
        record_order_on_success=record_order_on_success,
        order_label=order_label,
        order_note=order_note,
        order_currency=order_currency,
        order_store_dir=order_store_dir,
    )


def _order_label_from_result(config: Config, result: dict[str, object]) -> str:
    explicit = str(config.order_label or "").strip()
    if explicit:
        return explicit
    for key in ("selectionLabel", "giftCardDenomination", "query"):
        value = result.get(key)
        if str(value or "").strip():
            return str(value).strip()
    store_domain = str(result.get("storeDomain") or "").strip()
    provider = str(result.get("provider") or "").strip()
    if store_domain and provider:
        return f"{store_domain} checkout via {provider}"
    if store_domain:
        return f"{store_domain} checkout purchase"
    return "checkout purchase"


def record_paid_order_if_needed(config: Config, result: dict[str, object]) -> dict[str, object] | None:
    if not config.record_order_on_success:
        return None
    if config.mode != "execute":
        return None
    if result.get("phase") != "execute_success":
        return None

    try:
        store = OrderStore(config.order_store_dir)
        order_record = store.save_paid_order(
            {
                "order_id": f"CHK-{result['runId']}",
                "product": _order_label_from_result(config, result),
                "price": result.get("displayedTotal"),
                "currency": config.order_currency,
                "merchant": result.get("storeDomain"),
                "provider": result.get("provider"),
                "route": result.get("route"),
                "purchase_reason": result.get("hint"),
                "note": config.order_note or result.get("hint"),
                "paid_at": result.get("finishedAt"),
                "metadata": {
                    "scriptVersion": result.get("scriptVersion"),
                    "runId": result.get("runId"),
                    "query": result.get("query"),
                    "entryUrl": result.get("entryUrl"),
                    "candidateChosen": result.get("candidateChosen"),
                    "productUrl": result.get("productUrl"),
                    "checkoutUrl": result.get("checkoutUrl"),
                    "filledCheckoutScreenshot": result.get("filledCheckoutScreenshot"),
                    "handoffRequired": result.get("handoffRequired"),
                    "outcome": result.get("outcome"),
                },
            }
        )
    except Exception as exc:
        result["orderRecorded"] = False
        result["orderRecordError"] = str(exc)
        return None

    result["orderRecorded"] = True
    result["orderId"] = order_record["order_id"]
    result["orderPath"] = str(store.order_path(order_record["order_id"]))
    result["orderStorageDir"] = str(store.root_dir)
    result["orderRecord"] = order_record
    return order_record


async def run_flow(
    config: Config,
    sensitive_data: dict[str, str],
    user_data_dir: Path | None,
    run_id: str,
    traces_dir: Path,
    videos_dir: Path | None,
    conversations_dir: Path,
) -> tuple[dict[str, object], dict[str, object], str]:
    entry_candidates = normalize_entry_candidates(config.candidate_urls_json)
    if not entry_candidates:
        final_payload = normalize_checkout_state(config.mode, default_payload(), max_total_usd=config.max_total_usd)
        return (
            final_payload,
            {"artifacts": {"conversation": None, "screenshots": [], "urls": [], "errors": [], "rankedCandidates": [], "method": "none"}},
            "none",
        )

    route = choose_route(entry_candidates)
    if route == "shopify_navigation":
        navigation_result = await run_navigation_playwright_fallback(
            config=config,
            run_id=run_id,
            user_data_dir=user_data_dir,
            traces_dir=traces_dir,
            videos_dir=videos_dir,
            conversations_dir=conversations_dir,
            ranked_candidates=entry_candidates[:3],
            fallback_reason="checkout_playwright_handoff",
        )
    else:
        navigation_result = build_direct_navigation_result(entry_candidates[0])

    navigation_payload = navigation_result["navigation"]
    navigation_status = navigation_payload.get("outcome", {}).get("status")
    if navigation_status in {"no_shopify_candidate", "candidate_selected", "product_selected", "needs_manual_verification"}:
        final_payload = normalize_checkout_state(config.mode, navigation_payload, max_total_usd=config.max_total_usd)
        return final_payload, navigation_result, route

    final_payload = await run_checkout_flow(
        config=config,
        navigation_payload=navigation_payload,
        sensitive_data=sensitive_data,
        user_data_dir=user_data_dir,
        run_id=run_id,
        traces_dir=traces_dir,
        videos_dir=videos_dir,
        runtime=navigation_result.get("runtime"),
    )
    return final_payload, navigation_result, route


async def async_main() -> int:
    if load_dotenv is not None:
        load_dotenv()
    parser = build_parser()
    args = parser.parse_args()
    try:
        config = resolve_config(args)
        secrets = load_secrets(config.secrets_path)
        sensitive_data = build_sensitive_data(secrets)
        if config.resident_id_number:
            sensitive_data["resident_id_number"] = config.resident_id_number
        if config.mode == "execute" and sensitive_data.get("delivery_name") and not config.confirm_delivery:
            raise ValueError("Execute mode requires explicit delivery confirmation. Re-run with --confirm-delivery after the user confirms the Delivery information.")
    except Exception as exc:
        result = {
            "ok": False,
            "phase": "config_error",
            "startedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "finishedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "scriptVersion": SCRIPT_VERSION,
            "userMessage": "Checkout cannot start yet because required information is missing or the local setup is not ready. Once the missing details are provided, I can continue from there.",
            "hint": str(exc),
        }
        print(json_dumps(result))
        return 1

    started_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    run_id = str(int(time.time() * 1000))
    out_dir = config.out_dir
    out_dir.mkdir(parents=True, exist_ok=True)
    traces_dir = out_dir / "traces"
    conversations_dir = out_dir / "conversations"
    traces_dir.mkdir(parents=True, exist_ok=True)
    conversations_dir.mkdir(parents=True, exist_ok=True)
    videos_dir = out_dir / "videos" / run_id if config.record_video else None
    if videos_dir:
        videos_dir.mkdir(parents=True, exist_ok=True)

    temp_profile_dir: Path | None = None
    user_data_dir = config.user_data_dir
    if config.fresh_profile:
        temp_profile_dir = Path(tempfile.mkdtemp(prefix="agentic-checkout-"))
        user_data_dir = temp_profile_dir
    elif user_data_dir:
        user_data_dir.mkdir(parents=True, exist_ok=True)

    try:
        final_payload, navigation_result, route = await asyncio.wait_for(
            run_flow(
                config=config,
                sensitive_data=sensitive_data,
                user_data_dir=user_data_dir,
                run_id=run_id,
                traces_dir=traces_dir,
                videos_dir=videos_dir,
                conversations_dir=conversations_dir,
            ),
            timeout=config.max_run_seconds,
        )
        result = {
            "ok": final_payload["phase"] in {"preview_complete", "execute_success"},
            "phase": final_payload["phase"],
            "startedAt": started_at,
            "finishedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "scriptVersion": SCRIPT_VERSION,
            "runId": run_id,
            "mode": config.mode,
            "query": config.query,
            "route": route,
            "entryUrl": final_payload.get("entryUrl"),
            "candidateChosen": final_payload["candidateChosen"],
            "storeDomain": final_payload["storeDomain"],
            "productUrl": final_payload["productUrl"],
            "checkoutUrl": final_payload["checkoutUrl"],
            "provider": final_payload["provider"],
            "displayedTotal": final_payload["displayedTotal"],
            "selectionLabel": final_payload.get("selectionLabel"),
            "giftCardDenomination": final_payload["giftCardDenomination"],
            "contactFilled": final_payload["contactFilled"],
            "deliveryFilled": final_payload.get("deliveryFilled"),
            "billingIdentityFilled": final_payload.get("billingIdentityFilled"),
            "postalFilled": final_payload["postalFilled"],
            "residentIdFilled": final_payload.get("residentIdFilled"),
            "paymentFieldVerification": final_payload.get("paymentFieldVerification"),
            "legalConsentChecked": final_payload.get("legalConsentChecked"),
            "saveInfoUnchecked": final_payload["saveInfoUnchecked"],
            "handoffRequired": final_payload.get("handoffRequired", False),
            "handoffReason": final_payload.get("handoffReason"),
            "userMessage": final_payload.get("userMessage"),
            "filledCheckoutScreenshot": final_payload.get("filledCheckoutScreenshot"),
            "hint": final_payload["hint"],
            "outcome": final_payload["outcome"],
            "traceDir": str(traces_dir) if config.record_trace else None,
            "videoDir": str(videos_dir) if config.record_video and videos_dir else None,
            "userDataDirUsed": str(user_data_dir) if user_data_dir else None,
            "freshProfileUsed": config.fresh_profile,
            "navigationArtifacts": navigation_result["artifacts"],
            "orderRecorded": False,
        }
        record_paid_order_if_needed(config, result)
    except asyncio.TimeoutError:
        result = {
            "ok": False,
            "phase": "timeout",
            "startedAt": started_at,
            "finishedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "scriptVersion": SCRIPT_VERSION,
            "runId": run_id,
            "mode": config.mode,
            "query": config.query,
            "route": "unknown",
            **default_payload(),
            "userMessage": f"I have been working on this for a while, but the page still could not complete this step within {config.max_run_seconds} seconds. You can try again later, or send me the current page link so I can keep investigating.",
            "hint": f"Checkout run exceeded MAX_RUN_SECONDS={config.max_run_seconds}.",
            "traceDir": str(traces_dir) if config.record_trace else None,
            "videoDir": str(videos_dir) if config.record_video and videos_dir else None,
            "orderRecorded": False,
        }
    except Exception as exc:
        result = {
            "ok": False,
            "phase": "exception",
            "startedAt": started_at,
            "finishedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "scriptVersion": SCRIPT_VERSION,
            "runId": run_id,
            "mode": config.mode,
            "query": config.query,
            "route": "unknown",
            **default_payload(),
            "userMessage": "The page encountered an unexpected issue while continuing the checkout, so I could not complete this step for you yet. Send me the link and I can continue investigating.",
            "hint": str(exc),
            "traceDir": str(traces_dir) if config.record_trace else None,
            "videoDir": str(videos_dir) if config.record_video and videos_dir else None,
            "orderRecorded": False,
        }
    finally:
        if config.keep_open_seconds > 0:
            await asyncio.sleep(config.keep_open_seconds)
        if temp_profile_dir:
            shutil.rmtree(temp_profile_dir, ignore_errors=True)

    result_path = out_dir / f"result-{run_id}.json"
    latest_path = out_dir / "result-latest.json"
    result_path.write_text(json_dumps(result) + "\n", encoding="utf-8")
    latest_path.write_text(json_dumps(result) + "\n", encoding="utf-8")
    print(json_dumps(result))
    return 0 if result.get("ok") else 1


def main() -> int:
    return asyncio.run(async_main())


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Operator-visible execute demo for the checkout handoff skill."""

from __future__ import annotations

import argparse
import subprocess
import sys
from pathlib import Path


SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parent
RUNNER = SCRIPT_DIR / "checkout_playwright_handoff.py"

DEFAULT_ENTRY_URL = "https://gracie-designs.myshopify.com/products/gift-card"
DEFAULT_OUT_DIR = SKILL_ROOT / "artifacts" / "execute-demo-headed"
DEFAULT_SECRETS_CANDIDATES = [
    Path.home() / ".clawdbot" / "credentials" / "kofi_tip_demo.json",
    Path.home() / ".clawdbot" / "credentials" / "shopify_giftcard_demo.json",
    Path.home() / ".clawdbot" / "credentials" / "buy_coffee_demo.json",
]


def default_secrets_path() -> str:
    for candidate in DEFAULT_SECRETS_CANDIDATES:
        if candidate.exists():
            return str(candidate)
    return str(DEFAULT_SECRETS_CANDIDATES[0])


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Run the execute-mode, headed checkout demo with visible browser automation."
    )
    parser.add_argument("--entry-url", default=DEFAULT_ENTRY_URL)
    parser.add_argument("--secrets-path", default=default_secrets_path())
    parser.add_argument("--out-dir", default=str(DEFAULT_OUT_DIR))
    parser.add_argument("--browser-channel", default=None)
    parser.add_argument("--max-total-usd", type=float, default=10.0)
    parser.add_argument("--max-run-seconds", type=int, default=600)
    parser.add_argument("--keep-open-seconds", type=float, default=30.0)
    parser.add_argument("--record-trace", action="store_true", default=False)
    parser.add_argument("--no-record-trace", dest="record_trace", action="store_false")
    parser.add_argument("--record-video", action="store_true")
    parser.add_argument("--print-only", action="store_true", help="Print the resolved command without running it.")
    return parser


def build_command(args: argparse.Namespace) -> list[str]:
    command = [
        sys.executable,
        str(RUNNER),
        "--entry-url",
        str(args.entry_url),
        "--mode",
        "execute",
        "--secrets-path",
        str(args.secrets_path),
        "--out-dir",
        str(args.out_dir),
        "--headed",
        "--max-total-usd",
        str(args.max_total_usd),
        "--max-run-seconds",
        str(args.max_run_seconds),
        "--keep-open-seconds",
        str(args.keep_open_seconds),
    ]
    if args.browser_channel:
        command.extend(["--browser-channel", str(args.browser_channel)])
    if args.record_trace:
        command.append("--record-trace")
    if args.record_video:
        command.append("--record-video")
    return command


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    secrets_path = Path(args.secrets_path).expanduser()
    if not secrets_path.exists():
        looked_in = ", ".join(str(path) for path in DEFAULT_SECRETS_CANDIDATES)
        parser.error(
            f"Secrets file not found: {secrets_path}. "
            f"Looked for demo credentials in: {looked_in}"
        )

    command = build_command(args)
    print("Running headed execute demo:", flush=True)
    print(" ".join(command), flush=True)

    if args.print_only:
        return 0

    completed = subprocess.run(command, cwd=str(SKILL_ROOT), check=False)
    return completed.returncode


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Manage paid orders recorded by the checkout and shopping skills."""

from __future__ import annotations

import argparse
import json
import sys
from typing import Any

from order_store import OrderStore


if sys.platform == "win32":
    import io

    sys.stdin = io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8")
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")


def _json_print(payload: dict[str, Any]) -> None:
    print(json.dumps(payload, ensure_ascii=False, indent=2))


def _load_create_payload(args: argparse.Namespace) -> dict[str, Any]:
    raw = sys.stdin.read().strip()
    if raw:
        try:
            payload = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise ValueError(f"Invalid JSON input: {exc}") from exc
        if not isinstance(payload, dict):
            raise ValueError("create expects a JSON object")
        return payload
    return {
        "order_id": args.order_id,
        "product": args.product,
        "price": args.price,
        "currency": args.currency,
        "merchant": args.merchant,
        "provider": args.provider,
        "platform": args.platform,
        "language": args.language,
        "route": args.route,
        "purchase_reason": args.purchase_reason,
        "user_input": args.user_input,
        "source": args.source,
        "note": args.note,
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Manage paid orders stored under data/paid_orders.")
    subparsers = parser.add_subparsers(dest="command", required=True)

    create_parser = subparsers.add_parser("create", help="Create or upsert a paid order record.")
    create_parser.add_argument("--order-id")
    create_parser.add_argument("--product")
    create_parser.add_argument("--price", type=float)
    create_parser.add_argument("--currency")
    create_parser.add_argument("--merchant")
    create_parser.add_argument("--provider")
    create_parser.add_argument("--platform")
    create_parser.add_argument("--language")
    create_parser.add_argument("--route")
    create_parser.add_argument("--purchase-reason")
    create_parser.add_argument("--user-input")
    create_parser.add_argument("--source", default="manual")
    create_parser.add_argument("--note")

    get_parser = subparsers.add_parser("get", help="Get one paid order by local order ID.")
    get_parser.add_argument("--order-id", required=True)

    list_parser = subparsers.add_parser("list", help="List paid orders.")
    list_parser.add_argument("--limit", type=int, default=20)
    list_parser.add_argument("--merchant")
    list_parser.add_argument("--provider")
    list_parser.add_argument("--platform")
    list_parser.add_argument("--status")
    list_parser.add_argument("--payment-status")
    list_parser.add_argument("--language")
    list_parser.add_argument("--keyword")
    list_parser.add_argument("--start-date")
    list_parser.add_argument("--end-date")

    search_parser = subparsers.add_parser("search", help="Search paid orders by keyword.")
    search_parser.add_argument("--keyword", required=True)
    search_parser.add_argument("--limit", type=int, default=20)
    search_parser.add_argument("--merchant")
    search_parser.add_argument("--provider")
    search_parser.add_argument("--platform")
    search_parser.add_argument("--status")
    search_parser.add_argument("--payment-status")
    search_parser.add_argument("--language")

    summary_parser = subparsers.add_parser("summary", help="Summarize paid orders.")
    summary_parser.add_argument("--days", type=int)

    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    store = OrderStore()

    try:
        if args.command == "create":
            order = store.save_paid_order(_load_create_payload(args))
            _json_print({"ok": True, "order": order, "storage_dir": str(store.root_dir)})
            return 0

        if args.command == "get":
            order = store.get_order(args.order_id)
            if order is None:
                _json_print({"ok": False, "error": f"Order not found: {args.order_id}", "storage_dir": str(store.root_dir)})
                return 1
            _json_print({"ok": True, "order": order, "storage_dir": str(store.root_dir)})
            return 0

        if args.command == "list":
            orders = store.list_orders(
                limit=args.limit,
                merchant=args.merchant,
                provider=args.provider,
                platform=args.platform,
                status=args.status,
                payment_status=args.payment_status,
                language=args.language,
                keyword=args.keyword,
                start_date=args.start_date,
                end_date=args.end_date,
            )
            _json_print({"ok": True, "count": len(orders), "orders": orders, "storage_dir": str(store.root_dir)})
            return 0

        if args.command == "search":
            orders = store.search_orders(
                keyword=args.keyword,
                limit=args.limit,
                merchant=args.merchant,
                provider=args.provider,
                platform=args.platform,
                status=args.status,
                payment_status=args.payment_status,
                language=args.language,
            )
            _json_print({"ok": True, "count": len(orders), "orders": orders, "storage_dir": str(store.root_dir)})
            return 0

        if args.command == "summary":
            _json_print({"ok": True, "summary": store.summary(days=args.days), "storage_dir": str(store.root_dir)})
            return 0
    except ValueError as exc:
        _json_print({"ok": False, "error": str(exc), "storage_dir": str(store.root_dir)})
        return 1

    _json_print({"ok": False, "error": f"Unsupported command: {args.command}", "storage_dir": str(store.root_dir)})
    return 1


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Filesystem-backed paid-order storage shared by shopping and checkout flows."""

from __future__ import annotations

import json
import os
import re
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any


SKILL_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_ORDER_DIR = SKILL_ROOT / "data" / "paid_orders"
ORDER_DIR_ENVS = ("CHECKOUT_HANDOFF_ORDER_DIR", "SMART_SHOPPER_ORDER_DIR")


def _now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def _parse_iso(value: str | None) -> datetime | None:
    if not value:
        return None
    normalized = value.strip()
    if not normalized:
        return None
    if normalized.endswith("Z"):
        normalized = normalized[:-1] + "+00:00"
    try:
        return datetime.fromisoformat(normalized)
    except ValueError:
        return None


def _coerce_float(value: Any) -> float | None:
    if value in (None, ""):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _clean_text(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _ensure_list(value: Any) -> list[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def _sanitize_filename(order_id: str) -> str:
    return re.sub(r"[^A-Za-z0-9._-]+", "_", order_id).strip("._") or "order"


def _matches_keyword(order: dict[str, Any], keyword: str | None) -> bool:
    if not keyword:
        return True
    needle = keyword.lower().strip()
    haystacks = [
        order.get("order_id"),
        order.get("product"),
        order.get("merchant"),
        order.get("provider"),
        order.get("platform"),
        order.get("route"),
        order.get("purchase_reason"),
        order.get("user_input"),
        order.get("source"),
        order.get("note"),
    ]
    metadata = order.get("metadata")
    if metadata:
        haystacks.append(json.dumps(metadata, ensure_ascii=False))
    return any(needle in str(item).lower() for item in haystacks if item)


def _infer_currency(
    language: str | None,
    platform: str | None,
    merchant: str | None,
    currency: str | None,
) -> str:
    if currency:
        return currency.upper()
    if (language or "").lower() == "zh":
        return "CNY"
    source = " ".join(part for part in (platform, merchant) if part).lower()
    if any(token in source for token in ("jd", "taobao", "tmall", "douyin", "pdd")):
        return "CNY"
    return "USD"


class OrderStore:
    """Persist paid checkout orders inside the skill folder or configured override directory."""

    def __init__(self, root_dir: str | os.PathLike[str] | None = None) -> None:
        configured = root_dir
        if not configured:
            for env_name in ORDER_DIR_ENVS:
                env_value = os.environ.get(env_name)
                if env_value:
                    configured = env_value
                    break
        self.root_dir = Path(configured).expanduser() if configured else DEFAULT_ORDER_DIR
        self.root_dir.mkdir(parents=True, exist_ok=True)

    def order_path(self, order_id: str) -> Path:
        return self.root_dir / f"{_sanitize_filename(order_id)}.json"

    def save_paid_order(self, payload: dict[str, Any]) -> dict[str, Any]:
        order_id = _clean_text(payload.get("order_id"))
        product = _clean_text(payload.get("product"))
        if not order_id:
            raise ValueError("order_id is required")
        if not product:
            raise ValueError("product is required")

        existing = self.get_order(order_id)
        now_iso = _now_iso()
        paid_at = _clean_text(payload.get("paid_at")) or (existing or {}).get("paid_at") or now_iso
        price = _coerce_float(payload.get("price"))
        merchant = _clean_text(payload.get("merchant")) or _clean_text(payload.get("store_domain"))
        platform = _clean_text(payload.get("platform")) or (existing or {}).get("platform")
        language = _clean_text(payload.get("language")) or (existing or {}).get("language")

        record = {
            "order_id": order_id,
            "status": _clean_text(payload.get("status")) or (existing or {}).get("status") or "paid",
            "payment_status": _clean_text(payload.get("payment_status")) or (existing or {}).get("payment_status") or "paid",
            "product": product,
            "price": price,
            "currency": _infer_currency(
                language,
                platform,
                merchant or (existing or {}).get("merchant"),
                _clean_text(payload.get("currency")) or (existing or {}).get("currency"),
            ),
            "merchant": merchant or (existing or {}).get("merchant"),
            "provider": _clean_text(payload.get("provider")) or (existing or {}).get("provider"),
            "platform": platform,
            "language": language,
            "route": _clean_text(payload.get("route")) or (existing or {}).get("route"),
            "purchase_reason": _clean_text(payload.get("purchase_reason")) or (existing or {}).get("purchase_reason"),
            "user_input": _clean_text(payload.get("user_input")) or (existing or {}).get("user_input"),
            "source": _clean_text(payload.get("source")) or (existing or {}).get("source") or "manual",
            "note": _clean_text(payload.get("note")) or (existing or {}).get("note"),
            "notes": _ensure_list(payload.get("notes")) or (existing or {}).get("notes") or [],
            "metadata": payload.get("metadata") if isinstance(payload.get("metadata"), dict) else (existing or {}).get("metadata") or {},
            "paid_at": paid_at,
            "created_at": (existing or {}).get("created_at") or now_iso,
            "updated_at": now_iso,
        }

        with self.order_path(order_id).open("w", encoding="utf-8") as handle:
            json.dump(record, handle, ensure_ascii=False, indent=2)
        return record

    def get_order(self, order_id: str) -> dict[str, Any] | None:
        path = self.order_path(order_id)
        if not path.exists():
            return None
        with path.open("r", encoding="utf-8") as handle:
            return json.load(handle)

    def list_orders(
        self,
        *,
        limit: int | None = None,
        merchant: str | None = None,
        provider: str | None = None,
        platform: str | None = None,
        status: str | None = None,
        payment_status: str | None = None,
        language: str | None = None,
        keyword: str | None = None,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> list[dict[str, Any]]:
        start_dt = _parse_iso(start_date)
        end_dt = _parse_iso(end_date)
        merchant_value = (merchant or "").strip().lower()
        provider_value = (provider or "").strip().lower()
        platform_value = (platform or "").strip().lower()
        status_value = (status or "").strip().lower()
        payment_status_value = (payment_status or "").strip().lower()
        language_value = (language or "").strip().lower()

        orders: list[dict[str, Any]] = []
        for path in self.root_dir.glob("*.json"):
            with path.open("r", encoding="utf-8") as handle:
                order = json.load(handle)
            if merchant_value and (order.get("merchant") or "").lower() != merchant_value:
                continue
            if provider_value and (order.get("provider") or "").lower() != provider_value:
                continue
            if platform_value and (order.get("platform") or "").lower() != platform_value:
                continue
            if status_value and (order.get("status") or "").lower() != status_value:
                continue
            if payment_status_value and (order.get("payment_status") or "").lower() != payment_status_value:
                continue
            if language_value and (order.get("language") or "").lower() != language_value:
                continue
            if not _matches_keyword(order, keyword):
                continue
            paid_at = _parse_iso(order.get("paid_at"))
            if start_dt and (paid_at is None or paid_at < start_dt):
                continue
            if end_dt and (paid_at is None or paid_at > end_dt):
                continue
            orders.append(order)

        orders.sort(
            key=lambda item: (
                _parse_iso(item.get("paid_at")) or datetime.min.replace(tzinfo=timezone.utc),
                item.get("order_id") or "",
            ),
            reverse=True,
        )
        if limit is not None and limit >= 0:
            return orders[:limit]
        return orders

    def search_orders(
        self,
        *,
        keyword: str,
        limit: int | None = None,
        merchant: str | None = None,
        provider: str | None = None,
        platform: str | None = None,
        status: str | None = None,
        payment_status: str | None = None,
        language: str | None = None,
    ) -> list[dict[str, Any]]:
        return self.list_orders(
            keyword=keyword,
            limit=limit,
            merchant=merchant,
            provider=provider,
            platform=platform,
            status=status,
            payment_status=payment_status,
            language=language,
        )

    def summary(self, *, days: int | None = None) -> dict[str, Any]:
        start_date = None
        if days is not None and days > 0:
            start_date = (datetime.now(timezone.utc) - timedelta(days=days)).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        orders = self.list_orders(start_date=start_date)

        by_merchant = Counter()
        by_provider = Counter()
        by_platform = Counter()
        by_currency: dict[str, dict[str, Any]] = defaultdict(lambda: {"count": 0, "total_paid": 0.0})
        total_paid = 0.0
        latest_paid_at = None

        for order in orders:
            merchant = order.get("merchant") or "unknown"
            provider = order.get("provider") or "unknown"
            platform = order.get("platform") or "unknown"
            currency = order.get("currency") or "UNKNOWN"
            amount = _coerce_float(order.get("price")) or 0.0

            by_merchant[merchant] += 1
            by_provider[provider] += 1
            by_platform[platform] += 1
            by_currency[currency]["count"] += 1
            by_currency[currency]["total_paid"] += amount
            total_paid += amount

            paid_at = order.get("paid_at")
            if paid_at and (latest_paid_at is None or paid_at > latest_paid_at):
                latest_paid_at = paid_at

        return {
            "count": len(orders),
            "total_paid": round(total_paid, 2),
            "by_merchant": dict(by_merchant),
            "by_provider": dict(by_provider),
            "by_platform": dict(by_platform),
            "by_currency": {
                key: {"count": value["count"], "total_paid": round(value["total_paid"], 2)}
                for key, value in by_currency.items()
            },
            "latest_paid_at": latest_paid_at,
        }

#!/usr/bin/env python3
"""Interactive setup flow for checkout payment, delivery, and billing profiles."""

from __future__ import annotations

import argparse
import getpass
import json
import os
from pathlib import Path
from typing import Any

try:  # pragma: no cover - interactive helper only
    from shopify.runtime import _normalize_profile_section as normalize_profile_section_for_setup
except Exception:  # pragma: no cover
    normalize_profile_section_for_setup = None


DEFAULT_OUTPUT = Path.home() / ".clawdbot" / "credentials" / "real_card.json"


def normalize_text(value: Any) -> str:
    return " ".join(str(value or "").strip().split())


def prompt_text(label: str, default: str | None = None, required: bool = False, secret: bool = False) -> str:
    suffix = f" [{default}]" if default else ""
    prompt = f"{label}{suffix}: "
    while True:
        raw = getpass.getpass(prompt) if secret else input(prompt)
        value = normalize_text(raw) if raw else normalize_text(default)
        if value or not required:
            return value
        print(f"{label} is required.", flush=True)


def prompt_bool(label: str, default: bool = True) -> bool:
    suffix = "Y/n" if default else "y/N"
    while True:
        raw = normalize_text(input(f"{label} [{suffix}]: ")).lower()
        if not raw:
            return default
        if raw in {"y", "yes"}:
            return True
        if raw in {"n", "no"}:
            return False
        print("Please answer yes or no.", flush=True)


def split_full_name(value: str) -> tuple[str, str]:
    parts = normalize_text(value).split(" ", 1)
    if not parts or not parts[0]:
        return "", ""
    if len(parts) == 1:
        return parts[0], parts[0]
    return parts[0], parts[1]


def collect_address_block(label: str, default_country: str) -> dict[str, str]:
    print(f"\n{label}", flush=True)
    full_name = prompt_text("Full name", required=True)
    first_name, last_name = split_full_name(full_name)
    address1 = prompt_text("Address line 1", required=True)
    address2 = prompt_text("Address line 2")
    city = prompt_text("City", required=True)
    state = prompt_text("State / Province", required=True)
    country = prompt_text("Country", default=default_country, required=True)
    phone = prompt_text("Phone", required=True)
    ascii_defaults: dict[str, str] = {}
    postal_default = ""
    if normalize_profile_section_for_setup is not None:
        inferred = normalize_profile_section_for_setup(
            {
                "name": full_name,
                "address1": address1,
                "address2": address2,
                "city": city,
                "state": state,
                "country": country,
                "phone": phone,
            },
            default_country=country,
        )
        ascii_defaults = inferred
        postal_default = inferred.get("postal", "")
    address1_ascii_default = ascii_defaults.get("address1_ascii") or (address1 if address1.isascii() else "")
    address1_ascii = prompt_text("Address line 1 (English / Latin fallback)", default=address1_ascii_default)
    city_ascii_default = ascii_defaults.get("city_ascii") or (city if city.isascii() else "")
    city_ascii = prompt_text("City (English / Latin fallback)", default=city_ascii_default)
    state_ascii_default = ascii_defaults.get("state_ascii") or (state if state.isascii() else "")
    state_ascii = prompt_text("State / Province (English / Latin fallback)", default=state_ascii_default)
    postal = prompt_text("ZIP / Postal code", default=postal_default, required=True)
    return {
        "name": full_name,
        "first_name": first_name,
        "last_name": last_name,
        "address1": address1,
        "address1_ascii": address1_ascii,
        "address2": address2,
        "city": city,
        "city_ascii": city_ascii,
        "state": state,
        "state_ascii": state_ascii,
        "postal": postal,
        "country": country,
        "phone": phone,
    }


def build_profile(output_path: Path, default_country: str) -> dict[str, Any]:
    print("Checkout profile setup", flush=True)
    print(f"Output file: {output_path}", flush=True)

    email = prompt_text("Checkout email", required=True)
    delivery = collect_address_block("Delivery profile", default_country)
    same_billing = prompt_bool("Use the same address for billing", default=True)
    billing = delivery.copy() if same_billing else collect_address_block("Billing profile", default_country)
    billing["same_as_delivery"] = same_billing

    print("\nPayment card", flush=True)
    cardholder_default = billing.get("name") or delivery.get("name")
    cardholder_name = prompt_text("Name on card", default=cardholder_default, required=True)
    card_number = prompt_text("Card number", required=True, secret=True)
    exp = prompt_text("Expiry (MM/YY or MM/YYYY)", required=True, secret=True)
    cvc = prompt_text("CVC", required=True, secret=True)
    card_postal = prompt_text("Card postal code", default=billing.get("postal"), required=True)
    card_country = prompt_text("Card country", default=billing.get("country") or default_country, required=True)
    resident_id_number = prompt_text("Resident ID number (optional)")

    profile = {
        "email": email,
        "payment": {
            "email": email,
            "card_number": card_number,
            "exp": exp,
            "cvc": cvc,
            "postal": card_postal,
            "country": card_country,
            "name": cardholder_name,
        },
        "delivery": delivery,
        "billing": billing,
    }
    if resident_id_number:
        profile["additional_information"] = {"resident_id_number": resident_id_number}
    return profile


def write_profile(path: Path, profile: dict[str, Any], overwrite: bool) -> None:
    if path.exists() and not overwrite:
        raise FileExistsError(f"Refusing to overwrite existing file without --overwrite: {path}")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(profile, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    os.chmod(path, 0o600)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Collect payment, delivery, and billing details into one checkout profile JSON.")
    parser.add_argument("--output", default=str(DEFAULT_OUTPUT))
    parser.add_argument("--country", default="United States", help="Default country for delivery, billing, and card prompts.")
    parser.add_argument("--overwrite", action="store_true")
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    output_path = Path(os.path.expanduser(os.path.expandvars(str(args.output))))
    try:
        profile = build_profile(output_path, default_country=str(args.country))
        write_profile(output_path, profile, overwrite=bool(args.overwrite))
    except FileExistsError as exc:
        parser.error(str(exc))
    except KeyboardInterrupt:
        print("\nSetup cancelled.", flush=True)
        return 130

    print(f"\nSaved checkout profile to {output_path}", flush=True)
    print("Use this file as --secrets-path for preview or execute.", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""One-command setup for checkout runtime and optional profile creation."""

from __future__ import annotations

import argparse
import platform
import subprocess
import sys
from pathlib import Path


SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parent
PROFILE_SETUP = SCRIPT_DIR / "setup_checkout_profile.py"
VENV_DIR = SKILL_ROOT / ".venv"


def run_command(command: list[str]) -> None:
    completed = subprocess.run(command, cwd=str(SKILL_ROOT), check=False)
    if completed.returncode != 0:
        raise RuntimeError(f"Command failed with exit code {completed.returncode}: {' '.join(command)}")


def ensure_runtime(venv_dir: Path) -> Path:
    if not venv_dir.exists():
        try:
            run_command([sys.executable, "-m", "venv", str(venv_dir)])
        except RuntimeError as exc:
            raise RuntimeError(
                "Python virtualenv bootstrap failed. "
                "If this is a fresh Docker/OpenClaw environment, install the base runtime first with: "
                "apt-get update && apt-get install -y python3-pip python3-venv xauth"
            ) from exc
    python_bin = venv_dir / "bin" / "python"
    if not python_bin.exists():
        raise RuntimeError(f"Virtualenv python not found: {python_bin}")
    run_command([str(python_bin), "-m", "pip", "install", "-r", "requirements.txt"])
    install_command = [str(python_bin), "-m", "playwright", "install", "chromium"]
    if platform.system().lower() == "linux" and hasattr(sys, "getuid") and sys.getuid() == 0:
        install_command = [str(python_bin), "-m", "playwright", "install", "--with-deps", "chromium"]
    try:
        run_command(install_command)
    except RuntimeError as exc:
        raise RuntimeError(
            "Playwright browser install failed. "
            "If system libraries are missing, run: python3 -m playwright install --with-deps chromium"
        ) from exc
    return python_bin


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Install checkout skill runtime upfront and optionally collect payment, delivery, and billing profile data."
    )
    parser.add_argument("--skip-runtime", action="store_true", help="Skip venv / pip / Playwright setup.")
    parser.add_argument("--skip-profile", action="store_true", help="Skip the interactive profile collection flow.")
    parser.add_argument("--venv-dir", default=str(VENV_DIR))
    parser.add_argument("--profile-output", default=str(Path.home() / ".clawdbot" / "credentials" / "real_card.json"))
    parser.add_argument("--country", default="United States")
    parser.add_argument("--overwrite-profile", action="store_true")
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    try:
        if not args.skip_runtime:
            python_bin = ensure_runtime(Path(args.venv_dir).expanduser())
            print(f"Checkout runtime ready: {python_bin}", flush=True)
        if not args.skip_profile:
            command = [
                sys.executable,
                str(PROFILE_SETUP),
                "--output",
                str(Path(args.profile_output).expanduser()),
                "--country",
                str(args.country),
            ]
            if args.overwrite_profile:
                command.append("--overwrite")
            run_command(command)
    except Exception as exc:
        print(str(exc), file=sys.stderr, flush=True)
        return 1

    print("Checkout skill setup complete.", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Shopify gift card browser automation helpers."""

"""Payment adapters for Shopify checkout."""

#!/usr/bin/env python3
"""Shared Playwright helpers for Shopify payment adapters."""

from __future__ import annotations

from typing import Any, Callable
import asyncio
import re
import sys


def normalize_text(value: Any) -> str:
    return re.sub(r"\s+", " ", str(value or "")).strip()


def digits(value: Any) -> str:
    return re.sub(r"\D+", "", str(value or ""))


async def pause(config: Any, seconds: float | None = None) -> None:
    delay = seconds if seconds is not None else float(getattr(config, "action_delay_seconds", 0) or 0)
    if delay > 0:
        await asyncio.sleep(delay)


def _select_all_shortcut() -> str:
    return "Meta+A" if sys.platform == "darwin" else "Control+A"


async def _iter_contexts(page: Any) -> list[Any]:
    contexts = [page]
    try:
        for frame in page.frames:
            if frame == page.main_frame:
                continue
            contexts.append(frame)
    except Exception:
        pass
    return contexts


async def _read_locator_value(locator: Any) -> str:
    try:
        value = await locator.input_value()
    except Exception:
        value = ""
    if normalize_text(value):
        return normalize_text(value)
    try:
        raw = await locator.evaluate(
            """(el) => {
                if (!el) return "";
                const value = "value" in el ? el.value : "";
                if (value) return value;
                return el.textContent || "";
            }"""
        )
    except Exception:
        raw = ""
    return normalize_text(raw)


async def _looks_like_autocomplete(locator: Any) -> bool:
    try:
        role = normalize_text(await locator.get_attribute("role")).lower()
        aria_autocomplete = normalize_text(await locator.get_attribute("aria-autocomplete")).lower()
        aria_haspopup = normalize_text(await locator.get_attribute("aria-haspopup")).lower()
        if role == "combobox" or aria_autocomplete in {"list", "both"} or aria_haspopup == "listbox":
            return True
    except Exception:
        return False
    return False


async def _is_interactable(locator: Any) -> bool:
    try:
        if not await locator.is_visible() or not await locator.is_enabled():
            return False
    except Exception:
        return False
    try:
        aria_hidden = normalize_text(await locator.get_attribute("aria-hidden")).lower()
        field_type = normalize_text(await locator.get_attribute("type")).lower()
        readonly = normalize_text(await locator.get_attribute("readonly")).lower()
        if aria_hidden == "true" or field_type == "hidden" or readonly in {"true", "readonly"}:
            return False
    except Exception:
        return False
    return True


async def fill_text_exact(
    locator: Any,
    value: str,
    config: Any,
    verifier: Callable[[str], bool] | None = None,
    commit_autocomplete: bool = False,
) -> bool:
    target = str(value)
    verify = verifier or (lambda raw: normalize_text(raw) == normalize_text(target))
    for _attempt in range(3):
        try:
            await locator.scroll_into_view_if_needed(timeout=2000)
        except Exception:
            pass
        try:
            await locator.click(timeout=3000)
        except Exception:
            pass
        try:
            await locator.press(_select_all_shortcut())
            await locator.press("Backspace")
        except Exception:
            pass
        try:
            await locator.fill("")
        except Exception:
            pass
        try:
            await locator.type(target, delay=35)
        except Exception:
            try:
                await locator.fill(target)
            except Exception:
                pass
        await pause(config, 0.2)
        current = await _read_locator_value(locator)
        if verify(current):
            return True
        if commit_autocomplete and await _looks_like_autocomplete(locator):
            for key in ("ArrowDown", "Enter", "Tab"):
                try:
                    await locator.press(key)
                except Exception:
                    continue
                await pause(config, 0.25)
                current = await _read_locator_value(locator)
                if verify(current):
                    return True
    return False


async def fill_first_matching(
    page: Any,
    selectors: list[str],
    value: str,
    verifier: Callable[[str], bool],
    config: Any,
    timeout_seconds: int = 20,
    commit_autocomplete: bool = False,
) -> tuple[bool, str]:
    deadline = asyncio.get_running_loop().time() + max(timeout_seconds, 1)
    last_value = ""
    while asyncio.get_running_loop().time() < deadline:
        for context in await _iter_contexts(page):
            for selector in selectors:
                try:
                    locator = context.locator(selector)
                    count = min(await locator.count(), 8)
                except Exception:
                    continue
                for index in range(count):
                    candidate = locator.nth(index)
                    if not await _is_interactable(candidate):
                        continue
                    if await fill_text_exact(
                        candidate,
                        value,
                        config,
                        verifier=verifier,
                        commit_autocomplete=commit_autocomplete,
                    ):
                        last_value = await _read_locator_value(candidate)
                        if verifier(last_value):
                            return True, last_value
        await asyncio.sleep(0.25)
    return False, last_value


async def select_first_matching(
    page: Any,
    selectors: list[str],
    values: list[str],
    timeout_seconds: int = 10,
) -> bool:
    targets = [normalize_text(value) for value in values if normalize_text(value)]
    if not targets:
        return False
    deadline = asyncio.get_running_loop().time() + max(timeout_seconds, 1)
    while asyncio.get_running_loop().time() < deadline:
        for context in await _iter_contexts(page):
            for selector in selectors:
                try:
                    locator = context.locator(selector)
                    count = min(await locator.count(), 6)
                except Exception:
                    continue
                for index in range(count):
                    candidate = locator.nth(index)
                    if not await _is_interactable(candidate):
                        continue
                    for target in targets:
                        for mode in ("label", "value"):
                            try:
                                await candidate.select_option(timeout=2000, **{mode: target})
                                return True
                            except Exception:
                                continue
        await asyncio.sleep(0.25)
    return False


async def click_first_visible(page: Any, selectors: list[str], timeout_ms: int = 3000) -> bool:
    for selector in selectors:
        try:
            locator = page.locator(selector)
            count = min(await locator.count(), 8)
        except Exception:
            continue
        for index in range(count):
            candidate = locator.nth(index)
            if not await _is_interactable(candidate):
                continue
            try:
                try:
                    await candidate.scroll_into_view_if_needed(timeout=2000)
                except Exception:
                    pass
                await candidate.click(timeout=timeout_ms)
                return True
            except Exception:
                try:
                    await candidate.click(timeout=timeout_ms, force=True)
                    return True
                except Exception:
                    continue
    return False


async def visible_button_texts(page: Any) -> list[str]:
    texts: list[str] = []
    try:
        locator = page.locator("button, [role='button'], input[type='submit']")
        count = min(await locator.count(), 20)
    except Exception:
        return texts
    for index in range(count):
        try:
            candidate = locator.nth(index)
            if not await candidate.is_visible():
                continue
            text = normalize_text(await candidate.inner_text())
            if not text:
                try:
                    text = normalize_text(await candidate.get_attribute("value"))
                except Exception:
                    text = ""
            if text:
                texts.append(text)
        except Exception:
            continue
    return texts


async def iframe_urls(page: Any) -> list[str]:
    values: list[str] = []
    try:
        for frame in page.frames:
            if frame == page.main_frame:
                continue
            if frame.url:
                values.append(frame.url)
    except Exception:
        pass
    return values


CHECKBOX_STATE_SCRIPT = """(payload) => {
  const { phrases, desiredState } = payload;
  const normalize = (text) => String(text || '').replace(/\\s+/g, ' ').trim().toLowerCase();
  const visible = (el) => {
    if (!el || !(el instanceof Element)) return false;
    const style = window.getComputedStyle(el);
    if (style.visibility === 'hidden' || style.display === 'none' || Number(style.opacity || '1') === 0) return false;
    const rect = el.getBoundingClientRect();
    return rect.width > 0 && rect.height > 0;
  };
  const labelledByText = (node) => {
    const ids = String(node?.getAttribute?.('aria-labelledby') || '').trim().split(/\\s+/).filter(Boolean);
    return ids
      .map((id) => document.getElementById(id))
      .filter(Boolean)
      .map((el) => el.innerText || el.textContent || '')
      .join(' ');
  };
  const relatedNodes = (box) => {
    const nodes = [];
    const push = (node) => {
      if (!node || nodes.includes(node)) return;
      nodes.push(node);
    };
    push(box);
    const id = box.getAttribute && box.getAttribute('id');
    if (id) push(document.querySelector(`label[for="${id}"]`));
    push(box.closest && box.closest('label'));
    push(box.nextElementSibling);
    push(box.previousElementSibling);
    let current = box.parentElement;
    for (let depth = 0; current && depth < 4; depth += 1, current = current.parentElement) {
      push(current);
    }
    return nodes;
  };
  const combinedText = (box) => {
    return relatedNodes(box)
      .map((node) => [node.innerText || node.textContent || '', labelledByText(node)].filter(Boolean).join(' '))
      .filter(Boolean)
      .join(' ');
  };
  const matches = (box) => {
    const hay = normalize(combinedText(box));
    return hay && phrases.some((phrase) => hay.includes(phrase));
  };
  const isChecked = (box) => {
    if ('checked' in box) return !!box.checked;
    return box.getAttribute('aria-checked') === 'true';
  };
  const interact = (node) => {
    if (!node || typeof node.click !== 'function') return;
    node.click();
    node.dispatchEvent(new Event('input', { bubbles: true }));
    node.dispatchEvent(new Event('change', { bubbles: true }));
  };
  const setChecked = (box, shouldCheck) => {
    if (isChecked(box) === shouldCheck) return isChecked(box);
    for (const node of relatedNodes(box)) {
      if (!visible(node) && node !== box) continue;
      interact(node);
      if (isChecked(box) === shouldCheck) return true;
    }
    interact(box);
    return isChecked(box) === shouldCheck;
  };
  for (const box of Array.from(document.querySelectorAll('input[type="checkbox"], [role="checkbox"]'))) {
    if (!matches(box)) continue;
    const visibleNodes = relatedNodes(box).filter((node) => visible(node));
    if (!visibleNodes.length && !visible(box)) continue;
    let checked = isChecked(box);
    if (desiredState !== null && checked !== desiredState) {
      setChecked(box, desiredState);
      checked = isChecked(box);
    }
    return JSON.stringify({ found: true, checked });
  }
  return JSON.stringify({ found: false, checked: null });
}"""


async def disable_save_info(page: Any, phrases: list[str], config: Any) -> bool | None:
    phrases = [normalize_text(phrase).lower() for phrase in phrases]
    for context in await _iter_contexts(page):
        try:
            raw = await context.evaluate(CHECKBOX_STATE_SCRIPT, {"phrases": phrases, "desiredState": False})
        except Exception:
            continue
        lowered = normalize_text(raw)
        if not lowered:
            continue
        if "\"unchecked\": true" in lowered.lower():
            await pause(config, 0.2)
            return True
        if "\"unchecked\": false" in lowered.lower():
            return False
    return None


async def ensure_checkbox_state(page: Any, phrases: list[str], config: Any, should_check: bool) -> bool | None:
    normalized_phrases = [normalize_text(phrase).lower() for phrase in phrases if normalize_text(phrase)]
    for context in await _iter_contexts(page):
        try:
            raw = await context.evaluate(
                CHECKBOX_STATE_SCRIPT,
                {"phrases": normalized_phrases, "desiredState": bool(should_check)},
            )
        except Exception:
            continue
        lowered = normalize_text(raw)
        if not lowered:
            continue
        if "\"found\": false" in lowered.lower():
            continue
        await pause(config, 0.2)
        if "\"checked\": true" in lowered.lower():
            return True
        if "\"checked\": false" in lowered.lower():
            return False
    return None