From 3d04b314efb71b80db3784febddaa9a7fa018c01 Mon Sep 17 00:00:00 2001 From: Henrik Lissner Date: Tue, 3 Oct 2017 19:05:00 +0200 Subject: [PATCH] Refactor python scripts --- bin/htmlent | 48 ++++------ bin/shekel | 247 +++++++++++++++++++++++++++++++--------------------- bin/time-in | 124 ++++++++++++++------------ bin/uri | 49 +++++------ 4 files changed, 251 insertions(+), 217 deletions(-) diff --git a/bin/htmlent b/bin/htmlent index e5d04d9c..0351aa10 100755 --- a/bin/htmlent +++ b/bin/htmlent @@ -1,16 +1,18 @@ #!/usr/bin/env python3 -# Encode/decode HTML entities in stdin or given file(s). +""" +Encode/decode HTML entities in stdin or given file(s). +""" -import getopt -import sys -import os +import argparse import fileinput import html from html.entities import entitydefs +from pathlib import Path ENTITIES = {value: key for key, value in entitydefs.items()} + def encode(text): t = "" for c in text: @@ -23,33 +25,19 @@ def encode(text): def decode(text): return html.unescape(text) -def usage(): - # TODO Improve this - print("Usage: {} [-edh] file1[ file2[ ...]]".format(os.path.basename(sys.argv[0]))) - if __name__ == '__main__': - COMMANDS = { - "-e": encode, - "-d": decode - } + opts = argparse.ArgumentParser(description="Encodes (or decodes) HTML entities in text.") + opts.add_argument('file', nargs="?", help="File to encode/decode") + opts.add_argument('-d', '--decode', action='store_true', help="Decode the given files/input") + args = opts.parse_args() - try: - optlist, args = getopt.getopt(sys.argv[1:], 'edh') - sys.argv = [sys.argv[0]] + args - except getopt.GetoptError as err: - print(err) - sys.exit(1) + if "file" in args and args.file: + text = Path(args.file).read_text() + else: + text = "\n".join([line for line in fileinput.input()]).strip() - action = encode - for o, a in optlist: - if o in COMMANDS: - action = COMMANDS[o] - elif o == '-h': - usage() - sys.exit() - else: - assert False, "unhandled option" - - input = "\n".join([line for line in fileinput.input()]).strip() - print(action(input)) + if args.decode: + print(decode(text)) + else: + print(encode(text)) diff --git a/bin/shekel b/bin/shekel index 7125e95b..41647c6d 100755 --- a/bin/shekel +++ b/bin/shekel @@ -1,130 +1,177 @@ #!/usr/bin/env python3 -# A shell script that converts one currency to another. -# -# Usage: shekel AMOUNT FROM 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) +""" +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 os import json from urllib import request from datetime import datetime -from base64 import b64encode +from pathlib import Path try: from iterfzf import iterfzf -except: +except ImportError: pass -API_KEY = "92a7ccb3b6038a7151fc15e99f2de83d" # please use your own! +API_KEY = "92a7ccb3b6038a7151fc15e99f2de83d" # please use your own! API_URL = "http://apilayer.net/api/live?access_key={}¤cies={},{}" -DB_URL = "https://raw.githubusercontent.com/hlissner/lb6-actions/master/Convert%20Currency.lbaction/Contents/Resources/currencies.json" +DB_URL = "https://gist.githubusercontent.com/hlissner/f843823949d7c7453b51c6b52b2e1c16/raw/15a1bc7590be40cd78acfd133ecededbc0871610/currencies.json" -DEFAULT_CODE = "CAD" -CACHE_FILE = os.path.expanduser("~/.cache/shekel-hist") +DEFAULT_CODES = ('CAD', 'USD') +CACHE_FILE = "/tmp/shekel-hist" CACHE_TTL = 3600 -CACHE = {} -def _request(url, *args): - try: - real_url = url.format(*map(request.quote, map(str, args))) - resp = request.urlopen(real_url).read().decode('utf-8') - data = json.loads(resp) - if data['success']: - return data["quotes"] - raise Exception(data['error']['info']) - except IOError: - print("IOError") +class CurrencyCache(): + def __init__(self, path=CACHE_FILE, ttl=CACHE_TTL, db_url=DB_URL): + self.path = Path(path) + self.ttl = ttl -def _cache(key, value): - CACHE[key] = { - 'time': datetime.now().timestamp(), - 'value': value - } + 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 cache(from_code, to_code, fn, ttl=CACHE_TTL): - key = "-".join([from_code, to_code]) - if key not in CACHE or CACHE[key]['time'] > CACHE[key]['time'] + ttl: - value = fn() - if value is not None: - _cache(key, value) - _cache("-".join([to_code, from_code]), value) - return CACHE[key]['value'] + def __enter__(self): + return self -def get_rate(from_code, to_code=DEFAULT_CODE): - if from_code == to_code: - raise ValueError("Trying to convert currency to itself ({})".format(from_code)) - elif from_code is None or to_code is None: + 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 - rates = cache( - from_code, - to_code, - lambda: _request(API_URL, API_KEY, from_code, to_code) - ) - src = rates["USD" + from_code] - dest = rates["USD" + to_code] - return (1 / src) * dest -def parse(codes, db): - for i, code in enumerate(codes): - if len(code) == 3 and code.isupper(): - continue - resp = iterfzf(currencies(), query=code) - if resp is None: - codes[i] = None - else: - codes[i] = resp.split(" ")[0] + def _set(self, key, value): + self.cache[key] = { + 'time': datetime.now().timestamp(), + 'value': value + } - return codes -def currencies(): - for curr in CACHE['_DB']: - yield "{} {}".format(curr['id'], curr['currencyName']) +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 usage(): - print("Usage: {} AMOUNT FROM [TO]".format(sys.argv[0])) -# -if __name__ == '__main__': - refresh = False - if "-f" in sys.argv: - refresh = True - sys.argv.remove('-f') +def format_currency(curr): + """Format a currency entry for the FZF prompt.""" + return "{} {}".format(curr['id'], curr['currencyName']) - if len(sys.argv) < 3: - usage() + +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 refresh: - CACHE = {} - else: - if os.path.isfile(CACHE_FILE): - with open(CACHE_FILE, 'r') as f: - CACHE = json.load(f) - if "_DB" not in CACHE: - CACHE['_DB'] = json.loads(request.urlopen(DB_URL).read().decode('utf-8')) - amount, codes = float(sys.argv[1]), parse(sys.argv[2:4], CACHE['_DB']) - rate = get_rate(*codes) - if rate is None: - print("Aborted") - else: - converted_amount = amount * rate - print("{:.6f} {}".format(converted_amount, codes[1])) - - with open(CACHE_FILE, 'w') as f: - json.dump(CACHE, f) - - sys.exit(0) +if __name__ == '__main__': + main() diff --git a/bin/time-in b/bin/time-in index eb611acb..5f1a1e61 100755 --- a/bin/time-in +++ b/bin/time-in @@ -1,18 +1,20 @@ #!/usr/bin/env python3 -# Retrieves the time (and time difference) at a location (can be an address, -# postal code, country name, etc). Uses Google Maps APIs to map locations to -# coordinates, and coordinates to timezones. -# -# Requires: pytz. tzlocal -# Usage: time-in Copenhagen "Toronto, ON" Vancouver +""" +Retrieves the time (and time difference) at a location (can be an address, +postal code, country name, etc). Uses Google Maps APIs to map locations to +coordinates, and coordinates to timezones. + +Requires: pytz, tzlocal +Usage: time-in Copenhagen "Toronto, ON" Vancouver +""" import sys -import os import json from base64 import b64encode from datetime import datetime from urllib import request +from pathlib import Path from pytz import timezone from tzlocal import get_localzone @@ -30,28 +32,23 @@ def _generate_url(url, *args): return url.format(*map(request.quote, map(str, args))) def _request(url, *args): - try: - real_url = url.format(*map(request.quote, map(str, args))) - resp = request.urlopen(real_url).read() - result = json.loads(resp.decode('utf-8')) - if result['status'] == 'OK': - return result - elif result['status'] != 'UNKNOWN_ERROR': - raise Exception(result['error_message']) - except IOError: - print("IOError") # FIXME Better error handling + real_url = url.format(*map(request.quote, map(str, args))) + resp = request.urlopen(real_url).read() + result = json.loads(resp.decode('utf-8')) + if result['status'] == 'OK': + return result + elif result['status'] != 'UNKNOWN_ERROR': + raise Exception(result['error_message']) -def get_timezone(lat, lng, timestamp): - return _request(URL_TZ, lat, lng, timestamp)['timeZoneId'] +def get_timezone(coords, timestamp): + return _request(URL_TZ, coords['lat'], coords['lng'], timestamp)['timeZoneId'] def get_coords(desc): result = _request(URL_GEOCODE, desc) - if result['results']: - return { - "lat": result['results'][0]['geometry']['location']['lat'], - "lng": result['results'][0]['geometry']['location']['lng'], - "name": result['results'][0]['formatted_address'] - } + if 'results' in result: + return {"lat": result['results'][0]['geometry']['location']['lat'], + "lng": result['results'][0]['geometry']['location']['lng'], + "name": result['results'][0]['formatted_address']} def cache(key, fn, ttl): if key in CACHE and CACHE[key]['time'] < CACHE[key]['time'] + ttl: @@ -60,52 +57,63 @@ def cache(key, fn, ttl): if value is not None: CACHE[key] = { 'time': datetime.now().timestamp(), - 'value': fn() + 'value': value } return value -def diff(a, b): +def diff(a, b, absolute=False): + """Returns a timedelta object""" a = a.replace(tzinfo=None) b = b.replace(tzinfo=None) - return max(a, b) - min(a, b) + return (max(a, b) - min(a, b), b > a) def encode(s): return str(b64encode(s.encode('ascii'))) # -if __name__ == '__main__': - if not sys.argv[1:]: - print("Usage: {} [LOCATION]".format(sys.argv[0])) - sys.exit(1) - - if os.path.isfile(CACHE_FILE): - with open(CACHE_FILE, 'r') as f: - CACHE = json.load(f) - +def each_location(locations): + """ + Enumerates over and maps a list of location strings to time difference data. + """ here = get_localzone().localize(datetime.now()) - for location in sys.argv[1:]: - d_coords = cache("coords-" + encode(location), lambda: get_coords(location), 86400) - if d_coords: - d_tzname = cache( - "tz-" + encode(location), - lambda: get_timezone(d_coords['lat'], d_coords['lng'], here.timestamp()), - 1800 - ) + for location in locations: + coords = cache("coords-" + encode(location), + lambda l=location: get_coords(l), + 86400) + if coords: + tzname = cache("tz-" + encode(location), + lambda c=coords: get_timezone(c, here.timestamp()), + 1800) + then = here.astimezone(timezone(tzname)) + tdiff, direction = diff(here, then) + if tdiff.seconds == 0: + suffix = "(same timezone)" + elif direction: + suffix = "ahead" + else: + suffix = "behind" - then = here.astimezone(timezone(d_tzname)) - diff = diff(here, then) - - print("{:24} {}\t{} {}".format( - d_coords['name'], - then, - diff, - "ago" if then < here else "ahead" - )) + yield { + "name": coords['name'], + "then": then.strftime("%b %d, %H:%M:%S %Z%z"), + "diff": int(tdiff.seconds / 60 / 60), + "suffix": suffix, + } else: - print("Couldn't find {}".format(location)) + yield None - with open(CACHE_FILE, 'w') as f: - json.dump(CACHE, f) - sys.exit(0) +if __name__ == '__main__': + assert sys.argv[1:], "Usage: {} [LOCATION]".format(sys.argv[0]) + + cachefp = Path(CACHE_FILE) + CACHE = json.loads(cachefp.read_text()) if cachefp.exists() else {} + + for data in each_location(sys.argv[1:]): + if data: + print("{name:24} {then}\t{diff}h {suffix}".format(**data)) + else: + print("Couldn't find {name} or something went wrong".format(**data), file=sys.stderr) + + cachefp.write_text(json.dumps(CACHE)) diff --git a/bin/uri b/bin/uri index e81bd57a..afaaa67d 100755 --- a/bin/uri +++ b/bin/uri @@ -1,6 +1,8 @@ #!/usr/bin/env python -# Encode/decode url strings in stdin or the given file(s). +""" +Encode/decode url strings in stdin or the given file(s). +""" import getopt import fileinput @@ -11,32 +13,13 @@ try: except ImportError: from urllib import quote_plus, quote, unquote_plus, unquote -# -def encode(text, plus=False): - return quote(text.strip()) - -def encode_plus(text): - return quote_plus(text.strip()) - -def decode(text): - return unquote(text.strip()) - -def decode_plus(text): - return unquote_plus(text.strip()) def usage(): - # TODO Improve this + """Print out help for this script""" print("Usage: {} [-edEDh] file1[ file2[ ...]]".format(os.path.basename(sys.argv[0]))) -# -if __name__ == '__main__': - COMMANDS = { - "-e": encode, - "-E": encode_plus, - "-d": decode, - "-D": decode_plus - } +def main(commands): try: optlist, args = getopt.getopt(sys.argv[1:], 'edEDh') sys.argv = [sys.argv[0]] + args @@ -44,15 +27,23 @@ if __name__ == '__main__': print(err) sys.exit(1) - action = encode - for o, a in optlist: - if o in COMMANDS: - action = COMMANDS[o] - elif o == '-h': + action = quote + for flag, _ in optlist: + if flag in commands: + action = commands[flag] + elif flag == '-h': usage() sys.exit() else: assert False, "unhandled option" - input = "\n".join([line for line in fileinput.input()]) - print(action(input)) + print(action("\n".join([line for line in fileinput.input()]).strip())) + + +if __name__ == '__main__': + main({ + "-e": quote, + "-E": quote_plus, + "-d": unquote, + "-D": unquote_plus + })