lbry-sdk/lbrynet/core/utils.py

120 lines
3 KiB
Python
Raw Normal View History

import base64
import datetime
import logging
2015-08-20 17:27:15 +02:00
import random
import socket
2016-12-30 19:35:17 +01:00
import string
import json
2015-08-20 17:27:15 +02:00
2016-11-19 23:58:40 +01:00
import pkg_resources
from lbrynet.core.cryptoutils import get_lbry_hash_obj
2016-09-30 06:06:07 +02:00
# digest_size is in bytes, and blob hashes are hex encoded
blobhash_length = get_lbry_hash_obj().digest_size * 2
2015-08-20 17:27:15 +02:00
log = logging.getLogger(__name__)
2016-10-05 21:16:20 +02:00
# defining these time functions here allows for easier overriding in testing
2016-09-30 06:06:07 +02:00
def now():
return datetime.datetime.now()
2016-10-05 21:16:20 +02:00
2016-09-30 06:06:07 +02:00
def utcnow():
return datetime.datetime.utcnow()
2016-10-05 21:16:20 +02:00
2016-09-30 06:06:07 +02:00
def isonow():
"""Return utc now in isoformat with timezone"""
return utcnow().isoformat() + 'Z'
2016-09-30 06:06:07 +02:00
def today():
return datetime.datetime.today()
2017-01-02 20:52:24 +01:00
def timedelta(**kwargs):
return datetime.timedelta(**kwargs)
def datetime_obj(*args, **kwargs):
return datetime.datetime(*args, **kwargs)
def call_later(delay, func, *args, **kwargs):
# Import here to ensure that it gets called after installing a reator
# see: http://twistedmatrix.com/documents/current/core/howto/choosing-reactor.html
from twisted.internet import reactor
return reactor.callLater(delay, func, *args, **kwargs)
2015-08-20 17:27:15 +02:00
def generate_id(num=None):
h = get_lbry_hash_obj()
if num is not None:
h.update(str(num))
else:
h.update(str(random.getrandbits(512)))
return h.digest()
2016-10-05 21:16:20 +02:00
def is_valid_hashcharacter(char):
return char in "0123456789abcdef"
2015-08-20 17:27:15 +02:00
def is_valid_blobhash(blobhash):
2016-10-19 00:25:16 +02:00
"""Checks whether the blobhash is the correct length and contains only
valid characters (0-9, a-f)
2015-08-20 17:27:15 +02:00
@param blobhash: string, the blobhash to check
2016-10-19 00:25:16 +02:00
@return: True/False
2015-08-20 17:27:15 +02:00
"""
2016-10-19 00:25:16 +02:00
return len(blobhash) == blobhash_length and all(is_valid_hashcharacter(l) for l in blobhash)
def version_is_greater_than(a, b):
"""Returns True if version a is more recent than version b"""
2016-11-19 23:58:40 +01:00
return pkg_resources.parse_version(a) > pkg_resources.parse_version(b)
def deobfuscate(obfustacated):
return base64.b64decode(obfustacated.decode('rot13'))
def obfuscate(plain):
return base64.b64encode(plain).encode('rot13')
def check_connection(server="www.lbry.io", port=80):
"""Attempts to open a socket to server:port and returns True if successful."""
try:
2016-11-11 17:10:00 +01:00
log.debug('Checking connection to %s:%s', server, port)
host = socket.gethostbyname(server)
s = socket.create_connection((host, port), 2)
2016-11-11 17:10:00 +01:00
log.debug('Connection successful')
return True
except Exception as ex:
log.info(
"Failed to connect to %s:%s. Maybe the internet connection is not working",
server, port, exc_info=True)
return False
2016-12-30 19:35:17 +01:00
def random_string(length=10, chars=string.ascii_lowercase):
return ''.join([random.choice(chars) for _ in range(length)])
def short_hash(hash_str):
return hash_str[:6]
2017-03-09 16:39:17 +01:00
def get_sd_hash(stream_info):
if not stream_info:
return None
return stream_info['sources']['lbry_sd_hash']
def json_dumps_pretty(obj, **kwargs):
return json.dumps(obj, sort_keys=True, indent=2, separators=(',', ': '), **kwargs)