Add CSV writing feature

This commit is contained in:
Christopher Teutsch 2020-07-18 20:22:33 +02:00
parent add06ef20d
commit 486c86766a
4 changed files with 51 additions and 20 deletions

View File

@ -1,7 +1,7 @@
# CLI for First Data conversion rates # CLI for First Data conversion rates
## Requirements ## Requirements
* python3 with modules `PyPDF3`, `appdirs`, `mechanize` * python3 with modules `PyPDF3`, `appdirs`, `mechanize`, `dateutil`
## Usage: ## Usage:
`python3 crawl.py [-t {VISA,MC}] [-g ISO_DATE] [-r] {-i | CURRENCY AMOUNT}` `python3 crawl.py [-t {VISA,MC}] [-g ISO_DATE] [-r] {-i | CURRENCY AMOUNT}`
@ -10,6 +10,7 @@
#### `AMOUNT` #### `AMOUNT`
This must be a number. This must be a number.
#### `CURRENCY` #### `CURRENCY`
This must be the three-letter currency abbreviation, case is irrelevant. This must be the three-letter currency abbreviation, case is irrelevant.
@ -22,16 +23,26 @@ Format: ISO date
#### `-r`, `--direction` #### `-r`, `--direction`
Reverse conversion direction (EUR to specified currency, instead of specified currency to EUR) Reverse conversion direction (EUR to specified currency, instead of specified currency to EUR)
#### `-c`, `--csv`
Write the currency results to standard output, formatted as CSV:
|ISO4217 abbreviation|Full German name|Asking rate|Bidding rate|Date the rate was valid on|
|:---|:---|:---|:---|:---|
#### `-i`, `--interactive` #### `-i`, `--interactive`
Calculate interactively on stdin Calculate interactively on stdin
##### `q`, `exit`, `quit` ##### `q`, `exit`, `quit`
Quit the program. Quit the program.
##### `AMOUNT CURRENCY` ##### `AMOUNT CURRENCY`
Convert AMOUNT euros to CURRENCY. Convert AMOUNT euros to CURRENCY.
##### `CURRENCY AMOUNT` ##### `CURRENCY AMOUNT`
Convert AMOUNT CURRENCY to euros. Convert AMOUNT CURRENCY to euros.
##### `d`, `date` ##### `d`, `date`
Print the date the data is from. Print the date the data is from.

View File

@ -8,6 +8,7 @@ import pathlib
from datetime import date as DTDate from datetime import date as DTDate
from datetime import datetime as DTDateTime from datetime import datetime as DTDateTime
import appdirs import appdirs
import csv
import utils import utils
@ -15,7 +16,7 @@ import utils
DIRECTION_TO_EUR = 0 DIRECTION_TO_EUR = 0
DIRECTION_FROM_EUR = 1 DIRECTION_FROM_EUR = 1
## Argument parsing # Argument parsing
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description='Currency conversion using First Data cards.') description='Currency conversion using First Data cards.')
parser.add_argument( parser.add_argument(
@ -38,6 +39,12 @@ parser.add_argument(
action='store_true', action='store_true',
help='Reverse direction (EUR -> currency)' 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( parser.add_argument(
'--cache-dir', '--cache-dir',
dest='cache_dir', dest='cache_dir',
@ -58,13 +65,13 @@ vals_group.add_argument(
type=str, type=str,
help='Currency abbreviation to convert from/to (e.g. EUR)', help='Currency abbreviation to convert from/to (e.g. EUR)',
nargs='?' nargs='?'
) )
vals_group.add_argument( vals_group.add_argument(
'amt', 'amt',
type=float, type=float,
help='Amount', help='Amount',
nargs='?' nargs='?'
) )
def _process_stdin(argv: str, res: utils.CurrencyResult) -> None: def _process_stdin(argv: str, res: utils.CurrencyResult) -> None:
@ -103,7 +110,7 @@ d | date: Print the date which the data is from.
dir_use = DIRECTION_TO_EUR dir_use = DIRECTION_TO_EUR
else: else:
raise ValueError raise ValueError
# check that the currency exists # check that the currency exists
if cur_use not in res.rates: if cur_use not in res.rates:
raise ValueError raise ValueError
@ -143,7 +150,8 @@ def calc_result(amt: float, rate: utils.Rate, direction: int, duty: float = 0) -
elif direction == DIRECTION_TO_EUR: elif direction == DIRECTION_TO_EUR:
result = amt / rate.bid * 1+duty result = amt / rate.bid * 1+duty
else: else:
raise ValueError('direction must be DIRECTION_FROM_EUR or DIRECTION_TO_EUR') raise ValueError(
'direction must be DIRECTION_FROM_EUR or DIRECTION_TO_EUR')
return result return result
@ -159,11 +167,12 @@ def fmt_and_calc(amt: float, cur: str, res: utils.CurrencyResult, direction: str
else: else:
return 'Currency %s could not be found' % cur return 'Currency %s could not be found' % cur
# args = parser.parse_args('USD 1000'.split()) # args = parser.parse_args('USD 1000'.split())
args = parser.parse_args() args = parser.parse_args()
#logger = logging.getLogger('mechanize') #logger = logging.getLogger('mechanize')
#logger.addHandler(logging.StreamHandler(sys.stdout)) # logger.addHandler(logging.StreamHandler(sys.stdout))
#logger.setLevel(logging.DEBUG) # logger.setLevel(logging.DEBUG)
# determine card type # determine card type
if args.card_type == 'VISA': if args.card_type == 'VISA':
@ -186,7 +195,8 @@ else:
if args.cache_dir is not None: if args.cache_dir is not None:
filepath = pathlib.Path(args.cache_dir).resolve() filepath = pathlib.Path(args.cache_dir).resolve()
else: else:
filepath = pathlib.Path(appdirs.user_cache_dir('FirstDataCrawler', 'iwonder')) filepath = pathlib.Path(appdirs.user_cache_dir(
'FirstDataCrawler', 'iwonder'))
if not filepath.exists(): if not filepath.exists():
filepath.mkdir(parents=True) filepath.mkdir(parents=True)
filename = filepath / utils.mk_filename(retrieve_date, use_card_type) filename = filepath / utils.mk_filename(retrieve_date, use_card_type)
@ -209,5 +219,9 @@ if args.interactive:
_process_stdin(input('> '), results) _process_stdin(input('> '), results)
except (KeyboardInterrupt, EOFError): except (KeyboardInterrupt, EOFError):
sys.exit() 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: else:
print(fmt_and_calc(args.amt, args.currency, results, direction)) print(fmt_and_calc(args.amt, args.currency, results, direction))

View File

@ -1,3 +1,4 @@
PyPDF3 PyPDF3
appdirs appdirs
mechanize mechanize
python-dateutil

View File

@ -7,13 +7,15 @@ from collections import namedtuple
from datetime import date as DTDate from datetime import date as DTDate
from datetime import timedelta as DTTimeDelta from datetime import timedelta as DTTimeDelta
from datetime import datetime as DTDateTime from datetime import datetime as DTDateTime
from typing import BinaryIO, List from typing import BinaryIO, List, Dict
from collections import OrderedDict
from sys import stderr
import mechanize as m import mechanize as m
import PyPDF3 import PyPDF3
from dateutil.relativedelta import FR, relativedelta from dateutil.relativedelta import FR, relativedelta
Rate = namedtuple('Rate', ['abbr', 'full_name', 'ask', 'bid']) Rate = namedtuple('Rate', ['abbr', 'full_name', 'ask', 'bid', 'date'])
# Constants # Constants
CARD_MASTERCARD = ['0'] CARD_MASTERCARD = ['0']
@ -22,7 +24,7 @@ CARD_VISA = ['1']
class CurrencyResult: class CurrencyResult:
def __init__(self): def __init__(self):
self.rates = list() self.rates = Dict[str, Rate]
self.card_type = str() self.card_type = str()
self.date = None self.date = None
@ -63,7 +65,7 @@ def _array_remove_empty(obj: list) -> List[str]:
return obj return obj
def _parse_line(line: str) -> Rate or None: def _parse_line(line: str, ctx: CurrencyResult) -> Rate or None:
arr = line.split(" ") # 3 spaces = minimum separation in PDF arr = line.split(" ") # 3 spaces = minimum separation in PDF
arr = _array_remove_empty(arr) arr = _array_remove_empty(arr)
# process currency name # process currency name
@ -72,13 +74,14 @@ def _parse_line(line: str) -> Rate or None:
abbr=names[0], abbr=names[0],
full_name=names[1].strip("()"), full_name=names[1].strip("()"),
ask=_parse_rate(arr[1]), ask=_parse_rate(arr[1]),
bid=_parse_rate(arr[2]) bid=_parse_rate(arr[2]),
date=ctx.date
) )
return rate return rate
def get_results_from_text(text: str, currency: str = None) -> CurrencyResult: def get_results_from_text(text: str, currency: str = None) -> CurrencyResult:
rates = {} rates = OrderedDict()
result = CurrencyResult() result = CurrencyResult()
lines = text.splitlines() lines = text.splitlines()
# skip intro lines # skip intro lines
@ -92,25 +95,25 @@ def get_results_from_text(text: str, currency: str = None) -> CurrencyResult:
# now the rates begin # now the rates begin
if currency is None: if currency is None:
for line in lines: for line in lines:
line_result = _parse_line(line) line_result = _parse_line(line, result)
rates[line_result.abbr] = line_result rates[line_result.abbr] = line_result
else: else:
pattern = re.compile("^"+currency) pattern = re.compile("^"+currency)
for line in lines: for line in lines:
if pattern.match(line): if pattern.match(line):
line_result = _parse_line(line) line_result = _parse_line(line, result)
rates[line_result.abbr] = line_result rates[line_result.abbr] = line_result
result.rates = rates result.rates = rates
return result return result
def get_results_from_pdf(buf: BinaryIO or str, currency: str = None) -> CurrencyResult: def get_results_from_pdf(buf: BinaryIO or str, currency: str = None) -> CurrencyResult:
print('Parsing data... ', end='') print('Parsing data... ', end='', file=stderr)
reader = PyPDF3.PdfFileReader(buf) reader = PyPDF3.PdfFileReader(buf)
text = str() text = str()
for num in range(0, reader.getNumPages()-1): for num in range(0, reader.getNumPages()-1):
text += reader.getPage(num).extractText() text += reader.getPage(num).extractText()
print('Done.') print('Done.', file=stderr)
return get_results_from_text(text, currency=currency) return get_results_from_text(text, currency=currency)
@ -164,3 +167,5 @@ def mk_filename(date: DTDate, card_type: List[str]) -> str:
else: else:
raise TypeError("not a valid card type") raise TypeError("not a valid card type")
return fn return fn