06/05 Optimize app: implement Plaid
This commit is contained in:
@@ -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:<transaction_id> 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
|
||||
Reference in New Issue
Block a user