gpc-generator/aviza/przelewy24_transactions.py

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.
    """
    with open(csv_file, 'rb') as f:
        raw_data = f.read(10000)  # Read first 10KB for encoding detection
    detected = chardet.detect(raw_data)
    encoding = detected['encoding']
    print(f"Detected encoding: {encoding}")

    df = pd.read_csv(csv_file, delimiter=',', dtype=str, encoding=encoding)

    return df


def search_bank_transaction(search_string):
    """
    Searches for a given string in the 10th column (position 9) of the loaded DataFrame.

    :param search_string: String to search for in the specified column.
    :return: The first matching row as a dictionary or None if not found.
    """
    global transactions_df
    # Create a boolean mask based on the 10th column (using positional index)
    mask = transactions_df.iloc[:, 9].str.contains(search_string, na=False, case=False)
    # Use .loc to filter rows; this will always return a DataFrame
    matching_rows = transactions_df.loc[mask]
    # Convert the DataFrame rows into a list of dictionaries
    records = matching_rows.to_dict('records')
    return records[0] if records else None


def extract_and_process_csv_przelewy24_transactions(csv_file_path, bank_statement_file_path, refunds_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)
    grouped_data_refunds = 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}")

    try:
        with open(refunds_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["ID wypłaty"]  # Get payout number from column "N"
                grouped_data_refunds[payout_number].append(row)  # Store row under its payout number

    except Exception as e:
        print(f"Error processing {refunds_file_path}: {e}")

    for payout_number, rows in grouped_data.items():
        try:
            transformed_data = convert_csv_to_gpc_przelewy24_transactions(rows, grouped_data_refunds[payout_number], 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, refunds, payout_id, account_number, currency, mapping):
    gpc_lines = []
    global transactions_df


    total_payout = 0
    total_payout_debet = 0
    total_payout_credit = 0
    total_payout_abs = 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

    corresponding_transaction = search_bank_transaction(payout_id)

    for ref in refunds:
        refund_reference = extract_order_number(ref['Opis transakcji'])
        gpc_data_refund = Data(
            account=account_number,
            payer_account=0,
            no=int(refund_reference) if refund_reference.isdigit() else 0,
            balance=Decimal(ref['Kwota']),
            code=TransactionCode.DEBET,
            variable=int(refund_reference) if refund_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_debet += abs(Decimal(ref['Kwota']))


    real_payout_amount = Decimal(corresponding_transaction['Kwota'].replace("+", "").replace("-", ""))
    real_payout_amount = real_payout_amount.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)
    real_payout_amount_cents = int(real_payout_amount * 100)

    # poplatky
    poplatek_data = Data(
        account=account_number,
        payer_account=0,
        no=int(total_payout_credit-real_payout_amount_cents-total_payout_debet),
        balance=int(total_payout_credit-real_payout_amount_cents-total_payout_debet),
        code=TransactionCode.DEBET,
        variable=0,
        # variable=corresponding_transaction['Zpráva pro příjemce'].split(',')[-1].strip(),
        constant_symbol=0,
        bank_code=0,
        specific_symbol=0,
        client_name="CG poplatek platba",
        currency="1001",
        date=created_on
    )

    gpc_lines.append(poplatek_data.to_string())

    # vyuctovani row
    payout_data = Data(
        account=account_number,
        payer_account=95102013900000630206821286,
        no=int(real_payout_amount_cents),
        balance=real_payout_amount_cents,
        code=TransactionCode.DEBET,
        variable=int(real_payout_amount_cents),
        # 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)