import csv import pandas as pd import chardet import os import io import zipfile import shutil from datetime import datetime, date 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 GLOBAL_CURRENCY = None transactions_df = None 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. """ global GLOBAL_CURRENCY df = pd.read_csv(csv_file, delimiter=',', dtype=str) if GLOBAL_CURRENCY == "HUF": if 'Reference' not in df.columns: raise ValueError("The CSV file does not contain the required column 'Reference'.") if GLOBAL_CURRENCY == "RON": if 'Reference' not in df.columns: raise ValueError("The CSV file does not contain the required column 'Reference'.") 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 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 GLOBAL_CURRENCY, transactions_df if GLOBAL_CURRENCY == "HUF": row_title = 'Reference' if GLOBAL_CURRENCY == "RON": row_title = 'Reference' matching_row = transactions_df[transactions_df[row_title].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_gls_auto(zip_file_path, bank_statement_file_path, output_file, account_number, currency): global GLOBAL_CURRENCY, transactions_df all_transformed_data = [] transactions_df = load_bank_transactions(bank_statement_file_path) GLOBAL_CURRENCY = currency 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 zip file. with zipfile.ZipFile(zip_file_path, 'r') as zip_ref: zip_ref.extractall(extract_folder) # Search for the folder named 'GLOBAL_CURRENCY' within the extracted contents. global_currency_folder = None for root, dirs, files in os.walk(os.path.join(extract_folder, 'Aviza GLS')): if os.path.basename(root) == GLOBAL_CURRENCY: global_currency_folder = root break if global_currency_folder: # Process every CSV file found in the GLOBAL_CURRENCY folder. for filename in os.listdir(global_currency_folder): if filename.endswith(".xlsx"): csv_path = os.path.join(global_currency_folder, filename) transformed_data = convert_csv_to_gpc_gls_auto(csv_path, account_number, currency) all_transformed_data.append(transformed_data) else: print(f"'GLOBAL_CURRENCY' folder not found in zip file") # Clean up extracted files. for root, dirs, files in os.walk(extract_folder, topdown=False): for file in files: os.remove(os.path.join(root, file)) for dir in dirs: os.rmdir(os.path.join(root, dir)) os.rmdir(extract_folder) print(f"Processed and cleaned up") # Write all collected transformed data to the output file. return write_output_gpc_to_zip(output_file, all_transformed_data) def convert_csv_to_gpc_gls_auto(csv_file_path, account_number, currency): gpc_lines = [] global transactions_df df = pd.read_excel(csv_file_path, skiprows=7, dtype=str) if currency == "HUF": payment_date = datetime.strptime(csv_file_path.split("_" + currency + "_")[1].split("_")[0], "%Y%m%d").strftime( "%Y.%m.%d") elif currency == "RON": payment_date = datetime.strptime(csv_file_path.split("_" + currency + "_")[1].replace(".xlsx", ""), "%d%m%Y").strftime("%Y.%m.%d") df.iloc[:, 4] = pd.to_numeric(df.iloc[:, 4].str.replace(',', '.'), errors='coerce').fillna(0) cumsum = 0.00 transformed_data = [] total_rows = len(df) total_payout = 0.0 total_payout_debet = 0.0 total_payout_credit = 0.0 total_payout_abs = 0.0 for index, row in df.iterrows(): reference = "0" if pd.isna(row.iloc[2]) else extract_order_number(row.iloc[2]) transaction_id = extract_numbers(row.iloc[0]) direction = None source_name = "" payer_account = 0 created_on = parse_date(row.iloc[3]) # Convert the value using Decimal instead of float source_str = row.iloc[4] 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 if index == total_rows - 2: break corresponding_transaction = search_bank_transaction(payment_date) # 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=parse_date(corresponding_transaction['Created on']) if corresponding_transaction is not None else datetime.now() ) 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=int(date.today().strftime("%j")), 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") def transform_csv(input_file): global transactions_df, GLOBAL_CURRENCY df = pd.read_excel(input_file, skiprows=7, dtype=str) payment_date = datetime.strptime(input_file.split("_" + GLOBAL_CURRENCY + "_")[1].split("_")[0], "%Y%m%d").strftime("%Y.%m.%d") df.iloc[:, 4] = pd.to_numeric(df.iloc[:, 4].str.replace(',', '.'), errors='coerce').fillna(0) cumsum = 0.00 transformed_data = [] total_rows = len(df) for index, row in df.iterrows(): amount = row.iloc[4] typ_operace = "t" if amount >= 0 else "c" cumsum += amount transformed_row = [ typ_operace, row.iloc[3], row.iloc[0], amount, cumsum, "TRUE", row.iloc[2], f"Dobirka za FA s VS {row.iloc[2]}", "", "", "", "", "", "", row.iloc[1], 0, GLOBAL_CURRENCY ] transformed_data.append(transformed_row) progress = (index + 1) / total_rows * 100 sys.stdout.write(f"\rProcessing: {progress:.2f}%") sys.stdout.flush() if index == total_rows - 2: break total_sum = cumsum corresponding_transaction = search_bank_transaction(payment_date) final_row = ["w", datetime.strptime(corresponding_transaction['Created on'], "%Y-%m-%d %H:%M:%S").strftime("%Y-%m-%d"), corresponding_transaction['ID'].split('-')[-1].strip(), -total_sum, 0, "TRUE", "", "Vyrovnání zůstatku", "12600016-16965466-28438156", "", "", "", "", "", "NEWLINE BREAK", "", GLOBAL_CURRENCY] transformed_data.append(final_row) return transformed_data