Compare commits

..

1 Commits

Author SHA1 Message Date
245b46135b wip 2020-04-28 16:23:53 +02:00
5 changed files with 36 additions and 84 deletions

View File

@ -1,7 +1,7 @@
# CLI for First Data conversion rates
## Requirements
* python3 with modules `PyPDF3`, `appdirs`, `mechanize`, `dateutil`
* python3 with modules `PyPDF3`, `appdirs`, `mechanize`
## Usage:
`python3 crawl.py [-t {VISA,MC}] [-g ISO_DATE] [-r] {-i | CURRENCY AMOUNT}`
@ -10,7 +10,6 @@
#### `AMOUNT`
This must be a number.
#### `CURRENCY`
This must be the three-letter currency abbreviation, case is irrelevant.
@ -23,29 +22,16 @@ Format: ISO date
#### `-r`, `--direction`
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|
|:---|:---|:---|:---|:---|
#### `-q`, `--quiet`
Do not output informational messages such as "Parsing..." or "Downloading..."
#### `-i`, `--interactive`
Calculate interactively on stdin
##### `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 the data is from.

0
backend.py Normal file
View File

View File

@ -8,7 +8,6 @@ import pathlib
from datetime import date as DTDate
from datetime import datetime as DTDateTime
import appdirs
import csv
import utils
@ -16,7 +15,7 @@ import utils
DIRECTION_TO_EUR = 0
DIRECTION_FROM_EUR = 1
# Argument parsing
## Argument parsing
parser = argparse.ArgumentParser(
description='Currency conversion using First Data cards.')
parser.add_argument(
@ -39,18 +38,6 @@ parser.add_argument(
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(
'-q', '--quiet',
dest='quiet',
action='store_true',
help='Do not output the \'Downloading...\' and \'Parsing...\' messages.'
)
parser.add_argument(
'--cache-dir',
dest='cache_dir',
@ -71,13 +58,13 @@ vals_group.add_argument(
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:
@ -116,7 +103,7 @@ d | date: Print the date which the data is from.
dir_use = DIRECTION_TO_EUR
else:
raise ValueError
# check that the currency exists
if cur_use not in res.rates:
raise ValueError
@ -130,9 +117,7 @@ d | date: Print the date which the data is from.
else:
print("Not implemented: '" + " ".join(argv) + "'")
except IndexError:
if argv is None:
pass
else:
if argv:
print("Too few arguments: '" + " ".join(argv) + "'")
except ValueError:
print("The currency specified does not exist.")
@ -147,8 +132,7 @@ def is_float(string: str) -> bool:
def _parse_date_from_args(date_str: str) -> DTDate:
from dateutil.parser import isoparse
return isoparse(date_str).date()
return DTDateTime.strptime(date_str).date()
def calc_result(amt: float, rate: utils.Rate, direction: int, duty: float = 0) -> float:
@ -157,8 +141,7 @@ def calc_result(amt: float, rate: utils.Rate, direction: int, duty: float = 0) -
elif direction == DIRECTION_TO_EUR:
result = amt / rate.bid * 1+duty
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
@ -174,12 +157,11 @@ def fmt_and_calc(amt: float, cur: str, res: utils.CurrencyResult, direction: str
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)
#logger.addHandler(logging.StreamHandler(sys.stdout))
#logger.setLevel(logging.DEBUG)
# determine card type
if args.card_type == 'VISA':
@ -202,36 +184,28 @@ else:
if args.cache_dir is not None:
filepath = pathlib.Path(args.cache_dir).resolve()
else:
filepath = pathlib.Path(appdirs.user_cache_dir(
'FirstDataCrawler', 'iwonder'))
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, quiet=args.quiet)
results = utils.get_results_from_pdf(f)
else:
buf = utils.get_fileio(retrieve_date, card_type=use_card_type, quiet=args.quiet)
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, quiet=args.quiet)
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:
if args.amt and args.currency:
print(fmt_and_calc(args.amt, args.currency, results, direction))
else:
print("Missing arguments. Exiting", file=sys.stderr)
exit(1)
print(fmt_and_calc(args.amt, args.currency, results, direction))

View File

@ -1,4 +1,4 @@
PyPDF3
appdirs
mechanize
python-dateutil
money-lib

View File

@ -7,23 +7,22 @@ from collections import namedtuple
from datetime import date as DTDate
from datetime import timedelta as DTTimeDelta
from datetime import datetime as DTDateTime
from typing import BinaryIO, List, Dict
from collections import OrderedDict
from sys import stderr
from typing import BinaryIO, List
import mechanize as m
import PyPDF3
from dateutil.relativedelta import FR, relativedelta
Rate = namedtuple('Rate', ['abbr', 'full_name', 'ask', 'bid', 'date'])
Rate = namedtuple('Rate', ['abbr', 'full_name', 'ask', 'bid'])
# Constants
CARD_MASTERCARD = ['0']
CARD_VISA = ['1']
class CurrencyResult:
def __init__(self):
self.rates = Dict[str, Rate]
self.rates = list()
self.card_type = str()
self.date = None
@ -64,7 +63,7 @@ def _array_remove_empty(obj: list) -> List[str]:
return obj
def _parse_line(line: str, ctx: CurrencyResult) -> Rate or None:
def _parse_line(line: str) -> Rate or None:
arr = line.split(" ") # 3 spaces = minimum separation in PDF
arr = _array_remove_empty(arr)
# process currency name
@ -73,14 +72,13 @@ def _parse_line(line: str, ctx: CurrencyResult) -> Rate or None:
abbr=names[0],
full_name=names[1].strip("()"),
ask=_parse_rate(arr[1]),
bid=_parse_rate(arr[2]),
date=ctx.date
bid=_parse_rate(arr[2])
)
return rate
def get_results_from_text(text: str, currency: str = None, quiet: bool = False) -> CurrencyResult:
rates = OrderedDict()
def get_results_from_text(text: str, currency: str = None) -> CurrencyResult:
rates = {}
result = CurrencyResult()
lines = text.splitlines()
# skip intro lines
@ -94,46 +92,43 @@ def get_results_from_text(text: str, currency: str = None, quiet: bool = False)
# now the rates begin
if currency is None:
for line in lines:
line_result = _parse_line(line, result)
line_result = _parse_line(line)
rates[line_result.abbr] = line_result
else:
pattern = re.compile("^"+currency)
for line in lines:
if pattern.match(line):
line_result = _parse_line(line, result)
line_result = _parse_line(line)
rates[line_result.abbr] = line_result
result.rates = rates
return result
def get_results_from_pdf(buf: BinaryIO or str, currency: str = None, quiet: bool = False) -> CurrencyResult:
if not quiet:
print('Parsing data... ', end='', file=stderr)
def get_results_from_pdf(buf: BinaryIO or str, currency: str = None) -> CurrencyResult:
print('Parsing data... ', end='')
reader = PyPDF3.PdfFileReader(buf)
text = str()
for num in range(0, reader.getNumPages()-1):
text += reader.getPage(num).extractText()
if not quiet:
print('Done.', file=stderr)
return get_results_from_text(text, currency=currency, quiet=quiet)
print('Done.')
return get_results_from_text(text, currency=currency)
def get_fileio(date: DTDate, card_type: List[str] = CARD_VISA, quiet: bool = False) -> BinaryIO: # pylint: disable=dangerous-default-value
def get_fileio(date: DTDate, card_type: List[str] = CARD_VISA) -> BinaryIO: # pylint: disable=dangerous-default-value
# pylint: disable=no-member # mechanize.Browser has some lazy-loading methods that pylint doesn't see
if not quiet:
print('Downloading rates for ' + date.strftime('%Y-%m-%d') + '... ', end='', file=stderr)
print('Downloading rates for ' + date.strftime('%Y-%m-%d') + '... ', end='')
b = m.Browser()
# Firefox 64 User-Agent
# ua = 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:64.0) Gecko/20100101 Firefox/64.0'
# b.set_header('User-Agent', ua)
# Ignore robots.txt
b.set_handle_robots(False)
# b.set_handle_robots(False)
# Debugging flags
# b.set_debug_http(True)
# b.set_debug_redirects(True)
# b.set_debug_responses(True)
# PDF URL
b.open('https://online.firstdata.com/CurrencyCalculator/fremdwaehrungskurse/pdf')
b.open('https://misc.firstdata.eu/CurrencyCalculator/fremdwaehrungskurse/pdf')
fm = b.forms()[0]
# This must be done because I can't change the options otherwise
fm.set_all_readonly(False)
@ -145,8 +140,7 @@ def get_fileio(date: DTDate, card_type: List[str] = CARD_VISA, quiet: bool = Fal
rq = fm.click(name='submitButton', coord=(random.randint(1, 114), random.randint(1, 20)))
rq.add_header('Accept', '*/*')
rp = b.retrieve(rq)
if not quiet:
print(' Done.', file=stderr)
print(' Done.')
# Returns an open file-like object with the PDF as contents
return open(rp[0], 'rb')
@ -170,5 +164,3 @@ def mk_filename(date: DTDate, card_type: List[str]) -> str:
else:
raise TypeError("not a valid card type")
return fn