NPO-Accounting/conservancy_beancount Changeset - ed0bc469ce5a

Changeset - ed0bc469ce5a

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Ben Sturmfels (bsturmfels) - 2 years ago 2022-02-04 08:15:11
ben@sturm.com.au

reconcile: Add type checking information to new prototype reconcilers.

2 files changed with 17 insertions and 13 deletions:

conservancy_beancount/reconcile/helper.py

conservancy_beancount/reconcile/prototype_amex_reconciler.py

0 comments (0 inline, 0 general)

conservancy_beancount/reconcile/helper.py

➞

Show inline comments

@@ ... @@
 Not implemented:
  - --report-group-regex
  - git branch selection from bean-query-goofy-daemon.plx
 """
 import argparse
 from dateutil.relativedelta import relativedelta
 import datetime
 import io
 import tempfile
 import textwrap
 import typing
 from typing import List
 import os
 from beancount import loader
 from beancount.query.query import run_query
 def end_of_month(date):
+def end_of_month(date: datetime.date) -> datetime.date:
     """Given a date, return the last day of the month."""
     # Using 'day' replaces, rather than adds.
     return date + relativedelta(day=31)
 def format_record_for_grep(row, homedir):
+def format_record_for_grep(row: typing.List, homedir: str) -> typing.List:
     """Return a line in a grep-style.
     This is so the line can be fed into Emacs grep-mode for quickly jumping to
     the relevant lines in the books.
     """
     file = row[0].replace(homedir, '~')
     return [f'{file}:{row[1]}:'] + row[2:]
 def max_column_widths(rows):
+def max_column_widths(rows: List) -> List[int]:
     """Return the max width for each column in a table of data."""
     if not rows:
         return []
     else:
         maxes = [0] * len(rows[0])
         for row in rows:
             for i, val in enumerate(row):
                 length = len(str(val))
                 maxes[i] = max(maxes[i], length)
         return maxes
 def tabulate(rows, headers=None):
+def tabulate(rows: List, headers: List=None) -> str:
     """Format a table of data as a string.
     Implemented here to avoid adding dependency on "tabulate" package.
     """
     output = io.StringIO()
     if headers:
         rows = [headers] + rows
     widths = max_column_widths(rows)
     for row in rows:
         for i, col in enumerate(row):
             width = widths[i]
             if col is None:
@@ ... / @@ -92,26 +94,27 @@ parser.add_argument('--grep-output-filename') @@
 args = parser.parse_args()
 if args.month or args.period:
     if not (args.month and args.period):
         parser.error('--month and --period must be used together')
     parsed_date = datetime.datetime.strptime(args.month, '%Y-%m').date()
     preDate = end_of_month(parsed_date - relativedelta(months=args.period)).isoformat()
     lastDateInPeriod = end_of_month(parsed_date).isoformat()
     month = args.month
 else:
     if not (args.cur_end_date and args.prev_end_date):
         parser.error(' --prev-end-date and --cur-end-date must be used together')
     preDate = args.prev_end_date
     lastDateInPeriod = args.cur_end_date
     month = lastDateInPeriod.strftime('%Y-%m')
     lastDateInPeriod = args.cur_end_date.isoformat()
     month = args.cur_end_date.strftime('%Y-%m')
 grep_output_file: typing.IO
 if args.grep_output_filename:
     grep_output_file = open(args.grep_output_filename, 'w')
 else:
     grep_output_file = tempfile.NamedTemporaryFile(prefix='bc-reconcile-grep-output_', mode='w', delete=False)
 beancount_file = args.beancount_file
 account = args.account
 cost_function = args.cost_function
 statement_match = args.statement_match if args.statement_match else month
 QUERIES = {
     f"00: CLEARED BAL ENDING DAY BEFORE {preDate}":
     # $CONLEDGER -V -C -e "$preDate" bal "/$acct/"
@@ ... / @@ -159,23 +162,23 @@ QUERIES = { @@
     WHERE account = "{account}"
     and META("bank-statement") ~  "{statement_match}" and number({cost_function}(position)) > 0""",
+}
 # Run Beancount queries.
 print(f"START RECONCILIATION FOR {account} ENDING {lastDateInPeriod} (previous end date {preDate})")
 entries, _, options = loader.load_file(beancount_file)
 for desc, query in QUERIES.items():
     rtypes, rrows = run_query(entries, options, query, numberify=True)
     if not rrows:
         print(f'{desc:<55} {"N/A":>11}')
     elif desc.startswith('04'):
         homedir = os.getenv('HOME')
+        homedir = os.getenv('HOME', '')
         print(f'{desc}\n   See {grep_output_file.name}')
         grep_rows = [format_record_for_grep(row, homedir) for row in rrows]
         print(tabulate(grep_rows), file=grep_output_file)
     elif len(rrows) == 1:
         result = rrows[0][0]
         print(f'{desc:<55} {result:>11,.2f}')
     else:
         headers = [c[0].capitalize() for c in rtypes]
         print(desc)
         print(textwrap.indent(tabulate(rrows, headers=headers), '    '))

conservancy_beancount/reconcile/prototype_amex_reconciler.py

➞

Show inline comments

 """A prototype AMEX statement reconciler.
 Run like this:
 $ python3 -m pip install thefuzz
 $ python3 conservancy_beancount/reconcile/prototype_amex_reconciler.py --beancount-file=$HOME/conservancy/beancount/books/2021.beancount --amex-csv=$HOME/conservancy/confidential/2021-09-10_AMEX_activity.csv
 """
 import argparse
 import csv
 import datetime
 import decimal
 from typing import Dict, List, Tuple
 from beancount import loader
 from beancount.query.query import run_query
 from thefuzz import fuzz
+from thefuzz import fuzz  # type: ignore
 # NOTE: Statement doesn't seem to give us a running balance or a final total.
 def standardize_amex_record(row):
+def standardize_amex_record(row: Dict) -> Dict:
     return {
         'date': datetime.datetime.strptime(row['Date'], '%m/%d/%Y').date(),
         'amount': -1 * decimal.Decimal(row['Amount']),
         'payee': row['Description'],
+    }
 def standardize_beancount_record(row):
+def standardize_beancount_record(row) -> Dict:  # type: ignore[no-untyped-def]
     return {
         'date': row.date,
         'amount': row.number_cost_position,
         'payee': row.payee if row.payee else row.narration,
+    }
 def format_record(record):
+def format_record(record: Dict) -> str:
     return f"{record['date'].isoformat()}: {record['amount']:>8} {record['payee'][:20]:<20}"
 def sort_records(records):
+def sort_records(records: List) -> List:
     return sorted(records, key=lambda x: (x['date'], x['amount']))
 def records_match(r1, r2):
+def records_match(r1: Dict, r2: Dict) -> Tuple[bool, str]:
     """Do these records represent the same transaction?"""
     date_matches = r1['date'] >= r2['date'] - datetime.timedelta(days=1) and r1['date'] <= r2['date'] + datetime.timedelta(days=1)
     amount_matches = r1['amount'] == r2['amount']
     payee_match_quality = fuzz.token_set_ratio(r1['payee'], r2['payee'])
     payee_matches = payee_match_quality > 50
     if date_matches and amount_matches and payee_matches:
         return True, ''
     elif date_matches and amount_matches:
         return True, f'Low payee match ({payee_match_quality}%)'
     else:
         return False, ''

0 comments (0 inline, 0 general)