|
|
import pdfplumber |
|
|
import pandas as pd |
|
|
import re |
|
|
from typing import List, Dict, Optional |
|
|
import logging |
|
|
from datetime import datetime |
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
class PDFTransactionProcessor: |
|
|
"""Process transaction data from PDF bank statements.""" |
|
|
|
|
|
def __init__(self): |
|
|
self.transaction_patterns = { |
|
|
'standard': re.compile( |
|
|
r'(?P<date>\d{2}/\d{2}/\d{2,4})\s+(?P<description>.+?)\s+(?P<amount>-?\$?\d{1,3}(?:,\d{3})*\.\d{2})' |
|
|
), |
|
|
'bank_specific': { |
|
|
'chase': re.compile( |
|
|
r'(?P<date>\d{2}/\d{2})\s+(?P<description>.+?)\s+(?P<amount>-?\$?\d{1,3}(?:,\d{3})*\.\d{2})' |
|
|
), |
|
|
'bank_of_america': re.compile( |
|
|
r'(?P<date>\d{2}/\d{2}/\d{4})\s+(?P<description>.+?)\s+(?P<amount>-?\d{1,3}(?:,\d{3})*\.\d{2})' |
|
|
) |
|
|
} |
|
|
} |
|
|
|
|
|
def extract_text_from_pdf(self, pdf_path: str) -> str: |
|
|
"""Extract text from PDF file.""" |
|
|
try: |
|
|
full_text = [] |
|
|
with pdfplumber.open(pdf_path) as pdf: |
|
|
for page in pdf.pages: |
|
|
text = page.extract_text() |
|
|
if text: |
|
|
full_text.append(text) |
|
|
return "\n".join(full_text) |
|
|
except Exception as e: |
|
|
logger.error(f"Error extracting text from PDF: {str(e)}") |
|
|
raise |
|
|
|
|
|
def identify_bank(self, text: str) -> Optional[str]: |
|
|
"""Identify bank from PDF text.""" |
|
|
bank_indicators = { |
|
|
'chase': ['chase', 'jpmorgan'], |
|
|
'bank_of_america': ['bank of america', 'bofa'], |
|
|
'wells_fargo': ['wells fargo'], |
|
|
'citi': ['citibank', 'citi'] |
|
|
} |
|
|
|
|
|
text_lower = text.lower() |
|
|
for bank, indicators in bank_indicators.items(): |
|
|
if any(indicator in text_lower for indicator in indicators): |
|
|
return bank |
|
|
return None |
|
|
|
|
|
def parse_transactions(self, text: str, bank: Optional[str] = None) -> List[Dict]: |
|
|
"""Parse transactions from extracted text.""" |
|
|
transactions = [] |
|
|
lines = text.split('\n') |
|
|
|
|
|
|
|
|
if bank and bank in self.transaction_patterns['bank_specific']: |
|
|
pattern = self.transaction_patterns['bank_specific'][bank] |
|
|
transactions = self._parse_with_pattern(lines, pattern) |
|
|
|
|
|
|
|
|
if not transactions: |
|
|
pattern = self.transaction_patterns['standard'] |
|
|
transactions = self._parse_with_pattern(lines, pattern) |
|
|
|
|
|
return transactions |
|
|
|
|
|
def _parse_with_pattern(self, lines: List[str], pattern) -> List[Dict]: |
|
|
"""Helper method to parse transactions using a regex pattern.""" |
|
|
transactions = [] |
|
|
for line in lines: |
|
|
match = pattern.search(line) |
|
|
if match: |
|
|
try: |
|
|
amount = match.group('amount').replace('$', '').replace(',', '') |
|
|
transactions.append({ |
|
|
'date': match.group('date'), |
|
|
'description': match.group('description').strip(), |
|
|
'amount': float(amount) |
|
|
}) |
|
|
except (ValueError, AttributeError) as e: |
|
|
logger.debug(f"Skipping line due to parsing error: {line}") |
|
|
return transactions |
|
|
|
|
|
def process_pdf(self, pdf_path: str) -> pd.DataFrame: |
|
|
"""Process PDF file and return transactions as DataFrame.""" |
|
|
try: |
|
|
text = self.extract_text_from_pdf(pdf_path) |
|
|
bank = self.identify_bank(text) |
|
|
transactions = self.parse_transactions(text, bank) |
|
|
|
|
|
if not transactions: |
|
|
raise ValueError("No transactions found in PDF") |
|
|
|
|
|
df = pd.DataFrame(transactions) |
|
|
|
|
|
|
|
|
df['date'] = pd.to_datetime(df['date'], errors='coerce') |
|
|
df = df.dropna(subset=['date', 'amount']) |
|
|
df['description'] = df['description'].str.strip() |
|
|
|
|
|
return df |
|
|
|
|
|
except Exception as e: |
|
|
logger.error(f"Error processing PDF: {str(e)}") |
|
|
raise |
|
|
|
|
|
|
