Интеграция API платежей USDT на Python

"""Python 3.11+ server-side example; no calls or credentials at import time."""
import hashlib
import hmac
import json
import random
import re
import time
from urllib.error import HTTPError, URLError
from urllib.request import HTTPRedirectHandler, Request, build_opener

API = 'https://api.dedyx.com/api/v1'


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None  # Never forward an API key to a redirected host.


def create_payment(api_key, wallet, amount, description, idempotency_key):
    """Persist the key and these fields with your order BEFORE calling this function."""
    if not re.fullmatch(r'[A-Za-z0-9._:-]{8,128}', idempotency_key):
        raise ValueError('Use a persisted 8–128 character idempotency key.')
    body = json.dumps({'merchant_wallet': wallet, 'amount': amount,
                       'description': description}, separators=(',', ':')).encode()
    request = Request(API + '/payments', data=body, method='POST', headers={
        'X-API-Key': api_key, 'Idempotency-Key': idempotency_key,
        'Content-Type': 'application/json'})
    opener = build_opener(NoRedirect())
    for attempt in range(3):
        delay = 0.5 * (2 ** attempt) + random.uniform(0, 0.25)
        try:
            with opener.open(request, timeout=10) as response:
                return json.load(response)
        except HTTPError as error:
            if error.code not in (429, 500, 502, 503, 504) or attempt == 2:
                raise
            retry_after = error.headers.get('Retry-After', '')
            if retry_after.isascii() and retry_after.isdigit():
                seconds = int(retry_after)
                if seconds > 60:
                    raise  # Schedule later with the SAME stored key/body.
                delay = max(delay, seconds)
            error.close()  # nginx may return HTML; do not assume an error JSON body.
        except (URLError, TimeoutError):
            if attempt == 2:
                raise  # Outcome may be uncertain; preserve the key and reconcile.
        time.sleep(delay)

import os
from dedyx_client import create_payment

# These fields must come from a persisted order record.
payment = create_payment(
    api_key=os.environ["DEDYX_API_KEY"],
    wallet=os.environ["DEDYX_WALLET"],
    amount="49.00",
    description="Order #1024",
    idempotency_key="order-1024-v1",
)
# Persist payment["payment_id"] and the response before redirecting.
print(payment["payment_url"])

import hashlib
import hmac
import json
import re
import time

def verify_webhook(raw_body, timestamp, signature, header_event_id, key_id, keys, now=None):
    """keys maps saved X-Dedyx-Key-ID values to the full whsec_... string."""
    if not re.fullmatch(r'[0-9]{1,12}', timestamp or ''):
        raise ValueError('Invalid timestamp')
    if abs((time.time() if now is None else now) - int(timestamp)) > 300:
        raise ValueError('Stale signature')
    if key_id not in keys or not re.fullmatch(r'v1=[0-9a-f]{64}', signature or ''):
        raise ValueError('Unknown signing key or signature format')
    expected = 'v1=' + hmac.new(keys[key_id].encode(), timestamp.encode() + b'.' + raw_body,
                               hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        raise ValueError('Invalid signature')
    event = json.loads(raw_body)
    if not isinstance(event, dict) or not isinstance(event.get('event_id'), str) or not event['event_id'] or event['event_id'] != header_event_id:
        raise ValueError('Event ID mismatch')
    # A valid signature is only the first check; validate your own order before fulfilment.
    return event