bonus-import-tools/dinect_api.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
# __author__ = 'szhdanoff@gmail.com'
__version__ = '1.0.4'
import os
import requests
import json
import time
from functools import wraps

from dotenv import load_dotenv
load_dotenv()


def retry_on_error(max_retries=3, delay=1, backoff=2):
    """
    Decorator for retrying API calls on transient errors.
    
    Args:
        max_retries: Maximum number of retry attempts
        delay: Initial delay between retries in seconds
        backoff: Multiplier for delay after each retry
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            current_delay = delay
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except (requests.exceptions.ConnectionError, 
                        requests.exceptions.Timeout,
                        requests.exceptions.HTTPError) as e:
                    last_exception = e
                    if attempt < max_retries:
                        print(f"Retry {attempt + 1}/{max_retries} after {current_delay}s - Error: {e}")
                        time.sleep(current_delay)
                        current_delay *= backoff
                    else:
                        print(f"All {max_retries} retries exhausted. Last error: {e}")
            raise last_exception
        return wrapper
    return decorator


def handle_api_error(func):
    """
    Decorator for handling API errors consistently.
    Returns tuple (False, error_message) on any exception.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.RequestException as e:
            return False, f"Network error: {str(e)}"
        except (ValueError, KeyError) as e:
            return False, f"Data error: {str(e)}"
        except Exception as e:
            return False, f"Unexpected error: {str(e)}"
    return wrapper

is_prod = bool(os.getenv('PRODUCTION', False))
if is_prod:
    url = 'https://pos-api.dinect.com/20130701/'
else:
    url = 'https://pos-api-ote.dinect.com/20130701/'

print('PRODUCTION:', is_prod, '| API URL:', url)

# APP_TOKEN = os.getenv('APP_TOKEN')
# POS_TOKEN = os.getenv('POS_TOKEN')

app_token = os.getenv('APP_TOKEN')
pos_token = os.getenv('POS_TOKEN')
currency = os.getenv('CURRENCY', 'RUB')

HEADERS = {
    'Authorization': f'dmtoken {pos_token}',
    'DM-Authorization': f'dmapptoken {app_token}',
    'User-Agent': f'bonus-import-tools-2024 v.{__version__}',
    'Accept': 'application/json',
    # 'Accept-Language': 'ru,ru-RU;q=0.8,en-gb;q=0.5,en;q=0.3',
    'Accept-Charset': 'utf-8',
    'Connection': 'close',
    'Content-Type': 'application/json',
    # 'Content-Type': 'application/x-www-form-urlencoded',
}


def get_user(search_id, get_type='auto', headers=None) -> tuple:
    """
    A function to get user information based on the search_id and get_type.

    Parameters:
        search_id (str): The search id for the user.
        get_type (str, optional): The type of search (default is 'auto').
        headers (dict, optional): The headers for the request (default is None).

    Returns:
        tuple: A tuple containing a boolean value and the response data.
            - The boolean value indicating success or failure.
            - The response data based on the request made.
    """
    return _get_user_impl(search_id, get_type, headers)


@retry_on_error(max_retries=3, delay=1, backoff=2)
@handle_api_error
def _get_user_impl(search_id, get_type='auto', headers=None) -> tuple:
    if headers is None:
        headers = HEADERS

    base_url = url + 'users/'
    # get_type = auto, card, phone, email, foreigncard
    r = requests.get(
        base_url,
        headers=headers,
        params={
            get_type: search_id
        }
    )
    if r.status_code == 200:
        if len(r.json()) == 0:
            return False, None, None, None, r.text
        else:
            return True, r.json()[0]['id'], r.json()[0]['card'], r.json()[0]['purchases_url'], r.json()
    else:
        return False, None, None, None, r.text
        # return False, r.text


def new_user(full_name, phone, gender=1, foreign_card=None, email=None, headers=None):
    """
    A function that creates a new user with optional headers.

    Args:
        headers (dict, optional): The headers to include in the request. Defaults to None.

    Returns:
        tuple: A tuple containing a boolean indicating the success of the request and the JSON response.
               If the request is successful, the boolean is True and the JSON response is returned.
               If the request is unsuccessful, the boolean is False and the JSON response is returned.
               :param email:
               :param gender:
               :param foreign_card:
               :param headers:
               :param phone:
               :param full_name :
    """
    if headers is None:
        headers = HEADERS

    base_url = url + '/users/'

    params = {
        # "short_name": nickname,
        "full_name": full_name,
        "phone": phone,
        "email": email,
        "gender": gender,
    }

    r = requests.post(base_url, headers=headers, json=params)
    if r.status_code == 201:
        return True, r.json()
    else:
        return False, r.json()


def new_user_by_card(external_card, full_name, phone, gender=1, email=None, headers=None, use_existing=True):
    """
    Creates a new user by card.

    Args:
        external_card (str): The external card code.
        full_name (str): The full name of the user.
        phone (str): The phone number of the user.
        gender (int, optional): The gender of the user. Defaults to 1.
        email (str, optional): The email of the user. Defaults to None.
        headers (dict, optional): The headers to include in the request. Defaults to None.
        use_existing (bool, optional): Whether to bind the user to an existing card. Defaults to True.

    Returns:
        tuple: A tuple containing a boolean indicating the success of the request and the JSON response.
               If the request is successful, the boolean is True and the JSON response is returned.
               If the request is unsuccessful, the boolean is False and the error message is returned.
    """
    return _new_user_by_card_impl(external_card, full_name, phone, gender, email, headers, use_existing)


@retry_on_error(max_retries=3, delay=1, backoff=2)
@handle_api_error
def _new_user_by_card_impl(external_card, full_name, phone, gender=1, email=None, headers=None, use_existing=True):
    if headers is None:
        headers = HEADERS

    base_url = url + 'cards/'
    params = {
        'full_name': full_name,
        'phone': phone,
        'email': email,
        'gender': gender,
        'bind': use_existing,
        'code': external_card,
        'format': 'qrcode',
    }
    r = requests.post(base_url, headers=headers, json=params)
    if r.status_code == 201:
        return True, r.json()
    else:
        return False, r.text


    if headers is None:
        headers = HEADERS

    base_url = url + '/users/'

    params = {
        # "short_name": nickname,
        "full_name": full_name,
        "phone": phone,
        "email": email,
        "gender": gender,
    }

    r = requests.post(base_url, headers=headers, json=params)
    if r.status_code == 201:
        return True, r.json()
    else:
        return False, r.json()


def add_external_card(user_id, card, headers=None):
    """
    Adds an external card to a user.

    Args:
        user_id (int): The ID of the user.
        card (str): The card number.
        headers (dict, optional): The headers to include in the request. Defaults to None.

    Returns:
        tuple: A tuple containing a boolean indicating the success of the request and the JSON response.
               If the request is successful, the boolean is True and the JSON response is returned.
               If the request is unsuccessful, the boolean is False and the JSON response is returned.
    """
    if headers is None:
        headers = HEADERS

    base_url = url + 'users/cards/'
    params = {
        "card": card
    }
    r = requests.post(base_url, headers=headers, json=params)
    if r.status_code == 201:
        return True, r.json()
    else:
        return False, r.text if r.text else f"HTTP {r.status_code}"


def bonuses_update(
        user_id,
        summ_total,
        bonus_amount,
        sum_with_discount,
        sum_discount,
        doc_id,
        headers=None,
        dry_run=False
):
    """
    Updates user bonuses (adds or subtracts bonus points).
    
    Args:
        user_id: User ID in Dinect system
        summ_total: Total sum of transaction
        bonus_amount: Bonus amount to add (positive) or pay with bonuses (negative)
        sum_with_discount: Sum after discount
        sum_discount: Discount amount
        doc_id: Document ID for the transaction
        headers: Optional request headers
        dry_run: If True, simulates the transaction without actual processing
        
    Returns:
        tuple: (success: bool, response_data: dict|str)
    """
    return _bonuses_update_impl(user_id, summ_total, bonus_amount, sum_with_discount, sum_discount, doc_id, headers, dry_run)


@retry_on_error(max_retries=3, delay=1, backoff=2)
@handle_api_error
def _bonuses_update_impl(
        user_id,
        summ_total,
        bonus_amount,
        sum_with_discount,
        sum_discount,
        doc_id,
        headers=None,
        dry_run=False
):
    if headers is None:
        headers = HEADERS

    base_url = url + 'users/' + str(user_id) + '/purchases/'
    params = {
        "doc_id": doc_id,
#        "bonus_amount": bonus_amount,
        "sum_total": summ_total,
        "sum_discount": sum_discount,
        "sum_with_discount": sum_with_discount,
        "commit": 'True',
        "curr_iso_name": currency,
        "override": 'True'
        # "date": '2024-08-03 12:53:07',
    }
    bonus_amount_numeric = float(bonus_amount)
    if bonus_amount_numeric >= 0:
        params["bonus_amount"] = str(int(bonus_amount_numeric))
    else:
        # Negative value - pay with bonuses (use absolute value)
        params["bonus_payment"] = str(int(abs(bonus_amount_numeric)))

    r = requests.post(base_url, headers=headers, json=params)
    if r.status_code == 201:
        return True, r.json()
    else:
        return False, r.text