1
0
Fork 0

Compare commits

..

No commits in common. "ca8bff8fc553d87923d9a359cab435e884ec01bc" and "43bf08eae81e89c63180a53fbefa886307dc2c2f" have entirely different histories.

1 changed files with 22 additions and 74 deletions

View File

@ -1,4 +1,3 @@
import base64
import copy
import datetime
import json
@ -9,7 +8,6 @@ import shlex
import shutil
import subprocess
import tempfile
import urllib.parse
from pathlib import Path
from types import TracebackType
from typing import Any, Optional, Type
@ -53,13 +51,7 @@ def ntfy(
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
headers['Message'] = message.replace('\n', '\\n')
r = requests.put(
url,
headers=headers,
@ -125,13 +117,6 @@ def rbw_code(
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 firefly_import(csv: Path, config: dict[str, Any], token: str) -> None:
log.debug('Importing transactions from %s to Firefly III', csv)
env = {
@ -397,10 +382,6 @@ class CommerceBank:
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
@ -426,29 +407,21 @@ class CommerceBank:
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()
idx = self.page.url.rstrip('/').split('/')[-1]
href = (
f'Download.ashx?Index={idx}'
f'&From={from_date}&To={to_date}'
f'&Type=csv'
'&DurationOfMonths=6'
)
log.debug('Navigating to %s', href)
with self.page.expect_download() as di:
self.page.get_by_role('button', name='Download').click()
self.page.evaluate(f'window.location.href = "{href}";')
log.debug('Waiting for download to complete')
self.page.wait_for_timeout(random.randint(1000, 3000))
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, token: str) -> None:
@ -470,7 +443,6 @@ class Chase:
'skip_form': False,
'add_import_tag': True,
'roles': [
'_ignore',
'date_transaction',
'date_process',
'description',
@ -479,16 +451,7 @@ class Chase:
'amount',
'note',
],
'do_mapping': [
False,
False,
False,
False,
False,
False,
False,
False,
],
'do_mapping': [False, False, False, True, False, False, False],
'mapping': [],
'duplicate_detection_method': 'classic',
'ignore_duplicate_lines': True,
@ -578,9 +541,8 @@ class Chase:
).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
)
self.page.get_by_text('Amazon Rewards points').wait_for(timeout=60000)
self.page.get_by_role('button', name='Open an account').wait_for()
log.info('Successfully logged in to Chase')
self._logged_in = True
@ -589,19 +551,15 @@ class Chase:
) -> 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()
href = '#/dashboard/accountDetails/downloadAccountTransactions/index'
self.page.evaluate(f'window.location.href = "{href}";')
log.debug('Waiting for page to load')
s = self.page.locator('button#select-downloadActivityOptionId')
s.wait_for()
log.debug('Filling account activity download form')
self.page.locator('#select-downloadActivityOptionId-label').click()
self.page.locator('button#select-account-selector').click()
self.page.get_by_text('CREDIT CARD').nth(1).locator('../..').click()
s.click()
self.page.get_by_text('Choose a date range').nth(1).locator(
'../..'
).click()
@ -638,16 +596,6 @@ class Chase:
def firefly_import(self, csv: Path, account: int, token: str) -> 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}')
firefly_import(csv, config, token)