104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
import camelot
|
|
import pandas as pd
|
|
import re
|
|
|
|
def extract_bank_data(file_path):
|
|
print(f"Reading {file_path}... shifting Payment Details parsing one row lower.")
|
|
|
|
# We use 'stream' flavor for the NLB report layout
|
|
tables = camelot.read_pdf(file_path, pages='1-end', flavor='stream')
|
|
|
|
final_transactions = []
|
|
current_tx = None
|
|
|
|
# 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-Z0-9]{8,11}\b'
|
|
|
|
for table in tables:
|
|
df = table.df
|
|
for _, row in df.iterrows():
|
|
cells = [str(c).strip() for c in row]
|
|
row_text = " ".join(cells)
|
|
|
|
amount_match = re.search(amount_pattern, row_text)
|
|
date_match = re.search(date_pattern, row_text)
|
|
|
|
# TRIGGER: New transaction starts
|
|
if amount_match and date_match:
|
|
if current_tx:
|
|
final_transactions.append(current_tx)
|
|
|
|
amt_val = amount_match.group(1)
|
|
|
|
# Debit/Credit alignment logic
|
|
is_credit = False
|
|
for i, c in enumerate(cells):
|
|
if '=' in c and i >= 2:
|
|
is_credit = True
|
|
|
|
current_tx = {
|
|
"value date": date_match.group(1),
|
|
"Creditor / Debtor": cells[0].split('\n')[0].strip(),
|
|
"account": "",
|
|
"payment details": "",
|
|
"debit": "" if is_credit else amt_val,
|
|
"credit": amt_val if is_credit else "",
|
|
"skip_rows": 0 # Counter to shift parsing lower
|
|
}
|
|
|
|
elif current_tx:
|
|
text_line = cells[0]
|
|
if not text_line: continue
|
|
|
|
# 1. Identify IBAN
|
|
iban_search = re.search(iban_pattern, text_line)
|
|
if iban_search:
|
|
current_tx["account"] = iban_search.group(0).replace(" ", "")
|
|
# When IBAN is found, the line below it is usually BIC.
|
|
# We set skip_rows to 1 to skip that BIC line.
|
|
current_tx["skip_rows"] = 1
|
|
continue
|
|
|
|
# 2. Skip the "BIC" line (one row lower logic)
|
|
if current_tx["skip_rows"] > 0:
|
|
current_tx["skip_rows"] -= 1
|
|
continue
|
|
|
|
# 3. Capture Payment Details (after skipping)
|
|
# Filter out obvious address noise
|
|
is_address = any(k in text_line.upper() for k in ["CESTA", "ULICA", "TRG", "LJUBLJANA"]) or re.search(r'\b\d{4}\b', text_line)
|
|
|
|
if not is_address and len(text_line) > 2:
|
|
# Scrub labels and internal bank noise
|
|
clean_text = re.sub(r'^(purpose|payment details|reference)', '', text_line, flags=re.I).strip()
|
|
clean_text = re.sub(r'\b[CD]R\d+\b|\bNRC\b', '', clean_text).strip()
|
|
|
|
if clean_text:
|
|
# Append to details
|
|
if not current_tx["payment details"]:
|
|
current_tx["payment details"] = clean_text
|
|
else:
|
|
current_tx["payment details"] += " " + clean_text
|
|
|
|
if current_tx:
|
|
final_transactions.append(current_tx)
|
|
|
|
# Export
|
|
df_final = pd.DataFrame(final_transactions).drop(columns=['skip_rows'])
|
|
cols = ["value date", "Creditor / Debtor", "account", "payment details", "debit", "credit"]
|
|
df_final = df_final[cols]
|
|
|
|
# Final cleanup
|
|
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.")
|
|
|
|
if __name__ == "__main__":
|
|
extract_bank_data("MSDataReport (1).pdf") |