import csv import chardet 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 import re def convert_csv_to_gpc(csv_file_path, gpc_file_path, account_number, currency, mapping): gpc_lines = [] # 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=78449, old_sign='+', new_balance=6215449, new_sign='+', turnover_debet=6585600, turnover_debet_sign='0', turnover_credit=127226, turnover_credit_sign='0', transaction_list_no=int(date.today().strftime("%j")), date=datetime.now() ) gpc_lines.append(header.to_string()) 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) for row in reader: 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 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 = int(source_amount * 100) created_on = parse_date(row[mapping['created_on']]) if reference == '0' and row[mapping['source_name']].lower() == "zasilkovna sro": match = re.compile(r'\b\d{10}\b').search(row[mapping['reference']]) if match: reference = match.group(0) if reference == '0' and ("gls general" in row[mapping['source_name']].lower()): reference = str(source_amount) if reference == '0' and mapping['forced_encoding'] == 'iso-8859-2': reference = extract_order_number(row[8]) if "p24w-" in row[mapping['reference']].lower(): reference = row[mapping['reference']].split("TCH ")[-1].split(" PAYPRO")[0] # Determine transaction type if(direction is None): if source_amount > 0: transaction_code = TransactionCode.CREDIT else: transaction_code = TransactionCode.DEBET else: if direction == "out" or direction == "neutral": transaction_code = TransactionCode.DEBET else: transaction_code = TransactionCode.CREDIT # Convert currency currency_code = CURRENCIES_GPC.get("CZK", "0000") # Create GPC Data object gpc_data = Data( account=account_number, payer_account=payer_account, no=transaction_id, balance=source_amount, code=transaction_code, variable=int(reference) if reference.isdigit() else 0, constant_symbol=0, bank_code=0, specific_symbol=0, client_name=source_name, currency=currency_code, date=created_on ) gpc_lines.append(gpc_data.to_string()) # with open(gpc_file_path, mode='w', encoding='utf-8') as gpc_file: # gpc_file.writelines(gpc_lines) file_content = "".join(gpc_lines) print(f"GPC file content successfully created for: {gpc_file_path}") return file_content.encode("utf-8") # Example mappings mapping_wise = { 'transaction_id': 'ID', 'reference': 'Reference', 'direction': 'Direction', 'source_name': 'Source name', 'source_amount': 'Source amount (after fees)', 'source_currency': 'Source currency', 'payer_account': 'Source currency', 'created_on': 'Created on', 'delimiter': ",", 'is_dict_mapping': True, 'use_transaction_id': False, 'payer_number_exists': False, 'use_source_name': False, 'forced_encoding': None } mapping_pko = { 'transaction_id': 2, 'reference': 9, 'direction': None, 'source_name': 7, 'source_amount': 3, 'source_currency': 4, 'payer_account': 6, 'created_on': 1, 'delimiter': ",", 'is_dict_mapping': False, 'use_transaction_id': False, 'payer_number_exists': True, 'use_source_name': False, 'forced_encoding': 'iso-8859-2' } mapping_sparkasse = { 'transaction_id': 7, 'reference': 4, 'direction': None, 'source_name': 11, 'source_amount': 14, 'source_currency': 15, 'payer_account': 12, 'created_on': 1, 'delimiter': ";", 'is_dict_mapping': False, 'use_transaction_id': False, 'payer_number_exists': True, 'use_source_name': False, 'forced_encoding': None } # Example usage: # convert_csv_to_gpc("leden-2025-huf.csv", "wise-huf-2025.gpc", account_number=330005602964780100, currency="HUF", mapping=mapping_wise) # 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)