117 lines
5.1 KiB
Python
117 lines
5.1 KiB
Python
import camelot
|
|
import pandas as pd
|
|
import re
|
|
|
|
def extract_bank_data(file_path):
|
|
print(f"Reading {file_path}... enforcing strict Debit/Credit alignment.")
|
|
|
|
# We use 'stream' flavor for the NLB PDF layout
|
|
tables = camelot.read_pdf(file_path, pages='1-end', flavor='stream')
|
|
|
|
final_transactions = []
|
|
current_tx = None
|
|
|
|
# Precise Regex Patterns
|
|
iban_pattern = r'SI56\s?\d{4}\s?\d{4}\s?\d{4}\s?\d{3}'
|
|
amount_pattern = r'=(\d+[\.,]\d{2})'
|
|
date_pattern = r'(\d{2}\.\d{2}\.\d{2})'
|
|
bic_pattern = r'\b[A-Z]{8,11}\b' # Standard BIC pattern
|
|
|
|
for table in tables:
|
|
df = table.df
|
|
for _, row in df.iterrows():
|
|
cells = [str(c).strip() for c in row]
|
|
row_text = " ".join(cells)
|
|
|
|
# TRIGGER: A new transaction starts with an amount (=xx,xx) and a date
|
|
amount_match = re.search(amount_pattern, row_text)
|
|
date_match = re.search(date_pattern, row_text)
|
|
|
|
if amount_match and date_match:
|
|
# Save previous transaction before starting new one
|
|
if current_tx:
|
|
final_transactions.append(current_tx)
|
|
|
|
amt_val = amount_match.group(1)
|
|
|
|
# --- STRICT DEBIT/CREDIT LOGIC ---
|
|
# Based on the NLB layout, columns are indexed 0 to N.
|
|
# If the '=' sign is found in index 1, it is DEBIT.
|
|
# If the '=' sign is found in index 2 or higher, it is CREDIT.
|
|
is_credit = False
|
|
for i, cell_content in enumerate(cells):
|
|
if '=' in cell_content:
|
|
if i >= 2:
|
|
is_credit = True
|
|
break
|
|
|
|
# Take only the first line of the cell as the Name (to exclude address)
|
|
name_raw = cells[0].split('\n')[0].strip()
|
|
|
|
current_tx = {
|
|
"value date": date_match.group(1),
|
|
"Creditor / Debtor": name_raw,
|
|
"account": "",
|
|
"payment details": "",
|
|
"debit": "" if is_credit else amt_val,
|
|
"credit": amt_val if is_credit else "",
|
|
"collecting": True
|
|
}
|
|
|
|
# Check if IBAN is on this same row
|
|
ib = re.search(iban_pattern, cells[0])
|
|
if ib:
|
|
current_tx["account"] = ib.group(0)
|
|
|
|
elif current_tx:
|
|
text_line = cells[0]
|
|
if not text_line or text_line.lower() in ["address", "account"]:
|
|
continue
|
|
|
|
# 1. Look for IBAN
|
|
iban_search = re.search(iban_pattern, text_line)
|
|
if iban_search:
|
|
current_tx["account"] = iban_search.group(0)
|
|
# Anything after IBAN on the same line might be details
|
|
rem = text_line.replace(iban_search.group(0), "").strip()
|
|
text_line = rem
|
|
|
|
# 2. Filter out Address and Bank Noise
|
|
is_address = any(k in text_line.upper() for k in ["CESTA", "ULICA", "TRG", " LJUBLJANA", " LOGATEC"]) or re.search(r'\d{4}', text_line)
|
|
|
|
if not is_address and len(text_line) > 1:
|
|
# Clean up BIC/Routing codes (e.g. LJBASI2X)
|
|
clean_text = re.sub(bic_pattern, '', text_line).strip()
|
|
# Clean up internal bank references (CR.../DR...)
|
|
clean_text = re.sub(r'\b[CD]R\d+\b|\bNRC\b', '', clean_text).strip()
|
|
|
|
if clean_text:
|
|
if not current_tx["payment details"]:
|
|
current_tx["payment details"] = clean_text
|
|
else:
|
|
# Avoid repeating the name in details if it's already there
|
|
if clean_text not in current_tx["Creditor / Debtor"]:
|
|
current_tx["payment details"] += " " + clean_text
|
|
|
|
# Append last transaction
|
|
if current_tx:
|
|
final_transactions.append(current_tx)
|
|
|
|
# Convert to DataFrame
|
|
df_final = pd.DataFrame(final_transactions)
|
|
|
|
# Final Cleanup: ensure columns match your requested list exactly
|
|
cols = ["value date", "Creditor / Debtor", "account", "payment details", "debit", "credit"]
|
|
df_final = df_final[cols]
|
|
|
|
# Clean whitespace and handle empty values
|
|
for col in df_final.columns:
|
|
df_final[col] = df_final[col].astype(str).str.replace(r'\s+', ' ', regex=True).str.strip()
|
|
df_final[col] = df_final[col].replace('nan', '')
|
|
|
|
output_file = "bank_export_final_fixed.csv"
|
|
df_final.to_csv(output_file, sep=';', index=False, encoding='utf-8-sig')
|
|
print(f"Success! {len(df_final)} transactions exported to {output_file}")
|
|
|
|
if __name__ == "__main__":
|
|
extract_bank_data("MSDataReport (1).pdf") |