<commit_before>import os
import time
import apigen
import threading
import random
import binascii
import btctxstore
import crochet
from twisted.internet import defer
from storjkademlia.crawling import NodeSpiderCrawl
from collections import defaultdict
from pycoin.encoding import a2b_hashed_base58
from storjkademlia.storage import ForgetfulStorage
from storjkademlia.node import Node
from storjkademlia.network import Server
from pyp2p.lib import get_unused_port
from . import quasar
from . import protocol
from . version import __version__  # NOQA


# TODO add default bootstrap nodes


class StorjNet(apigen.Definition):

    def __init__(self,

                 # network options
                 node_key=None, node_port=None, bootstrap=None,
                 networkid="mainnet", rpc_call_timeout=120,

                 # bandwidth limits
                 limit_send_sec=None, limit_receive_sec=None,
                 limit_send_month=None, limit_receive_month=None,

                 # messaging options
                 message_queue_limit=protocol.MESSAGE_QUEUE_LIMIT,

                 # stream options
                 stream_max_count=protocol.STREAM_MAX_COUNT,
                 stream_buffer_limit=protocol.STREAM_BUFFER_LIMIT,

                 # quasar options
                 quasar_queue_limit=quasar.QUEUE_LIMIT,
                 quasar_history_limit=quasar.HISTORY_LIMIT,
                 quasar_size=quasar.SIZE, quasar_depth=quasar.DEPTH,
                 quasar_ttl=quasar.TTL, quasar_freshness=quasar.FRESHNESS,
                 quasar_refresh_time=quasar.REFRESH_TIME,
                 quasar_extra_propagations=quasar.EXTRA_PROPAGATIONS,

                 # logging options
                 log_statistics=False, quiet=False,
                 debug=False, verbose=False, noisy=False,

                 # other options
                 dht_find_cache_timeout=3600):

        # FIXME add host interface for network interface to listen on
        # TODO sanatize input
        # TODO add doc string

        self._log_stats = log_statistics
        self._dht_find_cache = {}  # nodeid -> (ip, port, timestamp)
        self._dht_find_cache_timeout = dht_find_cache_timeout
        self._call_timeout = rpc_call_timeout
        self._setup_node(node_key)
        self._setup_protocol(noisy, message_queue_limit,
                             stream_max_count, stream_buffer_limit)
        self._setup_kademlia(bootstrap, node_port)
        self._setup_quasar(
            quasar_queue_limit, quasar_history_limit, quasar_size,
            quasar_depth, quasar_ttl, quasar_freshness, quasar_refresh_time,
            quasar_extra_propagations, log_statistics
        )
        # TODO setup streams

    def _setup_node(self, node_key):
        self._btctxstore = btctxstore.BtcTxStore()
        node_key = node_key or self._btctxstore.create_key()
        is_hwif = self._btctxstore.validate_wallet(node_key)
        self._key = self._btctxstore.get_key(node_key) if is_hwif else node_key
        address = self._btctxstore.get_address(self._key)
        self._nodeid = a2b_hashed_base58(address)[1:]

    def _setup_protocol(self, noisy, message_queue_limit,
                        stream_max_count, stream_buffer_limit):
        storage = ForgetfulStorage()
        self._protocol = protocol.Protocol(
            Node(self._nodeid), storage, noisy=noisy,
            message_queue_limit=message_queue_limit,
            stream_max_count=stream_max_count,
            stream_buffer_limit=stream_buffer_limit
        )
        # TODO set rpc logger

    def _setup_kademlia(self, bootstrap, node_port):

        # ensure transport address is a tuple
        if bootstrap is not None:
            bootstrap = [(addr[0], addr[1]) for addr in bootstrap]

        self._port = node_port or get_unused_port()
        self._kademlia = Server(id=self._nodeid, protocol=self._protocol)
        self._kademlia.bootstrap(bootstrap or [])
        self._kademlia.listen(self._port)
        # TODO set kademlia logger

    def _setup_quasar(self, queue_limit, history_limit, size, depth, ttl,
                      freshness, refresh_time, extra_propagations,
                      log_statistics):
        self._quasar = quasar.Quasar(
            self._protocol, queue_limit=queue_limit,
            history_limit=history_limit, size=size, depth=depth, ttl=ttl,
            freshness=freshness, refresh_time=refresh_time,
            extra_propagations=extra_propagations,
            log_statistics=log_statistics
        )

    def dht_put_async(self, key, value):
        """Store key/value pair in DHT."""
        # TODO sanatize input
        return self._kademlia.set(key, value)

    @apigen.command()
    def dht_put(self, key, value):
        """Store key/value pair in DHT."""

        @crochet.wait_for(timeout=self._call_timeout)
        def func():
            return self.dht_put_async(key, value)
        return func()

    def dht_get_async(self, key):
        """Get value for given key in DHT."""
        # TODO sanatize input
        return self._kademlia.get(key)

    @apigen.command()
    def dht_get(self, key):
        """Get value for given key in DHT."""

        @crochet.wait_for(timeout=self._call_timeout)
        def func():
            return self.dht_get_async(key)
        return func()

    def dht_stun_async(self):
        """Stun random neighbor to see own wan ip/port."""
        # TODO cache result
        hexid, ip, port = random.choice(self.dht_peers())
        d = self._protocol.stun((ip, port))

        def func(result):
            if result[0]:
                return result[1]
            return None
        d.addCallback(func)
        return d

    @apigen.command()
    def dht_stun(self):
        """Stun random neighbor to see own wan ip/port."""

        @crochet.wait_for(timeout=self._call_timeout)
        def func():
            return self.dht_stun_async()
        return func()

    def _dht_find_cached(self, nodeid):
        if nodeid not in self._dht_find_cache:
            return None
        ip, port, timestamp = self._dht_find_cache[nodeid]

        # remove outdated entry
        if timestamp + self._dht_find_cache_timeout > time.time():
            del self._dht_find_cache[nodeid]
            return None

        return defer.succeed([ip, port])

    def _dht_find_add_to_cache(self, nodeid, result):

        # remove any previous entry
        if nodeid in self._dht_find_cache:
            del self._dht_find_cache[nodeid]

        # do not cache failed results
        if result is None:
            return None

        # add to cache
        ip, port = result
        self._dht_find_cache[nodeid] = (ip, port, time.time())

        return result

    def dht_find_async(self, hexnodeid):<commit_msg>fixed find cache timeout check<commit_after>import os
import time
import apigen
import threading
import random
import binascii
import btctxstore
import crochet
from twisted.internet import defer
from storjkademlia.crawling import NodeSpiderCrawl
from collections import defaultdict
from pycoin.encoding import a2b_hashed_base58
from storjkademlia.storage import ForgetfulStorage
from storjkademlia.node import Node
from storjkademlia.network import Server
from pyp2p.lib import get_unused_port
from . import quasar
from . import protocol
from . version import __version__  # NOQA


# TODO add default bootstrap nodes


class StorjNet(apigen.Definition):

    def __init__(self,

                 # network options
                 node_key=None, node_port=None, bootstrap=None,
                 networkid="mainnet", rpc_call_timeout=120,

                 # bandwidth limits
                 limit_send_sec=None, limit_receive_sec=None,
                 limit_send_month=None, limit_receive_month=None,

                 # messaging options
                 message_queue_limit=protocol.MESSAGE_QUEUE_LIMIT,

                 # stream options
                 stream_max_count=protocol.STREAM_MAX_COUNT,
                 stream_buffer_limit=protocol.STREAM_BUFFER_LIMIT,

                 # quasar options
                 quasar_queue_limit=quasar.QUEUE_LIMIT,
                 quasar_history_limit=quasar.HISTORY_LIMIT,
                 quasar_size=quasar.SIZE, quasar_depth=quasar.DEPTH,
                 quasar_ttl=quasar.TTL, quasar_freshness=quasar.FRESHNESS,
                 quasar_refresh_time=quasar.REFRESH_TIME,
                 quasar_extra_propagations=quasar.EXTRA_PROPAGATIONS,

                 # logging options
                 log_statistics=False, quiet=False,
                 debug=False, verbose=False, noisy=False,

                 # other options
                 dht_find_cache_timeout=3600):

        # FIXME add host interface for network interface to listen on
        # TODO sanatize input
        # TODO add doc string

        self._log_stats = log_statistics
        self._dht_find_cache = {}  # nodeid -> (ip, port, timestamp)
        self._dht_find_cache_timeout = dht_find_cache_timeout
        self._call_timeout = rpc_call_timeout
        self._setup_node(node_key)
        self._setup_protocol(noisy, message_queue_limit,
                             stream_max_count, stream_buffer_limit)
        self._setup_kademlia(bootstrap, node_port)
        self._setup_quasar(
            quasar_queue_limit, quasar_history_limit, quasar_size,
            quasar_depth, quasar_ttl, quasar_freshness, quasar_refresh_time,
            quasar_extra_propagations, log_statistics
        )
        # TODO setup streams

    def _setup_node(self, node_key):
        self._btctxstore = btctxstore.BtcTxStore()
        node_key = node_key or self._btctxstore.create_key()
        is_hwif = self._btctxstore.validate_wallet(node_key)
        self._key = self._btctxstore.get_key(node_key) if is_hwif else node_key
        address = self._btctxstore.get_address(self._key)
        self._nodeid = a2b_hashed_base58(address)[1:]

    def _setup_protocol(self, noisy, message_queue_limit,
                        stream_max_count, stream_buffer_limit):
        storage = ForgetfulStorage()
        self._protocol = protocol.Protocol(
            Node(self._nodeid), storage, noisy=noisy,
            message_queue_limit=message_queue_limit,
            stream_max_count=stream_max_count,
            stream_buffer_limit=stream_buffer_limit
        )
        # TODO set rpc logger

    def _setup_kademlia(self, bootstrap, node_port):

        # ensure transport address is a tuple
        if bootstrap is not None:
            bootstrap = [(addr[0], addr[1]) for addr in bootstrap]

        self._port = node_port or get_unused_port()
        self._kademlia = Server(id=self._nodeid, protocol=self._protocol)
        self._kademlia.bootstrap(bootstrap or [])
        self._kademlia.listen(self._port)
        # TODO set kademlia logger

    def _setup_quasar(self, queue_limit, history_limit, size, depth, ttl,
                      freshness, refresh_time, extra_propagations,
                      log_statistics):
        self._quasar = quasar.Quasar(
            self._protocol, queue_limit=queue_limit,
            history_limit=history_limit, size=size, depth=depth, ttl=ttl,
            freshness=freshness, refresh_time=refresh_time,
            extra_propagations=extra_propagations,
            log_statistics=log_statistics
        )

    def dht_put_async(self, key, value):
        """Store key/value pair in DHT."""
        # TODO sanatize input
        return self._kademlia.set(key, value)

    @apigen.command()
    def dht_put(self, key, value):
        """Store key/value pair in DHT."""

        @crochet.wait_for(timeout=self._call_timeout)
        def func():
            return self.dht_put_async(key, value)
        return func()

    def dht_get_async(self, key):
        """Get value for given key in DHT."""
        # TODO sanatize input
        return self._kademlia.get(key)

    @apigen.command()
    def dht_get(self, key):
        """Get value for given key in DHT."""

        @crochet.wait_for(timeout=self._call_timeout)
        def func():
            return self.dht_get_async(key)
        return func()

    def dht_stun_async(self):
        """Stun random neighbor to see own wan ip/port."""
        # TODO cache result
        hexid, ip, port = random.choice(self.dht_peers())
        d = self._protocol.stun((ip, port))

        def func(result):
            if result[0]:
                return result[1]
            return None
        d.addCallback(func)
        return d

    @apigen.command()
    def dht_stun(self):
        """Stun random neighbor to see own wan ip/port."""

        @crochet.wait_for(timeout=self._call_timeout)
        def func():
            return self.dht_stun_async()
        return func()

    def _dht_find_cached(self, nodeid):
        if nodeid not in self._dht_find_cache:
            return None
        ip, port, timestamp = self._dht_find_cache[nodeid]

        # remove outdated entry
        if timestamp + self._dht_find_cache_timeout < time.time():
            del self._dht_find_cache[nodeid]
            return None

        return defer.succeed([ip, port])

    def _dht_find_add_to_cache(self, nodeid, result):

        # remove any previous entry
        if nodeid in self._dht_find_cache:
            del self._dht_find_cache[nodeid]

        # do not cache failed results
        if result is None:
            return None

        # add to cache
        ip, port = result
        self._dht_find_cache[nodeid] = (ip, port, time.time())

        return result

    def dht_find_async(self, hexnodeid):
{'fraction_non_alphanumeric': 0.07470038967722961, 'fraction_numerical': 0.0027939122123373282, 'mean_word_length': 2.3248594475678317, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5238265', 'n_tokens_mistral': 4131, 'n_tokens_neox': 3964, 'n_words': 881}