HDFC bank statement importer for Beancount

config.py

import sys
from typing import List

from beancount.core.number import D
from beancount.ingest import importer
from beancount.core import amount, flags
from beancount.core.data import Posting, Transaction, new_metadata
from smart_importer import apply_hooks
from smart_importer.detector import DuplicateDetector

from datetime import datetime

import pandas as pd

import os

HDFC_CUSTOMER_ID = ""

UNCLASSIFIED_EXPENSES = "Expenses:Unclassified"
UNCLASSIFIED_INCOME = "Income:Unclassified"
BANK_ACC = "Assets:BankAccount:HDFC:Savings"

acc_map: dict[str, str] = {
    "swiggy": "Expenses:Food",
    "roopen": "Expenses:Transportation",
    "neftin": "Income:Salary:Refyne",
    "bbpsb": "Expenses:Electricity",
    "rent": "Liabilities:HouseRent",
    "groceries": "Expenses:Groceries",
    "medicines": "Expenses:Medicines",
    "food": "Expenses:Food",
}


def get_acc_from_narration(narration: str, fallback_acc: str) -> str:
    for k, v in acc_map.items():
        if k in narration.lower():
            return v
    return fallback_acc


def get_from_to_acc(amt_paid, amt_received, narration: str) -> (str, str):
    if amt_paid:
        return BANK_ACC, get_acc_from_narration(narration, UNCLASSIFIED_EXPENSES)
    if amt_received:
        return get_acc_from_narration(narration, UNCLASSIFIED_INCOME), BANK_ACC
    else:
        raise Exception("wtf!")


def get_txn_from_details(
    trans_date, amt_paid, amt_received, narration: str, meta: dict, uniq_id: str
) -> Transaction:
    from_acc, to_acc = get_from_to_acc(amt_paid, amt_received, narration)
    final_amount = amt_paid if amt_paid else amt_received

    meta["ref"] = uniq_id

    txn = Transaction(
        meta=meta,
        date=trans_date,
        flag=flags.FLAG_OKAY,
        payee=None,
        narration=narration,
        tags=set(),
        links=set(),
        postings=[
            Posting(
                from_acc,
                -amount.Amount(D(final_amount), "INR"),
                None,
                None,
                None,
                None,
            ),
            Posting(
                to_acc,
                amount.Amount(D(final_amount), "INR"),
                None,
                None,
                None,
                None,
            ),
        ],
    )
    return txn


class ReferenceDuplicatesComparator:
    def __call__(self, entry1, entry2):
        return (
            "ref" in entry1.meta
            and "ref" in entry2.meta
            and entry1.meta["ref"] == entry2.meta["ref"]
        )


class HDFCCSVImporter(importer.ImporterProtocol):
    def identify(self, file) -> bool:
        return HDFC_CUSTOMER_ID in file.name.lower()

    def extract(self, file, existing_entries=None) -> List[Transaction]:
        df = pd.read_csv(file.name)
        df.columns = df.columns.str.replace(" ", "")

        df = df.dropna()

        return [
            get_txn_from_details(
                trans_date=datetime.strptime(row["Date"].strip(), "%d/%m/%y").date(),
                amt_paid=row["DebitAmount"],
                amt_received=row["CreditAmount"],
                narration=row["Narration"].strip(),
                meta=new_metadata(file.name, index),
                uniq_id=row["Chq/RefNumber"].strip(),
            )
            for index, row in enumerate(df.to_dict(orient="records"))
        ]


sys.path.append(os.path.dirname(__file__))
CONFIG = [
    apply_hooks(
        HDFCCSVImporter(),
        [
            DuplicateDetector(),
            DuplicateDetector(comparator=ReferenceDuplicatesComparator),
        ],
    )
]