Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_steem_nodes(), num_retries=10))
cls.bts = Steem(
node=nodelist.get_steem_nodes(),
nobroadcast=True,
num_retries=10
)
set_shared_steem_instance(cls.bts)
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(hive=True), num_retries=10))
cls.bts = Steem(
node=nodelist.get_nodes(hive=True),
nobroadcast=True,
num_retries=10,
timeout=30,
use_condenser=False,
keys={"active": wif},
)
# from getpass import getpass
# self.bts.wallet.unlock(getpass())
set_shared_steem_instance(cls.bts)
cls.bts.set_default_account("test")
b = Blockchain(steem_instance=cls.bts)
num = b.get_current_block_num()
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(hive=True), num_retries=10))
cls.bts = Steem(
node=nodelist.get_nodes(hive=True),
nobroadcast=True,
keys={"active": wif},
num_retries=10
)
cls.test_block_id = 19273700
# from getpass import getpass
# self.bts.wallet.unlock(getpass())
set_shared_steem_instance(cls.bts)
cls.bts.set_default_account("test")
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(normal=True, appbase=True), num_retries=3))
cls.nodes = nodelist.get_nodes()
if "https://api.steemit.com" in cls.nodes:
cls.nodes.remove("https://api.steemit.com")
cls.nodes_steemit = ["https://api.steemit.com"]
cls.appbase = Steem(
node=cls.nodes,
nobroadcast=True,
keys={"active": wif, "owner": wif, "memo": wif},
num_retries=10
)
cls.rpc = SteemNodeRPC(urls=cls.nodes_steemit)
# from getpass import getpass
# self.bts.wallet.unlock(getpass())
set_shared_steem_instance(cls.nodes_steemit)
def setUpClass(cls):
nodelist = NodeList()
nodelist.update_nodes(steem_instance=Steem(node=nodelist.get_nodes(exclude_limited=False, appbase=True), num_retries=10))
node_list = nodelist.get_nodes(exclude_limited=True)
cls.bts = Steem(
node=node_list,
use_condenser=True,
nobroadcast=True,
keys={"active": wif},
num_retries=10
)
# from getpass import getpass
# self.bts.wallet.unlock(getpass())
set_shared_steem_instance(cls.bts)
cls.bts.set_default_account("test")
# Advote Text
advotetext = "Great read! Your high-quality travel content was selected by @travelfeed curator @{}. We just gave you a small upvote together with over 60 followers of the @travelfeed curation trail. <br> Have you heard of @travelfeed? Using the #travelfeed tag rewards authors and content creators who produce exceptional travel related articles, so be sure use our tag to get much bigger upvotes, resteems and be featured in our curation posts! <br> <center>[![TravelFeed](https://ipfs.busy.org/ipfs/QmNTkoKQNzuQbQGbcZ1exTMjvxYUprdnVczxnvib9VUSqB)](http://travelfeed.io/travelfeed/@travelfeed/introducing-travelfeed-featuring-steemit-s-best-travel-content) <br> **Learn more about our travel project by clicking on the banner above and join our community on [Discord](https://discord.gg/jWWu73H)**</center>"
# Manual comment text for short posts
manualshorttext = "Hi @{}, \n Thank you for participating in the #travelfeed curated tag. To maintain a level of quality on the project we have certain criteria that must be met for participation. Please review the following: https://travelfeed.io/@travelfeed/how-to-participate-use-travelfeed-in-your-posts \n **We require at least 250 words.** \n Thank you very much for your interest and we hope to read some great travel articles from you soon! \n If you believe that you have received this comment by mistake or have updated your post to fit our criteria, please reply to this comment with <code>!tfreview</code>. For further questions, please contact us on the [TravelFeed Discord](https://discord.gg/jWWu73H). \n Regards, @travelfeed"
# Manual comment text for posts that are not in English
manuallangtext = "Hi @{}, \n Thank you for participating in the #travelfeed curated tag. To maintain a level of quality on the project we have certain criteria that must be met for participation. Please review the following: https://travelfeed.io/@travelfeed/how-to-participate-use-travelfeed-in-your-posts \n We require at least 250 words **in English**. \n Thank you very much for your interest and we hope to read some great travel articles from you soon! \n If you believe that you have received this comment by mistake or have updated your post to fit our criteria, please reply to this comment with <code>!tfreview</code>. For further questions, please contact us on the [TravelFeed Discord](https://discord.gg/jWWu73H). \n Regards, @travelfeed"
# Copyright text
copyrighttext = "Hi @{}, \n Thank you for participating in the #travelfeed curated tag. To maintain a level of quality on the project we have certain criteria that must be met for participation. Please review the following: https://travelfeed.io/@travelfeed/how-to-participate-use-travelfeed-in-your-posts \n We require **proper sourcing** for all media and text that is not your own. \n If you have updated your post with sources, please reply to this comment with <code>!tfreview</code>. For further questions, please contact us on the [TravelFeed Discord](https://discord.gg/jWWu73H). \n Thank you very much for your interest and we hope to read some great travel articles from you soon! \n Regards, @travelfeed"
honours = {}
resteems = {}
walletpw = os.environ.get('UNLOCK') #Beem wallet passphrase must be set as environment variable
TOKEN = os.environ.get('TOKEN') #Discord secret token must be set as environment variable
logger = logging.getLogger(__name__)
logging.basicConfig(filename=logpath, format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
nl = NodeList()
weights = {'block': 0.1, 'history': 1, 'apicall': 0.1, 'config': 0.1}
node_list = nl.update_nodes(weights)
steem = Steem(nodes=node_list)
steem.set_default_nodes(node_list)
steem.wallet.unlock(walletpw)
blockchain = Blockchain()
acc = Account(curationaccount)
# blacklist = acc.get_mutings(raw_name_list=True)
blacklist = []
"""
Discord functions
"""
async def send_discord(msg, cnl):
"""Sends the message *msg* to the Discord channel *cnl*"""
await bot.wait_until_ready()
for op in b.stream(start=23483000, stop=23483200, threading=threading, thread_num=thread_num,
opNames=['vote']):
sys.stdout.write("\r%s" % op['block_num'])
opcount += 1
now = time.time()
total_duration = now - start_time
print(" votes: %d, time %.2f" % (opcount, total_duration))
return opcount, total_duration
if __name__ == "__main__":
node_setup = 1
threading = True
thread_num = 8
timeout = 10
nodes = NodeList()
nodes.update_nodes(weights={"block": 1})
node_list_wss = nodes.get_nodes(https=False)[:5]
node_list_https = nodes.get_nodes(wss=False)[:5]
vote_result = []
duration = []
stm_wss = Steem(node=node_list_wss, timeout=timeout)
stm_https = Steem(node=node_list_https, timeout=timeout)
print("Without threading wss")
opcount_wot_wss, total_duration_wot_wss = stream_votes(stm_wss, False, 8)
print("Without threading https")
opcount_wot_https, total_duration_wot_https = stream_votes(stm_https, False, 8)
if threading:
print("\n Threading with %d threads is activated now." % thread_num)
stm = Steem(node=node_list_wss, timeout=timeout)
result = report_node[benchmark]
rank = result["rank"]
if not result["ok"]:
rank = max_score + 1
score = (max_score - rank) / (max_score - 1) * 100
weighted_score = score * weights_dict[benchmark]
scores.append(weighted_score)
sum_score = 0
for score in scores:
sum_score += score
new_node["score"] = sum_score
for node_failing in failing_nodes:
if node["url"] == node_failing:
new_node["score"] = -1
new_nodes.append(new_node)
super(NodeList, self).__init__(new_nodes)