178 lines
		
	
	
		
			5.5 KiB
		
	
	
	
		
			Python
		
	
	
		
			Executable file
		
	
	
	
	
			
		
		
	
	
			178 lines
		
	
	
		
			5.5 KiB
		
	
	
	
		
			Python
		
	
	
		
			Executable file
		
	
	
	
	
#!/usr/bin/env python3
 | 
						|
 | 
						|
"""
 | 
						|
A shell script that converts one currency to another.
 | 
						|
 | 
						|
Usage: shekel [-rh] AMOUNT FROM TO [TO ...]
 | 
						|
 | 
						|
FROM and TO are either currency codes or a fuzzy substring (requires iterfzf and
 | 
						|
fzf).
 | 
						|
 | 
						|
  Examples:
 | 
						|
    shekel 200 USD CAD
 | 
						|
  Don't remember the currency code? Use fuzzy search:
 | 
						|
    shekel 200 USD british
 | 
						|
  This will spawn an fzf session (requires the iterfzf package and fzf)
 | 
						|
 | 
						|
Requires: iterfzf (optional)
 | 
						|
"""
 | 
						|
 | 
						|
import argparse
 | 
						|
import sys
 | 
						|
import json
 | 
						|
from urllib import request
 | 
						|
from datetime import datetime
 | 
						|
from pathlib import Path
 | 
						|
 | 
						|
try:
 | 
						|
    from iterfzf import iterfzf
 | 
						|
except ImportError:
 | 
						|
    pass
 | 
						|
 | 
						|
 | 
						|
API_KEY = "92a7ccb3b6038a7151fc15e99f2de83d"  # please use your own!
 | 
						|
API_URL = "http://apilayer.net/api/live?access_key={}¤cies={},{}"
 | 
						|
DB_URL = "https://gist.githubusercontent.com/hlissner/f843823949d7c7453b51c6b52b2e1c16/raw/15a1bc7590be40cd78acfd133ecededbc0871610/currencies.json"
 | 
						|
 | 
						|
DEFAULT_CODES = ('CAD', 'USD')
 | 
						|
CACHE_FILE = "/tmp/shekel-hist"
 | 
						|
CACHE_TTL = 3600
 | 
						|
 | 
						|
class CurrencyCache():
 | 
						|
    def __init__(self, path=CACHE_FILE, ttl=CACHE_TTL, db_url=DB_URL):
 | 
						|
        self.path = Path(path)
 | 
						|
        self.ttl = ttl
 | 
						|
 | 
						|
        cache = {}
 | 
						|
        if self.path.exists():
 | 
						|
            cache = json.loads(self.path.read_text())
 | 
						|
        if "_DB" not in cache:
 | 
						|
            cache['_DB'] = json.loads(request.urlopen(db_url).read().decode('utf-8'))
 | 
						|
        self.cache = cache
 | 
						|
        self.data = self.cache['_DB']
 | 
						|
 | 
						|
    def __enter__(self):
 | 
						|
        return self
 | 
						|
 | 
						|
    def __exit__(self, *args):
 | 
						|
        self.path.write_text(json.dumps(self.cache))
 | 
						|
 | 
						|
    def clear(self):
 | 
						|
        db = self.cache['_DB']
 | 
						|
        self.cache.clear()
 | 
						|
        self.cache['_DB'] = db
 | 
						|
 | 
						|
    def store(self, from_code, to_code, rate):
 | 
						|
        self._set("-".join([from_code, to_code]), rate)
 | 
						|
        self._set("-".join([to_code, from_code]), 1 / rate)
 | 
						|
 | 
						|
    def get(self, from_code, to_code):
 | 
						|
        key = "-".join([from_code, to_code])
 | 
						|
        if key in self.cache:
 | 
						|
            return self.cache[key]['value']
 | 
						|
        return None
 | 
						|
 | 
						|
    def _set(self, key, value):
 | 
						|
        self.cache[key] = {
 | 
						|
            'time': datetime.now().timestamp(),
 | 
						|
            'value': value
 | 
						|
        }
 | 
						|
 | 
						|
 | 
						|
def _request_currency_rate(url=API_URL, *args):
 | 
						|
    """Fetch currency rates from remote API."""
 | 
						|
    real_url = url.format(*map(request.quote, map(str, args)))
 | 
						|
    data = json.loads(request.urlopen(real_url).read().decode('utf-8'))
 | 
						|
    if not data['success']:
 | 
						|
        raise Exception(data['error']['info'])
 | 
						|
    return data["quotes"]
 | 
						|
 | 
						|
 | 
						|
def format_currency(curr):
 | 
						|
    """Format a currency entry for the FZF prompt."""
 | 
						|
    return "{}  {}".format(curr['id'], curr['currencyName'])
 | 
						|
 | 
						|
 | 
						|
def parse(query, db={}):
 | 
						|
    """
 | 
						|
    Parses QUERY to see if it's an ISO 4217 currency code. Otherwise, treat it
 | 
						|
    like a fuzzy query and prompt the user with a completion list of currencies.
 | 
						|
 | 
						|
    :param query: An ISO 4217 currency code or a search query
 | 
						|
    :param db: The currency database to use
 | 
						|
    :returns: An ISO 4217 currency code
 | 
						|
    """
 | 
						|
    if len(query) == 3 and query.isupper():
 | 
						|
        return query
 | 
						|
    code = iterfzf(map(format_currency, db), query=query)
 | 
						|
    if code is None:
 | 
						|
        raise KeyboardInterrupt
 | 
						|
    return code.split("  ")[0]
 | 
						|
 | 
						|
 | 
						|
def currency_rate(from_code, to_code):
 | 
						|
    """
 | 
						|
    Retrieves the exchange rate between two currencies.
 | 
						|
 | 
						|
    :param from_code: an ISO 4217 currency code of the source currency
 | 
						|
    :param to_code: an ISO 4217 currency code of the destination currency
 | 
						|
    :returns: an exchange rate (a float)
 | 
						|
    """
 | 
						|
    if from_code == to_code:
 | 
						|
        return 1.0
 | 
						|
    elif not from_code or not to_code:
 | 
						|
        return None
 | 
						|
    rates = _request_currency_rate(API_URL, API_KEY, from_code, to_code)
 | 
						|
    return (1 / rates["USD" + from_code]) * rates["USD" + to_code]
 | 
						|
 | 
						|
 | 
						|
def each_currency_rate(from_code, to_codes, refresh=False):
 | 
						|
    """
 | 
						|
    Returns a cached generator for each exchange rate.
 | 
						|
 | 
						|
    :param from_code: the ISO 4217 currency code of the source currency
 | 
						|
    :param to_codes: a list of ISO 4217 currency codes to convert to
 | 
						|
    :yields: a tuple containg (TO_CODE, EXCHANGE_RATE)
 | 
						|
    """
 | 
						|
    with CurrencyCache(path=CACHE_FILE, ttl=CACHE_TTL, db_url=DB_URL) as cache:
 | 
						|
        if refresh:
 | 
						|
            cache.clear()
 | 
						|
        for code in to_codes:
 | 
						|
            code = parse(code, db=cache.data)
 | 
						|
            rate = cache.get(from_code, code)
 | 
						|
            if not rate:
 | 
						|
                rate = currency_rate(from_code, code)
 | 
						|
                cache.store(from_code, code, rate)
 | 
						|
            if not rate:
 | 
						|
                yield (None, None)
 | 
						|
            else:
 | 
						|
                yield (code, rate)
 | 
						|
 | 
						|
 | 
						|
def main():
 | 
						|
    opts = argparse.ArgumentParser(
 | 
						|
        description="Convert one amount of currency into another, or multiple.")
 | 
						|
    opts.add_argument('amount', type=float, help="The amount to convert")
 | 
						|
    opts.add_argument('src', metavar='from', type=str,
 | 
						|
                      help="Currency to convert from")
 | 
						|
    opts.add_argument('dest', metavar='to', type=str, nargs='+',
 | 
						|
                      help="Currency(ies) to convert to")
 | 
						|
    opts.add_argument('-r', '--refresh', action='store_true',
 | 
						|
                      help="Refresh the cache")
 | 
						|
    args = opts.parse_args()
 | 
						|
 | 
						|
    try:
 | 
						|
        for code, rate in each_currency_rate(args.src, args.dest, refresh=args.refresh):
 | 
						|
            if code:
 | 
						|
                print("{:.4f} {}".format(rate * args.amount, code))
 | 
						|
            else:
 | 
						|
                print("Failed to convert {} to {}".format(args.src, code),
 | 
						|
                      file=sys.stderr)
 | 
						|
    except KeyboardInterrupt:
 | 
						|
        print("Aborted!")
 | 
						|
        sys.exit(1)
 | 
						|
 | 
						|
 | 
						|
if __name__ == '__main__':
 | 
						|
    main()
 |