mirror of
https://git.auk.su/Dinect/bonus-import-tools.git
synced 2025-12-28 18:10:01 +00:00
114 lines
4.9 KiB
Python
114 lines
4.9 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
# __author__ = 'szhdanoff@gmail.com'
|
|
import os
|
|
import time
|
|
import csv
|
|
import phonenumbers
|
|
|
|
from email_validator import validate_email, EmailNotValidError
|
|
from dotenv import load_dotenv
|
|
# local imports
|
|
from dinect_api import get_user
|
|
from dinect_api import new_user
|
|
|
|
load_dotenv()
|
|
|
|
is_prod = bool(os.getenv('PRODUCTION', False))
|
|
COUNTRY = os.getenv('COUNTRY', 'RU')
|
|
|
|
|
|
csv_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'csv')
|
|
|
|
files = []
|
|
# r=root, d=directories, f = files
|
|
for r, d, f in os.walk(csv_path):
|
|
for file in f:
|
|
filename, file_extension = os.path.splitext(file)
|
|
if file_extension == '.csv':
|
|
files.append(os.path.join(r, file))
|
|
|
|
for f in files:
|
|
with open(f + f'-{time.strftime("%Y%m%d-%H%M%S", time.localtime())}.log', "w", encoding="utf-8") as log_file:
|
|
with open(f, "r", encoding="utf-8") as csv_file:
|
|
file_name = os.path.basename(f)
|
|
# USERS file
|
|
if 'users' in file_name:
|
|
print(f'Processing "users" file: {f}')
|
|
csv_reader = csv.reader(csv_file, delimiter=',')
|
|
line_count = 0
|
|
for row in csv_reader:
|
|
if line_count == 0:
|
|
line_count += 1
|
|
else:
|
|
try:
|
|
print(f'Processing line {line_count}: {row}')
|
|
nickname, full_name, card, phone, email, gender = row
|
|
# strip whitespaces
|
|
nickname = nickname.strip()
|
|
full_name = full_name.strip()
|
|
card = card.strip()
|
|
# validate phone
|
|
phone = phone.strip()
|
|
try:
|
|
parsed_phone = phonenumbers.parse(phone, region=COUNTRY)
|
|
if phonenumbers.is_valid_number(parsed_phone):
|
|
phone = phonenumbers.format_number(parsed_phone, phonenumbers.PhoneNumberFormat.E164)
|
|
except:
|
|
print(f'error in line: [{line_count}]- Invalid phone number: {phone}')
|
|
log_file.write(f'error in line: [{line_count}]- Invalid phone number: {phone}\n')
|
|
# validate email
|
|
email = email.strip()
|
|
try:
|
|
email_info = validate_email(email, check_deliverability=False)
|
|
email = email_info.normalized
|
|
except EmailNotValidError as e:
|
|
print(f'error in line: [{line_count}]- Invalid email: {email}')
|
|
log_file.write(f'error in line: [{line_count}]- Invalid email: {email}\n')
|
|
|
|
# validate / set gender
|
|
gender = gender.strip()
|
|
if gender not in ['M', 'F']:
|
|
gender = 'M'
|
|
|
|
line_count += 1
|
|
except ValueError as e:
|
|
ret = f'Unexpected error in line: [{line_count}] {repr(e)}'
|
|
log_file.write(f'{ret}\n')
|
|
# Updating the database via the API
|
|
user_found, user_id, user_card, purchases_url, data = get_user(card)
|
|
|
|
if not user_found:
|
|
user_created, data = new_user('Иван тестов', '79039426495')
|
|
if user_created:
|
|
# log_file.write(f'error in line: [{line_count}]- Invalid user data: {data}\n')
|
|
print('User created with', data['ID'], data['DIN'])
|
|
else:
|
|
log_file.write(f'error in line: [{line_count}]- Invalid user data: {data}\n')
|
|
|
|
|
|
|
|
|
|
# TRANSACTIONS file
|
|
if 'transaction' in file_name:
|
|
print(f'Processing "transaction" file: {f}')
|
|
csv_reader = csv.reader(csv_file, delimiter=',')
|
|
line_count = 0
|
|
for row in csv_reader:
|
|
if line_count == 0:
|
|
line_count += 1
|
|
else:
|
|
try:
|
|
print(f'Processing line {line_count}: {row}')
|
|
user_id, card, phone, summ_total, summ_discount, sum_with_discount, bonus_amount, transaction_date, transaction_time = row
|
|
line_count += 1
|
|
|
|
except ValueError as e:
|
|
ret = f'error in line: [{line_count}] {repr(e)}'
|
|
log_file.write(f'{ret}\n')
|
|
|
|
csv_file.close()
|
|
log_file.close()
|
|
|
|
# os.rename(f, f + '.old')
|