first-data-crawler-pdf/crawl.py

#!/usr/bin/env python3
# pylint: disable=missing-docstring,invalid-name
# import logging
import argparse
import sys
import os
import pathlib
from datetime import date as DTDate
from datetime import datetime as DTDateTime
import appdirs
import csv

import utils


DIRECTION_TO_EUR = 0
DIRECTION_FROM_EUR = 1

# Argument parsing
parser = argparse.ArgumentParser(
    description='Currency conversion using First Data cards.')
parser.add_argument(
    '-t', '--card-type',
    # argument_default='VISA', # UnsupportedError
    choices=['VISA', 'MC'],
    dest='card_type',
    type=str,
    help='Card Type'
)
parser.add_argument(
    '-g', '--fetch-date',
    dest='date',
    type=str,
    help='Date to get values for (default: yesterday, Friday on Sat-Mon)'
)
parser.add_argument(
    '-r', '--direction',
    dest='reverse',
    action='store_true',
    help='Reverse direction (EUR -> currency)'
)
parser.add_argument(
    '-c', '--csv',
    dest='csv',
    action='store_true',
    help='Write the results to stdout as CSV'
)
parser.add_argument(
    '--cache-dir',
    dest='cache_dir',
    type=str,
    help='Override the default cache directory with your own path'
)
exc_group = parser.add_mutually_exclusive_group()
exc_group.add_argument(
    '-i', '--interactive',
    dest='interactive',
    action='store_true',
    help='Calculate interactively on stdin'
)
vals_group = exc_group.add_argument_group()

vals_group.add_argument(
    'currency',
    type=str,
    help='Currency abbreviation to convert from/to (e.g. EUR)',
    nargs='?'
)
vals_group.add_argument(
    'amt',
    type=float,
    help='Amount',
    nargs='?'
)


def _process_stdin(argv: str, res: utils.CurrencyResult) -> None:
    argv = argv.split()
    try:
        if argv[0] in ['h', 'man', 'help']:
            print("""Usage:
q | exit | quit: Quit the program.
AMOUNT CURRENCY: Convert AMOUNT euros to CURRENCY.
CURRENCY AMOUNT: Convert AMOUNT CURRENCY to euros.
d | date: Print the date which the data is from.
            """)
        elif argv[0] in ['q', 'exit', 'quit']:
            sys.exit()
        elif argv[0] in ['date', 'd']:
            print(res.date)
        elif len(argv[0]) == 3 or len(argv[1]) == 3:
            # more than 3 letter abbreviations are invalid
            cur_use = str
            amt_use = float
            dir_use = int
            if len(argv) == 1 and not is_float(argv[0]):
                # only currency specified
                cur_use = argv[0].upper()
                amt_use = 1
                dir_use = DIRECTION_TO_EUR
            elif is_float(argv[0]):
                # amount first -> convert EUR to currency in argv[1]
                cur_use = argv[1].upper()
                amt_use = float(argv[0])
                dir_use = DIRECTION_FROM_EUR
            elif is_float(argv[1]):
                # currency first -> convert argv[1] to EUR
                cur_use = argv[0].upper()
                amt_use = float(argv[1])
                dir_use = DIRECTION_TO_EUR
            else:
                raise ValueError

            # check that the currency exists
            if cur_use not in res.rates:
                raise ValueError
            else:
                print(fmt_and_calc(
                    cur=cur_use,
                    amt=amt_use,
                    res=res,
                    direction=dir_use
                ))
        else:
            print("Not implemented: '" + " ".join(argv) + "'")
    except IndexError:
        if argv is None:
            pass
        else:
            print("Too few arguments: '" + " ".join(argv) + "'")
    except ValueError:
        print("The currency specified does not exist.")


def is_float(string: str) -> bool:
    try:
        float(string)
        return True
    except ValueError:
        return False


def _parse_date_from_args(date_str: str) -> DTDate:
    return DTDateTime.strptime(date_str).date()


def calc_result(amt: float, rate: utils.Rate, direction: int, duty: float = 0) -> float:
    if direction == DIRECTION_FROM_EUR:
        result = amt * rate.ask / 1+duty
    elif direction == DIRECTION_TO_EUR:
        result = amt / rate.bid * 1+duty
    else:
        raise ValueError(
            'direction must be DIRECTION_FROM_EUR or DIRECTION_TO_EUR')
    return result


def fmt_and_calc(amt: float, cur: str, res: utils.CurrencyResult, direction: str) -> str:
    cur = cur.upper()
    if cur in res.rates:
        numeric_result = calc_result(amt, res.rates[cur], direction)
        if direction == DIRECTION_FROM_EUR:
            fmt_vals = ('EUR', round(amt, 2), cur, round(numeric_result, 2))
        else:
            fmt_vals = (cur, round(amt, 2), 'EUR', round(numeric_result, 2))
        return '%s %0.2f = %s %0.2f' % fmt_vals
    else:
        return 'Currency %s could not be found' % cur


# args = parser.parse_args('USD 1000'.split())
args = parser.parse_args()
#logger = logging.getLogger('mechanize')
# logger.addHandler(logging.StreamHandler(sys.stdout))
# logger.setLevel(logging.DEBUG)

# determine card type
if args.card_type == 'VISA':
    use_card_type = utils.CARD_VISA
elif args.card_type == 'MC':
    use_card_type = utils.CARD_MASTERCARD
elif args.card_type is None:
    use_card_type = utils.CARD_VISA
else:
    sys.exit('Unsupported card type ' + args.card_type)
if args.reverse:
    direction = DIRECTION_FROM_EUR
else:
    direction = DIRECTION_TO_EUR
if args.date:
    retrieve_date = _parse_date_from_args(args.date)
else:
    retrieve_date = utils.get_date()

if args.cache_dir is not None:
    filepath = pathlib.Path(args.cache_dir).resolve()
else:
    filepath = pathlib.Path(appdirs.user_cache_dir(
        'FirstDataCrawler', 'iwonder'))
if not filepath.exists():
    filepath.mkdir(parents=True)
filename = filepath / utils.mk_filename(retrieve_date, use_card_type)
if os.path.exists(filename):
    with open(filename, 'rb') as f:
        results = utils.get_results_from_pdf(f)
else:
    buf = utils.get_fileio(retrieve_date, card_type=use_card_type)
    with open(filename, 'wb') as f:
        f.write(buf.read())
        buf.seek(0)
    results = utils.get_results_from_pdf(buf)
#
# processing
#

if args.interactive:
    try:
        while True:
            _process_stdin(input('> '), results)
    except (KeyboardInterrupt, EOFError):
        sys.exit()
elif args.csv:
    w = csv.writer(sys.stdout)
    for rate in results.rates.values():
        w.writerow([rate.abbr, rate.full_name, rate.ask, rate.bid, rate.date])
else:
    print(fmt_and_calc(args.amt, args.currency, results, direction))