aynish/helm
1
0
Fork 0
Code Pull requests Packages Projects Releases Activity

Refactor python scripts

Browse source
This commit is contained in:
Henrik Lissner 2017-10-03 19:05:00 +02:00
parent d2114373f3
commit 3d04b314ef
No known key found for this signature in database
GPG key ID: 5F6C0EA160557395
4 changed files with 251 additions and 217 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
bin/htmlent
View file

@ -1,16 +1,18 @@
#!/usr/bin/env python3 #!/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 argparse
import sys
import os
import fileinput import fileinput
import html import html
from html.entities import entitydefs from html.entities import entitydefs
from pathlib import Path
ENTITIES = {value: key for key, value in entitydefs.items()} ENTITIES = {value: key for key, value in entitydefs.items()}
def encode(text): def encode(text):
t = "" t = ""
for c in text: for c in text:
@ -23,33 +25,19 @@ def encode(text):
def decode(text): def decode(text):
return html.unescape(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__': if __name__ == '__main__':
COMMANDS = { opts = argparse.ArgumentParser(description="Encodes (or decodes) HTML entities in text.")
"-e": encode, opts.add_argument('file', nargs="?", help="File to encode/decode")
"-d": decode opts.add_argument('-d', '--decode', action='store_true', help="Decode the given files/input")
} args = opts.parse_args()
try: if "file" in args and args.file:
optlist, args = getopt.getopt(sys.argv[1:], 'edh') text = Path(args.file).read_text()
sys.argv = [sys.argv[0]] + args else:
except getopt.GetoptError as err: text = "\n".join([line for line in fileinput.input()]).strip()
print(err)
sys.exit(1)
action = encode if args.decode:
for o, a in optlist: print(decode(text))
if o in COMMANDS: else:
action = COMMANDS[o] print(encode(text))
elif o == '-h':
usage()
sys.exit()
else:
assert False, "unhandled option"
input = "\n".join([line for line in fileinput.input()]).strip()
print(action(input))

247
bin/shekel
View file

@ -1,130 +1,177 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# A shell script that converts one currency to another. """
# 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)
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 sys
import os
import json import json
from urllib import request from urllib import request
from datetime import datetime from datetime import datetime
from base64 import b64encode from pathlib import Path
try: try:
from iterfzf import iterfzf from iterfzf import iterfzf
except: except ImportError:
pass pass
API_KEY = "92a7ccb3b6038a7151fc15e99f2de83d" # please use your own! API_KEY = "92a7ccb3b6038a7151fc15e99f2de83d" # please use your own!
API_URL = "http://apilayer.net/api/live?access_key={}&currencies={},{}" API_URL = "http://apilayer.net/api/live?access_key={}&currencies={},{}"
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" DEFAULT_CODES = ('CAD', 'USD')
CACHE_FILE = os.path.expanduser("~/.cache/shekel-hist") CACHE_FILE = "/tmp/shekel-hist"
CACHE_TTL = 3600 CACHE_TTL = 3600
CACHE = {}
def _request(url, *args): class CurrencyCache():
try: def __init__(self, path=CACHE_FILE, ttl=CACHE_TTL, db_url=DB_URL):
real_url = url.format(*map(request.quote, map(str, args))) self.path = Path(path)
resp = request.urlopen(real_url).read().decode('utf-8') self.ttl = ttl
data = json.loads(resp)
if data['success']:
return data["quotes"]
raise Exception(data['error']['info'])
except IOError:
print("IOError")
def _cache(key, value): cache = {}
CACHE[key] = { if self.path.exists():
'time': datetime.now().timestamp(), cache = json.loads(self.path.read_text())
'value': value 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): def __enter__(self):
key = "-".join([from_code, to_code]) return self
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 get_rate(from_code, to_code=DEFAULT_CODE): def __exit__(self, *args):
if from_code == to_code: self.path.write_text(json.dumps(self.cache))
raise ValueError("Trying to convert currency to itself ({})".format(from_code))
elif from_code is None or to_code is None: 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 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): def _set(self, key, value):
for i, code in enumerate(codes): self.cache[key] = {
if len(code) == 3 and code.isupper(): 'time': datetime.now().timestamp(),
continue 'value': value
resp = iterfzf(currencies(), query=code) }
if resp is None:
codes[i] = None
else:
codes[i] = resp.split(" ")[0]
return codes
def currencies(): def _request_currency_rate(url=API_URL, *args):
for curr in CACHE['_DB']: """Fetch currency rates from remote API."""
yield "{} {}".format(curr['id'], curr['currencyName']) 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]))
# def format_currency(curr):
if __name__ == '__main__': """Format a currency entry for the FZF prompt."""
refresh = False return "{} {}".format(curr['id'], curr['currencyName'])
if "-f" in sys.argv:
refresh = True
sys.argv.remove('-f')
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) 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']) if __name__ == '__main__':
rate = get_rate(*codes) main()
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)

124
bin/time-in
View file

@ -1,18 +1,20 @@
#!/usr/bin/env python3 #!/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 Retrieves the time (and time difference) at a location (can be an address,
# coordinates, and coordinates to timezones. 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 Requires: pytz, tzlocal
Usage: time-in Copenhagen "Toronto, ON" Vancouver
"""
import sys import sys
import os
import json import json
from base64 import b64encode from base64 import b64encode
from datetime import datetime from datetime import datetime
from urllib import request from urllib import request
from pathlib import Path
from pytz import timezone from pytz import timezone
from tzlocal import get_localzone from tzlocal import get_localzone
@ -30,28 +32,23 @@ def _generate_url(url, *args):
return url.format(*map(request.quote, map(str, args))) return url.format(*map(request.quote, map(str, args)))
def _request(url, *args): def _request(url, *args):
try: real_url = url.format(*map(request.quote, map(str, args)))
real_url = url.format(*map(request.quote, map(str, args))) resp = request.urlopen(real_url).read()
resp = request.urlopen(real_url).read() result = json.loads(resp.decode('utf-8'))
result = json.loads(resp.decode('utf-8')) if result['status'] == 'OK':
if result['status'] == 'OK': return result
return result elif result['status'] != 'UNKNOWN_ERROR':
elif result['status'] != 'UNKNOWN_ERROR': raise Exception(result['error_message'])
raise Exception(result['error_message'])
except IOError:
print("IOError") # FIXME Better error handling
def get_timezone(lat, lng, timestamp): def get_timezone(coords, timestamp):
return _request(URL_TZ, lat, lng, timestamp)['timeZoneId'] return _request(URL_TZ, coords['lat'], coords['lng'], timestamp)['timeZoneId']
def get_coords(desc): def get_coords(desc):
result = _request(URL_GEOCODE, desc) result = _request(URL_GEOCODE, desc)
if result['results']: if 'results' in result:
return { return {"lat": result['results'][0]['geometry']['location']['lat'],
"lat": result['results'][0]['geometry']['location']['lat'], "lng": result['results'][0]['geometry']['location']['lng'],
"lng": result['results'][0]['geometry']['location']['lng'], "name": result['results'][0]['formatted_address']}
"name": result['results'][0]['formatted_address']
}
def cache(key, fn, ttl): def cache(key, fn, ttl):
if key in CACHE and CACHE[key]['time'] < CACHE[key]['time'] + 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: if value is not None:
CACHE[key] = { CACHE[key] = {
'time': datetime.now().timestamp(), 'time': datetime.now().timestamp(),
'value': fn() 'value': value
} }
return value return value
def diff(a, b): def diff(a, b, absolute=False):
"""Returns a timedelta object"""
a = a.replace(tzinfo=None) a = a.replace(tzinfo=None)
b = b.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): def encode(s):
return str(b64encode(s.encode('ascii'))) return str(b64encode(s.encode('ascii')))
# #
if __name__ == '__main__': def each_location(locations):
if not sys.argv[1:]: """
print("Usage: {} [LOCATION]".format(sys.argv[0])) Enumerates over and maps a list of location strings to time difference data.
sys.exit(1) """
if os.path.isfile(CACHE_FILE):
with open(CACHE_FILE, 'r') as f:
CACHE = json.load(f)
here = get_localzone().localize(datetime.now()) here = get_localzone().localize(datetime.now())
for location in sys.argv[1:]: for location in locations:
d_coords = cache("coords-" + encode(location), lambda: get_coords(location), 86400) coords = cache("coords-" + encode(location),
if d_coords: lambda l=location: get_coords(l),
d_tzname = cache( 86400)
"tz-" + encode(location), if coords:
lambda: get_timezone(d_coords['lat'], d_coords['lng'], here.timestamp()), tzname = cache("tz-" + encode(location),
1800 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)) yield {
diff = diff(here, then) "name": coords['name'],
"then": then.strftime("%b %d, %H:%M:%S %Z%z"),
print("{:24} {}\t{} {}".format( "diff": int(tdiff.seconds / 60 / 60),
d_coords['name'], "suffix": suffix,
then, }
diff,
"ago" if then < here else "ahead"
))
else: 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))

49
bin/uri
View file

@ -1,6 +1,8 @@
#!/usr/bin/env python #!/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 getopt
import fileinput import fileinput
@ -11,32 +13,13 @@ try:
except ImportError: except ImportError:
from urllib import quote_plus, quote, unquote_plus, unquote 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(): def usage():
# TODO Improve this """Print out help for this script"""
print("Usage: {} [-edEDh] file1[ file2[ ...]]".format(os.path.basename(sys.argv[0]))) 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: try:
optlist, args = getopt.getopt(sys.argv[1:], 'edEDh') optlist, args = getopt.getopt(sys.argv[1:], 'edEDh')
sys.argv = [sys.argv[0]] + args sys.argv = [sys.argv[0]] + args
@ -44,15 +27,23 @@ if __name__ == '__main__':
print(err) print(err)
sys.exit(1) sys.exit(1)
action = encode action = quote
for o, a in optlist: for flag, _ in optlist:
if o in COMMANDS: if flag in commands:
action = COMMANDS[o] action = commands[flag]
elif o == '-h': elif flag == '-h':
usage() usage()
sys.exit() sys.exit()
else: else:
assert False, "unhandled option" assert False, "unhandled option"
input = "\n".join([line for line in fileinput.input()]) print(action("\n".join([line for line in fileinput.input()]).strip()))
print(action(input))
if __name__ == '__main__':
main({
"-e": quote,
"-E": quote_plus,
"-d": unquote,
"-D": unquote_plus
})