parsers: add CIBC payment parser
This commit is contained in:
parent
db22076d4c
commit
8b9b91ff88
56
parsers.py
56
parsers.py
@ -12,6 +12,10 @@ from bs4 import BeautifulSoup
|
|||||||
from model import Transaction
|
from model import Transaction
|
||||||
|
|
||||||
|
|
||||||
|
def parse_email_time(s: str) -> datetime:
|
||||||
|
return datetime.strptime(s, "%a, %d %b %Y %H:%M:%S %z")
|
||||||
|
|
||||||
|
|
||||||
class TransactionParser(ABC):
|
class TransactionParser(ABC):
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def match(self, msg: Message) -> bool:
|
def match(self, msg: Message) -> bool:
|
||||||
@ -155,8 +159,7 @@ class BMOParser(TransactionParser):
|
|||||||
amount = Decimal(matches[2].replace(",", ""))
|
amount = Decimal(matches[2].replace(",", ""))
|
||||||
if matches[1] == "withdrawal":
|
if matches[1] == "withdrawal":
|
||||||
amount = amount * -1
|
amount = amount * -1
|
||||||
date_raw = msg["Date"]
|
date = parse_email_time(msg["Date"]).date()
|
||||||
date = datetime.strptime(date_raw, "%a, %d %b %Y %H:%M:%S %z").date()
|
|
||||||
account_ref = int(matches[3])
|
account_ref = int(matches[3])
|
||||||
if account_ref not in self._account_map:
|
if account_ref not in self._account_map:
|
||||||
info("Account %s not in account map", account_ref)
|
info("Account %s not in account map", account_ref)
|
||||||
@ -170,3 +173,52 @@ class BMOParser(TransactionParser):
|
|||||||
notes="via email",
|
notes="via email",
|
||||||
imported_id=msg["Message-ID"],
|
imported_id=msg["Message-ID"],
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class CIBCParser(TransactionParser):
|
||||||
|
PAYMENT_EXTRACT_RE = re.compile(
|
||||||
|
r"recently received a \$([0-9,]+\.\d{2}) payment to your [^<]+ ending in (\d{4})",
|
||||||
|
flags=re.MULTILINE,
|
||||||
|
)
|
||||||
|
|
||||||
|
def __init__(self, account_map: dict[int, UUID]):
|
||||||
|
self._account_map = account_map
|
||||||
|
|
||||||
|
def match(self, msg: Message) -> bool:
|
||||||
|
return msg["From"] == "CIBC Banking <mailbox.noreply@cibc.com>"
|
||||||
|
|
||||||
|
def extract_payment(self, msg: EmailMessage):
|
||||||
|
body = msg.get_body(preferencelist=("html", "plain"))
|
||||||
|
if body is None:
|
||||||
|
raise TransactionParsingFailed("No body of message found")
|
||||||
|
content = body.get_content()
|
||||||
|
if content is None:
|
||||||
|
raise TransactionParsingFailed("No content of message found")
|
||||||
|
|
||||||
|
matches = self.PAYMENT_EXTRACT_RE.search(content)
|
||||||
|
if matches is None:
|
||||||
|
raise TransactionParsingFailed("no matches for extraction RE")
|
||||||
|
|
||||||
|
amount = Decimal(matches[1].replace(",", ""))
|
||||||
|
account_ref = int(matches[2])
|
||||||
|
date = parse_email_time(msg["Date"]).date()
|
||||||
|
|
||||||
|
if account_ref not in self._account_map:
|
||||||
|
info("Account %s not in account map", account_ref)
|
||||||
|
return None
|
||||||
|
account_id = self._account_map[account_ref]
|
||||||
|
|
||||||
|
return Transaction(
|
||||||
|
account=account_id,
|
||||||
|
date=date,
|
||||||
|
amount=amount,
|
||||||
|
payee="",
|
||||||
|
notes="via email",
|
||||||
|
imported_id=msg["Message-ID"],
|
||||||
|
)
|
||||||
|
|
||||||
|
def extract(self, msg: EmailMessage) -> Optional[Transaction]:
|
||||||
|
match msg["Subject"]:
|
||||||
|
case "New payment to your credit card":
|
||||||
|
return self.extract_payment(msg)
|
||||||
|
return None
|
||||||
|
Loading…
x
Reference in New Issue
Block a user