117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
#!/usr/bin/env python
|
|
"""Integration module to combine WellsFargo and YNAB transactions for reconciliation."""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import TypedDict
|
|
|
|
from wf_debit_extractor import WellsFargoPayloadExtractor
|
|
from transaction_processor import TransactionProcessor
|
|
from ynab_client import YNABClient
|
|
|
|
|
|
class ReconciliationReport(TypedDict):
|
|
"""Summary of WellsFargo and YNAB data."""
|
|
|
|
timestamp: str
|
|
wellsfargo_transactions: list[dict]
|
|
ynab_transactions: list[dict]
|
|
wellsfargo_total: float
|
|
ynab_total: float
|
|
transaction_count_wf: int
|
|
transaction_count_ynab: int
|
|
|
|
|
|
def reconcile_transactions() -> ReconciliationReport:
|
|
"""
|
|
Fetch and compare WellsFargo and YNAB transactions.
|
|
|
|
Returns:
|
|
Reconciliation report with both datasets
|
|
"""
|
|
print("Fetching WellsFargo transactions...")
|
|
wf_extractor = WellsFargoPayloadExtractor()
|
|
wf_payloads = wf_extractor.extract_from_file("posted.dat")
|
|
|
|
wf_processor = TransactionProcessor()
|
|
wf_transactions = wf_processor.extract_transactions(wf_payloads)
|
|
|
|
print(f"Found {len(wf_transactions)} WellsFargo transaction(s)")
|
|
|
|
print("\nFetching YNAB transactions...")
|
|
ynab_client = YNABClient("ynab.yml")
|
|
ynab_transactions = ynab_client.get_transactions_for_days(days=30)
|
|
|
|
print(f"Found {len(ynab_transactions)} YNAB transaction(s)")
|
|
|
|
# Calculate totals
|
|
wf_total = sum(txn["transaction_amount_cents"] for txn in wf_transactions) / 100
|
|
ynab_total = sum(txn["amount_milliCurrency"] for txn in ynab_transactions) / 1000
|
|
|
|
# Serialize transactions for JSON output
|
|
wf_serialized = [
|
|
{
|
|
"id": txn["id"],
|
|
"amount": txn["transaction_amount_cents"] / 100,
|
|
"amount_cents": txn["transaction_amount_cents"],
|
|
"date": txn["transaction_date_datetime"].isoformat(),
|
|
"post_date": txn["post_date_datetime"].isoformat(),
|
|
"description": txn["transaction_description"],
|
|
}
|
|
for txn in wf_transactions
|
|
]
|
|
|
|
ynab_serialized = [
|
|
{
|
|
"id": txn["id"],
|
|
"amount": txn["amount_milliCurrency"] / 1000,
|
|
"amount_milliCurrency": txn["amount_milliCurrency"],
|
|
"date": txn["date"],
|
|
"payee": txn["payee_name"],
|
|
"category": txn["category_name"],
|
|
"account": txn["account_name"],
|
|
}
|
|
for txn in ynab_transactions
|
|
]
|
|
|
|
report: ReconciliationReport = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"wellsfargo_transactions": wf_serialized,
|
|
"ynab_transactions": ynab_serialized,
|
|
"wellsfargo_total": wf_total,
|
|
"ynab_total": ynab_total,
|
|
"transaction_count_wf": len(wf_transactions),
|
|
"transaction_count_ynab": len(ynab_transactions),
|
|
}
|
|
|
|
return report
|
|
|
|
|
|
def main():
|
|
"""Run full reconciliation and save report."""
|
|
report = reconcile_transactions()
|
|
|
|
# Display summary
|
|
print("\n" + "=" * 60)
|
|
print("RECONCILIATION SUMMARY")
|
|
print("=" * 60)
|
|
print(f"WellsFargo transactions: {report['transaction_count_wf']}")
|
|
print(f"WellsFargo total: ${report['wellsfargo_total']:.2f}")
|
|
print(f"\nYNAB transactions: {report['transaction_count_ynab']}")
|
|
print(f"YNAB total: ${report['ynab_total']:.2f}")
|
|
print("=" * 60)
|
|
|
|
# Save report
|
|
output_file = Path("reconciliation_report.json")
|
|
with output_file.open("w", encoding="utf-8") as f:
|
|
json.dump(report, f, indent=2)
|
|
|
|
print(f"\nReport saved to {output_file}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
|