duplicates.py

#!/usr/bin/python3

from collections import namedtuple, deque
from datetime import datetime, timedelta
import argparse
import csv
from fpos import core
from fpos.groups import DynamicGroups

Transaction = namedtuple("Transaction", "date, amount, description, category")

typeify = lambda t: Transaction(date=datetime.strptime(t.date, core.date_fmt),
                                amount=float(t.amount),
                                description=t.description,
                                category=t.category)

window_size = timedelta(days=4)

parser = argparse.ArgumentParser()
parser.add_argument("source", type=argparse.FileType('r'))
args = parser.parse_args()
reader = csv.reader(args.source, dialect='excel')
history = deque()
with DynamicGroups() as grouper:
    for t in map(typeify, map(Transaction._make, reader)):
        print(t)

        t_grp = grouper.find_group(t.description)

        # Search for a match in the history buffer
        if t_grp:
            for e in t_grp:
                v = e.value()
                if v in history and v.amount == t.amount:
                    print("Found duplicate:\n\t{}\n\t{}".format(v, t))

        grouper.insert(t.description, t, t_grp)

        history.append(t)

        # Truncate the history buffer
        tail = (t.date - window_size)
        while len(history):
            v = history.popleft()
            if v.date > tail:
                history.appendleft(v)
                break
            print("\tDropping {}".format(v))