import re from abc import ABC from abc import abstractmethod from datetime import datetime from decimal import Decimal from email.message import EmailMessage from email.message import Message from uuid import UUID from model import Transaction class TransactionParser(ABC): @abstractmethod def match(self, msg: Message) -> bool: pass @abstractmethod def extract(self, msg: EmailMessage) -> Transaction: pass class TransactionParsingFailed(Exception): pass class RogersBankParser(TransactionParser): EXTRACT_RE = re.compile( r"Attempt of \$(\d+\.\d{2}) was made on ([A-z]{3} \d{1,2}, \d{4})[^<]*at ([^<]+) in" ) def __init__(self, account_id: UUID): self.account_id = account_id def match(self, msg: Message) -> bool: return ( msg["From"] == "Rogers Bank " and msg["Subject"] == "Purchase amount alert" ) def extract(self, msg: EmailMessage) -> Transaction: body = msg.get_body() if body is None: raise TransactionParsingFailed("No body of message found") matches = self.EXTRACT_RE.search(body.as_string()) if matches is None: raise TransactionParsingFailed("No matches for extraction RE") amount = Decimal(matches[1]) date_raw = matches[2] payee = matches[3] date = datetime.strptime(date_raw, "%b %d, %Y").date() return Transaction( account=self.account_id, date=date, amount=amount, payee=payee, notes="via email", imported_id=msg["Message-ID"], ) class MBNAParser(TransactionParser): EXTRACT_RE = re.compile( r"A purchase of \$(\d+\.\d{2}) from ([^<]+) was made at (\d{1,2}:\d{2} (AM|PM)) UTC on (\d{4}-\d{2}-\d{2})" # noqa: E501 ) def __init__(self, account_id: UUID): self.account_id = account_id def match(self, msg: Message): return ( msg["From"] == "MBNA Notifications " and msg["Subject"] == "MBNA - Transaction Alert" ) def extract(self, msg: EmailMessage) -> Transaction: body = msg.get_body() if body is None: raise TransactionParsingFailed("No body of message found") matches = self.EXTRACT_RE.search(body.as_string()) if matches is None: raise TransactionParsingFailed("No matches for extraction RE") amount = Decimal(matches[1]) payee = matches[2] date = matches[5] return Transaction( account=self.account_id, date=date, amount=amount, payee=payee, notes="via email", imported_id=msg["Message-ID"], )