diff --git a/xactfetch.py b/xactfetch.py index 586bd57..5c70326 100644 --- a/xactfetch.py +++ b/xactfetch.py @@ -1,14 +1,12 @@ import base64 import copy import datetime +import getpass import json import logging import os import random -import shlex -import shutil import subprocess -import tempfile import urllib.parse from pathlib import Path from types import TracebackType @@ -25,6 +23,7 @@ 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, @@ -34,6 +33,42 @@ ACCOUNTS = { } +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, @@ -132,48 +167,11 @@ def rfc2047_base64encode( return f"=?UTF-8?B?{encoded}?=" -def firefly_import(csv: Path, config: dict[str, Any], token: str) -> None: - log.debug('Importing transactions from %s to Firefly III', csv) - env = { - 'PATH': os.environ['PATH'], - 'FIREFLY_III_ACCESS_TOKEN': token, - 'IMPORT_DIR_ALLOWLIST': '/import', - 'FIREFLY_III_URL': FIREFLY_III_URL, - 'WEB_SERVER': 'false', - } - with tempfile.TemporaryDirectory() as tmpdir: - dest = Path(tmpdir) / 'import.csv' - log.debug('Copying %s to %s', csv, dest) - shutil.copyfile(csv, dest) - configfile = dest.with_suffix('.json') - log.debug('Saving config as %s', configfile) - with configfile.open('w', encoding='utf-8') as f: - json.dump(config, f) - cmd = [ - 'podman', - 'run', - '--rm', - '-it', - '-v', - f'{tmpdir}:/import:ro,z', - '--env-host', - 'docker.io/fireflyiii/data-importer', - ] - if log.isEnabledFor(logging.DEBUG): - log.debug( - 'Running command: %s', - ' '.join(shlex.quote(str(a)) for a in cmd), - ) - if os.environ.get('DEBUG_SKIP_IMPORT'): - cmd = ['true'] - p = subprocess.run(cmd, env=env, check=False) - if p.returncode == 0: - log.info( - 'Successfully imported transactions from %s to Firefly III', - csv, - ) - else: - log.error('Failed to import transactions from %s') +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: @@ -204,7 +202,9 @@ def get_last_transaction_date(key: int, token: str) -> datetime.date: return last_date.date() + datetime.timedelta(days=1) -def download_chase(page: Page, end_date: datetime.date, token: str) -> bool: +def download_chase( + page: Page, end_date: datetime.date, token: str, importer: FireflyImporter +) -> bool: with Chase(page) as c, ntfyerror('Chase', page) as r: c.login() key = ACCOUNTS['chase'] @@ -224,11 +224,16 @@ def download_chase(page: Page, end_date: datetime.date, token: str) -> bool: return True csv = c.download_transactions(start_date, end_date) log.info('Importing transactions from Chase into Firefly III') - c.firefly_import(csv, key, token) + c.firefly_import(csv, key, importer) return r.success -def download_commerce(page: Page, end_date: datetime.date, token: str) -> bool: +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: @@ -259,7 +264,7 @@ def download_commerce(page: Page, end_date: datetime.date, token: str) -> bool: 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, token) + c.firefly_import(csv, key, importer) return r.success @@ -451,10 +456,12 @@ class CommerceBank: modal.get_by_label('Close').click() return path - def firefly_import(self, csv: Path, account: int, token: str) -> None: + def firefly_import( + self, csv: Path, account: int, importer: FireflyImporter + ) -> None: config = copy.deepcopy(self.IMPORT_CONFIG) config['default_account'] = account - firefly_import(csv, config, token) + importer.import_csv(csv, config) class Chase: @@ -635,7 +642,9 @@ class Chase: self.page.get_by_role('button', name='Sign out').click() log.info('Logged out of Chase') - def firefly_import(self, csv: Path, account: int, token: str) -> None: + 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: @@ -648,7 +657,7 @@ class Chase: config['do_mapping'].pop(0) else: raise ValueError(f'Unexpected CSV schema: {headers}') - firefly_import(csv, config, token) + importer.import_csv(csv, config) def main() -> None: @@ -660,15 +669,25 @@ def main() -> None: ) 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): - failed = True - if not download_chase(page, end_date, token): + 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)