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: """ Determines if the given email message matches the criteria for this parser. Args: msg (Message): The email message to evaluate. Returns: bool: True if the message matches the parser's criteria, False otherwise. """ pass @abstractmethod def extract(self, msg: EmailMessage) -> Transaction: """ Extracts transaction details from the given email message. Args: msg (EmailMessage): The email message to parse. Returns: Transaction: A Transaction object containing the extracted details. Raises: TransactionParsingFailed: If the message cannot be parsed successfully. """ 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"], )