import base64 import copy import datetime import json import logging import os import random import socket import sys import urllib.parse from pathlib import Path from types import TracebackType from typing import Any, Optional, Type import requests from playwright.sync_api import Page from playwright.sync_api import sync_playwright log = logging.getLogger('xactfetch') NTFY_URL = os.environ['NTFY_URL'] NTFY_TOPIC = os.environ['NTFY_TOPIC'] FIREFLY_III_URL = os.environ['FIREFLY_III_URL'] FIREFLY_III_IMPORTER_URL = os.environ['FIREFLY_IMPORT_URL'] SECRET_SOCKET_PATH = os.environ.get('SECRET_SOCKET_PATH') XDG_RUNTIME_DIR = os.environ.get('XDG_RUNTIME_DIR') ACCOUNTS = { 'commerce': { '8357': 1, '7730': 67, }, 'chase': 15, } class FireflyImporter: def __init__( self, url: str, secret: str, auth: Optional[tuple[str, str]], ) -> None: self.url = url self.secret = secret self.auth = auth def import_csv( self, csv: Path, config: dict[str, Any], ) -> None: log.debug('Importing transactions from %s to Firefly III', csv) url = f'{self.url.rstrip("/")}/autoupload' with csv.open('rb') as f: r = requests.post( url, auth=self.auth, headers={ 'Accept': 'application/json', }, params={ 'secret': self.secret, }, files={ 'importable': ('import.csv', f), 'json': ('import.json', json.dumps(config)), }, ) r.raise_for_status() class SecretsClient: def __init__(self) -> None: self.sock: socket.socket def __enter__(self) -> 'SecretsClient': if not hasattr(self, 'sock'): self.connect() return self def __exit__( self, exc_type: Optional[Type[Exception]], exc_value: Optional[Exception], tb: Optional[TracebackType], ) -> bool: self.sock.close() return False def connect(self) -> None: if SECRET_SOCKET_PATH: path = Path(SECRET_SOCKET_PATH) elif XDG_RUNTIME_DIR: path = Path(XDG_RUNTIME_DIR) / 'secretsocket/.ss' else: path = '.secretsocket' self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.sock.connect(str(path)) def get_secret(self, key: str) -> bytes: self.sock.send(f'{key}\n'.encode('utf-8')) buf = self.sock.recv(64 * 2**10) return buf.rstrip(b'\n') def ntfy( message: Optional[str] = None, topic: str = NTFY_TOPIC, title: Optional[str] = None, tags: Optional[str] = None, attach: Optional[bytes] = None, filename: Optional[str] = None, ) -> None: assert message or attach headers = { 'Title': title or 'xactfetch', } if tags: headers['Tags'] = tags url = f'{NTFY_URL}/{topic}' if attach: if filename: headers['Filename'] = filename if message: try: message.encode("ascii") except UnicodeEncodeError: message = rfc2047_base64encode(message) else: message = message.replace('\n', '\\n') headers['Message'] = message r = requests.put( url, headers=headers, data=attach, ) else: r = requests.post( url, headers=headers, data=message, ) r.raise_for_status() def rfc2047_base64encode( message: str, ) -> str: encoded = base64.b64encode(message.encode("utf-8")).decode("ascii") return f"=?UTF-8?B?{encoded}?=" def get_last_transaction_date(key: int, token: str) -> datetime.date: url = f'{FIREFLY_III_URL}/api/v1/accounts/{key}/transactions' r = requests.get( url, headers={ 'Authorization': f'Bearer {token}', 'Accept': 'application/vnd.api+json', }, ) r.raise_for_status() last_date = datetime.datetime.min for xact in r.json()['data']: for split in xact['attributes']['transactions']: try: datestr = split['date'].split('T')[0] date = datetime.datetime.fromisoformat(datestr) except (KeyError, ValueError) as e: log.warning( 'Could not parse date from transaction %s: %s', xact['id'], e, ) continue if date > last_date: last_date = date return last_date.date() def download_chase( page: Page, secrets: SecretsClient, end_date: datetime.date, token: str, importer: FireflyImporter, ) -> bool: with Chase(page, secrets) as c, ntfyerror('Chase', page) as r: key = ACCOUNTS['chase'] try: start_date = get_last_transaction_date(key, token) except (OSError, ValueError) as e: log.error( 'Skipping Chase account: could not get last transaction: %s', e, ) return False if start_date > end_date: log.info( 'Skipping Chase account: last transaction was %s', start_date, ) return True c.login() csv = c.download_transactions(start_date, end_date) log.info('Importing transactions from Chase into Firefly III') c.firefly_import(csv, key, importer) return r.success def download_commerce( page: Page, secrets: SecretsClient, end_date: datetime.date, token: str, importer: FireflyImporter, ) -> bool: log.info('Downloading transaction lists from Commerce Bank') csvs = [] with CommerceBank(page, secrets) as c, ntfyerror( 'Commerce Bank', page ) as r: for name, key in ACCOUNTS['commerce'].items(): try: start_date = get_last_transaction_date(key, token) except (OSError, ValueError) as e: log.error( 'Skipping account %s: could not get last transaction: %s', name, e, ) continue if start_date > end_date: log.info( 'Skipping account %s: last transaction was %s', name, start_date, ) continue log.info( 'Getting transactions since %s for account xxx%s', start_date, name, ) c.login() c.open_account(name) csvs.append((key, c.download_transactions(start_date, end_date))) log.info('Importing transactions from Commerce Bank into Firefly III') for key, csv in csvs: c.firefly_import(csv, key, importer) return r.success class ntfyerror: def __init__(self, bank: str, page: Page) -> None: self.bank = bank self.page = page self.success = True def __enter__(self) -> 'ntfyerror': return self def __exit__( self, exc_type: Optional[Type[Exception]], exc_value: Optional[Exception], tb: Optional[TracebackType], ) -> bool: if exc_type and exc_value and tb: self.success = False log.exception( 'Swallowed exception:', exc_info=(exc_type, exc_value, tb) ) if os.environ.get('DEBUG_NTFY', '1') == '0': return True if ss := self.page.screenshot(): save_screenshot(ss) ntfy( message=str(exc_value), title=f'xactfetch failed for {self.bank}', tags='warning', attach=ss, filename='screenshot.png', ) return True def save_screenshot(screenshot: bytes): now = datetime.datetime.now() filename = now.strftime('screenshot_%Y%m%d%H%M%S.png') log.debug('Saving browser screenshot to %s', filename) try: with open(filename, 'wb') as f: f.write(screenshot) except Exception as e: log.error('Failed to save browser screenshot: %s', e) else: log.info('Browser screenshot saved as %s', filename) class CommerceBank: URL = 'https://banking.commercebank.com/CBI/Auth/Login' IMPORT_CONFIG = { 'version': 3, 'source': 'fidi-1.2.2', 'created_at': '2023-04-27T08:05:10-05:00', 'date': 'n/j/Y', 'delimiter': 'comma', 'headers': True, 'rules': True, 'skip_form': False, 'add_import_tag': True, 'roles': [ 'date_transaction', 'internal_reference', 'description', 'amount_debit', 'amount_credit', ], 'do_mapping': [ False, False, False, False, False, ], 'mapping': [], 'duplicate_detection_method': 'classic', 'ignore_duplicate_lines': False, 'unique_column_index': 0, 'unique_column_type': 'internal_reference', 'flow': 'file', 'identifier': '0', 'connection': '0', 'ignore_spectre_categories': False, 'map_all_data': False, 'accounts': [], 'date_range': '', 'date_range_number': 30, 'date_range_unit': 'd', 'date_not_before': '', 'date_not_after': '', 'nordigen_country': '', 'nordigen_bank': '', 'nordigen_requisitions': [], 'nordigen_max_days': '90', 'conversion': False, 'ignore_duplicate_transactions': True, } def __init__(self, page: Page, secrets: SecretsClient) -> None: self.page = page self.secrets = secrets self._logged_in = False def __enter__(self) -> 'CommerceBank': return self def __exit__( self, exc_type: Optional[Type[Exception]], exc_value: Optional[Exception], tb: Optional[TracebackType], ) -> None: self.logout() def login(self) -> None: if self._logged_in: return log.debug('Navigating to %s', self.URL) self.page.goto(self.URL) username = self.secrets.get_secret('bank.commerce.username').decode() password = self.secrets.get_secret('bank.commerce.password').decode() log.debug('Filling username/password login form') self.page.get_by_role('textbox', name='Customer ID').fill(username) self.page.get_by_role('textbox', name='Password').fill(password) self.page.get_by_role('button', name='Log In').click() log.debug('Waiting for OTP 2FA form') otp_input = self.page.locator('id=securityCodeInput') otp_input.wait_for() self.page.wait_for_timeout(random.randint(1000, 3000)) log.debug('Filling OTP 2FA form') otp = self.secrets.get_secret('bank.commerce.otp').decode() otp_input.fill(otp) with self.page.expect_event('load'): self.page.get_by_role('button', name='Continue').click() log.debug('Waiting for page load') self.page.wait_for_load_state() cur_url = urllib.parse.urlparse(self.page.url) if cur_url.path != '/CBI/Accounts/Summary': new_url = cur_url._replace(path='/CBI/Accounts/Summary', query='') self.page.goto(urllib.parse.urlunparse(new_url)) log.info('Successfully logged in to Commerce Bank') self._logged_in = True def logout(self) -> None: if not self._logged_in: return log.debug('Logging out of Commerce Bank') with self.page.expect_event('load'): self.page.get_by_test_id('navWrap').get_by_text('Logout').click() log.info('Logged out of Commerce Bank') def open_account(self, account: str) -> None: log.debug('Navigating to activity page for account %s', account) if '/Activity/' in self.page.url: self.page.get_by_role('button', name='My Accounts').click() with self.page.expect_event('load'): self.page.get_by_role('link', name=account).click() self.page.wait_for_load_state() self.page.wait_for_timeout(random.randint(1000, 3000)) log.info('Loaded activity page for account %s', account) def download_transactions( self, from_date: datetime.date, to_date: datetime.date ) -> Path: log.info('Downloading transactions from %s to %s', from_date, to_date) datefmt = '%m/%d/%Y' self.page.get_by_role('link', name='Download Transactions').click() self.page.wait_for_timeout(random.randint(750, 1250)) modal = self.page.locator('#download-transactions') input_from = modal.locator('input[data-qaid=fromDate]') input_from.click() self.page.keyboard.press('Control+A') self.page.keyboard.press('Delete') self.page.keyboard.type(from_date.strftime(datefmt)) input_to = modal.locator('input[data-qaid=toDate]') input_to.click() self.page.keyboard.press('Control+A') self.page.keyboard.press('Delete') self.page.keyboard.type(to_date.strftime(datefmt)) modal.get_by_role('button', name='Select Type').click() self.page.get_by_text('Comma Separated').click() with self.page.expect_download() as di: self.page.get_by_role('button', name='Download').click() log.debug('Waiting for download to complete') path = di.value.path() assert path log.info('Downloaded transactions to %s', path) modal.get_by_label('Close').click() return path def firefly_import( self, csv: Path, account: int, importer: FireflyImporter ) -> None: config = copy.deepcopy(self.IMPORT_CONFIG) config['default_account'] = account importer.import_csv(csv, config) class Chase: URL = 'https://secure26ea.chase.com/web/auth/dashboard' IMPORT_CONFIG = { 'version': 3, 'source': 'fidi-1.2.2', 'created_at': '2023-04-27T09:54:42-05:00', 'date': 'n/j/Y', 'delimiter': 'comma', 'headers': True, 'rules': True, 'skip_form': False, 'add_import_tag': True, 'roles': [ '_ignore', 'date_transaction', 'date_process', 'description', 'tags-comma', '_ignore', 'amount', 'note', ], 'do_mapping': [ False, False, False, False, False, False, False, False, ], 'mapping': [], 'duplicate_detection_method': 'classic', 'ignore_duplicate_lines': True, 'unique_column_index': 0, 'unique_column_type': 'internal_reference', 'flow': 'file', 'identifier': '0', 'connection': '0', 'ignore_spectre_categories': False, 'map_all_data': True, 'accounts': [], 'date_range': '', 'date_range_number': 30, 'date_range_unit': 'd', 'date_not_before': '', 'date_not_after': '', 'nordigen_country': '', 'nordigen_bank': '', 'nordigen_requisitions': [], 'nordigen_max_days': '90', 'conversion': False, 'ignore_duplicate_transactions': True, } def __init__(self, page: Page, secrets: SecretsClient) -> None: self.page = page self.secrets = secrets self.saved_cookies = Path('cookies.json') self._logged_in = False def __enter__(self) -> 'Chase': self.load_cookies() return self def __exit__( self, exc_type: Optional[Type[Exception]], exc_value: Optional[Exception], tb: Optional[TracebackType], ) -> None: try: self.logout() finally: self.save_cookies() def load_cookies(self) -> None: log.debug('Loading saved cookies from %s', self.saved_cookies) try: with self.saved_cookies.open(encoding='utf-8') as f: self.page.context.add_cookies(json.load(f)) except: log.warning( 'Could not load saved cookies, ' 'SMS verification will be required!' ) else: log.info('Successfully loaded saved cookies') def save_cookies(self) -> None: log.debug('Saving cookies from %s', self.saved_cookies) try: with self.saved_cookies.open('w', encoding='utf-8') as f: f.write(json.dumps(self.page.context.cookies())) except Exception as e: log.error('Failed to save cookies: %s', e) else: log.info('Successfully saved cookies to %s', self.saved_cookies) def login(self) -> None: if self._logged_in: return log.debug('Navigating to %s', self.URL) self.page.goto(self.URL) self.page.wait_for_load_state() self.page.wait_for_timeout(random.randint(2000, 4000)) username = self.secrets.get_secret('bank.chase.username').decode() password = self.secrets.get_secret('bank.chase.password').decode() log.debug('Filling username/password login form') logonbox = self.page.frame_locator('#logonbox') logonbox.get_by_label('Username').fill(username) logonbox.get_by_label('Password').fill(password) self.page.wait_for_timeout(random.randint(500, 750)) logonbox.get_by_role('button', name='Sign in').click() log.debug('Waiting for page load') self.page.wait_for_load_state() self.page.get_by_role('button', name='Pay Card').wait_for( timeout=120000 ) log.info('Successfully logged in to Chase') self._logged_in = True def download_transactions( self, from_date: datetime.date, to_date: datetime.date ) -> Path: log.info('Downloading transactions from %s to %s', from_date, to_date) fmt = '%m/%d/%Y' self.page.locator('#CARD_ACCOUNTS').get_by_role( 'button', name='(...2467)' ).first.click() fl = self.page.locator('#flyout') fl.wait_for() fl.get_by_role('button', name='Pay card', exact=True).wait_for() fl.get_by_role( 'button', name='Account activity', exact=True ).wait_for() fl.get_by_role('link', name='Show details').wait_for() fl.get_by_role('link', name='Activity since last statement').click() fl.get_by_role('link', name='All transactions').click() fl.get_by_text('See more activity').wait_for() fl.get_by_role('button', name='Download Account Activity').click() log.debug('Filling account activity download form') self.page.locator('#select-downloadActivityOptionId-label').click() self.page.get_by_text('Choose a date range').nth(1).locator( '../..' ).click() self.page.wait_for_timeout(random.randint(500, 1500)) self.page.locator('#accountActivityFromDate-input-input').fill( from_date.strftime(fmt) ) self.page.locator('#accountActivityFromDate-input-input').blur() self.page.wait_for_timeout(random.randint(500, 1500)) self.page.locator('#accountActivityToDate-input-input').fill( to_date.strftime(fmt) ) self.page.locator('#accountActivityToDate-input-input').blur() self.page.wait_for_timeout(random.randint(500, 1500)) with self.page.expect_download(timeout=5000) as di: self.page.get_by_role( 'button', name='Download', exact=True ).click() log.debug('Waiting for download to complete') self.page.wait_for_timeout(random.randint(1000, 2500)) path = di.value.path() assert path log.info('Downloaded transactions to %s', path) return path def logout(self) -> None: if not self._logged_in: return log.debug('Logging out of Chase') with self.page.expect_event('load'): self.page.get_by_role('button', name='Sign out').click() log.info('Logged out of Chase') def firefly_import( self, csv: Path, account: int, importer: FireflyImporter ) -> None: config = copy.deepcopy(self.IMPORT_CONFIG) config['default_account'] = account with csv.open('r', encoding='utf-8') as f: headers = f.readline() if headers.startswith('Card'): log.debug('Detected CSV schema with Card column') elif headers.count(',') == 6: log.debug('Detected CSV schema without Card column') config['roles'].pop(0) config['do_mapping'].pop(0) else: raise ValueError(f'Unexpected CSV schema: {headers}') importer.import_csv(csv, config) def main() -> None: logging.basicConfig(level=logging.DEBUG) secrets = SecretsClient() secrets.connect() log.debug('Getting Firefly III access token') token = secrets.get_secret('firefly.token').decode() import_secret = secrets.get_secret('firefly.import.secret').decode() import_auth = ( secrets.get_secret('firefly.import.username').decode(), secrets.get_secret('firefly.import.password').decode(), ) importer = FireflyImporter( FIREFLY_III_IMPORTER_URL, import_secret, import_auth ) end_date = datetime.date.today() - datetime.timedelta(days=1) with sync_playwright() as pw, secrets: headless = os.environ.get('DEBUG_HEADLESS_BROWSER', '1') == '1' browser = pw.firefox.launch(headless=headless) page = browser.new_page() failed = False banks = sys.argv[1:] or list(ACCOUNTS.keys()) if 'commerce' in banks: if not download_commerce(page, secrets, end_date, token, importer): failed = True if 'chase' in banks: if not download_chase(page, secrets, end_date, token, importer): failed = True raise SystemExit(1 if failed else 0) if __name__ == '__main__': main()