mirror of
https://git.auk.su/Dinect/bonus-import-tools.git
synced 2025-12-28 18:10:01 +00:00
258 lines
7.6 KiB
Python
258 lines
7.6 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
# __author__ = 'szhdanoff@gmail.com'
|
|
__version__ = '1.0.1'
|
|
|
|
import os
|
|
import requests
|
|
import json
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
# local imports
|
|
# import app
|
|
|
|
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(is_prod, 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',
|
|
}
|
|
|
|
|
|
# GET /20130701/tokens/?next=/20130701/logon
|
|
# https://pos-api.dinect.com/20130701/tokens/3b01228843d115ae8c03a4d3b20dcb545dbb228c
|
|
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.
|
|
"""
|
|
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 bonuses_update(
|
|
user_id,
|
|
summ_total,
|
|
bonus_amount,
|
|
sum_with_discount,
|
|
sum_discount,
|
|
doc_id,
|
|
headers=None):
|
|
"""
|
|
Updates the bonuses for a user.
|
|
|
|
Args:
|
|
user_id (int): The ID of the user.
|
|
summ_total (float): The total amount.
|
|
bonus_amount (float): The bonus amount.
|
|
sum_with_discount (float): The amount with discount.
|
|
sum_discount (float): The discount amount.
|
|
doc_id (str): The document ID.
|
|
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/' + 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": 'RUB',
|
|
"override": 'True',
|
|
# "date": '2024-08-03 12:53:07',
|
|
}
|
|
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):
|
|
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
|
|
|
|
|
|
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()
|
|
|
|
|
|
# user_date = new_user('Test2', '79039426493', email='YlEJp@example.com')
|
|
# print(user_date)
|
|
|
|
# 79039426498
|
|
# (True, {'DIN': 232113, 'ID': '4620011139016260791309380'})
|
|
|
|
# result, user_id, user_card, purchases_url, data = get_user('1234567890123')
|
|
# result, user_id, user_card, purchases_url, data = get_user('79039426493', get_type='phone')
|
|
# print(result, data)
|
|
# if result:
|
|
# # user_id = data[0].get('id')
|
|
# # user_card = data[0].get('card')
|
|
# # purchases_url = data[0].get('purchases_url')
|
|
# print('user_id', user_id)
|
|
# print('user_card', user_card)
|
|
# print('purchases_url', purchases_url)
|
|
|
|
|
|
# добавление внешней карты лояльности
|
|
|
|
|
|
# print(get_user('4620011139016689273132009'))
|
|
# print(get_user('1234567890123'))
|
|
# print(get_user('+79039406889'))
|
|
# print(new_user())
|
|
# (True, '{"DIN":3152300,"ID":"4620011139016570939672611"}')
|
|
|
|
|
|
# print(bonuses_update(
|
|
# user_id=int('1002'),
|
|
# summ_total=100.00,
|
|
# bonus_amount=10.00,
|
|
# doc_id='test12',
|
|
# sum_with_discount=90.00,
|
|
# sum_discount=10.00,
|
|
# ))
|
|
# print(get_user('79039426493', get_type='phone'))
|
|
# (True, {'DIN': 3155239, 'ID': '4620011139016802073627661'})
|
|
|
|
|
|
# print(new_user_by_card(
|
|
# external_card='1234567891235', full_name='Test', phone='79039426495', email='123321@example.com', use_existing=False
|
|
# ))
|