diff --git a/main.py b/main.py index e3fc768..b9a4fb2 100644 --- a/main.py +++ b/main.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import Any import ynab +from dateutil.relativedelta import relativedelta from category_suggester import CategoryHistory from wf_credit_extractor import WellsFargoPayloadExtractor as CreditExtractor @@ -601,27 +602,48 @@ def _execute_import( # --- Create new + duplicates --- to_create = classification.new + classification.duplicates if to_create: - new_txns = [ - ynab.NewTransaction( - account_id=t["account_id"], - date=date.fromisoformat(t["date"]), - amount=t["amount"], - payee_name=t["payee_name"], - category_id=t.get("category_id"), - cleared=ynab.TransactionClearedStatus(t["cleared"]), - flag_color=ynab.TransactionFlagColor(t["flag_color"]), - import_id=t["import_id"], - approved=t.get("approved", False), + today = date.today() + earliest_ynab_date = today - relativedelta(years=5) + eligible_to_create = [] + skipped_for_date = [] + for transaction in to_create: + transaction_date = date.fromisoformat(transaction["date"]) + if earliest_ynab_date <= transaction_date <= today: + eligible_to_create.append(transaction) + else: + skipped_for_date.append(transaction) + + if skipped_for_date: + print( + f" [ynab] Skipped {len(skipped_for_date)} transaction(s) outside " + f"the supported date range ({earliest_ynab_date} through {today}):" + ) + for transaction in skipped_for_date: + payee = transaction.get("payee_name") or "(no payee)" + print(f" {transaction['date']} | {payee}") + + if eligible_to_create: + new_txns = [ + ynab.NewTransaction( + account_id=t["account_id"], + date=date.fromisoformat(t["date"]), + amount=t["amount"], + payee_name=t["payee_name"], + category_id=t.get("category_id"), + cleared=ynab.TransactionClearedStatus(t["cleared"]), + flag_color=ynab.TransactionFlagColor(t["flag_color"]), + import_id=t["import_id"], + approved=t.get("approved", False), + ) + for t in eligible_to_create + ] + resp = ynab_client.create_transactions(new_txns) + created_ids = resp.transaction_ids or [] + dup_ids = resp.duplicate_import_ids or [] + print( + f" [ynab] Created {len(created_ids)} transaction(s)" + + (f", {len(dup_ids)} server-side duplicate(s) skipped" if dup_ids else "") ) - for t in to_create - ] - resp = ynab_client.create_transactions(new_txns) - created_ids = resp.transaction_ids or [] - dup_ids = resp.duplicate_import_ids or [] - print( - f" [ynab] Created {len(created_ids)} transaction(s)" - + (f", {len(dup_ids)} server-side duplicate(s) skipped" if dup_ids else "") - ) # --- Update pending→posted --- if classification.pending_to_posted: