703 lines
22 KiB
Python
703 lines
22 KiB
Python
import base64
|
|
import copy
|
|
import datetime
|
|
import getpass
|
|
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
import subprocess
|
|
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 = 'https://ntfy.pyrocufflink.net'
|
|
NTFY_TOPIC = 'dustin'
|
|
FIREFLY_III_URL = 'https://firefly.pyrocufflink.blue'
|
|
FIREFLY_III_IMPORTER_URL = 'https://dustin.import.firefly.pyrocufflink.blue'
|
|
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()
|
|
|
|
|
|
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 rbw_unlocked() -> bool:
|
|
log.debug('Checking if rbw vault is locked')
|
|
cmd = ['rbw', 'unlocked']
|
|
p = subprocess.run(cmd, check=False, stdout=subprocess.DEVNULL)
|
|
unlocked = p.returncode == 0
|
|
log.info('rbw vault is %s', 'unlocked' if unlocked else 'locked')
|
|
return unlocked
|
|
|
|
|
|
def rbw_get(
|
|
name: str, folder: Optional[str] = None, username: Optional[str] = None
|
|
) -> str:
|
|
log.info(
|
|
'Getting password for Bitwarden vault item '
|
|
'%s (folder: %s, username: %s)',
|
|
name,
|
|
folder,
|
|
username,
|
|
)
|
|
cmd = ['rbw', 'get']
|
|
if folder is not None:
|
|
cmd += ('--folder', folder)
|
|
cmd.append(name)
|
|
if username is not None:
|
|
cmd.append(username)
|
|
p = subprocess.run(cmd, check=True, capture_output=True, encoding='utf-8')
|
|
assert p.stdout is not None
|
|
return p.stdout.rstrip('\n')
|
|
|
|
|
|
def rbw_code(
|
|
name: str, folder: Optional[str] = None, username: Optional[str] = None
|
|
) -> str:
|
|
log.info(
|
|
'Getting OTP code for Bitwarden vault item '
|
|
'%s (folder: %s, username: %s)',
|
|
name,
|
|
folder,
|
|
username,
|
|
)
|
|
cmd = ['rbw', 'code']
|
|
if folder is not None:
|
|
cmd += ('--folder', folder)
|
|
cmd.append(name)
|
|
if username is not None:
|
|
cmd.append(username)
|
|
p = subprocess.run(cmd, check=True, capture_output=True, encoding='utf-8')
|
|
assert p.stdout is not None
|
|
return p.stdout.rstrip('\n')
|
|
|
|
|
|
def rfc2047_base64encode(
|
|
message: str,
|
|
) -> str:
|
|
encoded = base64.b64encode(message.encode("utf-8")).decode("ascii")
|
|
return f"=?UTF-8?B?{encoded}?="
|
|
|
|
|
|
def secret_from_file(env: str, default: str) -> str:
|
|
filename = os.environ.get(env, default)
|
|
log.debug('Loading secret value from %s', filename)
|
|
with open(filename, 'r', encoding='utf-8') as f:
|
|
return f.read().rstrip()
|
|
|
|
|
|
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, end_date: datetime.date, token: str, importer: FireflyImporter
|
|
) -> bool:
|
|
with Chase(page) 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,
|
|
end_date: datetime.date,
|
|
token: str,
|
|
importer: FireflyImporter,
|
|
) -> bool:
|
|
log.info('Downloading transaction lists from Commerce Bank')
|
|
csvs = []
|
|
with CommerceBank(page) 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) -> None:
|
|
self.page = page
|
|
self.username = 'admiraln3mo'
|
|
self.vault_item = 'Commerce Bank'
|
|
self.vault_folder = 'Websites'
|
|
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)
|
|
password = rbw_get(self.vault_item, self.vault_folder, self.username)
|
|
log.debug('Filling username/password login form')
|
|
self.page.get_by_role('textbox', name='Customer ID').fill(
|
|
self.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 = rbw_code(self.vault_item, self.vault_folder, self.username)
|
|
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) -> None:
|
|
self.page = page
|
|
self.username = 'AdmiralN3mo'
|
|
self.vault_item = 'Chase'
|
|
self.vault_folder = 'Websites'
|
|
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))
|
|
password = rbw_get(self.vault_item, self.vault_folder, self.username)
|
|
log.debug('Filling username/password login form')
|
|
self.page.frame_locator('#logonbox').locator(
|
|
'input[name=userId]'
|
|
).fill(self.username)
|
|
self.page.frame_locator('#logonbox').locator(
|
|
'input[name=password]'
|
|
).fill(password)
|
|
self.page.wait_for_timeout(random.randint(500, 750))
|
|
self.page.frame_locator('#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='CREDIT CARD (...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('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)
|
|
if not rbw_unlocked():
|
|
ntfy(
|
|
'xactfetch needs you to unlock the rbw vault',
|
|
tags='closed_lock_with_key',
|
|
)
|
|
log.debug('Getting Firefly III access token from rbw vault')
|
|
token = rbw_get('xactfetch')
|
|
import_secret = secret_from_file(
|
|
'FIREFLY_IMPORT_SECRET_FILE', 'import.secret'
|
|
)
|
|
import_auth = (
|
|
os.environ.get('FIREFLY_IMPORT_USER', getpass.getuser()),
|
|
secret_from_file('FIREFLY_IMPORT_PASSWORD_FILE', 'import.password'),
|
|
)
|
|
importer = FireflyImporter(
|
|
FIREFLY_III_IMPORTER_URL, import_secret, import_auth
|
|
)
|
|
end_date = datetime.date.today() - datetime.timedelta(days=1)
|
|
with sync_playwright() as pw:
|
|
headless = os.environ.get('DEBUG_HEADLESS_BROWSER', '1') == '1'
|
|
browser = pw.firefox.launch(headless=headless)
|
|
page = browser.new_page()
|
|
failed = False
|
|
if not download_commerce(page, end_date, token, importer):
|
|
failed = True
|
|
if not download_chase(page, end_date, token, importer):
|
|
failed = True
|
|
raise SystemExit(1 if failed else 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|