327 lines
13 KiB
Python
327 lines
13 KiB
Python
import logging
|
|
from datetime import date as date_cls, datetime
|
|
|
|
from flask import (Blueprint, render_template, redirect, url_for,
|
|
flash, request, session, current_app)
|
|
from flask_login import login_required
|
|
|
|
from app.extensions import db
|
|
from app.models.account import Account
|
|
from app.models.schwab_connection import SchwabConnection, SchwabAccount
|
|
from app.services.schwab_service import (
|
|
get_auth_url, exchange_code, _apply_token_data,
|
|
get_account_number_hashes, get_accounts,
|
|
sync_preview, import_transactions, sync_account_snapshot,
|
|
ACCOUNT_TYPE_MAP,
|
|
)
|
|
|
|
schwab_bp = Blueprint('schwab', __name__, url_prefix='/schwab')
|
|
log = logging.getLogger(__name__)
|
|
from app.utils.audit import audit
|
|
|
|
|
|
def _active_connection():
|
|
return SchwabConnection.query.filter_by(is_active=True).first()
|
|
|
|
|
|
# ── Connect ───────────────────────────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/connect')
|
|
@login_required
|
|
def connect():
|
|
if not current_app.config.get('SCHWAB_CLIENT_ID'):
|
|
flash('SCHWAB_CLIENT_ID is not set in .env — add it and restart.', 'danger')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
auth_url, state = get_auth_url()
|
|
session['schwab_oauth_state'] = state
|
|
return redirect(auth_url)
|
|
|
|
|
|
@schwab_bp.route('/callback')
|
|
@login_required
|
|
def callback():
|
|
error = request.args.get('error')
|
|
if error:
|
|
flash(f'Schwab connection cancelled: {error}', 'warning')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
code = request.args.get('code', '')
|
|
state = request.args.get('state', '')
|
|
|
|
if not code:
|
|
flash('No authorization code received from Schwab.', 'danger')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
if state != session.pop('schwab_oauth_state', None):
|
|
flash('OAuth state mismatch — possible CSRF. Please try again.', 'danger')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
try:
|
|
token_data = exchange_code(code)
|
|
except Exception as e:
|
|
log.error('[schwab] token exchange failed: %s', e, exc_info=True)
|
|
flash(f'Failed to connect to Schwab: {e}', 'danger')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
# Deactivate any previous connection
|
|
SchwabConnection.query.filter_by(is_active=True).update({'is_active': False})
|
|
|
|
conn = SchwabConnection(
|
|
access_token='',
|
|
refresh_token='',
|
|
token_expires_at=datetime.utcnow(),
|
|
)
|
|
_apply_token_data(conn, token_data)
|
|
db.session.add(conn)
|
|
db.session.flush()
|
|
|
|
# Fetch account hash mapping first (hashValue is required for all API paths)
|
|
try:
|
|
hash_map = get_account_number_hashes(conn)
|
|
raw_accounts = get_accounts(conn)
|
|
except Exception as e:
|
|
log.error('[schwab] get_accounts failed: %s', e, exc_info=True)
|
|
db.session.rollback()
|
|
flash(f'Connected but failed to fetch accounts: {e}', 'danger')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
for ra in raw_accounts:
|
|
sec = ra.get('securitiesAccount', {})
|
|
acct_num = sec.get('accountNumber', '')
|
|
acct_hash = hash_map.get(acct_num, acct_num)
|
|
if not acct_hash:
|
|
continue
|
|
masked = '…' + acct_num[-4:] if len(acct_num) >= 4 else acct_num
|
|
|
|
# Try to find an existing record (by new hash or old raw number) so we
|
|
# can update it in-place and preserve the pfm_account_id mapping.
|
|
existing = (SchwabAccount.query.filter_by(account_hash=acct_hash).first() or
|
|
SchwabAccount.query.filter_by(account_hash=acct_num).first() or
|
|
SchwabAccount.query.filter_by(account_number_display=masked).first())
|
|
|
|
if existing:
|
|
existing.connection = conn
|
|
existing.account_hash = acct_hash
|
|
existing.is_active = True
|
|
else:
|
|
db.session.add(SchwabAccount(
|
|
connection=conn,
|
|
account_hash=acct_hash,
|
|
account_number_display=masked,
|
|
account_type=sec.get('type', 'CASH'),
|
|
account_name=f'Schwab {sec.get("type","Account")} {masked}',
|
|
))
|
|
|
|
db.session.commit()
|
|
audit('schwab_connected')
|
|
flash('Schwab connected successfully. Map your accounts to get started.', 'success')
|
|
return redirect(url_for('schwab.map_accounts'))
|
|
|
|
|
|
# ── Index ─────────────────────────────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/')
|
|
@login_required
|
|
def index():
|
|
connection = _active_connection()
|
|
accounts = connection.accounts.filter_by(is_active=True).all() if connection else []
|
|
return render_template('schwab/index.html',
|
|
connection=connection,
|
|
accounts=accounts,
|
|
schwab_configured=bool(current_app.config.get('SCHWAB_CLIENT_ID')))
|
|
|
|
|
|
# ── Account mapping ───────────────────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/map', methods=['GET', 'POST'])
|
|
@login_required
|
|
def map_accounts():
|
|
connection = _active_connection()
|
|
if not connection:
|
|
flash('No active Schwab connection.', 'warning')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
schwab_accounts = connection.accounts.filter_by(is_active=True).all()
|
|
pfm_accounts = Account.query.filter_by(is_active=True).order_by(Account.name).all()
|
|
|
|
if request.method == 'POST':
|
|
for sa in schwab_accounts:
|
|
val = request.form.get(f'pfm_account_{sa.id}', '')
|
|
if val == 'new':
|
|
pfm_type = ACCOUNT_TYPE_MAP.get(sa.account_type, 'other')
|
|
new_acct = Account(
|
|
name=sa.account_name,
|
|
account_type=pfm_type,
|
|
color='#4F81C7',
|
|
icon='bi-bank',
|
|
balance=0,
|
|
)
|
|
db.session.add(new_acct)
|
|
db.session.flush()
|
|
sa.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():
|
|
sa.pfm_account_id = acct_id
|
|
db.session.commit()
|
|
|
|
mapped = [sa for sa in schwab_accounts if sa.pfm_account_id]
|
|
if not mapped:
|
|
flash('Select at least one account to map.', 'warning')
|
|
return redirect(url_for('schwab.map_accounts'))
|
|
|
|
flash(f'{len(mapped)} account(s) mapped. Ready to sync.', 'success')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
return render_template('schwab/map_accounts.html',
|
|
connection=connection,
|
|
schwab_accounts=schwab_accounts,
|
|
pfm_accounts=pfm_accounts)
|
|
|
|
|
|
# ── Sync: preview ─────────────────────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/sync/<int:schwab_account_id>')
|
|
@login_required
|
|
def sync_preview_view(schwab_account_id):
|
|
sa = db.get_or_404(SchwabAccount, schwab_account_id)
|
|
|
|
if not sa.pfm_account_id:
|
|
flash('Map this account to a PFM account first.', 'warning')
|
|
return redirect(url_for('schwab.map_accounts'))
|
|
|
|
try:
|
|
preview = sync_preview(sa)
|
|
except Exception as e:
|
|
log.error('[schwab] sync_preview failed for account id=%s: %s',
|
|
schwab_account_id, e, exc_info=True)
|
|
flash(f'Sync failed: {e}', 'danger')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
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}
|
|
|
|
session['schwab_preview'] = [
|
|
{
|
|
'schwab_id': p['schwab_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'],
|
|
'schwab_type': p['schwab_type'],
|
|
}
|
|
for p in preview
|
|
]
|
|
session['schwab_account_id'] = schwab_account_id
|
|
|
|
return render_template('schwab/preview.html',
|
|
sa=sa,
|
|
preview=preview,
|
|
count=len(preview),
|
|
cat_map=cat_map,
|
|
categories=cats)
|
|
|
|
|
|
# ── Sync: confirm ─────────────────────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/sync/confirm', methods=['POST'])
|
|
@login_required
|
|
def sync_confirm():
|
|
raw = session.pop('schwab_preview', [])
|
|
sa_id = session.pop('schwab_account_id', None)
|
|
|
|
if not raw or not sa_id:
|
|
flash('No pending import. Please sync again.', 'warning')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
sa = db.get_or_404(SchwabAccount, sa_id)
|
|
selected_ids = set(request.form.getlist('selected'))
|
|
|
|
parsed = []
|
|
for r in raw:
|
|
if selected_ids and r['schwab_id'] not in selected_ids:
|
|
continue
|
|
r['date'] = date_cls.fromisoformat(r['date'])
|
|
override = request.form.get(f'type_{r["schwab_id"]}')
|
|
if override in ('income', 'expense'):
|
|
r['transaction_type'] = override
|
|
cat_override = request.form.get(f'category_{r["schwab_id"]}', '')
|
|
if cat_override.isdigit():
|
|
r['category_id'] = int(cat_override)
|
|
parsed.append(r)
|
|
|
|
if not parsed:
|
|
flash('No transactions selected.', 'warning')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
imported, skipped = import_transactions(parsed, sa)
|
|
flash(f'Imported {imported} transaction(s) from {sa.account_name}. '
|
|
f'Skipped {skipped} duplicate(s).', 'success')
|
|
return redirect(url_for('transactions.index'))
|
|
|
|
|
|
# ── Full resync ───────────────────────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/resync/<int:schwab_account_id>', methods=['POST'])
|
|
@login_required
|
|
def full_resync(schwab_account_id):
|
|
sa = db.get_or_404(SchwabAccount, schwab_account_id)
|
|
sa.last_sync_date = None
|
|
sa.last_schwab_txn_id = None
|
|
db.session.commit()
|
|
flash(f'{sa.account_name}: reset to full resync. '
|
|
f'Next sync fetches 90 days — duplicates are skipped automatically.', 'info')
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
|
|
# ── Balance + position snapshot ───────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/snapshot/<int:schwab_account_id>', methods=['POST'])
|
|
@login_required
|
|
def sync_snapshot(schwab_account_id):
|
|
"""Pull live balance and investment positions from Schwab and write to PFM."""
|
|
sa = db.get_or_404(SchwabAccount, schwab_account_id)
|
|
if not sa.pfm_account_id:
|
|
flash('Map this account to a PFM account first.', 'warning')
|
|
return redirect(url_for('schwab.index'))
|
|
# Always use the currently active connection — the stored connection_id on the
|
|
# account record may reference an older connection after a reconnect.
|
|
active_conn = SchwabConnection.query.filter_by(is_active=True).first()
|
|
if active_conn:
|
|
sa.connection = active_conn
|
|
try:
|
|
bal_updated, pos_synced = sync_account_snapshot(sa)
|
|
flash(
|
|
f'{sa.account_name}: balance updated'
|
|
f'{f", {pos_synced} position(s) synced" if pos_synced else " (no positions found)"}.',
|
|
'success',
|
|
)
|
|
except Exception as e:
|
|
log.error('[schwab] sync_snapshot failed for id=%s: %s', schwab_account_id, e, exc_info=True)
|
|
flash(f'Snapshot sync failed: {e}', 'danger')
|
|
next_url = request.form.get('next', '')
|
|
if next_url and next_url.startswith('/'):
|
|
return redirect(next_url)
|
|
return redirect(url_for('schwab.index'))
|
|
|
|
|
|
# ── Disconnect ────────────────────────────────────────────────────────────────
|
|
|
|
@schwab_bp.route('/disconnect', methods=['POST'])
|
|
@login_required
|
|
def disconnect():
|
|
connection = _active_connection()
|
|
if connection:
|
|
connection.is_active = False
|
|
for sa in connection.accounts:
|
|
sa.is_active = False
|
|
db.session.commit()
|
|
audit('schwab_disconnected')
|
|
flash('Disconnected from Schwab. Your imported transactions are kept.', 'info')
|
|
return redirect(url_for('schwab.index'))
|