#!/usr/bin/env python3 from abc import abstractmethod, ABC from datetime import datetime import csv import argparse import sys def date_type(s): """Parse a date, used for the add_argument()'s type keyword argument.""" if s in ('today', 'now'): return datetime.today() try: return datetime.strptime(s, '%Y-%m-%d') except ValueError: raise argparse.ArgumentTypeError( f"`{s}' invalid format. Try now, today, or a ISO date (%Y-%m-%d)." ) def format_type(s): """Parse the csv format, used for add_argument()'s type keyword argument. """ if s not in ('credit', 'checking'): raise argparse.ArgumentTypeError( f"`{s}' invalid. Try `checking' or `credit'." ) return s def build_parser(): """Build a parser and return it.""" parser = argparse.ArgumentParser( description='Transform a Bank of XXXXXX csv to something ledger' ' can use. Try importing it into ledge as follows:' ' ledger -f your.ledger convert --account' ' Assets:Bank:Checking massaged.csv' ) parser.add_argument( '--begin', type=date_type, default=datetime.min, help='now, today, or a ISO date (%%Y-%%m-%%d). ' 'Defaults to earliest date possible.' ) parser.add_argument( '--format', type=format_type, required=True, help='checking or credit', ) return parser class Translation(ABC): @staticmethod def flip(s): """Flip the sign of a string currency value.""" if s.startswith('-'): return s[1:] else: return f'-{s}' def __init__(self, row): self.row = row @abstractmethod def date(self): pass @abstractmethod def amount(self): pass @abstractmethod def description(self): pass @abstractmethod def code(self): pass class CheckingTranslation(Translation): def date(self): return datetime.strptime(self.row['Date'], '%m/%d/%Y') def amount(self): if self.row['Credit']: return f'-{self.row["Credit"]}' return self.row['Debit'] def description(self): return self.row['Description'] def code(self): return self.row['ChkRef'] class CreditTranslation(Translation): def date(self): return datetime.strptime(self.row['PostDate'], '%m/%d/%y') def amount(self): return Translation.flip(self.row['Amount']) def description(self): return self.row['Description'] def code(self): return self.row['Reference'].strip("'") TRANSLATION = { 'checking': CheckingTranslation, 'credit': CreditTranslation, } def main(): """ETL tool entry point.""" args = build_parser().parse_args() Translation = TRANSLATION[args.format] reader = csv.DictReader(sys.stdin) writer = csv.DictWriter(sys.stdout, ('date', 'amount', 'code', 'description')) writer.writeheader() for r in reader: t = Translation(r) date = t.date() if date < args.begin: continue writer.writerow({ 'date': date.strftime('%Y-%m-%d'), 'code': t.code(), 'description': t.description(), 'amount': t.amount(), }) if __name__ == '__main__': main()