t0is/gpc-generator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

przelewy, packeta comgate

Browse Source
This commit is contained in:
t0is 2025-03-10 14:08:23 +01:00
parent 2b7bc988f9
commit 14a7239068
8 changed files with 907 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
app_main.py
View File

@ -7,7 +7,10 @@ from aviza.gls_single import extract_and_process_zip_gls_single
from csv2gpc import convert_csv_to_gpc, mapping_sparkasse, mapping_pko, mapping_wise
from allegro import convert_csv_to_gpc_allegro, mapping_allegro
from aviza.przelewy24_report_single import extract_and_process_zip_przelewy_report_single
from aviza.paynl_auto import convert_csv_to_gpc_paynl_auto, mapping_paynl_avizo
from aviza.paynl_auto import extract_and_process_zip_paynl_auto, mapping_paynl_avizo
from aviza.przelewy24_auto import extract_and_process_zip_przelewy24_auto, mapping_przelewy_avizo
from aviza.packeta_comgate_convert import extract_and_process_zip_packeta2comgate, mapping_payu
from aviza.przelewy24_transactions import extract_and_process_csv_przelewy24_transactions, mapping_przelewy_transactions
import tempfile
import datetime
@ -125,10 +128,52 @@ def index():
)
elif option == "paynl_auto":
result_data = convert_csv_to_gpc_paynl_auto(tmp_file_path, bank_statement_file_path, f"{option}_vypis_{day_of_year}.gpc",
result_data = extract_and_process_zip_paynl_auto(tmp_file_path, bank_statement_file_path, f"{option}_avizo_{day_of_year}.gpc",
account_number=3498710000999125,
currency="EUR", mapping=mapping_paynl_avizo)
return send_file(
result_data,
download_name=f"paynl_auto_{day_of_year}.zip",
as_attachment=True,
mimetype="application/zip"
)
elif option == "przelewy24_auto":
result_data = extract_and_process_zip_przelewy24_auto(tmp_file_path, bank_statement_file_path, f"{option}_avizo_{day_of_year}.gpc",
account_number=3498710000999133,
currency="PLN", mapping=mapping_przelewy_avizo)
return send_file(
result_data,
download_name=f"przelewy24_auto_{day_of_year}.zip",
as_attachment=True,
mimetype="application/zip"
)
elif option == "przelewy24_transactions":
result_data = extract_and_process_csv_przelewy24_transactions(tmp_file_path, bank_statement_file_path, f"{option}_avizo_{day_of_year}.gpc",
account_number=3498710000999133,
currency="PLN", mapping=mapping_przelewy_transactions)
return send_file(
result_data,
download_name=f"przelewy24_auto_{day_of_year}.zip",
as_attachment=True,
mimetype="application/zip"
)
elif option == "packeta2comgate":
result_data = extract_and_process_zip_packeta2comgate(tmp_file_path, f"{option}_avizo_{day_of_year}.gpc", mapping=mapping_payu)
return send_file(
result_data,
download_name=f"packeta_auto_{day_of_year}.zip",
as_attachment=True,
mimetype="application/zip"
)
if isinstance(result_data, str):
@ -142,4 +187,4 @@ def index():
if __name__ == '__main__':
app.run(debug=True)
app.run(debug=True, port=5002)

15
aviza/helpers.py
View File

@ -28,4 +28,19 @@ def write_output_csv_to_zip(output_file, transformed_data):
zf.writestr(filename, csv_content)
mem_zip.seek(0) # Reset pointer to the beginning of the ZIP archive
return mem_zip
def write_output_gpc_to_zip(output_file, transformed_data):
# Create an in-memory ZIP archive
mem_zip = io.BytesIO()
with zipfile.ZipFile(mem_zip, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
for idx, file_content in enumerate(transformed_data):
filename = f"{idx}_{output_file}"
# Check if file_content is a string; if so, encode it
if isinstance(file_content, str):
encoded_content = file_content.encode("windows-1250")
else:
encoded_content = file_content
zf.writestr(filename, encoded_content)
mem_zip.seek(0) # Reset pointer to the beginning of the ZIP archive
return mem_zip

228
aviza/packeta_comgate_convert.py Normal file
View File

@ -0,0 +1,228 @@
import csv
import pandas as pd
import chardet
import os
import io
import zipfile
import shutil
from datetime import datetime
from gpc import Data, Header, TransactionCode, CURRENCIES_GPC
from utils import extract_order_number, extract_numbers, parse_date
from decimal import Decimal, ROUND_HALF_UP
from aviza.helpers import write_output_gpc_to_zip
def extract_and_process_zip_packeta2comgate(zip_file_path, output_file, mapping):
all_transformed_data = []
# Create a temporary folder for extraction
base_dir = os.path.dirname(zip_file_path)
extract_folder = os.path.join(base_dir, "extracted_temp")
os.makedirs(extract_folder, exist_ok=True)
# Extract the provided outer zip file
with zipfile.ZipFile(zip_file_path, 'r') as zip_ref:
zip_ref.extractall(extract_folder)
currency_folders = {
"EUR": {"currency": "EUR", "account_number": "3498710000999125"},
"HUF": {"currency": "HUF", "account_number": "3498710000999141"},
"RON": {"currency": "RON", "account_number": "3498710000999149"},
}
all_transformed_data = []
# Walk through extract_folder to find the currency folders
for root, dirs, files in os.walk(extract_folder):
for folder in dirs:
if folder in currency_folders:
folder_path = os.path.join(root, folder)
currency_info = currency_folders[folder]
currency = currency_info["currency"]
account_number = currency_info["account_number"]
# Process CSV files inside the currency folder
for file in os.listdir(folder_path):
if file.endswith(".csv") and not file.startswith("._"):
file_path = os.path.join(folder_path, file)
try:
transformed_data = convert_payu2comgate(file_path, account_number, currency, mapping)
all_transformed_data.append(transformed_data)
except Exception as e:
print(f"Error processing {file}: {e}")
# Clean up the outer extraction folder
shutil.rmtree(extract_folder)
print(f"Processed and cleaned up: {zip_file_path}")
# Return a ZIP archive (in memory) containing all the output CSV files.
return write_output_gpc_to_zip(output_file, all_transformed_data)
def convert_payu2comgate(csv_file_path, account_number, currency, mapping):
gpc_lines = []
if mapping['forced_encoding'] is not None:
detected_encoding = mapping['forced_encoding']
print(f"Forced encoding: {detected_encoding}")
else:
with open(csv_file_path, 'rb') as f:
rawdata = f.read(1024) # Read a small part of the file
result = chardet.detect(rawdata)
detected_encoding = result['encoding']
print(f"Detected encoding: {detected_encoding}")
with open(csv_file_path, mode='r', encoding=detected_encoding) as csv_file:
reader = csv.DictReader(csv_file, delimiter=mapping['delimiter']) if mapping['is_dict_mapping'] else csv.reader(
csv_file, delimiter=mapping['delimiter'])
if not mapping['is_dict_mapping']:
next(reader)
rows = list(reader) # Convert to list to easily access the last row
if not rows: # Ensure there's at least one row
raise ValueError("CSV file is empty after processing.")
total_payout = 0.0
total_payout_debet = 0.0
total_payout_credit = 0.0
total_payout_abs = 0.0
first = True
created_on = None
for row in rows[:-1]:
if first:
first = False
reference = extract_order_number(row[mapping['reference']])
transaction_id = extract_numbers(row[mapping['transaction_id']]) if mapping['use_transaction_id'] else reference
direction = row[mapping['direction']].lower() if mapping['direction'] is not None else None
source_name = row[mapping['source_name']].replace("Nazwa nadawcy: ", "")[:20].ljust(20) if mapping['use_source_name'] else ""
payer_account = extract_numbers(row[mapping['payer_account']].replace("Rachunek nadawcy: ", "").replace(" ", ""))[:16].ljust(16) if mapping['payer_number_exists'] else 0
if payer_account != 0:
if payer_account.strip() == "":
payer_account = 0
created_on = parse_date(row[mapping['created_on']])
# Convert the value using Decimal instead of float
source_str = row[mapping['source_amount']].replace(',', '.')
source_amount = Decimal(source_str)
# Quantize to 2 decimal places (rounding if necessary)
source_amount = source_amount.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
# Multiply by 100 to get cents (if that's what you need) and convert to integer
source_amount_cents = int(source_amount * 100)
# Determine transaction type
if(direction is None):
if source_amount_cents > 0:
transaction_code = TransactionCode.CREDIT
else:
transaction_code = TransactionCode.DEBET
else:
if direction == "out":
transaction_code = TransactionCode.DEBET
else:
transaction_code = TransactionCode.CREDIT
# Convert currency
currency_code = "100" + str(transaction_code.value)
# Create GPC Data object
gpc_data = Data(
account=account_number,
payer_account=payer_account,
no=transaction_id,
balance=source_amount_cents,
code=transaction_code,
variable=int(reference) if reference.isdigit() else 0,
constant_symbol=0,
bank_code=0,
specific_symbol=0,
client_name="CG ABCD-EFGH-IJKL" if transaction_code == TransactionCode.CREDIT else "CG refundace",
currency=currency_code,
date=created_on
)
gpc_lines.append(gpc_data.to_string())
total_payout += source_amount_cents
total_payout_abs += abs(source_amount_cents)
total_payout_debet += abs(source_amount_cents) if transaction_code == TransactionCode.DEBET else 0
total_payout_credit += abs(source_amount_cents) if transaction_code == TransactionCode.CREDIT else 0
# break
# vyuctovani row
payout_data = Data(
account=account_number,
payer_account=account_number,
no=int(total_payout),
balance=total_payout,
code=TransactionCode.DEBET,
variable=int(total_payout),
# variable=corresponding_transaction['Zpráva pro příjemce'].split(',')[-1].strip(),
constant_symbol=0,
bank_code=0,
specific_symbol=0,
client_name="CG vyúčtování",
currency="1001",
date=created_on
)
total_payout_credit += abs(total_payout)
total_payout_abs += abs(total_payout)
gpc_lines.append(payout_data.to_string())
# Create and add the header
header = Header(
account=account_number,
account_name=currency.ljust(20),
old_date=datetime.strptime("01-03-20", "%d-%m-%y"),
old_balance=0,
old_sign='+',
new_balance=0,
new_sign='+',
turnover_debet=total_payout,
turnover_debet_sign='+',
turnover_credit=total_payout,
turnover_credit_sign='+',
transaction_list_no=3,
date=datetime.now()
)
gpc_lines.insert(0, header.to_string())
file_content = "".join(gpc_lines)
print(f"GPC file content successfully created")
return file_content.encode("windows-1250")
mapping_payu = {
'transaction_id': 'ID transakce',
'reference': 'ID objednávky',
'direction': None,
'source_name': 'Jméno a příjmení',
'source_amount': 'Částka',
'source_currency': 'Měna',
'payer_account': None,
'created_on': 'Datum',
'fees': 'Provize',
'delimiter': ";",
'is_dict_mapping': True,
'use_transaction_id': True,
'payer_number_exists': False,
'use_source_name': False,
'forced_encoding': "windows-1250"
}
# Example usage:
# convert_csv_to_gpc("../Specification clearing 2024-05-10.csv", "avizo_przelewy24_test.gpc", account_number=3498710000999117, currency="EUR", mapping=mapping_przelewy24_avizo)
# convert_csv_to_gpc("pko_input.csv", "pko_output.gpc", account_number=95102013900000630206821286, currency="PLN", mapping=mapping_pko)
# convert_csv_to_gpc("sparkasse_input.csv", "sparkasse_output.gpc", account_number=95850503000221267034, currency="EUR", mapping=mapping_sparkasse)

99
aviza/paynl_auto.py
View File

@ -1,10 +1,15 @@
import csv
import pandas as pd
import chardet
import os
import io
import zipfile
import shutil
from datetime import datetime
from gpc import Data, Header, TransactionCode, CURRENCIES_GPC
from utils import extract_order_number, extract_numbers, parse_date
from decimal import Decimal, ROUND_HALF_UP
from aviza.helpers import write_output_gpc_to_zip
def load_bank_transactions(csv_file):
"""
@ -35,10 +40,71 @@ def search_bank_transaction(search_string):
return matching_row.iloc[0].to_dict() if not matching_row.empty else None
def convert_csv_to_gpc_paynl_auto(csv_file_path, bank_statement_file_path, gpc_file_path, account_number, currency, mapping):
gpc_lines = []
def extract_and_process_zip_paynl_auto(zip_file_path, bank_statement_file_path, output_file, account_number, currency, mapping):
global transactions_df
transactions_df = load_bank_transactions(bank_statement_file_path)
all_transformed_data = []
# Create a temporary folder for extraction
base_dir = os.path.dirname(zip_file_path)
extract_folder = os.path.join(base_dir, "extracted_temp")
os.makedirs(extract_folder, exist_ok=True)
# Extract the provided outer zip file
with zipfile.ZipFile(zip_file_path, 'r') as zip_ref:
zip_ref.extractall(extract_folder)
# Check if a top-level "Transactions" folder exists
transactions_folder = os.path.join(extract_folder, "Transactions")
if os.path.exists(transactions_folder):
# Look for a CSV file inside the Transactions folder
csv_filename = next(
(f for f in os.listdir(transactions_folder)
if f.startswith("Specification clearing") and f.endswith(".csv")),
None
)
if csv_filename:
csv_path = os.path.join(transactions_folder, csv_filename)
transformed_data = convert_csv_to_gpc_paynl_auto(csv_path, account_number, currency, mapping)
all_transformed_data.append(transformed_data)
else:
# Otherwise, look for inner zip files named "Specification clearing*.zip"
for root, dirs, files in os.walk(extract_folder):
for file in files:
if file.startswith("Specification clearing") and file.endswith(".zip"):
inner_zip_path = os.path.join(root, file)
# Create a temporary folder for the inner zip extraction
inner_extract_folder = os.path.join(root, "inner_extracted")
os.makedirs(inner_extract_folder, exist_ok=True)
with zipfile.ZipFile(inner_zip_path, 'r') as inner_zip:
inner_zip.extractall(inner_extract_folder)
# Expect the inner zip to contain a "Transactions" folder with the CSV
inner_transactions_folder = os.path.join(inner_extract_folder, "Transactions")
if os.path.exists(inner_transactions_folder):
inner_csv_filename = next(
(f for f in os.listdir(inner_transactions_folder)
if f.startswith("Specification clearing") and f.endswith(".csv")),
None
)
if inner_csv_filename:
inner_csv_path = os.path.join(inner_transactions_folder, inner_csv_filename)
transformed_data = convert_csv_to_gpc_paynl_auto(inner_csv_path, account_number, currency, mapping)
all_transformed_data.append(transformed_data)
# Clean up inner extraction folder
shutil.rmtree(inner_extract_folder)
# Clean up the outer extraction folder
shutil.rmtree(extract_folder)
print(f"Processed and cleaned up: {zip_file_path}")
# Return a ZIP archive (in memory) containing all the output CSV files.
return write_output_gpc_to_zip(output_file, all_transformed_data)
def convert_csv_to_gpc_paynl_auto(csv_file_path, account_number, currency, mapping):
gpc_lines = []
global transactions_df
if mapping['forced_encoding'] is not None:
@ -151,10 +217,10 @@ def convert_csv_to_gpc_paynl_auto(csv_file_path, bank_statement_file_path, gpc_f
payout_data = Data(
account=account_number,
payer_account=2801379531,
no=666111222,
no=int(total_payout),
balance=total_payout,
code=TransactionCode.DEBET,
variable=666111222,
variable=int(total_payout),
# variable=corresponding_transaction['Zpráva pro příjemce'].split(',')[-1].strip(),
constant_symbol=0,
bank_code=0,
@ -187,13 +253,9 @@ def convert_csv_to_gpc_paynl_auto(csv_file_path, bank_statement_file_path, gpc_f
)
gpc_lines.insert(0, header.to_string())
# with open(gpc_file_path, mode='w', encoding='utf-8') as gpc_file:
# gpc_file.writelines(gpc_lines)
#
# print(f"GPC file successfully created: {gpc_file_path}")
file_content = "".join(gpc_lines)
print(f"GPC file content successfully created for: {gpc_file_path}")
print(f"GPC file content successfully created")
return file_content.encode("windows-1250")
# Example mappings
@ -216,6 +278,25 @@ mapping_paynl_avizo = {
'forced_encoding': None
}
mapping_przelewy_avizo = {
'transaction_id': 'Transaction id',
'reference': 'Title',
'direction': None,
'source_name': 'Payment method name',
'source_amount': 'Amount',
'source_currency': '',
'payer_account': 'Session id',
'created_on': 'Creation date',
'CLEARING_ID': 'CLEARING_ID',
'fees': 'Commission',
'delimiter': ",",
'is_dict_mapping': True,
'use_transaction_id': True,
'payer_number_exists': False,
'use_source_name': False,
'forced_encoding': None
}
# Example usage:
# convert_csv_to_gpc("../Specification clearing 2024-05-10.csv", "avizo_paynl_test.gpc", account_number=3498710000999117, currency="EUR", mapping=mapping_paynl_avizo)

264
aviza/przelewy24_auto.py Normal file
View File

@ -0,0 +1,264 @@
import csv
import pandas as pd
import chardet
import os
import io
import zipfile
import shutil
from datetime import datetime
from gpc import Data, Header, TransactionCode, CURRENCIES_GPC
from utils import extract_order_number, extract_numbers, parse_date
from decimal import Decimal, ROUND_HALF_UP
from aviza.helpers import write_output_gpc_to_zip
def load_bank_transactions(csv_file):
"""
Loads the bank transactions CSV file into a DataFrame and returns it.
:param csv_file: Path to the bank transactions CSV file.
:return: A pandas DataFrame containing the transactions.
"""
df = pd.read_csv(csv_file, delimiter=';', dtype=str)
# Ensure the required column exists
if 'Zpráva pro příjemce' not in df.columns:
raise ValueError("The CSV file does not contain the required column 'Zpráva pro příjemce'.")
return df
def search_bank_transaction(search_string):
"""
Searches for a given string in the 'Zpráva pro příjemce' column of the loaded DataFrame.
:param df: Pandas DataFrame containing bank transactions.
:param search_string: String to search for in the 'Zpráva pro příjemce' column.
:return: The first matching row as a dictionary or None if not found.
"""
global transactions_df
matching_row = transactions_df[transactions_df['Zpráva pro příjemce'].str.contains(search_string, na=False, case=False)]
return matching_row.iloc[0].to_dict() if not matching_row.empty else None
def extract_and_process_zip_przelewy24_auto(zip_file_path, bank_statement_file_path, output_file, account_number, currency, mapping):
global transactions_df
transactions_df = load_bank_transactions(bank_statement_file_path)
all_transformed_data = []
# Create a temporary folder for extraction
base_dir = os.path.dirname(zip_file_path)
extract_folder = os.path.join(base_dir, "extracted_temp")
os.makedirs(extract_folder, exist_ok=True)
# Extract the provided outer zip file
with zipfile.ZipFile(zip_file_path, 'r') as zip_ref:
zip_ref.extractall(extract_folder)
for files in os.walk(extract_folder):
for file in files:
try:
if file.endswith(".csv"):
transformed_data = convert_csv_to_gpc_przelewy24_auto(file, account_number, currency, mapping)
all_transformed_data.append(transformed_data)
except:
continue
# Clean up the outer extraction folder
shutil.rmtree(extract_folder)
print(f"Processed and cleaned up: {zip_file_path}")
# Return a ZIP archive (in memory) containing all the output CSV files.
return write_output_gpc_to_zip(output_file, all_transformed_data)
def convert_csv_to_gpc_przelewy24_auto(csv_file_path, account_number, currency, mapping):
gpc_lines = []
global transactions_df
if mapping['forced_encoding'] is not None:
detected_encoding = mapping['forced_encoding']
print(f"Forced encoding: {detected_encoding}")
else:
with open(csv_file_path, 'rb') as f:
rawdata = f.read(1024) # Read a small part of the file
result = chardet.detect(rawdata)
detected_encoding = result['encoding']
print(f"Detected encoding: {detected_encoding}")
with open(csv_file_path, mode='r', encoding=detected_encoding) as csv_file:
# Skip the first 3 rows
for _ in range(3):
next(csv_file)
reader = csv.DictReader(csv_file, delimiter=mapping['delimiter']) if mapping['is_dict_mapping'] else csv.reader(
csv_file, delimiter=mapping['delimiter'])
if not mapping['is_dict_mapping']:
next(reader)
total_payout = 0.0
total_payout_debet = 0.0
total_payout_credit = 0.0
total_payout_abs = 0.0
first = True
clearing_id = ""
created_on = None
for row in reader:
if first:
# clearing_id = row[mapping['CLEARING_ID']]
first = False
reference = extract_order_number(row[mapping['reference']])
transaction_id = extract_numbers(row[mapping['transaction_id']]) if mapping['use_transaction_id'] else reference
direction = row[mapping['direction']].lower() if mapping['direction'] is not None else None
source_name = row[mapping['source_name']].replace("Nazwa nadawcy: ", "")[:20].ljust(20) if mapping['use_source_name'] else ""
payer_account = extract_numbers(row[mapping['payer_account']].replace("Rachunek nadawcy: ", "").replace(" ", ""))[:16].ljust(16) if mapping['payer_number_exists'] else 0
if payer_account != 0:
if payer_account.strip() == "":
payer_account = 0
created_on = parse_date(row[mapping['created_on']])
# Convert the value using Decimal instead of float
source_str = row[mapping['source_amount']].replace(',', '.')
source_amount = Decimal(source_str)
# Quantize to 2 decimal places (rounding if necessary)
source_amount = source_amount.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
# Multiply by 100 to get cents (if that's what you need) and convert to integer
source_amount_cents = int(source_amount * 100)
# Determine transaction type
if(direction is None):
if source_amount_cents > 0:
transaction_code = TransactionCode.CREDIT
else:
transaction_code = TransactionCode.DEBET
else:
if direction == "out":
transaction_code = TransactionCode.DEBET
else:
transaction_code = TransactionCode.CREDIT
# Convert currency
currency_code = "100" + str(transaction_code.value)
# Create GPC Data object
gpc_data = Data(
account=account_number,
payer_account=payer_account,
no=transaction_id,
balance=source_amount_cents,
code=transaction_code,
variable=int(reference) if reference.isdigit() else 0,
constant_symbol=0,
bank_code=0,
specific_symbol=0,
client_name="CG ABCD-EFGH-IJKL" if transaction_code == TransactionCode.CREDIT else "CG refundace",
currency=currency_code,
date=created_on
)
gpc_lines.append(gpc_data.to_string())
total_payout += source_amount_cents
total_payout_abs += abs(source_amount_cents)
total_payout_debet += abs(source_amount_cents) if transaction_code == TransactionCode.DEBET else 0
total_payout_credit += abs(source_amount_cents) if transaction_code == TransactionCode.CREDIT else 0
# break
# add fees
# fees_data = Data(
# account=account_number,
# payer_account=0,
# no=0,
# balance=total_fees,
# code=transaction_code,
# variable=0,
# constant_symbol=0,
# bank_code=0,
# specific_symbol=0,
# client_name="",
# currency=CURRENCIES_GPC.get("CZK", "0000"),
# date=created_on
# )
#
# gpc_lines.append(gpc_data.to_string())
# corresponding_transaction = search_bank_transaction(clearing_id)
# vyuctovani row
payout_data = Data(
account=account_number,
payer_account=95102013900000630206821286,
no=int(total_payout),
balance=total_payout,
code=TransactionCode.DEBET,
variable=int(total_payout),
# variable=corresponding_transaction['Zpráva pro příjemce'].split(',')[-1].strip(),
constant_symbol=0,
bank_code=0,
specific_symbol=0,
client_name="CG vyúčtování",
currency="1001",
date=created_on
)
total_payout_credit += abs(total_payout)
total_payout_abs += abs(total_payout)
gpc_lines.append(payout_data.to_string())
# Create and add the header
header = Header(
account=account_number,
account_name=currency.ljust(20),
old_date=datetime.strptime("01-03-20", "%d-%m-%y"),
old_balance=0,
old_sign='+',
new_balance=0,
new_sign='+',
turnover_debet=total_payout,
turnover_debet_sign='+',
turnover_credit=total_payout,
turnover_credit_sign='+',
transaction_list_no=3,
date=datetime.now()
)
gpc_lines.insert(0, header.to_string())
file_content = "".join(gpc_lines)
print(f"GPC file content successfully created")
return file_content.encode("windows-1250")
mapping_przelewy_avizo = {
'transaction_id': 'Transaction id',
'reference': 'Title',
'direction': None,
'source_name': 'Payment method name',
'source_amount': 'Amount',
'source_currency': '',
'payer_account': 'Session id',
'created_on': 'Creation date',
'CLEARING_ID': 'CLEARING_ID',
'fees': 'Commission',
'delimiter': ",",
'is_dict_mapping': True,
'use_transaction_id': True,
'payer_number_exists': False,
'use_source_name': False,
'forced_encoding': "UTF-8"
}
# Example usage:
# convert_csv_to_gpc("../Specification clearing 2024-05-10.csv", "avizo_przelewy24_test.gpc", account_number=3498710000999117, currency="EUR", mapping=mapping_przelewy24_avizo)
# convert_csv_to_gpc("pko_input.csv", "pko_output.gpc", account_number=95102013900000630206821286, currency="PLN", mapping=mapping_pko)
# convert_csv_to_gpc("sparkasse_input.csv", "sparkasse_output.gpc", account_number=95850503000221267034, currency="EUR", mapping=mapping_sparkasse)

248
aviza/przelewy24_transactions.py Normal file
View File

@ -0,0 +1,248 @@
import csv
import pandas as pd
import chardet
import os
import io
import zipfile
import shutil
from datetime import datetime
from gpc import Data, Header, TransactionCode, CURRENCIES_GPC
from utils import extract_order_number, extract_numbers, parse_date
from decimal import Decimal, ROUND_HALF_UP
from aviza.helpers import write_output_gpc_to_zip
from collections import defaultdict
def load_bank_transactions(csv_file):
"""
Loads the bank transactions CSV file into a DataFrame and returns it.
:param csv_file: Path to the bank transactions CSV file.
:return: A pandas DataFrame containing the transactions.
"""
df = pd.read_csv(csv_file, delimiter=';', dtype=str)
# Ensure the required column exists
if 'Zpráva pro příjemce' not in df.columns:
raise ValueError("The CSV file does not contain the required column 'Zpráva pro příjemce'.")
return df
def search_bank_transaction(search_string):
"""
Searches for a given string in the 'Zpráva pro příjemce' column of the loaded DataFrame.
:param df: Pandas DataFrame containing bank transactions.
:param search_string: String to search for in the 'Zpráva pro příjemce' column.
:return: The first matching row as a dictionary or None if not found.
"""
global transactions_df
matching_row = transactions_df[transactions_df['Zpráva pro příjemce'].str.contains(search_string, na=False, case=False)]
return matching_row.iloc[0].to_dict() if not matching_row.empty else None
def extract_and_process_csv_przelewy24_transactions(csv_file_path, bank_statement_file_path, output_file, account_number, currency, mapping):
global transactions_df
transactions_df = load_bank_transactions(bank_statement_file_path)
all_transformed_data = []
grouped_data = defaultdict(list)
if mapping['forced_encoding'] is not None:
detected_encoding = mapping['forced_encoding']
print(f"Forced encoding: {detected_encoding}")
else:
with open(csv_file_path, 'rb') as f:
rawdata = f.read(1024) # Read a small part of the file
result = chardet.detect(rawdata)
detected_encoding = result['encoding']
print(f"Detected encoding: {detected_encoding}")
try:
with open(csv_file_path, mode="r", encoding=detected_encoding) as csv_file:
reader = csv.DictReader(csv_file, delimiter=mapping['delimiter'])
for row in reader:
payout_number = row["Numer wypłaty"] # Get payout number from column "N"
grouped_data[payout_number].append(row) # Store row under its payout number
except Exception as e:
print(f"Error processing {csv_file_path}: {e}")
for payout_number, rows in grouped_data.items():
try:
transformed_data = convert_csv_to_gpc_przelewy24_transactions(rows, payout_number, account_number, currency, mapping)
all_transformed_data.append(transformed_data)
except Exception as e:
print(f"Error processing payout number {payout_number}: {e}")
# Return a ZIP archive (in memory) containing all the output CSV files.
return write_output_gpc_to_zip(output_file, all_transformed_data)
def convert_csv_to_gpc_przelewy24_transactions(rows, payout_id, account_number, currency, mapping):
gpc_lines = []
global transactions_df
total_payout = 0.0
total_payout_debet = 0.0
total_payout_credit = 0.0
total_payout_abs = 0.0
first = True
clearing_id = ""
created_on = None
for row in rows:
if first:
# clearing_id = row[mapping['CLEARING_ID']]
first = False
reference = extract_order_number(row[mapping['reference']])
transaction_id = extract_numbers(row[mapping['transaction_id']]) if mapping['use_transaction_id'] else reference
direction = row[mapping['direction']].lower() if mapping['direction'] is not None else None
source_name = row[mapping['source_name']].replace("Nazwa nadawcy: ", "")[:20].ljust(20) if mapping['use_source_name'] else ""
payer_account = extract_numbers(row[mapping['payer_account']].replace("Rachunek nadawcy: ", "").replace(" ", ""))[:16].ljust(16) if mapping['payer_number_exists'] else 0
if payer_account != 0:
if payer_account.strip() == "":
payer_account = 0
created_on = parse_date(row[mapping['created_on']])
# Convert the value using Decimal instead of float
source_amount_cents = int(row[mapping['source_amount']])
refund_amount_cents = int(row[mapping['refund_amount']])
# Determine transaction type
if(direction is None):
if source_amount_cents > 0:
transaction_code = TransactionCode.CREDIT
else:
transaction_code = TransactionCode.DEBET
else:
if direction == "out":
transaction_code = TransactionCode.DEBET
else:
transaction_code = TransactionCode.CREDIT
# Convert currency
currency_code = "100" + str(TransactionCode.CREDIT)
# Create GPC Data object
gpc_data = Data(
account=account_number,
payer_account=payer_account,
no=transaction_id,
balance=source_amount_cents,
code=TransactionCode.CREDIT,
variable=int(reference) if reference.isdigit() else 0,
constant_symbol=0,
bank_code=0,
specific_symbol=0,
client_name="CG ABCD-EFGH-IJKL",
currency="1002",
date=created_on
)
gpc_lines.append(gpc_data.to_string())
total_payout += source_amount_cents
total_payout_abs += abs(source_amount_cents)
total_payout_debet += abs(source_amount_cents) if transaction_code == TransactionCode.DEBET else 0
total_payout_credit += abs(source_amount_cents) if transaction_code == TransactionCode.CREDIT else 0
if refund_amount_cents != 0:
gpc_data_refund = Data(
account=account_number,
payer_account=payer_account,
no=transaction_id,
balance=refund_amount_cents,
code=TransactionCode.DEBET,
variable=int(reference) if reference.isdigit() else 0,
constant_symbol=0,
bank_code=0,
specific_symbol=0,
client_name="CG refundace",
currency="1001",
date=created_on
)
gpc_lines.append(gpc_data_refund.to_string())
total_payout += refund_amount_cents
total_payout_abs += abs(refund_amount_cents)
total_payout_debet += abs(refund_amount_cents) if transaction_code == TransactionCode.DEBET else 0
# use payout_id instead as variable
# vyuctovani row
payout_data = Data(
account=account_number,
payer_account=95102013900000630206821286,
no=int(total_payout),
balance=total_payout,
code=TransactionCode.DEBET,
variable=int(total_payout),
# variable=corresponding_transaction['Zpráva pro příjemce'].split(',')[-1].strip(),
constant_symbol=0,
bank_code=0,
specific_symbol=0,
client_name="CG vyúčtování",
currency="1001",
date=created_on
)
total_payout_credit += abs(total_payout)
total_payout_abs += abs(total_payout)
gpc_lines.append(payout_data.to_string())
# Create and add the header
header = Header(
account=account_number,
account_name=currency.ljust(20),
old_date=datetime.strptime("01-03-20", "%d-%m-%y"),
old_balance=0,
old_sign='+',
new_balance=0,
new_sign='+',
turnover_debet=total_payout,
turnover_debet_sign='+',
turnover_credit=total_payout,
turnover_credit_sign='+',
transaction_list_no=3,
date=datetime.now()
)
gpc_lines.insert(0, header.to_string())
file_content = "".join(gpc_lines)
print(f"GPC file content successfully created")
return file_content.encode("windows-1250")
mapping_przelewy_transactions = {
'transaction_id': 'Numeryczne ID transakcji',
'reference': 'Opis',
'direction': None,
'source_name': 'Metoda płatności',
'source_amount': 'Kwota',
'refund_amount': 'Zwroty',
'source_currency': '',
'payer_account': 'Numeryczne ID transakcji',
'created_on': 'Przyjęcie',
'delimiter': ",",
'is_dict_mapping': True,
'use_transaction_id': True,
'payer_number_exists': False,
'use_source_name': False,
'forced_encoding': "UTF-8"
}
# Example usage:
# convert_csv_to_gpc("../Specification clearing 2024-05-10.csv", "avizo_przelewy24_test.gpc", account_number=3498710000999117, currency="EUR", mapping=mapping_przelewy24_avizo)
# convert_csv_to_gpc("pko_input.csv", "pko_output.gpc", account_number=95102013900000630206821286, currency="PLN", mapping=mapping_pko)
# convert_csv_to_gpc("sparkasse_input.csv", "sparkasse_output.gpc", account_number=95850503000221267034, currency="EUR", mapping=mapping_sparkasse)

14
templates/index.html
View File

@ -82,18 +82,30 @@
<p class="text-gray-600 text-sm italic">Pouzijte .zip se vsemi zip soubory z exportu</p>
<p class="text-gray-600 text-sm">.gpc aviza pro EUR ucet</p>
</div>
<div class="tile bg-white shadow-md rounded-lg p-4 cursor-pointer hover:bg-blue-50 opacity-25 pointer-events-none"
<div class="tile bg-white shadow-md rounded-lg p-4 cursor-pointer hover:bg-blue-50"
data-option="przelewy24_auto" data-section="aviza">
<h2 class="text-xl font-semibold mb-2">Przelewy24 .gpc</h2>
<p class="text-gray-600 text-sm italic">Pouzijte .zip se vsemi .csv soubory z reportu</p>
<p class="text-gray-600 text-sm">.gpc aviza pro Przelewy24</p>
</div>
<div class="tile bg-white shadow-md rounded-lg p-4 cursor-pointer hover:bg-blue-50"
data-option="przelewy24_transactions" data-section="aviza">
<h2 class="text-xl font-semibold mb-2">Przelewy24 transakce .gpc</h2>
<p class="text-gray-600 text-sm italic">Pouzijte jeden .csv soubor z transakci</p>
<p class="text-gray-600 text-sm">.gpc aviza pro Przelewy24</p>
</div>
<div class="tile bg-white shadow-md rounded-lg p-4 cursor-pointer hover:bg-blue-50 opacity-25 pointer-events-none"
data-option="gls_auto_huf" data-section="aviza">
<h2 class="text-xl font-semibold mb-2">GLS HU .gpc</h2>
<p class="text-gray-600 text-sm italic">Pouzijte .zip se vsemi zip soubory z exportu (z disku zazipovat slozku <i>Aviza GLS</i>)</p>
<p class="text-gray-600 text-sm">.gpc aviza pro HUF dobirky</p>
</div>
<div class="tile bg-white shadow-md rounded-lg p-4 cursor-pointer hover:bg-blue-50"
data-option="packeta2comgate" data-section="aviza-other">
<h2 class="text-xl font-semibold mb-2">Packeta .gpc</h2>
<p class="text-gray-600 text-sm italic">Pouzijte .zip se vsemi avizy ze zasilkovny</p>
<p class="text-gray-600 text-sm">.gpc aviza pro dobirky z packety</p>
</div>

1
utils.py
View File

@ -31,3 +31,4 @@ def parse_date(date_str):
continue # Try the next format
raise ValueError(f"Unsupported date format: {date_str}") # Raise error if none match
return datetime(Now)