From fbfb7ab33c52831b8b8d063ff01a4f70d82bd5da Mon Sep 17 00:00:00 2001 From: NguyenND Date: Fri, 5 Jun 2026 09:52:11 -0400 Subject: [PATCH] 06/05 Optimize app: implement Plaid --- .env.example | 4 + app/__init__.py | 3 + app/config.py | 5 + app/models/plaid_item.py | 77 ++++++ app/routes/accounts.py | 15 +- app/routes/plaid.py | 383 ++++++++++++++++++++++++++ app/services/plaid_service.py | 382 +++++++++++++++++++++++++ app/services/teller_service.py | 38 ++- app/templates/accounts/index.html | 112 ++++++++ app/templates/base.html | 2 +- app/templates/plaid/index.html | 250 +++++++++++++++++ app/templates/plaid/map_accounts.html | 52 ++++ app/templates/plaid/preview.html | 201 ++++++++++++++ app/templates/settings/index.html | 9 + scripts/add_plaid_tables.py | 94 +++++++ 15 files changed, 1612 insertions(+), 15 deletions(-) create mode 100644 app/models/plaid_item.py create mode 100644 app/routes/plaid.py create mode 100644 app/services/plaid_service.py create mode 100644 app/templates/plaid/index.html create mode 100644 app/templates/plaid/map_accounts.html create mode 100644 app/templates/plaid/preview.html create mode 100644 scripts/add_plaid_tables.py diff --git a/.env.example b/.env.example index cc377a5..f61a894 100644 --- a/.env.example +++ b/.env.example @@ -9,3 +9,7 @@ APP_CURRENCY=USD APP_CURRENCY_SYMBOL=$ APP_TIMEZONE=Asia/Ho_Chi_Minh FLASK_APP=wsgi:app +# Plaid (sandbox / development / production) +PLAID_CLIENT_ID=your-plaid-client-id +PLAID_SECRET=your-plaid-secret +PLAID_ENV=sandbox diff --git a/app/__init__.py b/app/__init__.py index 83a4a2b..7dfbb8e 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -104,6 +104,7 @@ def create_app(config_name=None): from app.routes.settings import settings_bp from app.routes.teller import teller_bp from app.routes.schwab import schwab_bp + from app.routes.plaid import plaid_bp from app.routes.logs import logs_bp from app.routes.bank_import import bank_import_bp @@ -120,6 +121,7 @@ def create_app(config_name=None): app.register_blueprint(settings_bp) app.register_blueprint(teller_bp) app.register_blueprint(schwab_bp) + app.register_blueprint(plaid_bp) app.register_blueprint(logs_bp) app.register_blueprint(bank_import_bp) @@ -132,6 +134,7 @@ def create_app(config_name=None): ) from app.models.teller_enrollment import TellerEnrollment, TellerAccount from app.models.schwab_connection import SchwabConnection, SchwabAccount + from app.models.plaid_item import PlaidItem, PlaidAccount, PlaidSyncPreview # ── Session idle timeout ────────────────────────────────────────────────── from flask import session as _session, request as _request diff --git a/app/config.py b/app/config.py index caa3dbc..ced6a1f 100644 --- a/app/config.py +++ b/app/config.py @@ -30,6 +30,11 @@ class Config: TELLER_KEY_PATH = os.environ.get('TELLER_KEY_PATH', '/home/pfm/teller/private_key.pem') TELLER_WEBHOOK_SECRET = os.environ.get('TELLER_WEBHOOK_SECRET', '') + # Plaid (sandbox / development / production) + PLAID_CLIENT_ID = os.environ.get('PLAID_CLIENT_ID', '') + PLAID_SECRET = os.environ.get('PLAID_SECRET', '') + PLAID_ENV = os.environ.get('PLAID_ENV', 'sandbox') + # Schwab Developer API (OAuth 2.0) SCHWAB_CLIENT_ID = os.environ.get('SCHWAB_CLIENT_ID', '') SCHWAB_CLIENT_SECRET = os.environ.get('SCHWAB_CLIENT_SECRET', '') diff --git a/app/models/plaid_item.py b/app/models/plaid_item.py new file mode 100644 index 0000000..bc3f1dd --- /dev/null +++ b/app/models/plaid_item.py @@ -0,0 +1,77 @@ +from datetime import datetime +from app.extensions import db +from app.utils.crypto import EncryptedText + + +class PlaidItem(db.Model): + """One Plaid Item = one bank connection (may contain multiple accounts).""" + __tablename__ = 'plaid_items' + + id = db.Column(db.Integer, primary_key=True) + item_id = db.Column(db.String(100), unique=True, nullable=False, index=True) + access_token = db.Column(EncryptedText, nullable=False) + institution_id = db.Column(db.String(50), nullable=True) + institution_name = db.Column(db.String(100), nullable=True) + cursor = db.Column(db.String(500), nullable=True) # /transactions/sync cursor + is_active = db.Column(db.Boolean, default=True) + created_at = db.Column(db.DateTime, default=datetime.utcnow) + last_synced_at = db.Column(db.DateTime, nullable=True) + + accounts = db.relationship('PlaidAccount', back_populates='item', + cascade='all, delete-orphan', lazy='dynamic') + + def __repr__(self): + return f'' + + +class PlaidAccount(db.Model): + """Maps one Plaid account inside an item to a PFM account.""" + __tablename__ = 'plaid_accounts' + + id = db.Column(db.Integer, primary_key=True) + item_id = db.Column(db.Integer, db.ForeignKey('plaid_items.id'), nullable=False) + plaid_account_id = db.Column(db.String(100), unique=True, nullable=False, index=True) + pfm_account_id = db.Column(db.Integer, db.ForeignKey('accounts.id'), nullable=True) + + account_name = db.Column(db.String(100), nullable=True) + account_type = db.Column(db.String(50), nullable=True) # depository / credit / investment + account_subtype = db.Column(db.String(50), nullable=True) # checking / savings / credit card + mask = db.Column(db.String(10), nullable=True) # last 4 digits + + last_sync_date = db.Column(db.Date, nullable=True) + + # Credit card billing info (populated from /liabilities/get) + cc_due_date = db.Column(db.Date, nullable=True) + cc_minimum_payment = db.Column(db.Numeric(12, 2), nullable=True) + cc_last_statement_balance = db.Column(db.Numeric(12, 2), nullable=True) + cc_is_overdue = db.Column(db.Boolean, default=False) + cc_updated_at = db.Column(db.DateTime, nullable=True) + + is_active = db.Column(db.Boolean, default=True) + created_at = db.Column(db.DateTime, default=datetime.utcnow) + + item = db.relationship('PlaidItem', back_populates='accounts') + pfm_account = db.relationship('Account') + + @property + def display_name(self): + suffix = f' ••••{self.mask}' if self.mask else '' + return f'{self.account_name}{suffix}' + + def __repr__(self): + return f'' + + +class PlaidSyncPreview(db.Model): + """Temporary server-side storage for Plaid sync preview data (per item).""" + __tablename__ = 'plaid_sync_previews' + + id = db.Column(db.Integer, primary_key=True) + item_id = db.Column(db.Integer, db.ForeignKey('plaid_items.id'), + nullable=False, unique=True, index=True) + data_json = db.Column(db.Text, nullable=False) # JSON array of parsed transactions + next_cursor = db.Column(db.String(500), nullable=True) # advance item cursor on confirm + created_at = db.Column(db.DateTime, default=datetime.utcnow) + + def __repr__(self): + return f'' diff --git a/app/routes/accounts.py b/app/routes/accounts.py index 021cc7f..5486e76 100644 --- a/app/routes/accounts.py +++ b/app/routes/accounts.py @@ -61,11 +61,12 @@ def index(): from app.models.teller_enrollment import TellerAccount from app.models.schwab_connection import SchwabAccount + from app.models.plaid_item import PlaidAccount all_accounts = Account.query.filter_by(is_active=True).order_by(Account.name).all() pfm_ids = [a.id for a in all_accounts] # Build sync-provider maps before the calc_balance loop. - # Accounts linked to Teller or Schwab get their balance from the provider, + # Accounts linked to Teller, Schwab, or Plaid get their balance from the provider, # not from transaction summation, so we skip calc_balance for them. teller_accounts = TellerAccount.query.filter( TellerAccount.pfm_account_id.in_(pfm_ids), @@ -79,7 +80,13 @@ def index(): ).all() schwab_map = {sa.pfm_account_id: sa for sa in schwab_accounts} - provider_ids = set(teller_map) | set(schwab_map) + plaid_accounts = PlaidAccount.query.filter( + PlaidAccount.pfm_account_id.in_(pfm_ids), + PlaidAccount.is_active == True, + ).all() + plaid_map = {pa.pfm_account_id: pa for pa in plaid_accounts} + + provider_ids = set(teller_map) | set(schwab_map) | set(plaid_map) for a in all_accounts: if a.id not in provider_ids: calc_balance(a.id) @@ -111,7 +118,9 @@ def index(): credit_count=len(credit_accounts), monthly_charges=monthly_charges, teller_map=teller_map, - schwab_map=schwab_map) + schwab_map=schwab_map, + plaid_map=plaid_map, + today=date.today()) @accounts_bp.route('/new', methods=['GET', 'POST']) diff --git a/app/routes/plaid.py b/app/routes/plaid.py new file mode 100644 index 0000000..71139a9 --- /dev/null +++ b/app/routes/plaid.py @@ -0,0 +1,383 @@ +import json +import logging +from datetime import datetime + +from flask import (Blueprint, render_template, redirect, url_for, flash, + request, jsonify, session, current_app) +from flask_login import login_required + +from app.extensions import db +from app.models.account import Account +from app.models.plaid_item import PlaidItem, PlaidAccount, PlaidSyncPreview +from app.services.plaid_service import ( + create_link_token, exchange_public_token, + get_accounts, get_balances, + refresh_liabilities, sync_preview, import_transactions, +) + +plaid_bp = Blueprint('plaid', __name__, url_prefix='/plaid') +log = logging.getLogger(__name__) + + +def _configured(): + return bool(current_app.config.get('PLAID_CLIENT_ID') and + current_app.config.get('PLAID_SECRET')) + + +# ── Index ───────────────────────────────────────────────────────────────────── + +@plaid_bp.route('/') +@login_required +def index(): + from datetime import date + items = PlaidItem.query.filter_by(is_active=True).all() + return render_template('plaid/index.html', + items=items, + plaid_configured=_configured(), + today=date.today()) + + +# ── Link Token (AJAX) ───────────────────────────────────────────────────────── + +@plaid_bp.route('/create-link-token', methods=['POST']) +@login_required +def create_link_token_view(): + """AJAX: create a Plaid Link token for the frontend widget.""" + if not _configured(): + return jsonify({'error': 'Plaid is not configured — add PLAID_CLIENT_ID and PLAID_SECRET to .env'}), 400 + try: + token = create_link_token() + return jsonify({'link_token': token}) + except Exception as e: + log.error('[plaid] create_link_token failed: %s', e, exc_info=True) + return jsonify({'error': str(e)}), 500 + + +# ── Token Exchange (AJAX, called by frontend after Link success) ────────────── + +@plaid_bp.route('/exchange-token', methods=['POST']) +@login_required +def exchange_token(): + """ + Called by the frontend after Plaid Link succeeds. + Body: { public_token, institution_id, institution_name, accounts: [...] } + """ + data = request.get_json() + if not data or not data.get('public_token'): + return jsonify({'error': 'Missing public_token'}), 400 + + public_token = data['public_token'] + institution_id = data.get('institution_id', '') + institution_name = data.get('institution_name', 'Unknown Bank') + + try: + token_data = exchange_public_token(public_token) + except Exception as e: + log.error('[plaid] exchange_public_token failed: %s', e, exc_info=True) + return jsonify({'error': f'Token exchange failed: {e}'}), 500 + + access_token = token_data['access_token'] + item_id = token_data['item_id'] + + # Upsert PlaidItem (reconnect preserves existing accounts) + item = PlaidItem.query.filter_by(item_id=item_id).first() + if item: + item.access_token = access_token + item.is_active = True + item.institution_name = institution_name + item.institution_id = institution_id + else: + item = PlaidItem( + item_id = item_id, + access_token = access_token, + institution_id = institution_id, + institution_name = institution_name, + ) + db.session.add(item) + db.session.flush() + + # Fetch accounts from Plaid and upsert PlaidAccount rows + try: + plaid_accounts = get_accounts(item) + except Exception as e: + log.error('[plaid] get_accounts failed: %s', e, exc_info=True) + db.session.rollback() + return jsonify({'error': f'Could not fetch accounts: {e}'}), 500 + + for pa_data in plaid_accounts: + pa = PlaidAccount.query.filter_by(plaid_account_id=pa_data['account_id']).first() + if pa: + pa.item = item + pa.is_active = True + else: + db.session.add(PlaidAccount( + item = item, + plaid_account_id = pa_data['account_id'], + account_name = pa_data.get('name', ''), + account_type = pa_data.get('type', ''), + account_subtype = pa_data.get('subtype', ''), + mask = pa_data.get('mask', ''), + )) + + db.session.commit() + log.info('[plaid] connected item %s (%s) with %d account(s)', + item_id, institution_name, len(plaid_accounts)) + + return jsonify({'redirect': url_for('plaid.map_accounts', item_db_id=item.id)}) + + +# ── Account Mapping ─────────────────────────────────────────────────────────── + +@plaid_bp.route('/map/', methods=['GET', 'POST']) +@login_required +def map_accounts(item_db_id): + item = db.get_or_404(PlaidItem, item_db_id) + plaid_accounts = item.accounts.filter_by(is_active=True).all() + pfm_accounts = Account.query.filter_by(is_active=True).order_by(Account.name).all() + + TYPE_MAP = { + 'depository': 'checking', + 'credit': 'credit_card', + 'investment': 'investment', + 'loan': 'other', + 'other': 'other', + } + SUBTYPE_MAP = { + 'checking': 'checking', + 'savings': 'savings', + 'credit card': 'credit_card', + 'money market':'savings', + 'cd': 'savings', + 'brokerage': 'investment', + 'ira': 'investment', + '401k': 'investment', + } + + if request.method == 'POST': + for pa in plaid_accounts: + val = request.form.get(f'pfm_account_{pa.id}', '') + if val == 'new': + pfm_type = SUBTYPE_MAP.get(pa.account_subtype or '', + TYPE_MAP.get(pa.account_type or '', 'other')) + new_acct = Account( + name = f'{item.institution_name} — {pa.display_name}', + account_type = pfm_type, + color = '#8B5CF6', + icon = 'bi-bank', + balance = 0, + ) + db.session.add(new_acct) + db.session.flush() + pa.pfm_account_id = new_acct.id + elif val.isdigit(): + acct_id = int(val) + if Account.query.filter_by(id=acct_id, is_active=True).first(): + pa.pfm_account_id = acct_id + # val == '' → skip (leave unmapped) + + db.session.commit() + mapped = [pa for pa in plaid_accounts if pa.pfm_account_id] + if not mapped: + flash('Select at least one account to map.', 'warning') + return redirect(url_for('plaid.map_accounts', item_db_id=item_db_id)) + + flash(f'{len(mapped)} account(s) mapped. Ready to sync.', 'success') + return redirect(url_for('plaid.index')) + + return render_template('plaid/map_accounts.html', + item=item, + plaid_accounts=plaid_accounts, + pfm_accounts=pfm_accounts) + + +# ── Sync Preview ────────────────────────────────────────────────────────────── + +@plaid_bp.route('/sync/') +@login_required +def sync_preview_view(item_db_id): + item = db.get_or_404(PlaidItem, item_db_id) + + try: + parsed, next_cursor = sync_preview(item) + except Exception as e: + log.error('[plaid] sync_preview failed for item id=%s: %s', item_db_id, e, exc_info=True) + flash(f'Sync failed: {e}', 'danger') + return redirect(url_for('plaid.index')) + + if not parsed: + flash(f'{item.institution_name}: no new transactions since last sync.', 'info') + # Still advance the cursor so we don't re-fetch old history + item.cursor = next_cursor + item.last_synced_at = datetime.utcnow() + db.session.commit() + return redirect(url_for('plaid.index')) + + # Persist preview to DB (avoid session size limits) + preview_data = [ + { + 'plaid_id': p['plaid_id'], + 'plaid_account_id': p['plaid_account_id'], + 'date': p['date'].isoformat(), + 'transaction_type': p['transaction_type'], + 'amount': float(p['amount']), + 'description': p['description'], + 'account_id': p['account_id'], + 'category_id': p['category_id'], + 'notes': p['notes'], + } + for p in parsed + ] + + sp = PlaidSyncPreview.query.filter_by(item_id=item_db_id).first() + if sp: + sp.data_json = json.dumps(preview_data) + sp.next_cursor = next_cursor + sp.created_at = datetime.utcnow() + else: + sp = PlaidSyncPreview( + item_id = item_db_id, + data_json = json.dumps(preview_data), + next_cursor = next_cursor, + ) + db.session.add(sp) + db.session.commit() + session['plaid_preview_id'] = sp.id + session['plaid_preview_item'] = item_db_id + + from app.models.category import Category + cats = Category.query.filter_by(is_active=True).order_by(Category.name).all() + cat_map = {c.id: c.name for c in cats} + + # Group transactions by PlaidAccount for display + plaid_accounts = { + pa.plaid_account_id: pa + for pa in item.accounts.filter_by(is_active=True).all() + } + + return render_template('plaid/preview.html', + item=item, + preview=parsed, + count=len(parsed), + cat_map=cat_map, + categories=cats, + plaid_accounts=plaid_accounts) + + +# ── Sync Confirm ────────────────────────────────────────────────────────────── + +@plaid_bp.route('/sync/confirm', methods=['POST']) +@login_required +def sync_confirm(): + preview_id = session.pop('plaid_preview_id', None) + item_db_id = session.pop('plaid_preview_item', None) + + if not preview_id or not item_db_id: + flash('No pending import. Please sync again.', 'warning') + return redirect(url_for('plaid.index')) + + sp = db.session.get(PlaidSyncPreview, preview_id) + if not sp: + flash('Preview expired or already imported. Please sync again.', 'warning') + return redirect(url_for('plaid.index')) + + raw = json.loads(sp.data_json) + next_cursor = sp.next_cursor + item = db.get_or_404(PlaidItem, item_db_id) + db.session.delete(sp) + + selected_ids = set(request.form.getlist('selected')) + from datetime import date as date_cls + parsed = [] + for r in raw: + if selected_ids and r['plaid_id'] not in selected_ids: + continue + r['date'] = date_cls.fromisoformat(r['date']) + # Type override + override = request.form.get(f'type_{r["plaid_id"]}') + if override in ('income', 'expense'): + r['transaction_type'] = override + # Category override + cat_override = request.form.get(f'category_{r["plaid_id"]}', '') + if cat_override.isdigit(): + r['category_id'] = int(cat_override) + parsed.append(r) + + if not parsed: + flash('No transactions selected. Nothing was imported.', 'warning') + db.session.commit() + return redirect(url_for('plaid.index')) + + imported, skipped = import_transactions(parsed, next_cursor, item) + flash(f'Imported {imported} transaction(s) from {item.institution_name}. ' + f'Skipped {skipped} duplicate(s).', 'success') + return redirect(url_for('transactions.index')) + + +# ── Balance Refresh (AJAX) ──────────────────────────────────────────────────── + +@plaid_bp.route('/balance/', methods=['POST']) +@login_required +def refresh_balance(pa_db_id): + pa = db.get_or_404(PlaidAccount, pa_db_id) + if not pa.pfm_account_id: + return jsonify({'error': 'Account not mapped'}), 400 + + try: + bal_map = get_balances(pa.item, [pa.plaid_account_id]) + bal = bal_map.get(pa.plaid_account_id, {}) + current = bal.get('current') + available = bal.get('available') + + if current is None and available is None: + return jsonify({'error': 'No balance data returned'}), 502 + + is_cc = pa.account_type == 'credit' or pa.account_subtype == 'credit card' + if is_cc: + balance = -abs(float(current)) + else: + balance = float(available if available is not None else current) + + pa.pfm_account.balance = balance + db.session.commit() + log.info('[plaid] balance refreshed for %s: %.2f', pa.account_name, balance) + return jsonify({'status': 'ok', 'balance': balance, 'account': pa.account_name}) + + except Exception as e: + log.error('[plaid] balance refresh failed for pa_id=%s: %s', pa_db_id, e, exc_info=True) + return jsonify({'error': str(e)}), 502 + + +# ── Liabilities Refresh ─────────────────────────────────────────────────────── + +@plaid_bp.route('/liabilities/', methods=['POST']) +@login_required +def refresh_liabilities_view(item_db_id): + item = db.get_or_404(PlaidItem, item_db_id) + try: + updated = refresh_liabilities(item) + if updated: + flash(f'Credit card billing details updated for {updated} account(s).', 'success') + else: + flash('No credit card accounts found, or liabilities not supported by this institution.', 'info') + except Exception as e: + log.error('[plaid] refresh_liabilities failed for item id=%s: %s', item_db_id, e, exc_info=True) + flash(f'Failed to fetch billing details: {e}', 'danger') + + next_url = request.form.get('next', '') + if next_url and next_url.startswith('/'): + return redirect(next_url) + return redirect(url_for('plaid.index')) + + +# ── Disconnect ──────────────────────────────────────────────────────────────── + +@plaid_bp.route('/disconnect/', methods=['POST']) +@login_required +def disconnect(item_db_id): + item = db.get_or_404(PlaidItem, item_db_id) + item.is_active = False + for pa in item.accounts: + pa.is_active = False + db.session.commit() + flash(f'Disconnected from {item.institution_name}. Imported transactions are kept.', 'info') + return redirect(url_for('plaid.index')) diff --git a/app/services/plaid_service.py b/app/services/plaid_service.py new file mode 100644 index 0000000..19caa44 --- /dev/null +++ b/app/services/plaid_service.py @@ -0,0 +1,382 @@ +""" +Plaid Integration Service + +Auth: PLAID-CLIENT-ID + PLAID-SECRET headers on every request. + +Key endpoints: + POST /link/token/create → link_token for the frontend widget + POST /item/public_token/exchange → access_token + item_id + POST /accounts/get → list accounts in an item + POST /accounts/balance/get → live balances + POST /liabilities/get → credit card due date, min payment + POST /transactions/sync → cursor-based incremental sync + +Sign convention (ALL account types): + positive amount → money OUT of the account → expense + negative amount → money INTO the account → income +""" + +import logging +from datetime import date, datetime, timedelta + +import requests +from flask import current_app + +log = logging.getLogger(__name__) + +PLAID_HOSTS = { + 'sandbox': 'https://sandbox.plaid.com', + 'development': 'https://development.plaid.com', + 'production': 'https://production.plaid.com', +} + +# Plaid top-level category → PFM category name +CATEGORY_MAP = { + 'Food and Drink': 'Food & Dining', + 'Travel': 'Transport', + 'Shops': 'Shopping', + 'Recreation': 'Entertainment', + 'Healthcare': 'Health', + 'Service': 'Other', + 'Community': 'Other', + 'Bank Fees': 'Other', + 'Cash Advance': 'Other', + 'Interest': 'Other', + 'Payment': 'Other', + 'Transfer': 'Other', + 'Tax': 'Other', + 'Payroll': 'Salary', + 'Deposit': 'Other Income', + 'Income': 'Other Income', + 'Investment Income': 'Investment', + 'Utilities': 'Utilities', + 'Telecommunication': 'Utilities', + 'Insurance': 'Insurance', + 'Education': 'Education', + 'Rent and Utilities': 'Housing', + 'Mortgage': 'Housing', + 'Home Improvement': 'Housing', + 'Government and Non-Profit': 'Other', +} + + +def _base(): + env = current_app.config.get('PLAID_ENV', 'sandbox').lower() + return PLAID_HOSTS.get(env, PLAID_HOSTS['sandbox']) + + +def _headers(): + return { + 'PLAID-CLIENT-ID': current_app.config.get('PLAID_CLIENT_ID', ''), + 'PLAID-SECRET': current_app.config.get('PLAID_SECRET', ''), + 'Content-Type': 'application/json', + } + + +def _post(path, payload): + url = f'{_base()}{path}' + resp = requests.post(url, json=payload, headers=_headers(), timeout=30) + if not resp.ok: + log.error('[plaid] API error %s %s — body=%r', + resp.status_code, path, resp.text[:500]) + resp.raise_for_status() + return resp.json() + + +# ── Link Token ──────────────────────────────────────────────────────────────── + +def create_link_token(): + """Create a Link token for the frontend Plaid Link widget.""" + data = _post('/link/token/create', { + 'user': {'client_user_id': 'pfm-user'}, + 'client_name': 'Personal Finance Manager', + 'products': ['transactions'], + 'additional_consented_products': ['liabilities'], + 'country_codes': ['US'], + 'language': 'en', + }) + return data['link_token'] + + +# ── Token Exchange ──────────────────────────────────────────────────────────── + +def exchange_public_token(public_token): + """Exchange the one-time public_token for a permanent access_token.""" + return _post('/item/public_token/exchange', {'public_token': public_token}) + # Returns: {'access_token': '...', 'item_id': '...'} + + +# ── Accounts ────────────────────────────────────────────────────────────────── + +def get_accounts(item): + """Return list of account dicts for the item (cached, not live balances).""" + data = _post('/accounts/get', {'access_token': item.access_token}) + return data.get('accounts', []) + + +def get_balances(item, plaid_account_ids=None): + """ + Return live balance data keyed by plaid_account_id. + Optionally filter to a subset of account IDs. + """ + payload = {'access_token': item.access_token} + if plaid_account_ids: + payload['options'] = {'account_ids': plaid_account_ids} + data = _post('/accounts/balance/get', payload) + return {a['account_id']: a['balances'] for a in data.get('accounts', [])} + + +# ── Liabilities (credit cards) ──────────────────────────────────────────────── + +def get_liabilities(item): + """ + Fetch credit card billing details for all credit accounts in the item. + Returns list of liability dicts; empty list if the item has no credit accounts + or if the product is not supported. + + Key fields per entry: + account_id, minimum_payment_amount, next_payment_due_date, + last_statement_balance, last_statement_issue_date, is_overdue + """ + try: + data = _post('/liabilities/get', {'access_token': item.access_token}) + return data.get('liabilities', {}).get('credit', []) + except requests.HTTPError as e: + # PRODUCTS_NOT_SUPPORTED or institution doesn't support liabilities + log.warning('[plaid] liabilities not available for item %s: %s', item.item_id, e) + return [] + + +def refresh_liabilities(item): + """ + Pull latest credit card liabilities and persist them onto PlaidAccount rows. + Returns number of accounts updated. + """ + from app.extensions import db + + liabs = get_liabilities(item) + if not liabs: + return 0 + + from app.models.plaid_item import PlaidAccount + updated = 0 + for lib in liabs: + pa = PlaidAccount.query.filter_by( + plaid_account_id=lib['account_id'], is_active=True + ).first() + if not pa: + continue + + due_raw = lib.get('next_payment_due_date') + pa.cc_due_date = ( + datetime.strptime(due_raw, '%Y-%m-%d').date() if due_raw else None + ) + pa.cc_minimum_payment = lib.get('minimum_payment_amount') + pa.cc_last_statement_balance = lib.get('last_statement_balance') + pa.cc_is_overdue = bool(lib.get('is_overdue', False)) + pa.cc_updated_at = datetime.utcnow() + updated += 1 + + db.session.commit() + log.info('[plaid] liabilities updated for %d account(s) in item %s', + updated, item.item_id) + return updated + + +# ── Transactions Sync ───────────────────────────────────────────────────────── + +def sync_transactions(item): + """ + Cursor-based transaction sync. Fetches ALL pages until has_more=False. + Returns (added, modified, removed, next_cursor). + Passing an empty/None cursor fetches full available history. + """ + added = [] + modified = [] + removed = [] + cursor = item.cursor or '' + has_more = True + + while has_more: + data = _post('/transactions/sync', { + 'access_token': item.access_token, + 'cursor': cursor, + }) + added.extend(data.get('added', [])) + modified.extend(data.get('modified', [])) + removed.extend(data.get('removed', [])) + cursor = data.get('next_cursor', cursor) + has_more = data.get('has_more', False) + + return added, modified, removed, cursor + + +def build_category_map(): + from app.models.category import Category + cats = Category.query.filter_by(is_active=True).all() + return {c.name: c.id for c in cats} + + +def _map_plaid_category(plaid_cats, txn_type): + """Map Plaid category array to a PFM category name.""" + if not plaid_cats: + return None + top = plaid_cats[0] + mapped = CATEGORY_MAP.get(top) + if mapped: + return mapped + # Auto-categorize by description is done downstream; fall back by type + return 'Other Income' if txn_type == 'income' else None + + +def parse_transaction(plaid_txn, plaid_account_map, cat_id_map): + """ + Convert a Plaid transaction dict to a PFM-ready dict. + + plaid_account_map: {plaid_account_id: pfm_account_id} + """ + amount_raw = float(plaid_txn.get('amount', 0)) + # Plaid: positive = expense (debit/outflow), negative = income (credit/inflow) + if amount_raw > 0: + txn_type = 'expense' + amount = amount_raw + else: + txn_type = 'income' + amount = abs(amount_raw) + + description = ( + plaid_txn.get('merchant_name') or + plaid_txn.get('name') or + 'Plaid transaction' + ).strip() + + # Auto-categorize via keyword match first, then Plaid category + from app.services.bank_import_service import auto_categorize + pfm_cat_name = auto_categorize(description) + if not pfm_cat_name: + pfm_cat_name = _map_plaid_category(plaid_txn.get('category') or [], txn_type) + category_id = cat_id_map.get(pfm_cat_name) if pfm_cat_name else None + + plaid_acct_id = plaid_txn.get('account_id', '') + pfm_account_id = plaid_account_map.get(plaid_acct_id) + + txn_id = plaid_txn.get('transaction_id', '') + return { + 'plaid_id': txn_id, + 'plaid_account_id': plaid_acct_id, + 'date': datetime.strptime(plaid_txn['date'], '%Y-%m-%d').date(), + 'transaction_type': txn_type, + 'amount': amount, + 'description': description, + 'account_id': pfm_account_id, + 'category_id': category_id, + 'notes': f'Plaid:{txn_id}', + 'pending': plaid_txn.get('pending', False), + } + + +def sync_preview(item): + """ + Fetch new transactions for an item and return a list of parsed preview dicts. + Does NOT write to DB or advance the cursor — call this before showing the preview. + Returns (parsed_list, next_cursor). + """ + from app.models.plaid_item import PlaidAccount + + added, _modified, _removed, next_cursor = sync_transactions(item) + + # Build maps + plaid_accounts = PlaidAccount.query.filter_by(item_id=item.id, is_active=True).all() + plaid_account_map = {pa.plaid_account_id: pa.pfm_account_id for pa in plaid_accounts} + cat_map = build_category_map() + + parsed = [] + for txn in added: + if txn.get('pending', False): + continue # skip pending transactions — import after they post + p = parse_transaction(txn, plaid_account_map, cat_map) + if p['account_id'] is None: + continue # skip accounts not mapped to a PFM account + parsed.append(p) + + return parsed, next_cursor + + +def import_transactions(parsed_txns, next_cursor, item): + """ + Write selected parsed transactions to the DB. + Skips duplicates by checking Plaid: in notes. + Advances item cursor to next_cursor after successful import. + Returns (imported_count, skipped_count). + """ + from app.extensions import db + from app.models.transaction import Transaction + from app.services.account_service import calc_balance + from app.models.plaid_item import PlaidAccount + + imported = skipped = 0 + affected_accounts = set() + plaid_account_ids_synced = set() + + for p in parsed_txns: + pid = p['plaid_id'] + if Transaction.query.filter(Transaction.notes.like(f'%Plaid:{pid}%')).first(): + skipped += 1 + continue + + db.session.add(Transaction( + account_id = p['account_id'], + category_id = p.get('category_id'), + transaction_type = p['transaction_type'], + amount = p['amount'], + description = p['description'], + date = p['date'], + notes = p['notes'], + )) + if p['account_id']: + affected_accounts.add(p['account_id']) + plaid_account_ids_synced.add(p['plaid_account_id']) + imported += 1 + + db.session.commit() + + # Advance cursor + item.cursor = next_cursor + item.last_synced_at = datetime.utcnow() + + # Update last_sync_date per PlaidAccount + today = date.today() + for pa in PlaidAccount.query.filter_by(item_id=item.id, is_active=True).all(): + if pa.plaid_account_id in plaid_account_ids_synced: + pa.last_sync_date = today + + db.session.commit() + + # Refresh balances for mapped accounts + if affected_accounts: + try: + mapped_pa = PlaidAccount.query.filter( + PlaidAccount.pfm_account_id.in_(affected_accounts), + PlaidAccount.is_active == True, + ).all() + if mapped_pa: + pa_ids = [pa.plaid_account_id for pa in mapped_pa] + bal_map = get_balances(item, pa_ids) + for pa in mapped_pa: + bal = bal_map.get(pa.plaid_account_id, {}) + current = bal.get('current') + available = bal.get('available') + if current is not None: + is_cc = pa.account_type == 'credit' or pa.account_subtype == 'credit card' + if is_cc: + # Plaid returns positive current balance = amount owed on card + pa.pfm_account.balance = -abs(float(current)) + else: + pa.pfm_account.balance = float(available if available is not None else current) + db.session.commit() + except Exception as e: + log.warning('[plaid] balance refresh after import failed: %s', e) + for acct_id in affected_accounts: + calc_balance(acct_id) + + return imported, skipped diff --git a/app/services/teller_service.py b/app/services/teller_service.py index 7604492..dac91cb 100644 --- a/app/services/teller_service.py +++ b/app/services/teller_service.py @@ -175,23 +175,32 @@ def get_transactions(access_token, account_id, start_date=None, end_date=None, return data -def parse_transaction(teller_txn, pfm_account_id, category_id_map): +def parse_transaction(teller_txn, pfm_account_id, category_id_map, is_credit_card=False): """ Convert a Teller transaction dict to a PFM transaction dict ready for import. Returns dict with keys matching Transaction model fields. - Teller amounts: - - Positive = money entering the account (income / credit) - - Negative = money leaving the account (expense / debit) + Teller sign convention differs by account type: + - Depository (checking/savings): positive = inflow (income), negative = outflow (expense) + - Credit (credit card): positive = charge/purchase (expense), negative = payment/credit (income) """ amount_raw = float(teller_txn.get('amount', 0)) - # Teller: positive = inflow/credit (income), negative = outflow/debit (expense) - if amount_raw < 0: - txn_type = 'expense' - amount = abs(amount_raw) + if is_credit_card: + # Credit cards: positive amount = purchase (expense), negative = payment (income) + if amount_raw > 0: + txn_type = 'expense' + amount = amount_raw + else: + txn_type = 'income' + amount = abs(amount_raw) else: - txn_type = 'income' - amount = amount_raw + # Depository: positive = deposit (income), negative = withdrawal (expense) + if amount_raw < 0: + txn_type = 'expense' + amount = abs(amount_raw) + else: + txn_type = 'income' + amount = amount_raw description = teller_txn.get('description', '').strip() or 'Teller transaction' # Use enriched counterparty name if available @@ -264,9 +273,16 @@ def sync_preview(teller_account, days_back=90): raise cat_map = build_category_map() + is_cc = ( + teller_account.account_type == 'credit' or + teller_account.account_subtype == 'credit_card' or + (teller_account.pfm_account and + teller_account.pfm_account.account_type == 'credit_card') + ) parsed = [] for txn in raw_txns: - p = parse_transaction(txn, teller_account.pfm_account_id, cat_map) + p = parse_transaction(txn, teller_account.pfm_account_id, cat_map, + is_credit_card=is_cc) parsed.append(p) return parsed diff --git a/app/templates/accounts/index.html b/app/templates/accounts/index.html index ce71e14..30ae50d 100644 --- a/app/templates/accounts/index.html +++ b/app/templates/accounts/index.html @@ -27,6 +27,7 @@ {% for acct in accounts %} {% set ta = teller_map.get(acct.id) %} {% set sa = schwab_map.get(acct.id) %} + {% set pa = plaid_map.get(acct.id) %}
@@ -45,6 +46,9 @@ {% if sa %} Schwab {% endif %} + {% if pa %} + Plaid + {% endif %}
{{ acct.account_type | replace('_',' ') | title }}
@@ -90,6 +94,35 @@ {% endif %} + + {% if pa and tab == 'credit' and (pa.cc_due_date or pa.cc_minimum_payment) %} +
+ {% if pa.cc_due_date %} + {% set days_left = (pa.cc_due_date - today).days %} +
+
Due Date
+
{{ pa.cc_due_date.strftime('%b %d') }}
+ {% if days_left <= 0 %} +
Overdue!
+ {% elif days_left <= 3 %} +
{{ days_left }} day{{ 's' if days_left != 1 }} left
+ {% else %} +
{{ days_left }} days left
+ {% endif %} +
+ {% endif %} + {% if pa.cc_minimum_payment %} +
+
Min Payment
+
{{ pa.cc_minimum_payment | currency }}
+ {% if pa.cc_last_statement_balance %} +
Stmt: {{ pa.cc_last_statement_balance | currency }}
+ {% endif %} +
+ {% endif %} +
+ {% endif %} + {% if acct.notes %}
{{ acct.notes }}
{% endif %} @@ -154,6 +187,37 @@ {% endif %} {% endif %} + + {% if pa %} +
+ + + Sync + + {% if pa.account_type == 'credit' %} +
+ + + +
+ {% endif %} +
+ {% if pa.last_sync_date %} +
+ Last synced {{ pa.last_sync_date.strftime('%b %d') }} +
+ {% endif %} + {% endif %} + {% if sa %}
@@ -206,6 +270,54 @@ + +{% endblock %} diff --git a/app/templates/plaid/map_accounts.html b/app/templates/plaid/map_accounts.html new file mode 100644 index 0000000..4b4e62a --- /dev/null +++ b/app/templates/plaid/map_accounts.html @@ -0,0 +1,52 @@ +{% extends "base.html" %} +{% block title %}Map Plaid Accounts{% endblock %} +{% block page_title %}Map Accounts — {{ item.institution_name }}{% endblock %} + +{% block content %} +
+ + +
+
{{ item.institution_name }}
+
+ Map each Plaid account to a PFM account, or create a new one automatically. + Accounts left unmapped will be skipped during sync. +
+
+ +{% for pa in plaid_accounts %} +
+
+
+
{{ pa.display_name }}
+
+ {{ pa.account_type | title }} · {{ pa.account_subtype | replace('_',' ') | title }} +
+
+
+ + +
+
+
+{% endfor %} + +
+ Cancel + +
+
+{% endblock %} diff --git a/app/templates/plaid/preview.html b/app/templates/plaid/preview.html new file mode 100644 index 0000000..a05617e --- /dev/null +++ b/app/templates/plaid/preview.html @@ -0,0 +1,201 @@ +{% extends "base.html" %} +{% block title %}Plaid Sync Preview{% endblock %} +{% block page_title %}Sync Preview — {{ item.institution_name }}{% endblock %} + +{% block content %} +
+ + +
+
+
+
{{ item.institution_name }}
+
+ {{ count }} of {{ count }} transactions selected +
+
+
+ Cancel + +
+
+ + +
+ Bulk set selected: + + +
+ + +
+
+
+ +{% set ns = namespace(cur_acct_id=None) %} +{% for txn in preview %} +{% set pa = plaid_accounts.get(txn.plaid_account_id) %} +{% set acct_id = pa.pfm_account_id if pa else None %} + +{% if acct_id != ns.cur_acct_id %} +{% if ns.cur_acct_id is not none %}
{% endif %} +{% set ns.cur_acct_id = acct_id %} +
+
+
+ {% if pa and pa.pfm_account %}{{ pa.pfm_account.name }}{% else %}Unknown Account{% endif %} +
+ {% if pa %} +
{{ pa.display_name }} · {{ pa.account_type | title }}
+ {% endif %} +
+
+ + + + + + + + + + + + +{% endif %} + + + + + + + + + +{% endfor %} +{% if ns.cur_acct_id is not none %}
+ + DateDescriptionTypeCategoryAmount
+ + + {{ txn.date.strftime('%b %d, %Y') }} + {{ txn.description }} + + + + + {% if txn.transaction_type == 'income' %}+{% else %}-{% endif %}{{ txn.amount | currency }} +
{% endif %} + +
+ +
+ + + +{% endblock %} diff --git a/app/templates/settings/index.html b/app/templates/settings/index.html index 2df4956..80a32b3 100644 --- a/app/templates/settings/index.html +++ b/app/templates/settings/index.html @@ -22,6 +22,15 @@
Connect Charles Schwab directly
+ +
+ +
+ +
Plaid Sync
+
12,000+ US banks · CC billing dates
+
+
diff --git a/scripts/add_plaid_tables.py b/scripts/add_plaid_tables.py new file mode 100644 index 0000000..44dca27 --- /dev/null +++ b/scripts/add_plaid_tables.py @@ -0,0 +1,94 @@ +""" +Migration: add Plaid tables. + +Run once after deploying the Plaid integration: + python scripts/add_plaid_tables.py + +Creates: + plaid_items — one row per connected bank (Item in Plaid terminology) + plaid_accounts — one row per Plaid account, mapped to a PFM account + plaid_sync_previews — temporary storage for transaction sync previews +""" + +import sys +import os +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app import create_app +from app.extensions import db + + +TABLES = [ + # plaid_items + """ + CREATE TABLE IF NOT EXISTS `plaid_items` ( + `id` INT NOT NULL AUTO_INCREMENT PRIMARY KEY, + `item_id` VARCHAR(100) NOT NULL UNIQUE, + `access_token` TEXT NOT NULL, + `institution_id` VARCHAR(50) DEFAULT NULL, + `institution_name` VARCHAR(100) DEFAULT NULL, + `cursor` VARCHAR(500) DEFAULT NULL, + `is_active` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `last_synced_at` DATETIME DEFAULT NULL, + INDEX `ix_plaid_items_item_id` (`item_id`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """, + + # plaid_accounts + """ + CREATE TABLE IF NOT EXISTS `plaid_accounts` ( + `id` INT NOT NULL AUTO_INCREMENT PRIMARY KEY, + `item_id` INT NOT NULL, + `plaid_account_id` VARCHAR(100) NOT NULL UNIQUE, + `pfm_account_id` INT DEFAULT NULL, + `account_name` VARCHAR(100) DEFAULT NULL, + `account_type` VARCHAR(50) DEFAULT NULL, + `account_subtype` VARCHAR(50) DEFAULT NULL, + `mask` VARCHAR(10) DEFAULT NULL, + `last_sync_date` DATE DEFAULT NULL, + `cc_due_date` DATE DEFAULT NULL, + `cc_minimum_payment` DECIMAL(12,2) DEFAULT NULL, + `cc_last_statement_balance` DECIMAL(12,2) DEFAULT NULL, + `cc_is_overdue` TINYINT(1) NOT NULL DEFAULT 0, + `cc_updated_at` DATETIME DEFAULT NULL, + `is_active` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (`item_id`) REFERENCES `plaid_items`(`id`) ON DELETE CASCADE, + FOREIGN KEY (`pfm_account_id`) REFERENCES `accounts`(`id`) ON DELETE SET NULL, + INDEX `ix_plaid_accounts_plaid_account_id` (`plaid_account_id`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """, + + # plaid_sync_previews + """ + CREATE TABLE IF NOT EXISTS `plaid_sync_previews` ( + `id` INT NOT NULL AUTO_INCREMENT PRIMARY KEY, + `item_id` INT NOT NULL UNIQUE, + `data_json` TEXT NOT NULL, + `next_cursor` VARCHAR(500) DEFAULT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (`item_id`) REFERENCES `plaid_items`(`id`) ON DELETE CASCADE, + INDEX `ix_plaid_sync_previews_item_id` (`item_id`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """, +] + + +def run(): + app = create_app() + with app.app_context(): + conn = db.engine.raw_connection() + cur = conn.cursor() + for ddl in TABLES: + name = ddl.strip().split('`')[1] + cur.execute(ddl) + print(f' ✓ {name}') + conn.commit() + cur.close() + conn.close() + print('Done.') + + +if __name__ == '__main__': + run()