add imap_ssl support, some style fixes

This commit is contained in:
ktims committed 2026-10-18 14:18:05 -07:00
1 parent d414d0cf09
commit 43d9a69707
2 files changed
+54 -42

No files matched your search

+28 -20
View File
@@ -1,17 +1,19 @@
import logging
import re
from abc import ABC, abstractmethod
from collections.abc import Sequence
from datetime import date, datetime
from decimal import Decimal
from email.message import EmailMessage, Message
from logging import debug, info, warning
from typing import Any, Optional
from typing import Any
from uuid import UUID
from bs4 import BeautifulSoup
from .model import Transaction
log = logging.getLogger("parse")
def parse_email_time(s: str) -> datetime:
return datetime.strptime(s, "%a, %d %b %Y %H:%M:%S %z")
@@ -29,10 +31,10 @@ class TransactionParser(ABC):
Returns:
bool: True if the message matches the parser's criteria, False otherwise.
"""
pass
...
@abstractmethod
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
def extract(self, msg: EmailMessage) -> Transaction | None:
"""
Extracts transaction details from the given email message.
@@ -45,7 +47,7 @@ class TransactionParser(ABC):
Raises:
TransactionParsingFailed: If the message cannot be parsed successfully.
"""
pass
...
@staticmethod
def get_content(
@@ -90,7 +92,7 @@ class RogersBankParser(TransactionParser):
and msg["Subject"] == "Purchase amount alert"
)
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
def extract(self, msg: EmailMessage) -> Transaction | None:
content = self.strip_html(msg)
matches = self.EXTRACT_RE.search(content)
if matches is None:
@@ -128,7 +130,7 @@ class MBNAParser(TransactionParser):
and msg["Subject"] == "MBNA - Transaction Alert"
)
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
def extract(self, msg: EmailMessage) -> Transaction | None:
content = self.get_content(msg)
matches = self.EXTRACT_RE.search(content)
if matches is None:
@@ -158,7 +160,7 @@ class BMOParser(TransactionParser):
def match(self, msg: Message) -> bool:
return msg["From"] == "bmoalerts@bmo.com"
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
def extract(self, msg: EmailMessage) -> Transaction | None:
content = self.strip_html(msg)
matches = self.EXTRACT_RE.search(content)
if matches is None:
@@ -170,7 +172,9 @@ class BMOParser(TransactionParser):
date = parse_email_time(msg["Date"]).date()
account_ref = int(matches[3])
if account_ref not in self._account_map:
warning("Account %s not in account map, skipping transaction", account_ref)
log.warning(
"Account %s not in account map, skipping transaction", account_ref
)
return None
account_id = self._account_map[account_ref]
return Transaction(
@@ -212,7 +216,9 @@ class CIBCParser(TransactionParser):
date = parse_email_time(msg["Date"]).date()
if account_ref not in self._account_map:
warning("Account %s not in account map, skipping transaction", account_ref)
log.warning(
"Account %s not in account map, skipping transaction", account_ref
)
return None
account_id = self._account_map[account_ref]
@@ -225,7 +231,7 @@ class CIBCParser(TransactionParser):
imported_id=msg["Message-ID"],
)
def extract_purchase(self, msg: EmailMessage) -> Optional[Transaction]:
def extract_purchase(self, msg: EmailMessage) -> Transaction | None:
content = self.get_content(msg)
matches = self.PURCHASE_EXTRACT_RE.search(content)
@@ -237,7 +243,9 @@ class CIBCParser(TransactionParser):
account_ref = int(matches["account"])
if account_ref not in self._account_map:
warning("Account %s not in account map, skipping transaction", account_ref)
log.warning(
"Account %s not in account map, skipping transaction", account_ref
)
return None
account_id = self._account_map[account_ref]
@@ -250,7 +258,7 @@ class CIBCParser(TransactionParser):
imported_id=msg["Message-ID"],
)
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
def extract(self, msg: EmailMessage) -> Transaction | None:
match msg["Subject"]:
case "New payment to your credit card":
return self.extract_payment(msg)
@@ -271,7 +279,7 @@ class ScotiaBankParser(TransactionParser):
def match(self, msg: Message) -> bool:
return msg["From"] == "Scotia InfoAlerts <infoalerts@scotiabank.com>"
def extract_transaction(self, msg: EmailMessage) -> Optional[Transaction]:
def extract_transaction(self, msg: EmailMessage) -> Transaction | None:
content = self.get_content(msg)
matches = self.PAYMENT_EXTRACT_RE.search(content)
@@ -282,7 +290,7 @@ class ScotiaBankParser(TransactionParser):
date = parse_email_time(msg["Date"]).date()
if matches["account"] not in self._account_map:
warning(
log.warning(
"Account %s not in account map, skipping transaction",
matches["account"],
)
@@ -300,14 +308,14 @@ class ScotiaBankParser(TransactionParser):
imported_id=msg["Message-ID"],
)
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
def extract(self, msg: EmailMessage) -> Transaction | None:
match msg["Subject"]:
case (
"Authorization on your credit account"
| "Authorization without credit card present"
):
return self.extract_transaction(msg)
info("Subject `%s` didn't match any extractors", msg["Subject"])
log.info("Subject `%s` didn't match any extractors", msg["Subject"])
return None
@@ -325,11 +333,11 @@ class NeoFinancialParser(TransactionParser):
"Subject"
].startswith("You earned")
def extract_purchase(self, msg: EmailMessage) -> Optional[Transaction]:
def extract_purchase(self, msg: EmailMessage) -> Transaction | None:
content = self.get_content(msg, preferencelist=["plain"])
matches = self.PURCHASE_EXTRACT_RE.search(content)
if matches is None:
debug(f"No match on\n\n{content}")
log.debug(f"No match on\n\n{content}")
raise TransactionParsingFailed("no matches for purchase extraction RE")
amount = Decimal(matches["amount"].replace(",", "")) * -1
@@ -344,7 +352,7 @@ class NeoFinancialParser(TransactionParser):
imported_id=msg["Message-ID"],
)
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
def extract(self, msg: EmailMessage) -> Transaction | None:
if msg["Subject"].startswith("You earned"):
return self.extract_purchase(msg)
return None