forked from LBRYCommunity/lbry-sdk
0edacbe4c8
Conflicts: lbrynet/conf.py lbrynet/lbrynet_daemon/Daemon.py lbrynet/lbrynet_daemon/DaemonCLI.py lbrynet/lbrynet_daemon/UIManager.py tests/functional/test_misc.py
121 lines
3 KiB
Python
121 lines
3 KiB
Python
import base64
|
|
import datetime
|
|
import distutils.version
|
|
import logging
|
|
import json
|
|
import random
|
|
import os
|
|
import socket
|
|
import yaml
|
|
|
|
from lbrynet import settings
|
|
from lbrynet.conf import AdjustableSettings
|
|
from lbrynet.core.cryptoutils import get_lbry_hash_obj
|
|
|
|
|
|
# digest_size is in bytes, and blob hashes are hex encoded
|
|
blobhash_length = get_lbry_hash_obj().digest_size * 2
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# defining these time functions here allows for easier overriding in testing
|
|
def now():
|
|
return datetime.datetime.now()
|
|
|
|
|
|
def utcnow():
|
|
return datetime.datetime.utcnow()
|
|
|
|
|
|
def isonow():
|
|
"""Return utc now in isoformat with timezone"""
|
|
return utcnow().isoformat() + 'Z'
|
|
|
|
def today():
|
|
return datetime.datetime.today()
|
|
|
|
|
|
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()
|
|
|
|
|
|
def is_valid_hashcharacter(char):
|
|
return char in "0123456789abcdef"
|
|
|
|
|
|
def is_valid_blobhash(blobhash):
|
|
"""Checks whether the blobhash is the correct length and contains only
|
|
valid characters (0-9, a-f)
|
|
|
|
@param blobhash: string, the blobhash to check
|
|
|
|
@return: True/False
|
|
"""
|
|
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"""
|
|
try:
|
|
return distutils.version.StrictVersion(a) > distutils.version.StrictVersion(b)
|
|
except ValueError:
|
|
return distutils.version.LooseVersion(a) > distutils.version.LooseVersion(b)
|
|
|
|
|
|
def deobfuscate(obfustacated):
|
|
return base64.b64decode(obfustacated.decode('rot13'))
|
|
|
|
|
|
def obfuscate(plain):
|
|
return base64.b64encode(plain).encode('rot13')
|
|
|
|
|
|
settings_decoders = {
|
|
'.json': json.loads,
|
|
'.yml': yaml.load
|
|
}
|
|
|
|
settings_encoders = {
|
|
'.json': json.dumps,
|
|
'.yml': yaml.safe_dump
|
|
}
|
|
|
|
ADJUSTABLE_SETTINGS = AdjustableSettings().get_dict()
|
|
|
|
|
|
def load_settings(path):
|
|
ext = os.path.splitext(path)[1]
|
|
with open(path, 'r') as settings_file:
|
|
data = settings_file.read()
|
|
decoder = settings_decoders.get(ext, False)
|
|
assert decoder is not False, "Unknown settings format .%s" % ext
|
|
return decoder(data)
|
|
|
|
|
|
def save_settings(path):
|
|
to_save = {k: v for k, v in settings.__dict__.iteritems() if k in ADJUSTABLE_SETTINGS}
|
|
ext = os.path.splitext(path)[1]
|
|
encoder = settings_encoders.get(ext, False)
|
|
assert encoder is not False, "Unknown settings format .%s" % ext
|
|
with open(path, 'w') as settings_file:
|
|
settings_file.write(encoder(to_save))
|
|
|
|
|
|
def check_connection(server="www.lbry.io", port=80):
|
|
"""Attempts to open a socket to server:port and returns True if successful."""
|
|
try:
|
|
host = socket.gethostbyname(server)
|
|
s = socket.create_connection((host, port), 2)
|
|
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
|