497 lines
20 KiB
Python
497 lines
20 KiB
Python
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, csrf as _csrf
|
|
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,
|
|
verify_webhook_token, auto_sync_item, update_item_webhook,
|
|
)
|
|
|
|
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/<int:item_db_id>', 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/<int:item_db_id>')
|
|
@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/<int:pa_db_id>', 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/<int:item_db_id>', 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'))
|
|
|
|
|
|
# ── Full Resync (clear cursor so next sync fetches full history) ──────────────
|
|
|
|
@plaid_bp.route('/resync/<int:item_db_id>', methods=['POST'])
|
|
@login_required
|
|
def full_resync(item_db_id):
|
|
item = db.get_or_404(PlaidItem, item_db_id)
|
|
item.cursor = None
|
|
item.last_synced_at = None
|
|
for pa in item.accounts.filter_by(is_active=True).all():
|
|
pa.last_sync_date = None
|
|
db.session.commit()
|
|
flash(
|
|
f'{item.institution_name}: cursor reset. '
|
|
f'Next sync will fetch full available history — duplicates will be skipped automatically.',
|
|
'info'
|
|
)
|
|
return redirect(url_for('plaid.index'))
|
|
|
|
|
|
# ── Webhook Receiver ─────────────────────────────────────────────────────────
|
|
|
|
@plaid_bp.route('/webhook', methods=['POST'])
|
|
@_csrf.exempt
|
|
def webhook():
|
|
"""
|
|
Receive Plaid transaction webhooks.
|
|
Plaid signs every request with a JWT in the Plaid-Verification header (ES256).
|
|
On TRANSACTIONS events: auto-import new transactions without requiring user confirmation.
|
|
On ITEM errors: log for operator visibility.
|
|
"""
|
|
token = request.headers.get('Plaid-Verification', '')
|
|
if not token:
|
|
log.warning('[plaid] webhook received without Plaid-Verification header')
|
|
return jsonify({'error': 'Missing verification token'}), 400
|
|
|
|
try:
|
|
verify_webhook_token(token)
|
|
except Exception as e:
|
|
log.warning('[plaid] webhook JWT verification failed: %s', e)
|
|
return jsonify({'error': 'Verification failed'}), 401
|
|
|
|
payload = request.get_json(silent=True) or {}
|
|
wh_type = payload.get('webhook_type', '')
|
|
wh_code = payload.get('webhook_code', '')
|
|
item_id = payload.get('item_id', '')
|
|
|
|
log.info('[plaid] webhook %s/%s item_id=%s', wh_type, wh_code, item_id)
|
|
|
|
if wh_type == 'TRANSACTIONS' and wh_code in (
|
|
'SYNC_UPDATES_AVAILABLE', 'DEFAULT_UPDATE',
|
|
'HISTORICAL_UPDATE', 'INITIAL_UPDATE',
|
|
):
|
|
item = PlaidItem.query.filter_by(item_id=item_id, is_active=True).first()
|
|
if not item:
|
|
log.warning('[plaid] webhook: no active item for item_id=%s', item_id)
|
|
return jsonify({'ok': True}) # 200 so Plaid doesn't keep retrying
|
|
|
|
try:
|
|
imported, skipped, removed = auto_sync_item(item)
|
|
log.info('[plaid] webhook auto-sync done: +%d skipped=%d removed=%d',
|
|
imported, skipped, removed)
|
|
except Exception as e:
|
|
log.error('[plaid] webhook auto-sync failed for item_id=%s: %s',
|
|
item_id, e, exc_info=True)
|
|
# Still return 200 — error is logged; retrying won't help an app-level error
|
|
|
|
elif wh_type == 'ITEM':
|
|
error = payload.get('error') or {}
|
|
if wh_code == 'ERROR':
|
|
log.error('[plaid] ITEM/ERROR item_id=%s code=%s msg=%s',
|
|
item_id, error.get('error_code'), error.get('error_message'))
|
|
elif wh_code == 'PENDING_EXPIRATION':
|
|
log.warning('[plaid] ITEM/PENDING_EXPIRATION item_id=%s — re-auth required soon', item_id)
|
|
elif wh_code == 'USER_PERMISSION_REVOKED':
|
|
log.warning('[plaid] ITEM/USER_PERMISSION_REVOKED item_id=%s', item_id)
|
|
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
# ── Update Webhook URL for Existing Items ─────────────────────────────────────
|
|
|
|
@plaid_bp.route('/update-webhook', methods=['POST'])
|
|
@login_required
|
|
def update_webhook():
|
|
"""
|
|
Tell Plaid to use the currently configured PLAID_WEBHOOK_URL for all active items.
|
|
Call this once after adding/changing PLAID_WEBHOOK_URL in .env for items that were
|
|
connected before the webhook was configured.
|
|
"""
|
|
webhook_url = current_app.config.get('PLAID_WEBHOOK_URL', '').strip()
|
|
if not webhook_url:
|
|
flash('PLAID_WEBHOOK_URL is not set in .env — nothing to update.', 'warning')
|
|
return redirect(url_for('plaid.index'))
|
|
|
|
items = PlaidItem.query.filter_by(is_active=True).all()
|
|
updated = 0
|
|
errors = 0
|
|
for item in items:
|
|
try:
|
|
update_item_webhook(item, webhook_url)
|
|
updated += 1
|
|
except Exception as e:
|
|
log.error('[plaid] update_item_webhook failed for item %s: %s', item.item_id, e)
|
|
errors += 1
|
|
|
|
if updated:
|
|
flash(f'Webhook URL updated for {updated} item(s).', 'success')
|
|
if errors:
|
|
flash(f'{errors} item(s) failed — check app logs.', 'warning')
|
|
return redirect(url_for('plaid.index'))
|
|
|
|
|
|
# ── Disconnect ────────────────────────────────────────────────────────────────
|
|
|
|
@plaid_bp.route('/disconnect/<int:item_db_id>', 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'))
|