first-data-crawler-pdf/crawl.py

#!/usr/bin/env python3
# pylint: disable=missing-docstring,invalid-name
# pylint: disable=no-member # mechanize.Browser has some lazy-loading methods that pylint doesn't see
# 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 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(
    '--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 ['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
            if is_float(argv[0]):
                # amount first -> convert to currency in argv[1]
                print(fmt_and_calc(
                    cur=argv[1].upper(),
                    amt=float(argv[0]),
                    res=res,
                    direction=DIRECTION_FROM_EUR))
            elif is_float(argv[1]):
                # currency first -> convert to EUR
                print(fmt_and_calc(
                    cur=argv[0].upper(),
                    amt=float(argv[1]),
                    res=res,
                    direction=DIRECTION_TO_EUR))
        else:
            print("Not implemented: '" + " ".join(argv) + "'")
    except IndexError:
        if argv is None:
            pass
        else:
            print("Too few arguments: '" + " ".join(argv) + "'")

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 / (retrieve_date.strftime('%Y%m%d') + '.pdf')
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()
else:
    print(fmt_and_calc(args.amt, args.currency, results, direction))