amex_json_download.py

#!/usr/bin/env python3
import os
import re
import requests
from autologin import AutoLogin
from configparser import SafeConfigParser
from glob import glob

# Helper definitions

DIR = os.path.dirname(os.path.realpath(__file__))
ROOT = DIR + '/..'
INBOX = ROOT + '/Inbox'
ARCHIVE = ROOT + '/Sources/Liabilities/Wilson/NL/AMEX'

def not_already_downloaded(statement_date):
  return len(glob(f"{INBOX}/{statement_date} - *.json"))   == 0 and \
         len(glob(f"{ARCHIVE}/{statement_date} - *.json")) == 0


# Read the configuration file
config = SafeConfigParser()
config.read(f"{DIR}/amex_download.ini")
loginURL = config['AMEX']['loginURL']
username = config['AMEX']['username']
password = config['AMEX']['password']


# Log into the AMEX Dashboard
al = AutoLogin()
print()
print(f"Logging into American Express Dashboard: {loginURL}")
cookies = al.auth_cookies_from_url(loginURL, username, password)

# Find the account_token, required for API requests
dashboard_response = requests.get('https://global.americanexpress.com/dashboard', cookies=cookies.jar)
token_regex = re.compile('account_token\\\\",\\\\"([^\\\\"]+)')
account_token = token_regex.findall(dashboard_response.text)[0]
headers = {'account_token': account_token}

# Download the current list of statement dates
statements_response = requests.get('https://global.americanexpress.com/account-data/v1/financials/statement_periods', cookies=cookies.jar, headers=headers)
statements = statements_response.json()

def download():
  first = True
  for statement in statements:
    statement_start_date = statement['statement_start_date']
    statement_end_date = statement['statement_end_date']
    
    # Always re-download the first statement, because it has the latest transactions
    if first or not_already_downloaded(statement_end_date):
      # Download the statement summary
      summary_url = f"https://global.americanexpress.com/account-data/v1/financials/transaction_summary?type=split_by_cardmember&statement_end_date={statement_end_date}"
      print(f"Downloading {statement_end_date} summary: {summary_url}")
      summary_response = requests.get(summary_url, cookies=cookies.jar, headers=headers)
      
      with open(f"{INBOX}/{statement_end_date} - Transaction summary since {statement_start_date}.json", "w") as text_file:
        text_file.write(summary_response.text)
      
      # Download the latest statement
      data_url = f"https://global.americanexpress.com/account-data/v1/financials/transactions?limit=1000&offset=0&statement_end_date={statement_end_date}&status=posted"
      print(f"Downloading {statement_end_date} transactions: {data_url}")
      data_response = requests.get(data_url, cookies=cookies.jar, headers=headers)

      with open(f"{INBOX}/{statement_end_date} - Transactions since {statement_start_date}.json", "w") as text_file:
        text_file.write(data_response.text)

      transactions = data_response.json()
      print()
      first = False

if __name__ == '__main__':
  download()