From 78a6d73c1149dc8b1817b88ac6c0fbb553e60b25 Mon Sep 17 00:00:00 2001 From: Ryan Kegel Date: Sun, 22 Mar 2026 23:30:38 -0400 Subject: [PATCH] first commit --- .gitignore | 1 + INDEX.md | 221 +++++++++++++ PROJECT_SUMMARY.md | 269 ++++++++++++++++ QUICKSTART.md | 144 +++++++++ README.md | 172 ++++++++++ YNAB_MODULE.md | 129 ++++++++ example_usage.py | 59 ++++ extractor.py | 88 ++++++ main.py | 658 +++++++++++++++++++++++++++++++++++++++ reconcile.py | 116 +++++++ requirements.txt | Bin 0 -> 452 bytes test_import.py | 152 +++++++++ transaction_processor.py | 77 +++++ wf_credit_extractor.py | 89 ++++++ wf_debit_extractor.py | 106 +++++++ wf_extractor.py | 105 +++++++ ynab_client.py | 266 ++++++++++++++++ ynab_example.py | 65 ++++ ynab_import.md | 15 + 19 files changed, 2732 insertions(+) create mode 100644 .gitignore create mode 100644 INDEX.md create mode 100644 PROJECT_SUMMARY.md create mode 100644 QUICKSTART.md create mode 100644 README.md create mode 100644 YNAB_MODULE.md create mode 100644 example_usage.py create mode 100644 extractor.py create mode 100644 main.py create mode 100644 reconcile.py create mode 100644 requirements.txt create mode 100644 test_import.py create mode 100644 transaction_processor.py create mode 100644 wf_credit_extractor.py create mode 100644 wf_debit_extractor.py create mode 100644 wf_extractor.py create mode 100644 ynab_client.py create mode 100644 ynab_example.py create mode 100644 ynab_import.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..82f0c3a --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/data/ diff --git a/INDEX.md b/INDEX.md new file mode 100644 index 0000000..db78841 --- /dev/null +++ b/INDEX.md @@ -0,0 +1,221 @@ +# FargoImport - File Index & Navigation + +## πŸ“– Start Here + +1. **New to the project?** β†’ Read `QUICKSTART.md` (5 min read) +2. **Want full details?** β†’ Read `README.md` (comprehensive) +3. **Need module reference?** β†’ Check `PROJECT_SUMMARY.md` + +--- + +## 🐍 Python Modules (7 files) + +### Core Extraction +- **`wf_debit_extractor.py`** - Extract Wells Fargo payloads from HAR JSON + - `WellsFargoPayloadExtractor` class + - Methods: `extract_from_file()`, `extract_from_data()`, `get_objects()` + - Usage: Extract proprietary JSON wrapped in `/*WellFargoProprietary%...%WellFargoProprietary*/` + +### Data Processing +- **`transaction_processor.py`** - Transform raw transactions into structured format + - `TransactionProcessor` class + - Methods: `extract_transactions()` + - Converts amounts to cents, timestamps to datetime, normalizes data + +### External Integration +- **`ynab_client.py`** - Connect to YNAB API and fetch transactions + - `YNABClient` class + - Methods: `get_transactions_for_days()`, `get_account_balance()` + - Configuration via `ynab.yml` + +### Reconciliation +- **`reconcile.py`** - Compare Wells Fargo and YNAB transactions + - `reconcile_transactions()` function + - `main()` CLI entry point + - Generates `reconciliation_report.json` + +### Credit Module +- **`wf_credit_extractor.py`** - Credit extractor module (renamed from `extractor.py`) + +### CLI & Examples +- **`main.py`** - Command-line interface for Wells Fargo extraction + - Args: `--input`, `--output`, `--show-errors` + - Usage: `python main.py --input posted.dat` + +- **`example_usage.py`** - Wells Fargo transaction processing example + - Demonstrates extraction β†’ processing β†’ JSON output + - Output: `processed_transactions.json` + +- **`ynab_example.py`** - YNAB transaction fetching example + - Demonstrates YNAB API integration + - Output: `ynab_transactions.json` + +--- + +## βš™οΈ Configuration (1 file) + +- **`ynab.yml`** - YNAB API credentials and account mapping + ```yaml + ynab: + - token: "your_ynab_api_token" + plan_id: "your_budget_id" + accounts: + - name: "Credit Card" + id: "account_uuid" + ``` + +--- + +## πŸ“š Documentation (4 files) + +| File | Purpose | Read Time | +|------|---------|-----------| +| `QUICKSTART.md` | Quick start guide with examples | 5 min | +| `README.md` | Full project documentation | 15 min | +| `PROJECT_SUMMARY.md` | Module reference & architecture | 10 min | +| `YNAB_MODULE.md` | YNAB client detailed docs | 5 min | + +--- + +## πŸ“Š Data Files (4 output files) + +These are generated by running the example scripts: + +- **`extracted.json`** - Raw Wells Fargo payload objects (from `main.py`) +- **`processed_transactions.json`** - Processed WF transactions (from `example_usage.py`) + - 179 sample transactions + - Fields: id, amount_cents, dates (timestamp & datetime), description + +- **`ynab_transactions.json`** - YNAB transactions (from `ynab_example.py`) + - Fetched from YNAB API + - Fields: id, date, payee, category, amount, account, memo + +- **`reconciliation_report.json`** - Comparison report (from `reconcile.py`) + - Summary of both data sources + - Totals and transaction counts + - Full transaction lists + +--- + +## 🎯 Common Tasks + +### Extract Wells Fargo Data +```bash +# Display in console +python main.py + +# Save to file +python main.py --output extracted.json + +# Process transactions +python example_usage.py +``` + +### Fetch YNAB Data +```bash +python ynab_example.py +``` + +### Reconcile Both Sources +```bash +python reconcile.py +``` + +### Programmatic Usage +```python +from wf_debit_extractor import WellsFargoPayloadExtractor +from transaction_processor import TransactionProcessor +from ynab_client import YNABClient + +# Wells Fargo +wf = WellsFargoPayloadExtractor() +payloads = wf.extract_from_file("posted.dat") +processor = TransactionProcessor() +wf_txns = processor.extract_transactions(payloads) + +# YNAB +ynab = YNABClient() +ynab_txns = ynab.get_transactions_for_days(days=30) + +# Access +for txn in wf_txns[:5]: + print(txn["transaction_description"]) +``` + +--- + +## πŸ” Module Dependencies + +``` +posted.dat (input) + ↓ +wf_debit_extractor.py + ↓ +transaction_processor.py + ↓ +example_usage.py β†’ processed_transactions.json + +ynab.yml (config) + ↓ +ynab_client.py + ↓ +ynab_example.py β†’ ynab_transactions.json + +Both ↓ +reconcile.py β†’ reconciliation_report.json +``` + +--- + +## βœ… File Status + +| Type | Count | Status | +|------|-------|--------| +| Core Modules | 4 | βœ… Tested & Working | +| CLI/Examples | 3 | βœ… Tested & Working | +| Configuration | 1 | βœ… Ready | +| Documentation | 4 | βœ… Complete | +| Data Files | 4 | βœ… Generated | +| **Total** | **16** | **βœ… Complete** | + +--- + +## πŸš€ Getting Started + +1. **Read** `QUICKSTART.md` (2 min) +2. **Update** `ynab.yml` with your API token +3. **Run** `python example_usage.py` (verify WF extraction) +4. **Run** `python ynab_example.py` (verify YNAB connection) +5. **Run** `python reconcile.py` (generate comparison) + +--- + +## πŸ“ž Documentation Map + +- Need quick answer? β†’ `QUICKSTART.md` +- How do I use X? β†’ `README.md` +- What does Y do? β†’ `PROJECT_SUMMARY.md` +- YNAB details? β†’ `YNAB_MODULE.md` +- Code docstrings? β†’ Open the `.py` file + +--- + +## πŸ”— Key Entry Points + +**CLI Interface:** +- `main.py` - Extract Wells Fargo data from command line + +**Programmatic API:** +- `wf_debit_extractor.py` - Wells Fargo extraction +- `transaction_processor.py` - Transaction processing +- `ynab_client.py` - YNAB integration +- `reconcile.py` - Reconciliation + +**Examples:** +- `example_usage.py` - Wells Fargo example +- `ynab_example.py` - YNAB example + +--- + +Generated: March 20, 2026 + diff --git a/PROJECT_SUMMARY.md b/PROJECT_SUMMARY.md new file mode 100644 index 0000000..f6e820a --- /dev/null +++ b/PROJECT_SUMMARY.md @@ -0,0 +1,269 @@ +# FargoImport Project - Complete Module Overview + +## Project Structure + +This project provides utilities for extracting Wells Fargo transaction data and integrating it with YNAB (You Need A Budget) for financial reconciliation. + +## Core Modules + +### 1. **wf_debit_extractor.py** - Wells Fargo Data Extraction +Extracts proprietary Wells Fargo JSON payloads from HAR-style JSON files. + +**Classes:** +- `WellsFargoPayloadExtractor` - Main extraction engine + - `extract_from_file(path)` - Load and parse HAR JSON file + - `extract_from_data(har_data)` - Parse raw HAR data + - `get_objects()` - Access parsed objects + - Properties: `objects`, `errors` + +**Functions:** +- `extract_wf_objects(har_data)` - Backward-compatible functional API + +**Usage:** +```python +from wf_debit_extractor import WellsFargoPayloadExtractor + +extractor = WellsFargoPayloadExtractor() +payloads = extractor.extract_from_file("posted.dat") +print(len(payloads)) # Number of extracted payloads +``` + +--- + +### 2. **transaction_processor.py** - Wells Fargo Transaction Processing +Transforms raw Wells Fargo payloads into structured transaction data. + +**Classes:** +- `TransactionProcessor` - Transaction extraction and transformation + - `extract_transactions(extracted_objects)` - Extract from payloads + +**Data Structure:** +- `ProcessedTransaction` TypedDict with fields: + - `id` - Transaction ID + - `transaction_amount_cents` - Amount as integer cents + - `transaction_date_timestamp` - Milliseconds since epoch + - `transaction_date_datetime` - Python datetime object + - `post_date_timestamp` - Milliseconds since epoch + - `post_date_datetime` - Python datetime object + - `transaction_description` - Transaction description + +**Usage:** +```python +from transaction_processor import TransactionProcessor +from wf_debit_extractor import WellsFargoPayloadExtractor + +extractor = WellsFargoPayloadExtractor() +payloads = extractor.extract_from_file("posted.dat") + +processor = TransactionProcessor() +transactions = processor.extract_transactions(payloads) +``` + +--- + +### 3. **ynab_client.py** - YNAB Integration +Authenticates with YNAB API and fetches transactions using official SDK. + +**Classes:** +- `YNABClient` - YNAB API wrapper + - `__init__(config_path)` - Initialize from YAML config + - `get_transactions_for_days(days)` - Fetch transactions for past N days + - `get_account_balance(account_id)` - Get account balance + +**Data Structure:** +- `YNABTransaction` TypedDict with fields: + - `id` - Transaction ID + - `date` - Date (YYYY-MM-DD format) + - `payee_name` - Payee name + - `category_name` - Category name + - `amount_milliCurrency` - Amount in milliCurrency units + - `memo` - Transaction memo + - `account_id` - YNAB account ID + - `account_name` - Account name + - `cleared` - Cleared status + +**Configuration (ynab.yml):** +```yaml +ynab: + - token: "your_api_token" + plan_id: "your_budget_id" + accounts: + - name: "Credit Card" + id: "account_id" +``` + +**Usage:** +```python +from ynab_client import YNABClient + +client = YNABClient("ynab.yml") +transactions = client.get_transactions_for_days(days=30) +balance = client.get_account_balance("account_id") +``` + +--- + +### 4. **reconcile.py** - Reconciliation Integration +Combines Wells Fargo and YNAB transactions for comparison and reporting. + +**Functions:** +- `reconcile_transactions()` - Fetch both data sources and create report +- `main()` - CLI interface with summary and JSON output + +**Data Structure:** +- `ReconciliationReport` TypedDict with: + - `timestamp` - Report timestamp + - `wellsfargo_transactions` - List of WF transactions + - `ynab_transactions` - List of YNAB transactions + - `wellsfargo_total` - Total WF amount + - `ynab_total` - Total YNAB amount + - `transaction_count_wf` - Count of WF transactions + - `transaction_count_ynab` - Count of YNAB transactions + +**Usage:** +```bash +python reconcile.py +``` + +Output: `reconciliation_report.json` with full comparison + +--- + +## Example Scripts + +### **example_usage.py** - Wells Fargo Transaction Example +Demonstrates extraction and processing of Wells Fargo data. + +**Output:** +- Console display of first 5 transactions +- Saves all transactions to `processed_transactions.json` + +**Run:** +```bash +python example_usage.py +``` + +### **ynab_example.py** - YNAB Transaction Example +Demonstrates fetching and displaying YNAB transactions. + +**Output:** +- Console display of first 10 transactions +- Saves all transactions to `ynab_transactions.json` + +**Run:** +```bash +python ynab_example.py +``` + +--- + +## CLI Interface + +### **main.py** - Command-line Entry Point +Thin wrapper around extraction functionality. + +**Arguments:** +- `--input` - Input HAR file (default: posted.dat) +- `--output` - Output JSON file (optional) +- `--show-errors` - Display parse errors + +**Examples:** +```bash +# Extract to console +python main.py --input posted.dat + +# Extract to file +python main.py --input posted.dat --output extracted.json + +# Show parsing errors +python main.py --input posted.dat --show-errors +``` + +--- + +## Data Flow + +``` +posted.dat (HAR JSON) + ↓ +wf_debit_extractor.py (Extract proprietary payloads) + ↓ +Extracted payload objects + ↓ +transaction_processor.py (Transform to structured format) + ↓ +Processed transaction list + ↓ +β”œβ”€β†’ example_usage.py (Save to processed_transactions.json) +└─→ reconcile.py (Compare with YNAB) + ↓ + ynab_client.py (Fetch from YNAB API) + ↓ + reconciliation_report.json +``` + +--- + +## Dependencies + +- **ynab** (v2.1.0) - Official YNAB Python SDK +- **pyyaml** (v6.0.3) - YAML configuration parsing + +**Install:** +```bash +pip install ynab pyyaml +``` + +--- + +## File Manifest + +| File | Purpose | Lines | +|------|---------|-------| +| `wf_debit_extractor.py` | Core extraction logic | 91 | +| `transaction_processor.py` | Transaction transformation | 70 | +| `ynab_client.py` | YNAB API integration | 150 | +| `reconcile.py` | Reconciliation orchestration | 117 | +| `main.py` | CLI wrapper | ~40 | +| `example_usage.py` | WF usage example | ~60 | +| `ynab_example.py` | YNAB usage example | ~60 | +| `ynab.yml` | YNAB configuration | 6 | +| `README.md` | Project documentation | ~100 | + +--- + +## Key Features + +βœ“ Extracts Wells Fargo proprietary JSON from HAR files +βœ“ Transforms transactions with amount normalization +βœ“ Converts millisecond timestamps to Python datetime objects +βœ“ Integrates with YNAB API for budget tracking +βœ“ Supports date range filtering (past X days) +βœ“ Generates reconciliation reports +βœ“ Error handling and reporting +βœ“ Type-safe with TypedDict definitions +βœ“ Configuration via YAML +βœ“ Modular, reusable architecture + +--- + +## Status + +- βœ… All modules created and tested +- βœ… No static errors +- βœ… Dependencies installed +- βœ… CLI functional +- βœ… Examples working +- βœ… Backward compatible +- ⚠️ YNAB integration requires valid API token + +--- + +## Next Steps (Optional) + +1. **Database Integration** - Store transactions in SQLite/PostgreSQL +2. **Matching Algorithm** - Auto-match WF↔YNAB transactions by amount/date +3. **Web Dashboard** - Flask/FastAPI interface for viewing reports +4. **Scheduling** - Celery/APScheduler for periodic reconciliation +5. **Export Formats** - CSV, Excel, PDF report generation + diff --git a/QUICKSTART.md b/QUICKSTART.md new file mode 100644 index 0000000..4a4e482 --- /dev/null +++ b/QUICKSTART.md @@ -0,0 +1,144 @@ +# Quick Start Guide + +## Installation + +1. **Dependencies already installed:** + - `ynab` (2.1.0) - YNAB API SDK + - `pyyaml` (6.0.3) - YAML parsing + +2. **Virtual environment active** in `.venv/` + +## Configuration + +Update `ynab.yml` with your YNAB credentials: + +```yaml +ynab: + - token: "your_actual_ynab_api_token" + plan_id: "your_budget_plan_id" + accounts: + - name: "Credit Card" + id: "account_uuid" +``` + +Get your credentials from: https://app.youneedabudget.com/settings/developer + +## Quick Examples + +### Extract Wells Fargo Transactions + +```bash +# Display in console +python main.py --input posted.dat + +# Save to file +python main.py --input posted.dat --output extracted.json + +# Process and save transactions +python example_usage.py +``` + +**Output:** `processed_transactions.json` with 179 sample transactions + +### Fetch YNAB Transactions + +```bash +python ynab_example.py +``` + +**Output:** `ynab_transactions.json` with transactions from past 30 days + +### Reconcile Both Sources + +```bash +python reconcile.py +``` + +**Output:** `reconciliation_report.json` comparing both datasets + +## Module Quick Reference + +| Task | Module | Method | +|------|--------|--------| +| Extract WF payload | `wf_debit_extractor.py` | `WellsFargoPayloadExtractor().extract_from_file()` | +| Process WF transactions | `transaction_processor.py` | `TransactionProcessor().extract_transactions()` | +| Fetch YNAB transactions | `ynab_client.py` | `YNABClient().get_transactions_for_days()` | +| Reconcile both | `reconcile.py` | `reconcile_transactions()` | + +## Common Code Patterns + +### Get Wells Fargo Transactions +```python +from wf_debit_extractor import WellsFargoPayloadExtractor +from transaction_processor import TransactionProcessor + +extractor = WellsFargoPayloadExtractor() +payloads = extractor.extract_from_file("posted.dat") + +processor = TransactionProcessor() +transactions = processor.extract_transactions(payloads) + +for txn in transactions: + print(f"{txn['id']}: ${txn['transaction_amount_cents']/100:.2f}") +``` + +### Get YNAB Transactions +```python +from ynab_client import YNABClient + +client = YNABClient("ynab.yml") +transactions = client.get_transactions_for_days(days=30) + +for txn in transactions: + print(f"{txn['date']}: {txn['payee_name']} ${txn['amount_milliCurrency']/1000:.2f}") +``` + +### Full Reconciliation +```python +from reconcile import reconcile_transactions + +report = reconcile_transactions() +print(f"WF: {report['transaction_count_wf']} transactions") +print(f"YNAB: {report['transaction_count_ynab']} transactions") +``` + +## File Locations + +- **Input:** `posted.dat` (Wells Fargo HAR export) +- **Config:** `ynab.yml` (YNAB credentials) +- **Output:** + - `extracted.json` - Raw WF payloads + - `processed_transactions.json` - Processed WF transactions + - `ynab_transactions.json` - YNAB transactions + - `reconciliation_report.json` - Comparison report + +## Troubleshooting + +**YNAB Authorization Error:** +- Verify token in `ynab.yml` is correct +- Check token hasn't expired +- Ensure token has budget access + +**Wells Fargo Extraction Returns 0 Objects:** +- Verify `posted.dat` contains `/*WellFargoProprietary%...%WellFargoProprietary*/` +- Check file is valid JSON + +**Import Errors:** +- Activate venv: `.venv\Scripts\Activate.ps1` +- Reinstall deps: `pip install ynab pyyaml` + +## Documentation + +See detailed docs in: +- `README.md` - Full project documentation +- `PROJECT_SUMMARY.md` - Module reference +- `YNAB_MODULE.md` - YNAB client details +- Module docstrings - Inline documentation + +## Next Steps + +1. Update `ynab.yml` with real credentials +2. Run `python example_usage.py` to test WF extraction +3. Run `python ynab_example.py` to test YNAB connection +4. Run `python reconcile.py` to generate comparison report + diff --git a/README.md b/README.md new file mode 100644 index 0000000..5a1b444 --- /dev/null +++ b/README.md @@ -0,0 +1,172 @@ +# FargoImport extractor + +This utility reads a HAR-style JSON file (for example `posted.dat`), scans: + +- `log.entries[*].response.content.text` + +It only processes text values that start with: + +- `/*WellFargoProprietary%` + +For each matching value, it extracts the JSON string between: + +- start: `/*WellFargoProprietary%` +- end: `%WellFargoProprietary*/` + +Then it parses that extracted string with Python `json.loads` and outputs the resulting Python objects. + +## Run + +```powershell +python .\main.py --input .\posted.dat +``` + +Optionally write extracted objects to a file: + +```powershell +python .\main.py --input .\posted.dat --output .\extracted.json +``` + +Show parse/skip errors while extracting: + +```powershell +python .\main.py --input .\posted.dat --show-errors +``` + +For HAR files where the payload is specifically at +`/log/entries/0/response/content/text`, use: + +```powershell +python .\main.py --input .\checking.dat --entry0-only --output .\checking_extracted.json +``` + +## Programmatic usage + +```python +from wf_debit_extractor import WellsFargoPayloadExtractor + +extractor = WellsFargoPayloadExtractor() +objects = extractor.extract_from_file("posted.dat") + +# For entry0-only HAR variants: +# objects = extractor.extract_from_file("checking.dat", entry0_only=True) + +# Access converted Python objects. +print(len(objects)) +print(extractor.get_objects()[0]) + +# Optional: review skipped/parse errors. +print(extractor.errors) +``` + +Legacy imports from `main` still work, but `wf_debit_extractor` is now the preferred module for programmatic use. + +## Transaction Processing + +Once you've extracted WellsFargo payloads, use `transaction_processor.py` to extract and transform transaction data. + +```python +from transaction_processor import TransactionProcessor +from wf_debit_extractor import WellsFargoPayloadExtractor + +# Extract payloads +extractor = WellsFargoPayloadExtractor() +payloads = extractor.extract_from_file("posted.dat") + +# Process transactions +processor = TransactionProcessor() +transactions = processor.extract_transactions(payloads) + +# Each transaction has: +# - id (string) +# - transaction_amount_cents (integer, e.g., 9.11 β†’ 911) +# - transaction_date_timestamp (milliseconds since epoch) +# - transaction_date_datetime (Python datetime object) +# - post_date_timestamp (milliseconds since epoch) +# - post_date_datetime (Python datetime object) +# - transaction_description (string) + +for txn in transactions: + print(txn["id"], txn["transaction_amount_cents"], txn["transaction_description"]) +``` + +See `example_usage.py` for a working example that also saves all transactions to `processed_transactions.json`. + +## YNAB Integration + +Use `ynab_client.py` to fetch transactions from your YNAB account. Configuration is loaded from `ynab.yml` which should contain your YNAB token, plan ID, and account IDs to track. + +```python +from ynab_client import YNABClient + +# Initialize with config file +client = YNABClient("ynab.yml") + +# Fetch transactions from past 30 days +transactions = client.get_transactions_for_days(days=30) + +# Each transaction includes: +# - id +# - date (YYYY-MM-DD) +# - payee_name +# - category_name +# - amount_milliCurrency (in milliCurrency units, divide by 1000 for dollars) +# - memo +# - account_id +# - account_name +# - cleared (cleared status) + +for txn in transactions: + print(f"{txn['date']} | {txn['payee_name']} | ${txn['amount_milliCurrency'] / 1000:.2f}") + +# Get account balance +balance = client.get_account_balance("7282c2e6-0470-423d-9748-ec36e29f5698") +print(f"Balance: ${balance / 1000:.2f}") +``` + +See `ynab_example.py` for a complete example that also saves all transactions to `ynab_transactions.json`. + +### ynab.yml Format + +```yaml +ynab: + - token: "your_ynab_api_token_here" + plan_id: "your_budget_id_here" + accounts: + - name: "Credit Card" + id: "account_id_here" +``` + +## Integration & Reconciliation + +Use `reconcile.py` to fetch both WellsFargo and YNAB transactions side-by-side for reconciliation. + +```powershell +python .\reconcile.py +``` + +This will: +1. Extract WellsFargo transactions from posted.dat +2. Fetch YNAB transactions for past 30 days +3. Display summary comparison +4. Save combined report to `reconciliation_report.json` + +### Example Output +``` +============================================================ +RECONCILIATION SUMMARY +============================================================ +WellsFargo transactions: 179 +WellsFargo total: $5,432.15 + +YNAB transactions: 42 +YNAB total: $3,210.50 +============================================================ +``` + +**Note:** Ensure your YNAB API token in `ynab.yml` is valid and has permission to access the budget/plan specified. + + + + + diff --git a/YNAB_MODULE.md b/YNAB_MODULE.md new file mode 100644 index 0000000..f23a20f --- /dev/null +++ b/YNAB_MODULE.md @@ -0,0 +1,129 @@ +# YNAB Client Module Summary + +## Overview +A Python module (`ynab_client.py`) that authenticates with YNAB using the official SDK and retrieves transactions from configured accounts. + +## Files Created + +### `ynab_client.py` - Main Module +- **YNABClient class**: Main interface for YNAB API interactions + - `__init__(config_path)`: Initialize from YAML config file + - `get_transactions_for_days(days=30)`: Fetch transactions for past N days + - `get_account_balance(account_id)`: Get current balance for an account + +- **YNABTransaction TypedDict**: Defines transaction data structure + - `id`: Transaction ID + - `date`: Transaction date (YYYY-MM-DD) + - `payee_name`: Payee name + - `category_name`: Category name + - `amount_milliCurrency`: Amount in milliCurrency units (divide by 1000 for dollars) + - `memo`: Transaction memo + - `account_id`: YNAB account ID + - `account_name`: Account name + - `cleared`: Cleared status + +### `ynab_example.py` - Example Usage +Demonstrates how to: +- Load YNAB client from config +- Fetch transactions for past 30 days +- Display transactions in human-readable format +- Save transactions to JSON file (`ynab_transactions.json`) + +### `ynab.yml` - Configuration File +Contains: +- YNAB API token +- Budget/Plan ID +- List of accounts to track (name, ID, and extractor) + +## Configuration File + +Create a `ynab.yml` file in the project root using the following structure: + +```yaml +ynab: + - token: "your-ynab-api-token" + plan_id: "your-budget-id" + accounts: + - name: "Credit Card" + id: "account-uuid" + extractor: "wf_credit_extractor" + - name: "Debit Card" + id: "account-uuid" + extractor: "wf_debit_extractor" +``` + +### Field Reference + +| Field | Required | Description | +|-------|----------|-------------| +| `ynab` | βœ“ | Top-level key. Contains a list of YNAB configurations (one per budget). | +| `token` | βœ“ | Your YNAB Personal Access Token. Generate one at **YNAB β†’ Account Settings β†’ Developer Settings**. Keep this secret β€” treat it like a password. | +| `plan_id` | βœ“ | The UUID of your YNAB budget. Find it in the URL when viewing your budget: `https://app.ynab.com//budget`. | +| `accounts` | βœ“ | List of bank accounts to sync. Each entry maps a YNAB account to a Wells Fargo data extractor. | +| `accounts[].name` | βœ“ | A human-readable label for the account (used in report output only). | +| `accounts[].id` | βœ“ | The UUID of the YNAB account. Find it in the URL when viewing the account in YNAB, or via the YNAB API. | +| `accounts[].extractor` | βœ“ | Which Wells Fargo extractor to use for this account. Must be one of: `wf_credit_extractor` (for credit cards) or `wf_debit_extractor` (for checking/debit accounts). | + +### How to Find Your IDs + +**API Token** +1. Log in to [app.ynab.com](https://app.ynab.com) +2. Go to **Account Settings β†’ Developer Settings** +3. Click **New Token**, give it a name, and copy the value + +**Budget (plan_id)** +- Open your budget in YNAB β€” the UUID is in the URL: + `https://app.ynab.com/`**`dc25f923-e6b4-45f5-8d42-65e53eed1f1e`**`/budget` + +**Account ID** +- Click on an account in YNAB β€” the UUID appears in the URL after `/accounts/` +- Or use the YNAB API Explorer at `https://api.ynab.com/v1/budgets//accounts` + +## Dependencies +- `ynab` (v2.1.0) - Official YNAB Python SDK +- `pyyaml` (v6.0.3) - YAML file parsing + +## Usage + +### Basic Usage +```python +from ynab_client import YNABClient + +# Initialize client +client = YNABClient("ynab.yml") + +# Get transactions from past 30 days +transactions = client.get_transactions_for_days(days=30) + +# Access transaction data +for txn in transactions: + amount_dollars = txn["amount_milliCurrency"] / 1000 + print(f"{txn['date']} | {txn['payee_name']} | ${amount_dollars:.2f}") +``` + +### Run Example +```powershell +python .\ynab_example.py +``` + +This will: +1. Load credentials from `ynab.yml` +2. Fetch transactions from configured accounts +3. Display first 10 transactions +4. Save all transactions to `ynab_transactions.json` + +## Key Features +- βœ“ Configuration from YAML file +- βœ“ Automatic date filtering (past X days) +- βœ“ Category name lookup +- βœ“ Account name mapping +- βœ“ Error handling for network/API issues +- βœ“ Both timestamp and formatted date fields +- βœ“ Balance retrieval for accounts + +## Status +- βœ“ Module created and tested +- βœ“ Configuration loading verified +- βœ“ Dependencies installed +- βœ“ No static errors + diff --git a/example_usage.py b/example_usage.py new file mode 100644 index 0000000..c113478 --- /dev/null +++ b/example_usage.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python +"""Example usage of WellsFargo extraction and transaction processing.""" + +import json +from pathlib import Path + +from transaction_processor import TransactionProcessor +from wf_debit_extractor import WellsFargoPayloadExtractor + + +def main(): + # Load or extract WellsFargo payloads + extractor = WellsFargoPayloadExtractor() + extracted_objects = extractor.extract_from_file("posted.dat") + + print(f"Extracted {len(extracted_objects)} WellsFargo payload(s)\n") + + # Process transactions + processor = TransactionProcessor() + transactions = processor.extract_transactions(extracted_objects) + + print(f"Found {len(transactions)} transaction(s)\n") + + # Display first few transactions + for i, txn in enumerate(transactions[:5]): + print(f"Transaction {i + 1}:") + print(f" ID: {txn['id']}") + print( + f" Amount: ${txn['transaction_amount_cents'] / 100:.2f} ({txn['transaction_amount_cents']} cents)" + ) + print(f" Description: {txn['transaction_description']}") + print(f" Transaction Date: {txn['transaction_date_datetime']} (ts: {txn['transaction_date_timestamp']})") + print(f" Post Date: {txn['post_date_datetime']} (ts: {txn['post_date_timestamp']})") + print() + + # Optionally save all processed transactions to JSON + output_file = Path("processed_transactions.json") + serializable = [ + { + "id": txn["id"], + "transaction_amount_cents": txn["transaction_amount_cents"], + "transaction_date_timestamp": txn["transaction_date_timestamp"], + "transaction_date_datetime": txn["transaction_date_datetime"].isoformat(), + "post_date_timestamp": txn["post_date_timestamp"], + "post_date_datetime": txn["post_date_datetime"].isoformat(), + "transaction_description": txn["transaction_description"], + } + for txn in transactions + ] + + with output_file.open("w", encoding="utf-8") as f: + json.dump(serializable, f, indent=2) + + print(f"Saved {len(transactions)} processed transaction(s) to {output_file}") + + +if __name__ == "__main__": + main() + diff --git a/extractor.py b/extractor.py new file mode 100644 index 0000000..6a799c7 --- /dev/null +++ b/extractor.py @@ -0,0 +1,88 @@ +import json +from pathlib import Path +from typing import Any + +START_MARKER = "/*WellFargoProprietary%" +END_MARKER = "%WellFargoProprietary*/" + + +class WellsFargoPayloadExtractor: + """Extract wrapped WellFargo payloads from HAR entries and parse them as JSON.""" + + def __init__(self) -> None: + self.objects: list[Any] = [] + self.errors: list[str] = [] + + def _normalize_text(self, raw_text: Any) -> str | None: + """Return a normalized text payload or None when input is not a string.""" + if not isinstance(raw_text, str): + return None + + text = raw_text.strip() + + # Some HAR exports store content.text as an escaped JSON string literal. + if len(text) >= 2 and text[0] == '"' and text[-1] == '"': + try: + decoded = json.loads(text) + if isinstance(decoded, str): + text = decoded + except json.JSONDecodeError: + self.errors.append("Could not decode an escaped text wrapper") + + return text + + def extract_from_data(self, har_data: dict[str, Any]) -> list[Any]: + """Parse and store objects from log.entries[*].response.content.text.""" + self.objects = [] + self.errors = [] + + entries = har_data.get("log", {}).get("entries", []) + if not isinstance(entries, list): + self.errors.append("Expected log.entries to be a list") + return self.objects + + for index, entry in enumerate(entries): + text_value = ( + entry.get("response", {}) + .get("content", {}) + .get("text") + if isinstance(entry, dict) + else None + ) + text = self._normalize_text(text_value) + if not text or not text.startswith(START_MARKER): + continue + + end_idx = text.find(END_MARKER, len(START_MARKER)) + if end_idx == -1: + self.errors.append(f"Entry {index}: missing end marker") + continue + + payload = text[len(START_MARKER) : end_idx] + if not payload: + self.errors.append(f"Entry {index}: payload between markers is empty") + continue + + try: + self.objects.append(json.loads(payload)) + except json.JSONDecodeError as ex: + self.errors.append(f"Entry {index}: invalid payload JSON ({ex.msg})") + + return self.objects + + def extract_from_file(self, input_path: str | Path) -> list[Any]: + """Load HAR JSON from a file path and extract payload objects.""" + path = Path(input_path) + with path.open("r", encoding="utf-8") as f: + har_data = json.load(f) + return self.extract_from_data(har_data) + + def get_objects(self) -> list[Any]: + """Return the latest parsed objects.""" + return self.objects + + +def extract_wf_objects(har_data: dict[str, Any]) -> list[Any]: + """Backward-compatible functional API.""" + return WellsFargoPayloadExtractor().extract_from_data(har_data) + diff --git a/main.py b/main.py new file mode 100644 index 0000000..41c5e25 --- /dev/null +++ b/main.py @@ -0,0 +1,658 @@ +import argparse +import json +from datetime import date, datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +import ynab + +from wf_credit_extractor import WellsFargoPayloadExtractor as CreditExtractor +from wf_debit_extractor import WellsFargoPayloadExtractor as DebitExtractor +from ynab_client import YNABClient, YNABTransaction + +# Maps the extractor name in ynab.yml to the correct extractor class +EXTRACTOR_MAP = { + "wf_credit_extractor": CreditExtractor, + "wf_debit_extractor": DebitExtractor, +} + +# Maps CLI flag name to the extractor name in ynab.yml +FLAG_TO_EXTRACTOR = { + "credit": "wf_credit_extractor", + "debit": "wf_debit_extractor", +} + +# Maps extractor name to the balance type to display +EXTRACTOR_BALANCE_TYPE = { + "wf_credit_extractor": "OUTSTANDING", + "wf_debit_extractor": "AVAILABLE", +} + + +def _default_output_path(input_path: Path) -> Path: + """Derive a JSON output path from the input file path.""" + return input_path.with_suffix(".json") + + +def _extract_raw_transactions(extracted_objects: list[Any]) -> list[dict]: + """Pull the raw transaction list out of extracted WF payload objects. + + Supports multiple HAR structures: + - /transactions/transactionData/transactions (posted.dat / checking.dat style) + - /accountModel/applicationData/transactionData/transactions (credit fetch style) + Each transaction is tagged with a "status" field: "POSTED" or "PENDING". + TEMP_AUTH transactionType in requestedCriteria marks pending transactions. + """ + transactions: list[dict] = [] + for obj in extracted_objects: + _collect_from_block(obj, transactions) + return transactions + + +def _collect_from_block(obj: dict, out: list[dict]) -> None: + """Recurse into a single extracted object to find all transactionData blocks.""" + # Path 1: /transactions/transactionData (old credit / debit style, and TEMP_AUTH) + txn_wrapper = obj.get("transactions", {}) + if isinstance(txn_wrapper, dict): + td = txn_wrapper.get("transactionData") + if isinstance(td, dict): + _append_transactions(td, out) + + # Path 2: /accountModel/applicationData/transactionData (new credit fetch style) + acct_td = ( + obj.get("accountModel", {}) + .get("applicationData", {}) + .get("transactionData") + ) + if isinstance(acct_td, dict): + _append_transactions(acct_td, out) + + +def _append_transactions(td: dict, out: list[dict]) -> None: + """Extract transactions from a transactionData block and tag POSTED/PENDING.""" + txn_type = ( + td.get("requestedCriteria", {}).get("transactionType", "") + or td.get("type", "") + ) + status = "PENDING" if txn_type == "TEMP_AUTH" else "POSTED" + + txn_list = td.get("transactions", []) + if isinstance(txn_list, list): + for txn in txn_list: + txn_copy = dict(txn) + txn_copy["status"] = status + if "transactionDescription" in txn_copy: + txn_copy["transactionDescription"] = " ".join(txn_copy["transactionDescription"].split()) + out.append(txn_copy) + + +def _wf_date_to_iso(timestamp_ms: int) -> str: + """Convert WF millisecond timestamp to YYYY-MM-DD string.""" + return datetime.fromtimestamp(timestamp_ms / 1000.0, tz=timezone.utc).strftime( + "%Y-%m-%d" + ) + + +def _wf_amount_to_cents(amount: float) -> int: + """Convert WF dollar amount to integer cents.""" + return round(amount * 100) + + +def _ynab_amount_to_cents(amount_milli: int) -> int: + """Convert YNAB milliCurrency to integer cents (divide by 10).""" + return round(amount_milli / 10) + + +def _extract_account_balance( + extracted_objects: list[dict], + balance_type: str, +) -> float | None: + """Extract the dollar balance from extracted WF objects. + + Searches each object's accountModel.applicationData.account.balance[] + for an entry whose 'type' matches *balance_type* and returns its 'amount'. + Returns None if not found. + """ + for obj in extracted_objects: + balances = ( + obj.get("accountModel", {}) + .get("applicationData", {}) + .get("account", {}) + .get("balance", []) + ) + for bal in balances: + if bal.get("type") == balance_type: + return bal.get("amount") + return None + + +def _run_extraction( + label: str, + input_file: str, + extractor: CreditExtractor | DebitExtractor, +) -> tuple[list[dict], list[dict]]: + """Run extractor, save JSON output, and return (raw_txns, extracted_objects).""" + input_path = Path(input_file) + extracted = extractor.extract_from_file(input_path) + output_path = _default_output_path(input_path) + + with output_path.open("w", encoding="utf-8") as f: + json.dump(extracted, f, indent=2) + + print(f"[{label}] Extracted {len(extracted)} object(s) to {output_path}") + + raw_txns = _extract_raw_transactions(extracted) + print(f"[{label}] Found {len(raw_txns)} raw transaction(s)") + + if extractor.errors: + print(f"[{label}] Errors:") + for error in extractor.errors: + print(f" - {error}") + + return raw_txns, extracted + + +def _run_ynab_transactions(days: int) -> None: + """Fetch and display YNAB transactions for the past N days, excluding reconciled.""" + client = YNABClient("ynab.yml") + transactions = client.get_transactions_for_days( + days=days, + exclude_cleared_statuses=["reconciled"], + ) + print( + f"[ynab] Found {len(transactions)} transaction(s) " + f"for the past {days} day(s), excluding reconciled" + ) + print(json.dumps(transactions, indent=2)) + + +def _filter_wf_from_date(wf_transactions: list[dict], start_iso: str) -> list[dict]: + """Keep only WF transactions whose transactionDate is on or after start_iso.""" + return [ + txn for txn in wf_transactions + if _wf_date_to_iso(txn["transactionDate"]) >= start_iso + ] + + +def _find_new_transactions( + wf_transactions: list[dict], + ynab_transactions: list[dict], +) -> list[dict]: + """ + Return WF transactions that do NOT already exist in YNAB. + + Match on (date, abs(amount_cents)): + - WF: transactionDate (ms) -> YYYY-MM-DD, transactionAmount * 100 -> cents + - YNAB: date (YYYY-MM-DD), abs(amount_milliCurrency / 10) -> cents + Uses a multiset so duplicate amounts on the same date are handled correctly. + """ + ynab_keys: dict[tuple[str, int], int] = {} + for txn in ynab_transactions: + key = (txn["date"], abs(_ynab_amount_to_cents(txn["amount_milliCurrency"]))) + ynab_keys[key] = ynab_keys.get(key, 0) + 1 + + new_transactions: list[dict] = [] + for txn in wf_transactions: + date_str = _wf_date_to_iso(txn["transactionDate"]) + cents = abs(_wf_amount_to_cents(txn["transactionAmount"])) + key = (date_str, cents) + + if ynab_keys.get(key, 0) > 0: + ynab_keys[key] -= 1 + else: + new_transactions.append(txn) + + return new_transactions + + +# --------------------------------------------------------------------------- +# YNAB Import helpers +# --------------------------------------------------------------------------- + +def _wf_to_ynab_millis(wf_txn: dict) -> int: + """Convert a WF transaction amount to YNAB milliCurrency. + + Sign logic (unified for credit & debit cards): + - debitCreditType == "CREDIT" β†’ inflow (positive) + - absent or "DEBIT" β†’ outflow (negative) + """ + amount = wf_txn["transactionAmount"] + millis = round(amount * 1000) + if wf_txn.get("debitCreditType") == "CREDIT": + return millis # inflow + return -millis # outflow + + +def _build_import_ids( + amounts: list[int], + dates: list[str], +) -> list[str]: + """Build YNAB import_id strings: ``YNAB:{amount}:{date}:{occurrence}``. + + Occurrence starts at 1 and increments for duplicate (amount, date) pairs + within the batch. + """ + seen: dict[tuple[int, str], int] = {} + import_ids: list[str] = [] + for amount, date_str in zip(amounts, dates): + key = (amount, date_str) + occurrence = seen.get(key, 0) + 1 + seen[key] = occurrence + import_ids.append(f"YNAB:{amount}:{date_str}:{occurrence}") + return import_ids + + +def _convert_wf_to_ynab_transactions( + wf_transactions: list[dict], + account_id: str, +) -> list[dict]: + """Convert raw WF transactions to YNAB-ready dicts. + + Each dict has the keys expected by ``ynab.NewTransaction`` plus a + ``_wf_source`` reference back to the original WF transaction. + """ + amounts: list[int] = [] + dates: list[str] = [] + for txn in wf_transactions: + amounts.append(_wf_to_ynab_millis(txn)) + dates.append(_wf_date_to_iso(txn["transactionDate"])) + + import_ids = _build_import_ids(amounts, dates) + + results: list[dict] = [] + for txn, amount, date_str, import_id in zip( + wf_transactions, amounts, dates, import_ids + ): + status = txn.get("status", "POSTED") + cleared = "cleared" if status == "POSTED" else "uncleared" + payee = " ".join(txn.get("transactionDescription", "").split()) + + results.append({ + "account_id": account_id, + "date": date_str, + "amount": amount, + "payee_name": payee[:200] if payee else None, + "cleared": cleared, + "flag_color": "blue", + "import_id": import_id, + "approved": False, + "_wf_source": txn, + }) + return results + + +class _ImportClassification: + """Holds the three buckets produced by ``_classify_for_import``.""" + + def __init__(self) -> None: + self.new: list[dict] = [] + self.pending_to_posted: list[tuple[dict, YNABTransaction]] = [] + self.duplicates: list[dict] = [] + + +def _classify_for_import( + ynab_ready: list[dict], + ynab_transactions: list[YNABTransaction], +) -> _ImportClassification: + """Classify incoming transactions into new / pendingβ†’posted / duplicate. + + Matching rules (all use (date, abs(amount)) via multiset): + 1. If a posted incoming txn matches an **uncleared** YNAB txn + β†’ pendingβ†’posted (update existing to cleared). + 2. If a cleared YNAB txn already has the same import_id + β†’ duplicate (create with RED flag). + 3. Otherwise β†’ new. + """ + result = _ImportClassification() + + # Build a set of existing cleared import_ids + cleared_import_ids: set[str] = set() + for t in ynab_transactions: + if t["cleared"] in ("cleared", "reconciled") and t.get("import_id"): + cleared_import_ids.add(t["import_id"]) + + # Build a multiset of uncleared YNAB txns keyed by (date, abs_amount) + # Each entry stores the list of YNAB transactions for that key. + uncleared_pool: dict[tuple[str, int], list[YNABTransaction]] = {} + for t in ynab_transactions: + if t["cleared"] == "uncleared": + key = (t["date"], abs(t["amount_milliCurrency"])) + uncleared_pool.setdefault(key, []).append(t) + + for txn in ynab_ready: + import_id = txn["import_id"] + abs_amount = abs(txn["amount"]) + txn_key = (txn["date"], abs_amount) + + # 1. Pendingβ†’Posted: incoming is cleared, matches an uncleared YNAB txn + if txn["cleared"] == "cleared" and uncleared_pool.get(txn_key): + matched_ynab = uncleared_pool[txn_key].pop(0) + if not uncleared_pool[txn_key]: + del uncleared_pool[txn_key] + result.pending_to_posted.append((txn, matched_ynab)) + + # 2. Duplicate: import_id already exists on a cleared YNAB txn + elif import_id in cleared_import_ids: + txn["flag_color"] = "red" + result.duplicates.append(txn) + + # 3. New + else: + result.new.append(txn) + + return result + + +def _format_amount_dollars(millis: int) -> str: + """Format milliCurrency as a dollar string with sign.""" + dollars = millis / 1000.0 + return f"${dollars:+,.2f}" + + +def _preview_import(classification: _ImportClassification) -> None: + """Print a formatted preview of the import classification.""" + col_date = 12 + col_amount = 14 + col_desc = 50 + header = f" {'Date':<{col_date}} {'Amount':>{col_amount}} {'Description':<{col_desc}}" + separator = f" {'-' * col_date} {'-' * col_amount} {'-' * col_desc}" + + # --- New --- + print(f"\n{'=' * 80}") + print(f" NEW TRANSACTIONS ({len(classification.new)}) β€” will be created with BLUE flag") + print(f"{'=' * 80}") + if classification.new: + print(header) + print(separator) + for txn in classification.new: + desc = (txn.get("payee_name") or "")[:col_desc] + print(f" {txn['date']:<{col_date}} {_format_amount_dollars(txn['amount']):>{col_amount}} {desc:<{col_desc}}") + else: + print(" (none)") + + # --- Pendingβ†’Posted --- + print(f"\n{'=' * 80}") + print(f" PENDING β†’ POSTED ({len(classification.pending_to_posted)}) β€” existing uncleared will be marked cleared") + print(f"{'=' * 80}") + if classification.pending_to_posted: + print(header) + print(separator) + for txn, matched in classification.pending_to_posted: + desc = (txn.get("payee_name") or "")[:col_desc] + print(f" {txn['date']:<{col_date}} {_format_amount_dollars(txn['amount']):>{col_amount}} {desc:<{col_desc}}") + else: + print(" (none)") + + # --- Duplicates --- + print(f"\n{'=' * 80}") + print(f" POSSIBLE DUPLICATES ({len(classification.duplicates)}) β€” will be created with RED flag") + print(f"{'=' * 80}") + if classification.duplicates: + print(header) + print(separator) + for txn in classification.duplicates: + desc = (txn.get("payee_name") or "")[:col_desc] + print(f" {txn['date']:<{col_date}} {_format_amount_dollars(txn['amount']):>{col_amount}} {desc:<{col_desc}}") + else: + print(" (none)") + + total = ( + len(classification.new) + + len(classification.pending_to_posted) + + len(classification.duplicates) + ) + print(f"\n Total: {total} transaction(s)") + print() + + +def _execute_import( + classification: _ImportClassification, + ynab_client: YNABClient, +) -> None: + """Send classified transactions to YNAB.""" + + # --- 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"], + 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 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: + updated = 0 + for _txn, matched_ynab in classification.pending_to_posted: + try: + ynab_client.update_transaction( + transaction_id=matched_ynab["id"], + cleared="cleared", + ) + updated += 1 + except Exception as e: + print(f" [ynab] Error updating {matched_ynab['id']}: {e}") + print(f" [ynab] Updated {updated} transaction(s) from uncleared β†’ cleared") + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Extract WF HAR files, compare with YNAB, find new transactions" + ) + parser.add_argument( + "--credit", + default="", + help="Optional credit HAR input file", + ) + parser.add_argument( + "--debit", + default="", + help="Optional debit HAR input file", + ) + parser.add_argument( + "--lookback", + type=int, + default=3, + metavar="DAYS", + help="Days before last reconciled YNAB transaction to start comparison (default: 3)", + ) + parser.add_argument( + "--ynab-transactions", + type=int, + default=None, + metavar="DAYS", + help="List YNAB transactions for the past DAYS (excluding reconciled)", + ) + parser.add_argument( + "--ynab-import", + action="store_true", + default=False, + help="Import new WF transactions into YNAB (requires --credit and/or --debit)", + ) + args = parser.parse_args() + + # Load YNAB config to map extractors -> account IDs + ynab_client = YNABClient("ynab.yml") + account_configs = ynab_client.account_configs + + # Build extractor_name -> [account_id, ...] mapping from ynab.yml + extractor_to_accounts: dict[str, list[str]] = {} + for acct in account_configs: + ext_name = acct.get("extractor", "") + if ext_name: + extractor_to_accounts.setdefault(ext_name, []).append(acct["id"]) + + # -- Extract WF transactions ------------------------------------------- + extracted_by_extractor: dict[str, list[dict]] = {} + objects_by_extractor: dict[str, list[dict]] = {} + + for flag, extractor_name in FLAG_TO_EXTRACTOR.items(): + input_file = getattr(args, flag, "") + if not input_file: + continue + extractor_cls = EXTRACTOR_MAP[extractor_name] + raw_txns, extracted_objects = _run_extraction(flag, input_file, extractor_cls()) + extracted_by_extractor[extractor_name] = raw_txns + objects_by_extractor[extractor_name] = extracted_objects + + # -- Compare with YNAB ------------------------------------------------- + if extracted_by_extractor: + # Gather all relevant account IDs across active extractors + all_account_ids = [ + aid + for ext_name in extracted_by_extractor + for aid in extractor_to_accounts.get(ext_name, []) + ] + + # Find the last reconciled transaction date in YNAB + last_reconciled = ynab_client.get_last_reconciled_date( + account_ids=all_account_ids + ) + + if last_reconciled is None: + print("\n[ynab] No reconciled transactions found; defaulting to 30 day lookback") + start_date = (datetime.now() - timedelta(days=30)).date() + else: + start_date = last_reconciled - timedelta(days=args.lookback) + print( + f"\n[ynab] Last reconciled date: {last_reconciled}" + f" β†’ start comparison from {start_date} " + f"({args.lookback} day lookback)" + ) + + start_iso = start_date.isoformat() + days_from_start = (datetime.now().date() - start_date).days + 1 + + # Fetch ALL YNAB transactions from start_date onward (including reconciled) + # so that reconciled matches also get consumed during comparison. + all_ynab_txns = ynab_client.get_transactions_for_days( + days=days_from_start, + ) + print(f"[ynab] Retrieved {len(all_ynab_txns)} transaction(s) from {start_iso} onward") + + # Compare per-extractor, scoped to the matching YNAB account IDs + for extractor_name, wf_txns in extracted_by_extractor.items(): + account_ids = set(extractor_to_accounts.get(extractor_name, [])) + scoped_ynab = [t for t in all_ynab_txns if t["account_id"] in account_ids] + + # Filter WF transactions to the same start date + filtered_wf = _filter_wf_from_date(wf_txns, start_iso) + + new_txns = _find_new_transactions(filtered_wf, scoped_ynab) + + label = extractor_name.replace("wf_", "").replace("_extractor", "") + print( + f"\n[{label}] {len(new_txns)} new transaction(s) " + f"not yet in YNAB (of {len(filtered_wf)} extracted from {start_iso}, " + f"{len(scoped_ynab)} in YNAB)" + ) + if new_txns: + print(json.dumps(new_txns, indent=2)) + + # -- YNAB Import ------------------------------------------------------- + if args.ynab_import and extracted_by_extractor: + all_ynab_ready: list[dict] = [] + balances: dict[str, float] = {} # label -> balance amount + + for extractor_name, wf_txns in extracted_by_extractor.items(): + account_ids_for_ext = extractor_to_accounts.get(extractor_name, []) + if not account_ids_for_ext: + continue + + # Use the first account ID for this extractor + target_account_id = account_ids_for_ext[0] + scoped_ynab = [ + t for t in all_ynab_txns if t["account_id"] in set(account_ids_for_ext) + ] + + # Filter WF to the comparison window, then find only new ones + filtered_wf = _filter_wf_from_date(wf_txns, start_iso) + new_wf = _find_new_transactions(filtered_wf, scoped_ynab) + + # Convert the new WF transactions to YNAB-ready dicts + ynab_ready = _convert_wf_to_ynab_transactions(new_wf, target_account_id) + + # Classify against existing YNAB transactions + classification = _classify_for_import(ynab_ready, scoped_ynab) + + label = extractor_name.replace("wf_", "").replace("_extractor", "") + print(f"\n--- {label.upper()} IMPORT PREVIEW ---") + + # Collect account balance for bottom-of-report summary + balance_type = EXTRACTOR_BALANCE_TYPE.get(extractor_name) + if balance_type: + balance = _extract_account_balance( + objects_by_extractor.get(extractor_name, []), + balance_type, + ) + if balance is not None: + balances[label] = balance + + _preview_import(classification) + + all_ynab_ready.append({ + "label": label, + "classification": classification, + }) + + # Display account balances at the bottom of the report + if balances: + parts = [] + for lbl, amt in balances.items(): + parts.append(f"{lbl.title()} ${amt:,.2f}") + print(f"{'=' * 80}") + print(" " + " ".join(parts)) + print(f"{'=' * 80}\n") + + # Check if there's anything to import + has_work = any( + entry["classification"].new + or entry["classification"].pending_to_posted + or entry["classification"].duplicates + for entry in all_ynab_ready + ) + + if has_work: + answer = input("Proceed with import? [y/N]: ").strip().lower() + if answer == "y": + for entry in all_ynab_ready: + label = entry["label"] + classification = entry["classification"] + print(f"\n[{label}] Importing...") + _execute_import(classification, ynab_client) + print("\n[ynab] Import complete.") + else: + print("\n[ynab] Import cancelled.") + else: + print("\n[ynab] Nothing to import.") + + elif args.ynab_import and not extracted_by_extractor: + print("[ynab-import] No extracted transactions. Use --credit and/or --debit.") + + # -- Standalone YNAB listing ------------------------------------------- + if args.ynab_transactions is not None: + _run_ynab_transactions(args.ynab_transactions) + + if not args.credit and not args.debit and args.ynab_transactions is None: + print("No input provided. Use --credit, --debit, --ynab-import, and/or --ynab-transactions.") + + +if __name__ == "__main__": + main() + + diff --git a/reconcile.py b/reconcile.py new file mode 100644 index 0000000..96da6d9 --- /dev/null +++ b/reconcile.py @@ -0,0 +1,116 @@ +#!/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() + + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..651abb25885fb178eafc18af55b6c45677c17f33 GIT binary patch literal 452 zcmZvZK?{OF5QX14=%=*kW<`e%-3z*Q5>qQdR8m3zeD!8l%|k51%FdgaZ=Robgo39; zgNTtL!HRc}ir+O3JRS^o3=c~z5cB;-MTR>=@|GEkEfrQ{qq>Sc680=q8xKkYUseA# zA-k9Co=)n3dV`<3!IO%DjM7_pTkW4u6?-K$5a?w#{wxRJ{F UYSdN7*mG9IoMfa{=-nUr0;S outflow (negative) + assert _wf_to_ynab_millis({"transactionAmount": 9.11}) == -9110 + # Credit card refund (debitCreditType=CREDIT) -> inflow (positive) + assert _wf_to_ynab_millis({"transactionAmount": 1.64, "debitCreditType": "CREDIT"}) == 1640 + # Debit card charge (debitCreditType=DEBIT) -> outflow (negative) + assert _wf_to_ynab_millis({"transactionAmount": 194.32, "debitCreditType": "DEBIT"}) == -194320 + # Debit card credit (debitCreditType=CREDIT) -> inflow (positive) + assert _wf_to_ynab_millis({"transactionAmount": 225.0, "debitCreditType": "CREDIT"}) == 225000 + # Pending (no debitCreditType) -> outflow (negative) + assert _wf_to_ynab_millis({"transactionAmount": 2.48}) == -2480 + print("test_wf_to_ynab_millis: PASS") + + +def test_build_import_ids(): + amounts = [-9110, -9110, -5000] + dates = ["2026-03-18", "2026-03-18", "2026-03-17"] + ids = _build_import_ids(amounts, dates) + assert ids == [ + "YNAB:-9110:2026-03-18:1", + "YNAB:-9110:2026-03-18:2", + "YNAB:-5000:2026-03-17:1", + ], f"Got {ids}" + print("test_build_import_ids: PASS") + + +def test_format_amount_dollars(): + assert _format_amount_dollars(-9110) == "$-9.11" + assert _format_amount_dollars(225000) == "$+225.00" + assert _format_amount_dollars(-194320) == "$-194.32" + print("test_format_amount_dollars: PASS") + + +def test_convert_wf_to_ynab_transactions(): + wf_txns = [ + { + "transactionAmount": 9.11, + "transactionDate": 1773817200000, + "transactionDescription": "WALGREENS #10276 MORRISVILLE NC", + "status": "POSTED", + }, + { + "transactionAmount": 2.48, + "transactionDate": 1774076400000, + "transactionDescription": "CULVERS", + "status": "PENDING", + }, + ] + result = _convert_wf_to_ynab_transactions(wf_txns, "acct-123") + + assert result[0]["amount"] == -9110 + assert result[0]["cleared"] == "cleared" + assert result[0]["flag_color"] == "blue" + assert result[0]["account_id"] == "acct-123" + assert result[0]["payee_name"] == "WALGREENS #10276 MORRISVILLE NC" + assert "YNAB:-9110:" in result[0]["import_id"] + + assert result[1]["amount"] == -2480 + assert result[1]["cleared"] == "uncleared" + assert result[1]["flag_color"] == "blue" + print("test_convert_wf_to_ynab_transactions: PASS") + + +def test_classify_for_import(): + wf_txns = [ + { + "transactionAmount": 9.11, + "transactionDate": 1773817200000, + "transactionDescription": "WALGREENS #10276", + "status": "POSTED", + }, + { + "transactionAmount": 2.48, + "transactionDate": 1774076400000, + "transactionDescription": "CULVERS", + "status": "PENDING", + }, + { + "transactionAmount": 50.00, + "transactionDate": 1773817200000, + "transactionDescription": "NEW TXN", + "status": "POSTED", + }, + ] + ynab_ready = _convert_wf_to_ynab_transactions(wf_txns, "acct-123") + + # Derive dates to build matching YNAB transactions + from main import _wf_date_to_iso + date0 = _wf_date_to_iso(1773817200000) + date1 = _wf_date_to_iso(1774076400000) + + ynab_txns = [ + # Uncleared txn matching posted WALGREENS -> pending_to_posted + { + "id": "ynab-1", "date": date0, "payee_name": "WALGREENS", + "category_name": None, "amount_milliCurrency": -9110, "memo": None, + "account_id": "acct-123", "account_name": "Credit", "cleared": "uncleared", + "import_id": "YNAB:-9110:" + date0 + ":1", + }, + # Cleared txn with same import_id as CULVERS -> duplicate + { + "id": "ynab-2", "date": date1, "payee_name": "CULVERS", + "category_name": None, "amount_milliCurrency": -2480, "memo": None, + "account_id": "acct-123", "account_name": "Credit", "cleared": "cleared", + "import_id": "YNAB:-2480:" + date1 + ":1", + }, + ] + + classification = _classify_for_import(ynab_ready, ynab_txns) + + assert len(classification.pending_to_posted) == 1 + assert classification.pending_to_posted[0][1]["id"] == "ynab-1" + + assert len(classification.duplicates) == 1 + assert classification.duplicates[0]["flag_color"] == "red" + + assert len(classification.new) == 1 + assert classification.new[0]["payee_name"] == "NEW TXN" + + print("test_classify_for_import: PASS") + + +def test_preview(): + """Visual test - just make sure it doesn't crash.""" + wf_txns = [ + {"transactionAmount": 9.11, "transactionDate": 1773817200000, + "transactionDescription": "WALGREENS #10276", "status": "POSTED"}, + {"transactionAmount": 50.00, "transactionDate": 1773817200000, + "transactionDescription": "NEW TXN", "status": "POSTED"}, + ] + ynab_ready = _convert_wf_to_ynab_transactions(wf_txns, "acct-123") + classification = _classify_for_import(ynab_ready, []) + _preview_import(classification) + print("test_preview: PASS") + + +if __name__ == "__main__": + test_wf_to_ynab_millis() + test_build_import_ids() + test_format_amount_dollars() + test_convert_wf_to_ynab_transactions() + test_classify_for_import() + test_preview() + print("\nALL TESTS PASSED") + diff --git a/transaction_processor.py b/transaction_processor.py new file mode 100644 index 0000000..5915bc8 --- /dev/null +++ b/transaction_processor.py @@ -0,0 +1,77 @@ +from datetime import datetime +from typing import Any, TypedDict + + +class ProcessedTransaction(TypedDict): + """Transaction data extracted and transformed to desired format.""" + + id: str + transaction_amount_cents: int + transaction_date_timestamp: int + transaction_date_datetime: datetime + post_date_timestamp: int + post_date_datetime: datetime + transaction_description: str + + +class TransactionProcessor: + """Extract and transform transaction data from WellsFargo payload objects.""" + + @staticmethod + def _timestamp_to_datetime(timestamp_ms: int) -> datetime: + """Convert millisecond timestamp to datetime object.""" + return datetime.fromtimestamp(timestamp_ms / 1000.0) + + @staticmethod + def _amount_to_cents(amount: float) -> int: + """Convert dollar amount to cents (integer).""" + return round(amount * 100) + + def extract_transactions( + self, extracted_objects: list[dict[str, Any]] + ) -> list[ProcessedTransaction]: + """ + Extract and transform transactions from extracted WellsFargo payloads. + + Path: /transactions/transactionData/transactions + """ + transactions: list[ProcessedTransaction] = [] + + for obj in extracted_objects: + # Navigate to transactions list + txn_list = ( + obj.get("transactions", {}) + .get("transactionData", {}) + .get("transactions", []) + ) + + if not isinstance(txn_list, list): + continue + + for txn in txn_list: + if not isinstance(txn, dict): + continue + + try: + processed: ProcessedTransaction = { + "id": txn["id"], + "transaction_amount_cents": self._amount_to_cents( + txn["transactionAmount"] + ), + "transaction_date_timestamp": txn["transactionDate"], + "transaction_date_datetime": self._timestamp_to_datetime( + txn["transactionDate"] + ), + "post_date_timestamp": txn["postDate"], + "post_date_datetime": self._timestamp_to_datetime( + txn["postDate"] + ), + "transaction_description": " ".join(txn["transactionDescription"].split()), + } + transactions.append(processed) + except (KeyError, TypeError, ValueError): + # Skip malformed transactions + continue + + return transactions + diff --git a/wf_credit_extractor.py b/wf_credit_extractor.py new file mode 100644 index 0000000..fbff08e --- /dev/null +++ b/wf_credit_extractor.py @@ -0,0 +1,89 @@ +import json +from pathlib import Path +from typing import Any + +START_MARKER = "/*WellFargoProprietary%" +END_MARKER = "%WellFargoProprietary*/" + + +class WellsFargoPayloadExtractor: + """Extract wrapped WellFargo payloads from HAR entries and parse them as JSON.""" + + def __init__(self) -> None: + self.objects: list[Any] = [] + self.errors: list[str] = [] + + def _normalize_text(self, raw_text: Any) -> str | None: + """Return a normalized text payload or None when input is not a string.""" + if not isinstance(raw_text, str): + return None + + text = raw_text.strip() + + # Some HAR exports store content.text as an escaped JSON string literal. + if len(text) >= 2 and text[0] == '"' and text[-1] == '"': + try: + decoded = json.loads(text) + if isinstance(decoded, str): + text = decoded + except json.JSONDecodeError: + self.errors.append("Could not decode an escaped text wrapper") + + return text + + def extract_from_data(self, har_data: dict[str, Any]) -> list[Any]: + """Parse and store objects from log.entries[*].response.content.text.""" + self.objects = [] + self.errors = [] + + entries = har_data.get("log", {}).get("entries", []) + if not isinstance(entries, list): + self.errors.append("Expected log.entries to be a list") + return self.objects + + for index, entry in enumerate(entries): + text_value = ( + entry.get("response", {}) + .get("content", {}) + .get("text") + if isinstance(entry, dict) + else None + ) + text = self._normalize_text(text_value) + if not text or not text.startswith(START_MARKER): + continue + + end_idx = text.find(END_MARKER, len(START_MARKER)) + if end_idx == -1: + self.errors.append(f"Entry {index}: missing end marker") + continue + + payload = text[len(START_MARKER) : end_idx] + if not payload: + self.errors.append(f"Entry {index}: payload between markers is empty") + continue + + try: + self.objects.append(json.loads(payload)) + except json.JSONDecodeError as ex: + self.errors.append(f"Entry {index}: invalid payload JSON ({ex.msg})") + + return self.objects + + def extract_from_file(self, input_path: str | Path) -> list[Any]: + """Load HAR JSON from a file path and extract payload objects.""" + path = Path(input_path) + with path.open("r", encoding="utf-8") as f: + har_data = json.load(f) + return self.extract_from_data(har_data) + + def get_objects(self) -> list[Any]: + """Return the latest parsed objects.""" + return self.objects + + +def extract_wf_objects(har_data: dict[str, Any]) -> list[Any]: + """Backward-compatible functional API.""" + return WellsFargoPayloadExtractor().extract_from_data(har_data) + + diff --git a/wf_debit_extractor.py b/wf_debit_extractor.py new file mode 100644 index 0000000..e66b83a --- /dev/null +++ b/wf_debit_extractor.py @@ -0,0 +1,106 @@ +import json +from pathlib import Path +from typing import Any + +START_MARKER = "/*WellFargoProprietary%" +END_MARKER = "%WellFargoProprietary*/" + + +class WellsFargoPayloadExtractor: + """Extract wrapped WellFargo payloads from HAR entries and parse them as JSON.""" + + def __init__(self) -> None: + self.objects: list[Any] = [] + self.errors: list[str] = [] + + def _normalize_text(self, raw_text: Any) -> str | None: + """Return a normalized text payload or None when input is not a string.""" + if not isinstance(raw_text, str): + return None + + text = raw_text.strip() + + # Some HAR exports store content.text as an escaped JSON string literal. + if len(text) >= 2 and text[0] == '"' and text[-1] == '"': + try: + decoded = json.loads(text) + if isinstance(decoded, str): + text = decoded + except json.JSONDecodeError: + self.errors.append("Could not decode an escaped text wrapper") + + return text + + def _parse_wrapped_text(self, text: str, context: str) -> Any | None: + """Parse a marker-wrapped WellFargo payload text into JSON.""" + if not text.startswith(START_MARKER): + return None + + end_idx = text.find(END_MARKER, len(START_MARKER)) + if end_idx == -1: + self.errors.append(f"{context}: missing end marker") + return None + + payload = text[len(START_MARKER) : end_idx] + if not payload: + self.errors.append(f"{context}: payload between markers is empty") + return None + + try: + return json.loads(payload) + except json.JSONDecodeError as ex: + self.errors.append(f"{context}: invalid payload JSON ({ex.msg})") + return None + + def extract_from_data( + self, har_data: dict[str, Any], entry0_only: bool = False + ) -> list[Any]: + """Parse payload objects from log.entries response/content/text fields.""" + self.objects = [] + self.errors = [] + + entries = har_data.get("log", {}).get("entries", []) + if not isinstance(entries, list): + self.errors.append("Expected log.entries to be a list") + return self.objects + + indexed_entries = list(enumerate(entries[:1] if entry0_only else entries)) + + for index, entry in indexed_entries: + text_value = ( + entry.get("response", {}) + .get("content", {}) + .get("text") + if isinstance(entry, dict) + else None + ) + text = self._normalize_text(text_value) + if not text: + continue + + parsed = self._parse_wrapped_text(text, f"Entry {index}") + if parsed is not None: + self.objects.append(parsed) + + return self.objects + + def extract_from_file( + self, input_path: str | Path, entry0_only: bool = False + ) -> list[Any]: + """Load HAR JSON from a file path and extract payload objects.""" + path = Path(input_path) + with path.open("r", encoding="utf-8") as f: + har_data = json.load(f) + return self.extract_from_data(har_data, entry0_only=entry0_only) + + def get_objects(self) -> list[Any]: + """Return the latest parsed objects.""" + return self.objects + + +def extract_wf_objects(har_data: dict[str, Any]) -> list[Any]: + """Backward-compatible functional API.""" + return WellsFargoPayloadExtractor().extract_from_data(har_data) + + + diff --git a/wf_extractor.py b/wf_extractor.py new file mode 100644 index 0000000..87bb0d1 --- /dev/null +++ b/wf_extractor.py @@ -0,0 +1,105 @@ +import json +from pathlib import Path +from typing import Any + +START_MARKER = "/*WellFargoProprietary%" +END_MARKER = "%WellFargoProprietary*/" + + +class WellsFargoPayloadExtractor: + """Extract wrapped WellFargo payloads from HAR entries and parse them as JSON.""" + + def __init__(self) -> None: + self.objects: list[Any] = [] + self.errors: list[str] = [] + + def _normalize_text(self, raw_text: Any) -> str | None: + """Return a normalized text payload or None when input is not a string.""" + if not isinstance(raw_text, str): + return None + + text = raw_text.strip() + + # Some HAR exports store content.text as an escaped JSON string literal. + if len(text) >= 2 and text[0] == '"' and text[-1] == '"': + try: + decoded = json.loads(text) + if isinstance(decoded, str): + text = decoded + except json.JSONDecodeError: + self.errors.append("Could not decode an escaped text wrapper") + + return text + + def _parse_wrapped_text(self, text: str, context: str) -> Any | None: + """Parse a marker-wrapped WellFargo payload text into JSON.""" + if not text.startswith(START_MARKER): + return None + + end_idx = text.find(END_MARKER, len(START_MARKER)) + if end_idx == -1: + self.errors.append(f"{context}: missing end marker") + return None + + payload = text[len(START_MARKER) : end_idx] + if not payload: + self.errors.append(f"{context}: payload between markers is empty") + return None + + try: + return json.loads(payload) + except json.JSONDecodeError as ex: + self.errors.append(f"{context}: invalid payload JSON ({ex.msg})") + return None + + def extract_from_data( + self, har_data: dict[str, Any], entry0_only: bool = False + ) -> list[Any]: + """Parse payload objects from log.entries response/content/text fields.""" + self.objects = [] + self.errors = [] + + entries = har_data.get("log", {}).get("entries", []) + if not isinstance(entries, list): + self.errors.append("Expected log.entries to be a list") + return self.objects + + indexed_entries = list(enumerate(entries[:1] if entry0_only else entries)) + + for index, entry in indexed_entries: + text_value = ( + entry.get("response", {}) + .get("content", {}) + .get("text") + if isinstance(entry, dict) + else None + ) + text = self._normalize_text(text_value) + if not text: + continue + + parsed = self._parse_wrapped_text(text, f"Entry {index}") + if parsed is not None: + self.objects.append(parsed) + + return self.objects + + def extract_from_file( + self, input_path: str | Path, entry0_only: bool = False + ) -> list[Any]: + """Load HAR JSON from a file path and extract payload objects.""" + path = Path(input_path) + with path.open("r", encoding="utf-8") as f: + har_data = json.load(f) + return self.extract_from_data(har_data, entry0_only=entry0_only) + + def get_objects(self) -> list[Any]: + """Return the latest parsed objects.""" + return self.objects + + +def extract_wf_objects(har_data: dict[str, Any]) -> list[Any]: + """Backward-compatible functional API.""" + return WellsFargoPayloadExtractor().extract_from_data(har_data) + + diff --git a/ynab_client.py b/ynab_client.py new file mode 100644 index 0000000..86614d1 --- /dev/null +++ b/ynab_client.py @@ -0,0 +1,266 @@ +import yaml +from datetime import datetime, timedelta +from pathlib import Path +from typing import TypedDict + +import ynab + + +class YNABTransaction(TypedDict): + """Transaction data from YNAB.""" + + id: str + date: str + payee_name: str | None + category_name: str | None + amount_milliCurrency: int + memo: str | None + account_id: str + account_name: str + cleared: str + import_id: str | None + + +class YNABConfig(TypedDict): + """Configuration loaded from ynab.yml.""" + + token: str + plan_id: str + accounts: list[dict[str, str]] + + +class YNABClient: + """Client for fetching transactions from YNAB.""" + + def __init__(self, config_path: str | Path = "ynab.yml") -> None: + """Initialize YNAB client from config file.""" + self.config = self._load_config(config_path) + self.token = self.config["token"] + self.plan_id = self.config["plan_id"] + self.account_configs = self.config["accounts"] + + # Use SDK-native access_token auth for bearer token handling. + config = ynab.Configuration(access_token=self.token) + self.api_client = ynab.ApiClient(config) + self.transactions_api = ynab.TransactionsApi(self.api_client) + self.accounts_api = ynab.AccountsApi(self.api_client) + self.categories_api = ynab.CategoriesApi(self.api_client) + + @staticmethod + def _load_config(config_path: str | Path) -> YNABConfig: + """Load YNAB configuration from YAML file.""" + path = Path(config_path) + with path.open("r", encoding="utf-8") as f: + data = yaml.safe_load(f) + + ynab_config = data.get("ynab", [{}])[0] + return { + "token": ynab_config.get("token", ""), + "plan_id": ynab_config.get("plan_id", ""), + "accounts": ynab_config.get("accounts", []), + } + + @staticmethod + def _normalize_cleared_status(cleared: object) -> str: + """Normalize YNAB cleared status values to lowercase strings.""" + if hasattr(cleared, "value"): + return str(getattr(cleared, "value")).strip().lower() + + status = str(cleared).strip() + if "." in status: + status = status.split(".")[-1] + return status.lower() + + def get_transactions_for_days( + self, + days: int = 30, + exclude_cleared_statuses: list[str] | None = None, + ) -> list[YNABTransaction]: + """ + Fetch transactions from configured accounts for the past x days. + + Args: + days: Number of days to look back (default: 30) + exclude_cleared_statuses: Optional list of normalized statuses to skip, + for example ["reconciled"] + + Returns: + List of transactions with transformed data + """ + transactions: list[YNABTransaction] = [] + cutoff_date = (datetime.now() - timedelta(days=days)).date() + excluded_statuses = { + status.strip().lower() for status in (exclude_cleared_statuses or []) + } + + # Fetch all accounts to build name lookup + accounts_response = self.accounts_api.get_accounts(self.plan_id) + account_map = { + str(acc.id): acc.name for acc in accounts_response.data.accounts + } + + # Fetch categories for name lookup + categories_response = self.categories_api.get_categories(self.plan_id) + category_map = {} + for category_group in categories_response.data.category_groups: + for cat in category_group.categories: + if cat.id: + category_map[str(cat.id)] = cat.name + + # If specific accounts are configured, use only those; otherwise use all + if self.account_configs: + account_ids = [acc["id"] for acc in self.account_configs] + else: + account_ids = list(account_map.keys()) + + # Fetch transactions for each account + for account_id in account_ids: + account_name = account_map.get(account_id, account_id) + + try: + txn_response = self.transactions_api.get_transactions_by_account( + plan_id=self.plan_id, + account_id=account_id, + since_date=cutoff_date, + ) + + for txn in txn_response.data.transactions: + txn_date = txn.var_date + if txn_date < cutoff_date: + continue + + cleared_status = self._normalize_cleared_status(txn.cleared) + if cleared_status in excluded_statuses: + continue + + category_name = None + if txn.category_id: + category_name = category_map.get(str(txn.category_id)) + + processed: YNABTransaction = { + "id": txn.id, + "date": str(txn.var_date), + "payee_name": txn.payee_name, + "category_name": category_name, + "amount_milliCurrency": txn.amount, + "memo": txn.memo, + "account_id": account_id, + "account_name": account_name, + "cleared": cleared_status, + "import_id": txn.import_id, + } + transactions.append(processed) + + except Exception as e: + print(f"Error fetching transactions for account {account_name}: {e}") + continue + + return transactions + + def get_account_balance(self, account_id: str) -> int: + """ + Get current balance for an account (in milliCurrency units). + + Args: + account_id: YNAB account ID + + Returns: + Account balance in milliCurrency + """ + accounts_response = self.accounts_api.get_accounts(self.plan_id) + for account in accounts_response.data.accounts: + if str(account.id) == account_id: + return account.balance + raise ValueError(f"Account {account_id} not found") + + def get_last_reconciled_date( + self, + account_ids: list[str] | None = None, + search_days: int = 90, + ) -> datetime | None: + """ + Find the date of the most recent reconciled transaction across accounts. + + Args: + account_ids: Account IDs to search. If None, uses all configured accounts. + search_days: How far back to search for reconciled transactions (default 90). + + Returns: + The date of the last reconciled transaction, or None if none found. + """ + from datetime import date as date_type + + cutoff_date = (datetime.now() - timedelta(days=search_days)).date() + + if account_ids is None: + account_ids = [acc["id"] for acc in self.account_configs] + + latest_date: date_type | None = None + + for account_id in account_ids: + try: + txn_response = self.transactions_api.get_transactions_by_account( + plan_id=self.plan_id, + account_id=account_id, + since_date=cutoff_date, + ) + for txn in txn_response.data.transactions: + status = self._normalize_cleared_status(txn.cleared) + if status == "reconciled": + txn_date = txn.var_date + if latest_date is None or txn_date > latest_date: + latest_date = txn_date + except Exception as e: + print(f"Error searching reconciled transactions for {account_id}: {e}") + continue + + return latest_date + + def create_transactions( + self, + transactions: list[ynab.NewTransaction], + ) -> ynab.SaveTransactionsResponseData: + """ + Create multiple transactions in YNAB. + + Args: + transactions: List of NewTransaction objects to create. + + Returns: + SaveTransactionsResponseData with created IDs and duplicate info. + """ + wrapper = ynab.PostTransactionsWrapper(transactions=transactions) + response = self.transactions_api.create_transaction( + plan_id=self.plan_id, + data=wrapper, + ) + return response.data + + def update_transaction( + self, + transaction_id: str, + cleared: str | None = None, + flag_color: str | None = None, + ) -> None: + """ + Update an existing YNAB transaction. + + Args: + transaction_id: The YNAB transaction ID to update. + cleared: New cleared status ('cleared', 'uncleared', 'reconciled'). + flag_color: New flag color (e.g. 'red', 'blue'). + """ + fields: dict = {} + if cleared is not None: + fields["cleared"] = ynab.TransactionClearedStatus(cleared) + if flag_color is not None: + fields["flag_color"] = ynab.TransactionFlagColor(flag_color) + + existing = ynab.ExistingTransaction(**fields) + wrapper = ynab.PutTransactionWrapper(transaction=existing) + self.transactions_api.update_transaction( + plan_id=self.plan_id, + transaction_id=transaction_id, + data=wrapper, + ) + diff --git a/ynab_example.py b/ynab_example.py new file mode 100644 index 0000000..2370fc3 --- /dev/null +++ b/ynab_example.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python +"""Example usage of YNAB client for fetching transactions.""" + +import json +from pathlib import Path + +from ynab_client import YNABClient + + +def main(): + # Initialize YNAB client from config + client = YNABClient("ynab.yml") + + # Fetch transactions from the past 30 days + print("Fetching transactions from the past 30 days...") + exclude_cleared_statuses = ["reconciled"] + transactions = client.get_transactions_for_days( + days=30, + exclude_cleared_statuses=exclude_cleared_statuses, + ) + + print( + f"Found {len(transactions)} transaction(s) " + f"(excluding: {', '.join(exclude_cleared_statuses) or 'none'})\n" + ) + + # Display first few transactions + for i, txn in enumerate(transactions[:10]): + print(f"Transaction {i + 1}:") + print(f" Date: {txn['date']}") + print(f" Payee: {txn['payee_name']}") + print(f" Category: {txn['category_name']}") + print(f" Amount: ${txn['amount_milliCurrency'] / 1000:.2f}") + print(f" Account: {txn['account_name']}") + print(f" Memo: {txn['memo']}") + print(f" Cleared: {txn['cleared']}") + print() + + # Optionally save all transactions to JSON + output_file = Path("ynab_transactions.json") + serializable = [ + { + "id": txn["id"], + "date": txn["date"], + "payee_name": txn["payee_name"], + "category_name": txn["category_name"], + "amount": txn["amount_milliCurrency"] / 1000, # Convert to regular currency + "amount_milliCurrency": txn["amount_milliCurrency"], + "memo": txn["memo"], + "account_id": txn["account_id"], + "account_name": txn["account_name"], + "cleared": txn["cleared"], + } + for txn in transactions + ] + + with output_file.open("w", encoding="utf-8") as f: + json.dump(serializable, f, indent=2) + + print(f"Saved {len(transactions)} YNAB transaction(s) to {output_file}") + + +if __name__ == "__main__": + main() + diff --git a/ynab_import.md b/ynab_import.md new file mode 100644 index 0000000..a2b8c63 --- /dev/null +++ b/ynab_import.md @@ -0,0 +1,15 @@ +## Overview +I need to import extracted transactions into YNAB. This module (`ynab_client.py`) will authenticate with YNAB using the official SDK and retrieve transactions from configured accounts. +YNAB API documentation: https://api.youneedabudget.com/v1 +https://github.com/ynab/ynab-sdk-python/blob/main/docs/TransactionsApi.md#create_transaction +https://github.com/ynab/ynab-sdk-python/blob/main/docs/NewTransaction.md + +- I want newly imported transactions to have a flag_color of BLUE +- Transactions that are 'posted' should have a cleared status of CLEARED +- Transactions that are 'pending' should have a cleared status of UNCLEARED +- The YNAB transaction import_id field should be set what is done in https://github.com/ynab/ynab-sdk-python/blob/main/docs/NewTransaction.md +- If a posted transaction comes in, and YNAB has an uncleared transaction with the same amount and date, then we should update the existing transaction to be cleared instead of creating a new one. This is to handle the case where a pending transaction becomes posted. We can use the import_id field to match transactions. Note: when a transaction moves from pending to posted, the description may change. +- If a transaction is already in YNAB with the same import_id and is cleared, we should still import the new transaction, but give it a flag_status of RED. +- Before importing, we should show the user in a presentable way the new transactions, the pending to posted matched transactions, and the possible duplicates (RED). Ask the user if they would like to import. + +This process should happen when the user runs with a parameter like '--ynab-import' \ No newline at end of file