| import csv |
| import io |
| from typing import Dict, Any |
|
|
| def parse_bank_csv(content: bytes) -> Dict[str, Any]: |
| """ |
| Parse a bank CSV file and return categorized transactions with summary statistics. |
| |
| Args: |
| content: Raw bytes of the CSV file |
| |
| Returns: |
| Dict with keys: |
| - transactions: List of transaction dicts with date, description, amount |
| - total_income: Sum of positive amounts |
| - total_expenses: Sum of absolute negative amounts |
| - net: Total income minus total expenses |
| - count: Number of transactions |
| - monthly_estimate: Average monthly expenses based on unique months in data |
| """ |
| text = content.decode('utf-8', errors='replace') |
| lines = text.strip().split('\n') |
|
|
| transactions = [] |
| total_income = 0 |
| total_expenses = 0 |
|
|
| |
| try: |
| |
| if lines: |
| dialect = csv.Sniffer().sniff(lines[0], delimiters=';,\t') |
| reader = csv.DictReader(io.StringIO(text), delimiter=dialect.delimiter) |
| else: |
| reader = csv.DictReader(io.StringIO(text), delimiter=';') |
| except csv.Error: |
| reader = csv.DictReader(io.StringIO(text), delimiter=';') |
|
|
| for row in reader: |
| amount = None |
| date_str = None |
| description = "" |
|
|
| for key, val in row.items(): |
| if not key: |
| continue |
| key_lower = key.lower().strip() |
|
|
| |
| if key_lower in ('betrag', 'amount', 'umsatz', 'betrag (eur)', 'wert'): |
| try: |
| amount = float(val.replace('.', '').replace(',', '.').strip().strip('"')) |
| except (ValueError, AttributeError): |
| pass |
|
|
| |
| if key_lower in ('buchungstag', 'datum', 'date', 'buchungsdatum', 'valuta'): |
| date_str = val.strip().strip('"') |
|
|
| |
| if key_lower in ('verwendungszweck', 'buchungstext', 'description', 'empfaenger', |
| 'auftraggeber / beguenstigter', 'zahlungsempfaenger', 'beschreibung'): |
| if val and val.strip(): |
| description = val.strip().strip('"')[:100] |
|
|
| if amount is not None: |
| if amount > 0: |
| total_income += amount |
| else: |
| total_expenses += abs(amount) |
| transactions.append({ |
| "date": date_str or "", |
| "description": description, |
| "amount": amount, |
| }) |
|
|
| |
| unique_months = len(set(t['date'][:7] for t in transactions if t['date'])) |
| monthly_estimate = total_expenses / max(1, unique_months) if transactions else 0 |
|
|
| return { |
| "transactions": transactions, |
| "total_income": round(total_income, 2), |
| "total_expenses": round(total_expenses, 2), |
| "net": round(total_income - total_expenses, 2), |
| "count": len(transactions), |
| "monthly_estimate": round(monthly_estimate, 2), |
| } |
|
|
