lbry-sdk/lbrynet/conf.py

import copy
import logging
import os
import sys

from appdirs import user_data_dir


log = logging.getLogger(__name__)


LINUX = 1
DARWIN = 2
WINDOWS = 3


if sys.platform.startswith("darwin"):
    platform = DARWIN
    default_download_directory = os.path.join(os.path.expanduser("~"), 'Downloads')
    default_data_dir = user_data_dir("LBRY")
    default_lbryum_dir = os.path.join(os.path.expanduser("~"), ".lbryum")
elif sys.platform.startswith("win"):
    platform = WINDOWS
    from lbrynet.winhelpers.knownpaths import get_path, FOLDERID, UserHandle
    default_download_directory = get_path(FOLDERID.Downloads, UserHandle.current)
    default_data_dir = os.path.join(
        get_path(FOLDERID.RoamingAppData, UserHandle.current), "lbrynet")
    default_lbryum_dir = os.path.join(
        get_path(FOLDERID.RoamingAppData, UserHandle.current), "lbryum")
else:
    platform = LINUX
    default_download_directory = os.path.join(os.path.expanduser("~"), 'Downloads')
    default_data_dir = os.path.join(os.path.expanduser("~"), ".lbrynet")
    default_lbryum_dir = os.path.join(os.path.expanduser("~"), ".lbryum")


def convert_setting(env_val, current_val):
    try:
        return _convert_setting(env_val, current_val)
    except Exception as exc:
        log.warning(
            'Failed to convert %s. Returning original: %s: %s',
            env_val, current_val, exc)
        return current_val


def _convert_setting(env_val, current_val):
    new_type = env_val.__class__
    current_type = current_val.__class__
    if current_type is bool:
        if new_type is bool:
            return env_val
        elif str(env_val).lower() == "false":
            return False
        elif str(env_val).lower() == "true":
            return True
        else:
            raise ValueError('{} is not a valid boolean value'.format(env_val))
    elif current_type is int:
        return int(env_val)
    elif current_type is float:
        return float(env_val)
    elif current_type is str:
        return str(env_val)
    elif current_type is dict:
        return dict(env_val)
    elif current_type is list:
        return list(env_val)
    elif current_type is tuple:
        return tuple(env_val)
    else:
        raise ValueError('Type {} cannot be converted'.format(current_type))


def convert_env_setting(setting, value):
    try:
        env_val = os.environ[setting]
    except KeyError:
        return value
    else:
        return convert_setting(env_val, value)


def get_env_settings(settings):
    for setting, value in settings.iteritems():
        setting = 'LBRY_' + setting.upper()
        yield convert_env_setting(setting, value)


def add_env_settings_to_dict(settings_dict):
    for setting, env_setting in zip(settings_dict, get_env_settings(settings_dict)):
        settings_dict.update({setting: env_setting})
    return settings_dict


class Setting(object):
    """A collection of configuration settings"""
    __fixed = []
    __excluded = ['get_dict', 'update']

    def __iter__(self):
        for k in self.__dict__.iterkeys():
            if k.startswith('_') or k in self.__excluded:
                continue
            yield k

    def __getitem__(self, item):
        assert item in self, IndexError
        return self.__dict__[item]

    def __setitem__(self, key, value):
        assert key in self and key not in self.__fixed, KeyError(key)
        old_value = self[key]
        new_value = convert_setting(value, old_value)
        self.__dict__[key] = new_value

    def __contains__(self, item):
        return item in iter(self)

    def get_dict(self):
        return {k: self[k] for k in self}

    def update(self, other):
        for k, v in other.iteritems():
            try:
                self.__setitem__(k, v)
            except (KeyError, AssertionError):
                pass


class AdjustableSettings(Setting):
    """Settings that are allowed to be overriden by the user"""
    def __init__(self):
        self.is_generous_host = True
        self.run_on_startup = False
        self.download_directory = default_download_directory
        self.max_upload = 0.0
        self.max_download = 0.0
        self.upload_log = True
        self.delete_blobs_on_remove = True
        self.use_upnp = True
        self.start_lbrycrdd = True
        self.run_reflector_server = False
        self.startup_scripts = []
        self.last_version = {'lbrynet': '0.0.1', 'lbryum': '0.0.1'}
        self.peer_port = 3333
        self.dht_node_port = 4444
        self.reflector_port = 5566
        self.download_timeout = 30
        self.max_search_results = 25
        self.search_timeout = 3.0
        self.cache_time = 150
        self.host_ui = True
        self.check_ui_requirements = True
        self.local_ui_path = False
        self.api_port = 5279
        self.search_servers = ['lighthouse1.lbry.io:50005']
        self.data_rate = .0001  # points/megabyte
        self.min_info_rate = .02  # points/1000 infos
        self.min_valuable_info_rate = .05  # points/1000 infos
        self.min_valuable_hash_rate = .05  # points/1000 infos
        self.max_connections_per_stream = 5
        self.known_dht_nodes = [
            ('104.236.42.182', 4000),
            ('lbrynet1.lbry.io', 4444),
            ('lbrynet2.lbry.io', 4444),
            ('lbrynet3.lbry.io', 4444)
        ]
        self.pointtrader_server = 'http://127.0.0.1:2424'
        self.reflector_servers = [("reflector.lbry.io", 5566)]
        self.wallet = "lbryum"
        self.ui_branch = "master"
        self.default_ui_branch = 'master'
        self.data_dir = default_data_dir
        self.lbryum_wallet_dir = default_lbryum_dir
        self.use_auth_http = False
        self.sd_download_timeout = 3
        self.max_key_fee = {'USD': {'amount': 25.0, 'address': ''}}


class ApplicationSettings(Setting):
    """Settings that are constants and shouldn't be overriden"""
    def __init__(self):
        self.MAX_HANDSHAKE_SIZE = 2**16
        self.MAX_REQUEST_SIZE = 2**16
        self.MAX_BLOB_REQUEST_SIZE = 2**16
        self.MAX_RESPONSE_INFO_SIZE = 2**16
        self.MAX_BLOB_INFOS_TO_REQUEST = 20
        self.BLOBFILES_DIR = "blobfiles"
        self.BLOB_SIZE = 2**21
        self.LOG_FILE_NAME = "lbrynet.log"
        self.LOG_POST_URL = "https://lbry.io/log-upload"
        self.CRYPTSD_FILE_EXTENSION = ".cryptsd"
        self.API_INTERFACE = "localhost"
        self.API_ADDRESS = "lbryapi"
        self.ICON_PATH = "icons" if platform is WINDOWS else "app.icns"
        self.APP_NAME = "LBRY"
        self.PROTOCOL_PREFIX = "lbry"
        self.wallet_TYPES = ["lbryum", "lbrycrd"]
        self.SOURCE_TYPES = ['lbry_sd_hash', 'url', 'btih']
        self.CURRENCIES = {
            'BTC': {'type': 'crypto'},
            'LBC': {'type': 'crypto'},
            'USD': {'type': 'fiat'},
        }
        self.LOGGLY_TOKEN = 'LJEzATH4AzRgAwxjAP00LwZ2YGx3MwVgZTMuBQZ3MQuxLmOv'
        self.ANALYTICS_ENDPOINT = 'https://api.segment.io/v1'
        self.ANALYTICS_TOKEN = 'Ax5LZzR1o3q3Z3WjATASDwR5rKyHH0qOIRIbLmMXn2H='


APPLICATION_SETTINGS = AdjustableSettings()
ADJUSTABLE_SETTINGS = AdjustableSettings()


class DefaultSettings(ApplicationSettings, AdjustableSettings):
    __fixed = APPLICATION_SETTINGS.get_dict().keys()

    def __init__(self):
        ApplicationSettings.__init__(self)
        AdjustableSettings.__init__(self)


DEFAULT_SETTINGS = DefaultSettings()


class Config(DefaultSettings):
    __shared_state = copy.deepcopy(DEFAULT_SETTINGS.get_dict())

    def __init__(self):
        self.__dict__ = add_env_settings_to_dict(self.__shared_state)

    @property
    def ORIGIN(self):
        return "http://%s:%i" % (DEFAULT_SETTINGS.API_INTERFACE, self.api_port)

    @property
    def REFERER(self):
        return "http://%s:%i/" % (DEFAULT_SETTINGS.API_INTERFACE, self.api_port)

    @property
    def API_CONNECTION_STRING(self):
        return "http://%s:%i/%s" % (
            DEFAULT_SETTINGS.API_INTERFACE, self.api_port, DEFAULT_SETTINGS.API_ADDRESS)

    @property
    def UI_ADDRESS(self):
        return "http://%s:%i" % (DEFAULT_SETTINGS.API_INTERFACE, self.api_port)


def get_data_dir():
    data_dir = default_data_dir
    if not os.path.isdir(data_dir):
        os.mkdir(data_dir)
    return data_dir


def get_log_filename():
    """Return the log file for this platform.

    Also ensure the containing directory exists
    """
    return os.path.join(get_data_dir(), settings.LOG_FILE_NAME)


# TODO: don't load the configuration automatically. The configuration
#       should be loaded at runtime, not at module import time. Module
#       import should have no side-effects. This is also bad because
#       it means that settings are read from the environment even for
#       tests, which is rarely what you want to happen.
settings = Config()