115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
import sys
|
|
|
|
if "/service_app_banks" not in list(sys.path):
|
|
sys.path.append("/service_app_banks")
|
|
|
|
import os
|
|
import json
|
|
import arrow
|
|
import typing
|
|
|
|
from logging import getLogger, basicConfig, INFO
|
|
from config_isbank import Config
|
|
from pydantic import BaseModel
|
|
|
|
from databases import AccountRecords, BuildIbans
|
|
|
|
from api_library.date_time_actions.date_functions import system_arrow
|
|
|
|
logger = getLogger(__name__)
|
|
basicConfig(filename=__name__, level=INFO)
|
|
|
|
|
|
class BankReceive(BaseModel):
|
|
import_file_name: str
|
|
iban: str
|
|
bank_date: str
|
|
channel_branch: str
|
|
currency: typing.Optional[str] = "TL"
|
|
currency_value: float
|
|
bank_balance: float
|
|
additional_balance: float
|
|
process_name: str
|
|
process_type: str
|
|
process_comment: str
|
|
bank_reference_code: str
|
|
|
|
|
|
def read_json_file(json_file_path):
|
|
if os.path.exists(json_file_path):
|
|
with open(json_file_path, "r") as json_file:
|
|
return json.load(json_file)
|
|
return {}
|
|
|
|
|
|
def write_json_file(json_file_path, data):
|
|
with open(json_file_path, "w") as json_file:
|
|
json.dump(data, json_file, indent=2)
|
|
|
|
|
|
def sender_service():
|
|
parsed_data = read_json_file(Config.PARSED_PATH)
|
|
if not parsed_data:
|
|
print("Parsed data can not found")
|
|
return
|
|
if is_bank_retrieve_account_records(bank_data=json.dumps(parsed_data)):
|
|
print("is_bank_retrieve_account_records is successful")
|
|
today_date = str(arrow.now("GMT+0"))
|
|
archive_path = str(Config.ARCHIVE_PATH + today_date + ".json").replace(" ", "")
|
|
already_archived = read_json_file(archive_path)
|
|
already_archived[today_date] = parsed_data
|
|
write_json_file(archive_path, already_archived)
|
|
write_json_file(Config.PARSED_PATH, {})
|
|
else:
|
|
print("is_bank_retrieve_account_records is failed")
|
|
|
|
|
|
def is_bank_retrieve_account_records(bank_data):
|
|
data_bulk = json.loads(bank_data)
|
|
new_record_list = []
|
|
for data_keys in data_bulk: # data_bulk is a dict
|
|
for data_dict in data_bulk[data_keys]: # data_bulk[data_keys] is a list
|
|
data_dict["bank_balance"] = data_dict.pop("balance")
|
|
data_dict["import_file_name"] = str(data_keys)
|
|
data_dict = BankReceive(**data_dict).model_dump()
|
|
bank_date = system_arrow.get(str(data_dict["bank_date"]))
|
|
data_dict["bank_date_w"] = bank_date.weekday()
|
|
data_dict["bank_date_m"] = bank_date.month
|
|
data_dict["bank_date_d"] = bank_date.day
|
|
data_dict["bank_date_y"] = bank_date.year
|
|
data_dict["bank_date"] = str(bank_date)
|
|
if build_iban := BuildIbans.filter_by_one(
|
|
iban=data_dict["iban"], system=True
|
|
).data:
|
|
data_dict.update(
|
|
{
|
|
"build_id": build_iban.build_id,
|
|
"build_uu_id": build_iban.build_uu_id,
|
|
}
|
|
)
|
|
if found_record := AccountRecords.filter_one(
|
|
AccountRecords.bank_date == data_dict["bank_date"],
|
|
AccountRecords.iban == data_dict["iban"],
|
|
AccountRecords.bank_reference_code == data_dict["bank_reference_code"],
|
|
AccountRecords.bank_balance == data_dict["bank_balance"],
|
|
system=True,
|
|
).data:
|
|
print("already @database record", found_record.id)
|
|
else:
|
|
new_account_record = AccountRecords.find_or_create(**data_dict)
|
|
new_account_record.save_and_confirm()
|
|
new_record_list.append(new_account_record.get_dict())
|
|
return new_record_list if new_record_list else []
|
|
|
|
|
|
# def send_request_to_api(data):
|
|
# try:
|
|
# headers = {"accept": "application/json", "Content-Type": "application/json"}
|
|
# res = requests.post(url=Config.IP_ADDRESS, json=data, headers=headers)
|
|
# if int(res.status_code) == 200:
|
|
# return res.ok, res.json(), res.status_code
|
|
# except Exception as e:
|
|
# err = e
|
|
# print('Exception',e)
|
|
# return False, {}, 500
|