pip install requests

import os
import requests

API_KEY = os.environ.get("LUXCORE_API_KEY")
BASE_URL = "https://api.lux-core.io/api/v1"

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json"
}

import os
import hmac
import hashlib
import time
import requests
from typing import Optional, Dict, Any

class LuxCoreClient:
    """LuxCore API Client for Python"""
    
    BASE_URL = "https://api.lux-core.io/api/v1"
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.environ.get("LUXCORE_API_KEY")
        if not self.api_key:
            raise ValueError("API key is required")
        
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        })
    
    def _request(self, method: str, path: str, data: Dict = None, params: Dict = None) -> Dict:
        """Make an API request"""
        url = f"{self.BASE_URL}{path}"
        
        response = self.session.request(
            method=method,
            url=url,
            json=data,
            params=params
        )
        
        if not response.ok:
            error = response.json()
            raise LuxCoreError(
                message=error.get("message", "Unknown error"),
                status_code=response.status_code,
                error_code=error.get("error")
            )
        
        return response.json()
    
    # Payments
    def create_payment(self, **kwargs) -> Dict:
        """Create a new payment"""
        return self._request("POST", "/payments", data=kwargs)
    
    def get_payment(self, payment_id: str) -> Dict:
        """Get payment by ID"""
        return self._request("GET", f"/payments/{payment_id}")
    
    def list_payments(self, **filters) -> Dict:
        """List payments with optional filters"""
        return self._request("GET", "/payments", params=filters)
    
    def cancel_payment(self, payment_id: str) -> Dict:
        """Cancel a pending payment"""
        return self._request("POST", f"/payments/{payment_id}/cancel")
    
    # Balance
    def get_balance(self, currency: str = None) -> Dict:
        """Get merchant balance"""
        params = {"currency": currency} if currency else {}
        return self._request("GET", "/balance", params=params)
    
    def get_all_balances(self) -> Dict:
        """Get all merchant balances"""
        return self._request("GET", "/balance/all")
    
    # Webhooks
    def create_webhook(self, **kwargs) -> Dict:
        """Create a webhook endpoint"""
        return self._request("POST", "/webhooks", data=kwargs)
    
    def list_webhooks(self) -> Dict:
        """List all webhooks"""
        return self._request("GET", "/webhooks")
    
    def delete_webhook(self, webhook_id: str) -> None:
        """Delete a webhook"""
        self._request("DELETE", f"/webhooks/{webhook_id}")
    
    @staticmethod
    def verify_webhook_signature(
        raw_body: str,
        signature: str,
        timestamp: int,
        secret: str
    ) -> bool:
        """Verify webhook signature using raw request body.

        IMPORTANT: Pass the raw request body string, not a parsed/re-serialized object.
        Re-serializing JSON can change key ordering or whitespace, causing signature mismatches.
        """
        # Check timestamp (reject if older than 5 minutes)
        current_time = int(time.time())
        if abs(current_time - timestamp) > 300:
            raise ValueError("Webhook timestamp too old")

        # Calculate expected signature using raw body
        payload_to_sign = f"{timestamp}.{raw_body}"
        expected = "hmac_sha256=" + hmac.new(
            secret.encode(),
            payload_to_sign.encode(),
            hashlib.sha256
        ).hexdigest()

        # Compare signatures
        if not hmac.compare_digest(signature, expected):
            raise ValueError("Invalid signature")
        return True


class LuxCoreError(Exception):
    """LuxCore API Error"""
    def __init__(self, message: str, status_code: int, error_code: str = None):
        self.message = message
        self.status_code = status_code
        self.error_code = error_code
        super().__init__(self.message)

client = LuxCoreClient()

# Create SPEI deposit
payment = client.create_payment(
    amount=100000,  # $1,000.00 MXN (in centavos)
    currency="MXN",
    method="spei",
    type="deposit",
    merchant_reference="order_12345",
    customer={
        "name": "Juan Perez",
        "email": "juan@example.com",
        "phone": "+525551234567"
    },
    metadata={
        "order_id": "12345",
        "source": "web"
    }
)

print(f"Payment created: {payment['transaction_id']}")
print(f"Status: {payment['status']}")
print(f"Bank details: {payment.get('bank_details')}")

# Create SPEI payout
payout = client.create_payment(
    amount=50000,  # $500.00 MXN
    currency="MXN",
    method="spei",
    type="withdrawal",
    merchant_reference="payout_001",
    customer={
        "name": "Finance Team",
        "email": "finance@company.com"
    },
    payout={
        "recipient_name": "Maria Garcia",
        "bank_account": "012345678901234567",
        "bank_code": "002",
        "reference": "PAYOUT-001",
        "account_type": "clabe",
        "beneficiary_tax_id": "XAXX010101000",
        "description": "Monthly payout"
    }
)

print(f"Payout created: {payout['transaction_id']}")

payment = client.get_payment("pay_1234567890_abcdefgh")
print(f"Status: {payment['status']}")

# List completed payments
payments = client.list_payments(
    status="completed",
    limit=20,
    created_at_from="2025-01-01T00:00:00Z"
)

print(f"Total pages: {payments.get('pagination', {}).get('total_pages')}")
for payment in payments['data']:
    print(f"{payment['transaction_id']}: {payment['amount']} {payment['currency']}")

# Get MXN balance
balance = client.get_balance(currency="MXN")
print(f"Balance type: {balance['balance_type']}")
print(f"Total: {balance['balance_amount']} MXN")
print(f"Available: {balance['available_amount']} MXN")
print(f"Frozen: {balance['frozen_amount']} MXN")

# Get all balances
all_balances = client.get_all_balances()
for b in all_balances['balances']:
    print(f"{b['currency']} ({b['balance_type']}): {b['available_amount']}")

from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = os.environ.get("LUXCORE_WEBHOOK_SECRET")

@app.route("/webhooks/luxcore", methods=["POST"])
def handle_webhook():
    # Get headers
    signature = request.headers.get("X-Webhook-Signature")
    timestamp = int(request.headers.get("X-Webhook-Timestamp", 0))
    raw_body = request.get_data(as_text=True)
    payload = request.json  # for business logic

    # Verify signature using raw body (not re-serialized JSON)
    try:
        LuxCoreClient.verify_webhook_signature(
            raw_body, signature, timestamp, WEBHOOK_SECRET
        )
    except ValueError as e:
        return jsonify({"error": str(e)}), 400
    
    # Process event — use envelope structure
    event = request.headers.get("X-Webhook-Event")
    webhook_id = request.headers.get("X-Webhook-Id")
    payment = payload["data"]["payment"]

    if event == "payment.completed":
        # Handle completed payment
        print(f"Payment {payment['id']} completed! (webhook: {webhook_id})")
        # Update your database, send confirmation, etc.

    elif event == "payment.failed":
        # Handle failed payment
        print(f"Payment {payment['id']} failed: {payment.get('failure_reason')}")

    elif event == "payment.cancelled":
        # Handle cancelled payment
        print(f"Payment {payment['id']} cancelled")

    return jsonify({"received": True}), 200

if __name__ == "__main__":
    app.run(port=3000)

from luxcore import LuxCoreClient, LuxCoreError

client = LuxCoreClient()

try:
    payment = client.create_payment(
        amount=100000,
        currency="MXN",
        method="spei",
        type="deposit",
        merchant_reference="order_123",
        customer={"name": "Juan", "email": "juan@example.com"}
    )
except LuxCoreError as e:
    if e.status_code == 401:
        print("Invalid API key")
    elif e.status_code == 400:
        print(f"Validation error: {e.message}")
    elif e.status_code == 429:
        print("Rate limited - wait and retry")
    else:
        print(f"API error: {e.message}")

#!/usr/bin/env python3
"""
LuxCore Payment Integration Example
"""
import os
from luxcore import LuxCoreClient, LuxCoreError

def main():
    # Initialize client
    client = LuxCoreClient()
    
    # Check balance first
    balance = client.get_balance(currency="MXN")
    print(f"Current balance: {balance['available_amount']} MXN")
    
    # Create a payment
    try:
        payment = client.create_payment(
            amount=100000,
            currency="MXN",
            method="spei",
            type="deposit",
            merchant_reference=f"order_{int(time.time())}",
            customer={
                "name": "Customer Name",
                "email": "customer@example.com"
            }
        )
        
        print(f"✅ Payment created: {payment['transaction_id']}")
        print(f"   Status: {payment['status']}")
        
        if payment.get("bank_details"):
            print(f"   Bank: {payment['bank_details']['bank_name']}")
            print(f"   Account: {payment['bank_details']['account_number']}")
            print(f"   Holder: {payment['bank_details']['account_holder']}")
        
    except LuxCoreError as e:
        print(f"❌ Error: {e.message}")

if __name__ == "__main__":
    main()