Extract transactions from PNC Virtual Wallet Statement PDF

pnc-vw-stmt.py

"""
Pulls balance summary from statement and validates that deposits and withdrawals
add up to match the balance summary. If they don't match the transactions will
be saved to <month>-<year>-invalid.csv so you can try and debug why.

Python and pdfplumber are the only dependencies
$ pip install pdfplumber

Extract transactions from all statements in a directory 
$ python pnc-vm-stmt.py ~/Documents/accounting/2020/statements/

Extract transactions from a single statement
$ python pnc-vm-stmt.py ~/Documents/accounting/2020/statements/Statement_Feb_19_2020.pdf
"""
import re
import sys
from dataclasses import dataclass
from datetime import date as Date
from decimal import Decimal
from pathlib import Path

import pdfplumber

YEAR = 2020


@dataclass
class BalanceSummary:
    beginning: Decimal
    deposits: Decimal
    deductions: Decimal
    ending: Decimal

    @classmethod
    def from_slice(cls, slice):
        if len(slice) != 4:
            raise ValueError("Should be exactly 4 values")
        return cls(
            Decimal(slice[0]), Decimal(slice[1]), Decimal(slice[2]), Decimal(slice[3])
        )


@dataclass
class Transaction:
    date: Date
    description: str
    deposit: Decimal = Decimal(0)
    withdrawal: Decimal = Decimal(0)
    account: str = "Assets:Current Assets:Checking Account"

    @classmethod
    def from_check(cls, num: int, date: Date, amount: Decimal):
        return cls(
            date=date,
            description=f"Check #{num}",
            withdrawal=amount,
        )

    @classmethod
    def from_raw(cls, date: Date, amount: Decimal, words: list[str]):
        description = " ".join(word for word in words)
        if (
            "Deposit" in description
            or "Debit Card Credit" in description
            or "Transfer From" in description
            or "Rtp Received Venmo" in description
        ):
            return cls(date, description, deposit=amount)
        return cls(date, description, withdrawal=amount)


def tokenize(text: str):
    fluff_words = [
        "Amount",
        "continued",
        "Description",
        "Primary",
        "account number",
        "Banking",
        "is",
        "Date",
        "Daily",
        "Balance",
        "totaling",
        "Additions",
        "Member FDIC Equal Housing Lender",
        "Deposits and Other",
        "on next page",
        "Virtual Wallet Spend Statement",
        "Online or Electronic",
        "Online and Electronic",
        "Deductions",
        "Detail",
        "other",
        "deductions",
        "For the period",
    ]
    token_specification = [
        ("DATE", r"\d\d/\d\d"),  # 01/12
        (
            "BALANCE_SUMMARY",
            r"(?:\d+\.\d+|\.\d+) (?:\d+\.\d+|\.\d+) (?:\d+\.\d+|\.\d+) (?:\d+\.\d+|\.\d+)",
        ),
        ("AMOUNT", r"(?:\d+\.\d+|\.\d+)"),  # 100.00 or .08
        ("ACCOUNT_NUMBER", r"(?<=Account Number: )\d+-\d+-\d+"),
        ("PERIOD", r"For the period \d\d/\d\d/\d\d\d\d to \d\d/\d\d/\d\d\d\d"),
        (
            "BOILERPLATE",
            r"Virtual Wallet Spend Statement PNC Bank.*Virtual Wallet Spend Account Summary",
        ),
        ("BOILERPLATE2", r"Overdraft Protection.*Balance Summary"),
        ("THERE", r"(?:There were \d+|There is \d+)"),
        ("TOTALING", r"totaling \$\d+\.\d+"),
        ("PAGE", r"Page \d+ of \d+"),
        ("FLUFF", r"(?:" + "|".join(fluff_words) + ")"),
        ("WORD", r"\w+"),
        ("DOLLAR", r"\$\d+\.\d+"),
        ("NEWLINE", r"\n"),  # Line endings
        ("MISMATCH", r"."),  # Any other character
    ]
    tok_regex = "|".join("(?P<%s>%s)" % pair for pair in token_specification)
    # for text in text_stream:
    for mo in re.finditer(tok_regex, text, re.DOTALL):
        kind = mo.lastgroup
        value = mo.group()
        if kind not in {
            "BOILERPLATE",
            "BOILERPLATE2",
            "THERE",
            "TOTALING",
            "FLUFF",
            "MISMATCH",
            "NEWLINE",
        }:
            yield kind, value


class VirtualWalletStatement:
    """Represents a PNC Virtual Wallet monthly PDF statement.

    ### API
    >>> stmt = VirtualWalletStatement("Statement_Apr_16_2020.pdf")
    >>> stmt.year
    >>> 2020
    >>> stmt.month
    >>> Apr
    >>> stmt.parse()
    """

    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self._tokens = []
        self._transactions = []
        self._balance_summary = None
        self._account_number = ""
        self._period = None

    @property
    def transactions(self) -> list[Transaction]:
        if not self._transactions:
            self._parse()
        return self._transactions

    @property
    def balance_summary(self):
        if not self._balance_summary:
            self._balance_summary = self._get_balance_summary()
        return self._balance_summary

    @property
    def tokens(self):
        if not self._tokens:
            self._tokens = list(tokenize(self.get_text()))
        return self._tokens

    @property
    def year(self) -> str:
        return self.path.stem.split("_")[-1]

    @property
    def month(self) -> str:
        return self.path.stem.split("_")[1]

    @property
    def account_number(self) -> str:
        if not self._account_number:
            self._account_number = self._get_account_number()
        return self._account_number

    @property
    def period(self) -> tuple[Date, Date]:
        """Returns start and end dates of statement period."""
        if not self._period:
            self._period = self._get_period()
        return self._period

    def get_text(self):
        """Returns text of the PDF as one big string stripped of newlines."""
        with pdfplumber.open(self.path) as pdf:
            text = ""
            for page in pdf.pages:
                # for text in page.extract_text():
                #     print(text)
                #     yield text
                text += page.extract_text().replace("\n", " ").replace(",", "")
        return text

    def _get_balance_summary(self) -> BalanceSummary:
        for kind, value in self.tokens:
            if kind == "BALANCE_SUMMARY":
                return BalanceSummary.from_slice(value.split(" "))

    def _get_account_number(self) -> str:
        for kind, value in self.tokens:
            if kind == "ACCOUNT_NUMBER":
                return value

    def _get_period(self) -> tuple[Date, Date]:
        for kind, value in self.tokens:
            if kind == "PERIOD":
                words = value.split(" ")
                _from, _to = words[3].split("/"), words[5].split("/")
                m, d, y = [int(v) for v in _from]
                start = Date(y, m, d)
                m, d, y = [int(v) for v in _to]
                end = Date(y, m, d)
                return start, end

    def _parse(self):
        transactions = []
        kinds, values = [], []

        for kind, value in self.tokens:
            kinds.append(kind)
            values.append(value)

        for n, kind in enumerate(kinds):
            if n == len(kinds) - 2:
                break

            elif kind == "WORD" and kinds[n + 1] == "AMOUNT" and kinds[n + 2] == "DATE":
                try:
                    num = int(values[n])  # check number
                    m, d = [int(v) for v in values[n + 2].split("/")]
                    start, end = self.period
                    if start.month == 12 and end.month == 1:
                        y = start.year if m == 12 else end.year
                    else:
                        y = int(self.year)
                    date = Date(y, m, d)
                    amount = Decimal(values[n + 1])
                    transactions.append(Transaction.from_check(num, date, amount))
                except ValueError:
                    continue

            elif kind == "DATE" and kinds[n + 1] == "AMOUNT" and kinds[n + 2] == "WORD":
                m, d = [int(v) for v in values[n].split("/")]
                start, end = self.period
                if start.month == 12 and end.month == 1:
                    y = start.year if m == 12 else end.year
                else:
                    y = int(self.year)
                date = Date(y, m, d)

                amount = Decimal(values[n + 1])

                words = []
                idx = 2
                for sub_kind in kinds[n + idx :]:
                    if sub_kind == "DATE":
                        break
                    word = values[n + idx]
                    if word in {"Checks", "Machine", "Withdrawals", "For"}:
                        break
                    words.append(word)
                    idx += 1
                transactions.append(Transaction.from_raw(date, amount, words))
        self._transactions = transactions

    def validate(self):
        """Returns whether the numbers add up."""
        deposits, withdrawals = Decimal(0), Decimal(0)
        for t in self.transactions:
            deposits += t.deposit
            withdrawals += t.withdrawal
        try:
            assert (
                self.balance_summary.beginning + deposits - withdrawals
            ) == self.balance_summary.ending
            assert deposits == self.balance_summary.deposits
            assert withdrawals == self.balance_summary.deductions
        except AssertionError:
            print("Validation Error")
            print(self.balance_summary)
            print(f"Parsed Deposits: {deposits}, Parsed Withdrawals: {withdrawals}")
            self.to_csv(f"{self.month}-{self.year}-invalid.csv")
            return False
        return True

    def to_csv(self, filename="transactions.csv"):
        with open(filename, "w") as f:
            for t in self.transactions:
                f.write(
                    "{},{},{},{},{}\n".format(
                        t.date, t.description, t.deposit, t.withdrawal, t.account
                    )
                )
        print(f"Saved {filename}")


if __name__ == "__main__":
    path = Path(sys.argv[1])
    # Directory of statement files
    Path("transactions").mkdir(exist_ok=True)
    if path.is_dir():
        for stmt in path.iterdir():
            statement = VirtualWalletStatement(stmt)
            if statement.validate():
                statement.to_csv(
                    filename=f"transactions/{statement.month}-{statement.year}.csv"
                )
    else:
        statement = VirtualWalletStatement(path)
        if statement.validate():
            statement.to_csv(f"transactions/{statement.month}-{statement.year}.csv")