From a1861a2f84bfcdbb8a9b70cf3e1ed1aad9d2a542 Mon Sep 17 00:00:00 2001 From: Jack Date: Mon, 14 Dec 2015 12:59:44 -0500 Subject: [PATCH] Update lbrynet-daemon Add auto fetcher to fetch new name claims --- lbrynet/core/client/AutoDownloader.py | 130 +++++++++++++++++++++++++- lbrynet/lbrynet_daemon/LBRYDaemon.py | 42 +++++++-- 2 files changed, 161 insertions(+), 11 deletions(-) diff --git a/lbrynet/core/client/AutoDownloader.py b/lbrynet/core/client/AutoDownloader.py index 10e06e6cf..939e543bb 100644 --- a/lbrynet/core/client/AutoDownloader.py +++ b/lbrynet/core/client/AutoDownloader.py @@ -1,7 +1,7 @@ import json import logging import os - +from datetime import datetime from twisted.internet import defer from twisted.internet.task import LoopingCall from lbrynet.core.Error import InvalidStreamInfoError, InsufficientFundsError @@ -145,11 +145,27 @@ class GetStream(object): self.description = self.stream_info['description'] if 'key_fee' in self.stream_info.keys(): self.key_fee = float(self.stream_info['key_fee']) - self.key_fee_address = self.stream_info['key_fee_address'] + if 'key_fee_address' in self.stream_info.keys(): + self.key_fee_address = self.stream_info['key_fee_address'] + else: + self.key_fee_address = None else: self.key_fee = None + self.key_fee_address = None self.stream_hash = self.stream_info['stream_hash'] + elif 'stream_hash' in json.loads(self.stream_info['value']): + self.resolved_name = self.stream_info.get('name', None) + self.description = json.loads(self.stream_info['value']).get('description', None) + + try: + if 'key_fee' in json.loads(self.stream_info['value']): + self.key_fee = float(json.loads(self.stream_info['value'])['key_fee']) + except ValueError: + self.key_fee = None + self.key_fee_address = json.loads(self.stream_info['value']).get('key_fee_address', None) + self.stream_hash = json.loads(self.stream_info['value'])['stream_hash'] else: + print 'InvalidStreamInfoError' raise InvalidStreamInfoError(self.stream_info) if self.key_fee > self.max_key_fee: @@ -212,7 +228,7 @@ class GetStream(object): def _set_downloader(downloader): self.downloader = downloader - print os.path.join(self.downloader.download_directory, self.downloader.file_name) + print "Downloading", self.stream_hash, " -->", os.path.join(self.downloader.download_directory, self.downloader.file_name) return self.downloader self.downloader = self.factory.make_downloader(self.metadata, [0.5, True], self.payment_rate_manager) @@ -278,9 +294,9 @@ class AutoFetcher(object): transactions = [self.rpc_conn.decoderawtransaction(self.rpc_conn.getrawtransaction(t)) for t in txids] for t in transactions: claims = self.rpc_conn.getclaimsfortx(t['txid']) - # uncomment to make it download lbry://yyyy on startup # if self.first_run: - # claims = self.rpc_conn.getclaimsfortx("43a784085949f7bebe5c2a2b74f4e2c6abec36219a5d04d285206b4056ea218b") + # #claims = self.rpc_conn.getclaimsfortx("96aca2c60efded5806b7336430c5987b9092ffbea9c6ed444e3bf8e008993e11") + # claims = self.rpc_conn.getclaimsfortx("cc9c7f5225ecb38877e6ca7574d110b23214ac3556b9d65784065ad3a85b4f74") # self.first_run = False if claims: for claim in claims: @@ -330,3 +346,107 @@ class AutoFetcher(object): print "maxkey=1.0" self.max_key_fee = settings["maxkey"] + + +class FetcherDaemon(object): + def __init__(self, session, lbry_file_manager, lbry_file_metadata_manager, wallet, sd_identifier, autofetcher_conf): + self.autofetcher_conf = autofetcher_conf + self.max_key_fee = 0.0 + self.sd_identifier = sd_identifier + self.wallet = wallet + self.session = session + self.lbry_file_manager = lbry_file_manager + self.lbry_metadata_manager = lbry_file_metadata_manager + self.seen = [] + self.lastbestblock = None + self.rpc_conn = self.wallet.get_rpc_conn_x() + self.search = None + self.first_run = True + self.is_running = False + self._get_autofetcher_conf() + + def start(self): + if not self.is_running: + self.is_running = True + self.search = LoopingCall(self._looped_search) + self.search.start(1) + else: + print "Autofetcher is already running" + + def stop(self): + if self.is_running: + self.search.stop() + self.is_running = False + else: + print "Autofetcher isn't running, there's nothing to stop" + + def check_if_running(self): + if self.is_running: + msg = "Autofetcher is running\n" + msg += "Last block hash: " + str(self.lastbestblock['bestblockhash']) + else: + msg = "Autofetcher is not running" + return msg + + def _get_names(self): + c = self.rpc_conn.getblockchaininfo() + rtn = [] + if self.lastbestblock != c: + block = self.rpc_conn.getblock(c['bestblockhash']) + txids = block['tx'] + transactions = [self.rpc_conn.decoderawtransaction(self.rpc_conn.getrawtransaction(t)) for t in txids] + for t in transactions: + claims = self.rpc_conn.getclaimsfortx(t['txid']) + # if self.first_run: + # # claims = self.rpc_conn.getclaimsfortx("96aca2c60efded5806b7336430c5987b9092ffbea9c6ed444e3bf8e008993e11") + # # claims = self.rpc_conn.getclaimsfortx("cc9c7f5225ecb38877e6ca7574d110b23214ac3556b9d65784065ad3a85b4f74") + # self.first_run = False + if claims: + for claim in claims: + if claim not in self.seen: + msg = "[" + str(datetime.now()) + "] New claim | lbry://" + str(claim['name']) + \ + " | stream hash: " + str(json.loads(claim['value'])['stream_hash']) + print msg + log.debug(msg) + rtn.append(claim) + self.seen.append(claim) + else: + # self.console.sendLine("No new claims in block #" + str(block['height'])) + pass + + self.lastbestblock = c + + if len(rtn): + return defer.succeed(rtn) + + def _download_claims(self, claims): + if claims: + for claim in claims: + download = defer.Deferred() + stream = GetStream(self.sd_identifier, self.session, self.wallet, self.lbry_file_manager, self.max_key_fee) + download.addCallback(lambda _: stream.start(claim)) + download.callback(None) + + return defer.succeed(None) + + def _looped_search(self): + d = defer.Deferred(None) + d.addCallback(lambda _: self._get_names()) + d.addCallback(self._download_claims) + d.callback(None) + + def _get_autofetcher_conf(self): + settings = {"maxkey": "0.0"} + if os.path.exists(self.autofetcher_conf): + conf = open(self.autofetcher_conf) + for l in conf: + if l.startswith("maxkey="): + settings["maxkey"] = float(l[7:].rstrip('\n')) + else: + print "Autofetcher using default max key price of 0.0" + print "To change this create the file:" + print str(self.autofetcher_conf) + print "Example contents of conf file:" + print "maxkey=1.0" + + self.max_key_fee = settings["maxkey"] \ No newline at end of file diff --git a/lbrynet/lbrynet_daemon/LBRYDaemon.py b/lbrynet/lbrynet_daemon/LBRYDaemon.py index 88291d902..543aac02b 100644 --- a/lbrynet/lbrynet_daemon/LBRYDaemon.py +++ b/lbrynet/lbrynet_daemon/LBRYDaemon.py @@ -1,7 +1,7 @@ from lbrynet.lbryfile.StreamDescriptor import LBRYFileStreamType from lbrynet.lbryfile.client.LBRYFileDownloader import LBRYFileSaverFactory, LBRYFileOpenerFactory from lbrynet.lbryfile.client.LBRYFileOptions import add_lbry_file_to_sd_identifier -from lbrynet.core.client.AutoDownloader import GetStream +from lbrynet.core.client.AutoDownloader import GetStream, FetcherDaemon from lbrynet.core.utils import generate_id from lbrynet.lbrynet_console.LBRYSettings import LBRYSettings from lbrynet.conf import MIN_BLOB_DATA_PAYMENT_RATE @@ -27,6 +27,7 @@ class LBRYDaemon(xmlrpc.XMLRPC): def setup(self): def _set_vars(): + self.fetcher = None self.current_db_revision = 1 self.run_server = True self.session = None @@ -67,6 +68,7 @@ class LBRYDaemon(xmlrpc.XMLRPC): self.wallet_type = "lbrycrd" self.lbrycrd_dir = os.path.join(os.path.expanduser("~"), ".lbrycrd") self.lbrycrd_conf = os.path.join(self.lbrycrd_dir, "lbrycrd.conf") + self.autofetcher_conf = os.path.join(self.lbrycrd_dir, "autofetcher.conf") self.rpc_conn = None self.files = [] self.created_data_dir = False @@ -86,10 +88,16 @@ class LBRYDaemon(xmlrpc.XMLRPC): d.addCallback(lambda _: self._setup_stream_identifier()) d.addCallback(lambda _: self._setup_lbry_file_manager()) d.addCallback(lambda _: self._setup_lbry_file_opener()) + d.addCallback(lambda _: self._setup_fetcher()) d.callback(None) return defer.succeed(None) + def _setup_fetcher(self): + self.fetcher = FetcherDaemon(self.session, self.lbry_file_manager, self.lbry_file_metadata_manager, + self.session.wallet, self.sd_identifier, self.autofetcher_conf) + return defer.succeed(None) + def _setup_data_directory(self): print "Loading databases..." if self.created_data_dir: @@ -194,13 +202,9 @@ class LBRYDaemon(xmlrpc.XMLRPC): self.rpc_conn = self.session.wallet.get_rpc_conn_x() dl = defer.DeferredList([d1, d2], fireOnOneErrback=True) - dl.addCallback(combine_results) - dl.addCallback(create_session) - dl.addCallback(lambda _: self.session.setup()) - return dl def get_lbrycrdd_path(self): @@ -254,10 +258,36 @@ class LBRYDaemon(xmlrpc.XMLRPC): self.sd_identifier.add_stream_downloader_factory(LBRYFileStreamType, downloader_factory) return defer.succeed(True) + def xmlrpc_start_fetcher(self): + """ + Start autofetcher + """ + + self.fetcher.start() + + return str('Started autofetching') + + def xmlrpc_stop_fetcher(self): + """ + Start autofetcher + """ + + self.fetcher.stop() + + return str('Started autofetching') + + def xmlrpc_fetcher_status(self): + """ + Start autofetcher + """ + + return str(self.fetcher.check_if_running()) + def xmlrpc_get_balance(self): """ Get LBC balance """ + return str(self.session.wallet.wallet_balance) def xmlrpc_stop(self): @@ -324,7 +354,7 @@ class LBRYDaemon(xmlrpc.XMLRPC): # d.addCallback(lambda _: self.files.append({'name': name, 'stream_hash': stream.stream_hash, # 'path': os.path.join(stream.downloader.download_directory, stream.downloader.file_name)})) d.addCallback(lambda _: {'ts': datetime.now(),'name': name}) - d.addErrback(lambda _: 'UnknownNameError') + d.addErrback(lambda err: str(err.getTraceback())) return d