How to use botometer - 10 common examples
To help you get started, we’ve selected a few botometer examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
internetlab-br / Twitter-Bots / prob_bot.py View on Github 
sql = "UPDATE `bot_followers`.`seguidores` SET `probabilidade`='"+str(prob)+"'WHERE `id`='"+str(id)+"';"
cursor.execute(sql)
db.commit()
db.close()
mashape_key = ""
twitter_app_auth = {
'consumer_key': '',
'consumer_secret': '',
'access_token': '',
'access_token_secret': '',
}
bom = botometer.Botometer(wait_on_ratelimit=True,
mashape_key=mashape_key,
**twitter_app_auth)
auth = tweepy.OAuthHandler('', '')
auth.set_access_token('',
'')
api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True, retry_count=3, retry_delay=60,
retry_errors=set([503]))
lista = create_lista()
random.shuffle(lista)
count = 0def bot_check_by_id(userId):
fileKeys = open('../credentials/credentialsTwitter.json').read()
keys = json.loads(fileKeys)
mashape_key = keys['mashape_key']
twitter_app_auth = {
'consumer_key': keys['consumer_key'],
'consumer_secret': keys['consumer_secret'],
'access_token': keys['access_token'],
'access_token_secret': keys['access_token_secret'],
}
bom = botometer.Botometer(wait_on_ratelimit=True,
mashape_key=mashape_key,
**twitter_app_auth)
# Check a single account by id
result = bom.check_account(userId)
return result['scores']def bot_check_by_screenname(userId):
fileKeys = open('../credentials/credentialsTwitter.json').read()
keys = json.loads(fileKeys)
mashape_key = keys['mashape_key']
twitter_app_auth = {
'consumer_key': keys['consumer_key'],
'consumer_secret': keys['consumer_secret'],
'access_token': keys['access_token'],
'access_token_secret': keys['access_token_secret'],
}
bom = botometer.Botometer(wait_on_ratelimit=True,
mashape_key=mashape_key,
**twitter_app_auth)
# Check a single account by id
result = bom.check_account('@'+userId)
return result['scores']def set_bot_or_not(cls, mspreaders):
"""Fetch bot score of top spreaders.
Parameters
----------
mspreaders : object
An instance of model Top20SpreaderMonthly.
"""
logger.info('Fetching bot score for top spreaders ...')
rapid_key = cls.conf['botometer']['rapid_key']
botometer_api_url = 'https://botometer-pro.p.rapidapi.com'
logger.info(rapid_key)
bon = botometer.Botometer(
botometer_api_url=botometer_api_url,
rapidapi_key=rapid_key,
wait_on_ratelimit=True,
**cls.conf['botometer']['twitter_app_credentials'])
max_retries = 3
num_retries = 0
for ms in mspreaders:
for num_retries in range(max_retries + 1):
result = None
try:
result = bon.check_account(ms.user_raw_id)
except (TweepError, NoTimelineError) as e:
err_msg = '{}: {}'.format(
type(e).__name__,
getattr(e, 'msg', '') or getattr(e, 'reason', ''),
)def create_botometer(mashape_key, twitter_api_auth):
try:
bom = Botometer(wait_on_rate_limit=True, mashape_key=mashape_key, **twitter_api_auth)
except tweepy.TweepError:
error(config_complete_path, 'Failed to connect to Botometer API')
return bomdef __init__(self, queue, api_hook, conn_number, out_dir, mashape_key):
multiprocessing.Process.__init__(self)
self.queue = queue
self.api_hook = api_hook
self.out_dir = out_dir
self.conn_number = conn_number
twitter_app_auth = {
'consumer_key':api_hook.CONSUMER_KEY,
'consumer_secret': api_hook.CONSUMER_SECRET,
'access_token': api_hook.access_token ,
'access_token_secret': api_hook.access_token_secret,
"wait_on_ratelimit": True
}
self.bom = botometer.Botometer(mashape_key=mashape_key,**twitter_app_auth)
self.mashape_key = mashape_keydef _get_result(self):
try:
result = self.botometer.check_account(self.user_id)
except NoTimelineError:
self.status = BotometerStatus.UNAVAILABLE
else:
self._probability = result.get("cap", {}).get("universal")
self.status = BotometerStatus.READYdef run(self):
print('Worker started')
# do some initialization here
while True:
data = self.queue.get(True)
try:
if data is None:
print('ALL FINISHED!!!!', self.conn_number)
break
print 'checking: ', data
result = self.bom._get_twitter_data(data)
json.dump(result, gzip.open(os.path.join(self.out_dir,str(data)+".gz"),"wb"))
sleep(7)
except botometer.NoTimelineError:
print 'no timeline for: ', data, ' continuing'
except TweepError as e:
print e
except KeyboardInterrupt as e:
print e
break
except Exception:
print('FAILED:: ', data)
exc_type, exc_value, exc_traceback = sys.exc_info()
print("*** print_tb:")
traceback.print_tb(exc_traceback, limit=50, file=sys.stdout)
print("*** print_exception:")
print 'finished: ', datarapid_key = cls.conf['botometer']['rapid_key']
botometer_api_url = 'https://botometer-pro.p.rapidapi.com'
logger.info(rapid_key)
bon = botometer.Botometer(
botometer_api_url=botometer_api_url,
rapidapi_key=rapid_key,
wait_on_ratelimit=True,
**cls.conf['botometer']['twitter_app_credentials'])
max_retries = 3
num_retries = 0
for ms in mspreaders:
for num_retries in range(max_retries + 1):
result = None
try:
result = bon.check_account(ms.user_raw_id)
except (TweepError, NoTimelineError) as e:
err_msg = '{}: {}'.format(
type(e).__name__,
getattr(e, 'msg', '') or getattr(e, 'reason', ''),
)
result = {'error': err_msg}
except (Timeout, ConnectionError, HTTPError) as e:
if num_retries >= max_retries:
raise
else:
time.sleep(2**num_retries)
if result is not None:
ms.bot_or_not = result
breakdef run(self):
self.status = BotometerStatus.RUNNING
try:
result = self.botometer.check_account(self.user_id)
except NoTimelineError:
self.status = BotometerStatus.UNAVAILABLE
self._probability = 0.0 # let's assume it's not a bot
else:
self.status = BotometerStatus.READY
self._probability = result.get("cap", {}).get("universal")botometer
Check Twitter accounts for bot behavior
