def limit_handled(cursor: tweepy.Cursor):
"""Wrap cursor access with rate limiting
:param cursor: The cursor to siphon
:returns: Cursor items
"""
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
time.sleep(15 * 60)
python类RateLimitError()的实例源码
def get_tweets(self):
"""
Generator used to retrieve tweets from the user
:return:
"""
while True:
try:
yield self.user_cursor.pages().next()
except tweepy.RateLimitError:
logging.info("[%s] Timeout Reached, Sleep for 15 minutes before restart" % self.process_name)
time.sleep(15 * 60)
logging.info("[%s] Waking up. Try again" % self.process_name)
except StopIteration:
logging.info("[%s] Stop Iteration, process complete" % self.process_name)
break
except Exception as e:
logging.info("[%s] Generic Error, restart in 60 seconds: %s" % (self.process_name, e))
time.sleep(60)
def _handle_twitter_rate_limit(cursor):
'''Handle twitter rate limits. If rate limit is reached, the next element
will be accessed again after sleep time'''
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
log.info('Twitter API rate limit error. Sleeping for {} secs.') \
.format(TWITTER_API_RATE_LIMIT_PERIOD)
sleep_time = TWITTER_API_RATE_LIMIT_PERIOD
time.sleep(sleep_time)
except tweepy.TweepError as e:
if str(e.api_code) == TWITTER_API_USER_NOT_FOUND_ERROR_CODE:
raise ValueError(
'Requested user was not found. Check your configuration')
raise e
def limit_handled(cursor):
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
timestamp = time.strftime("%d.%m.%Y %H:%M:%S", time.localtime())
print('Warning: Rate limit reached! ' + timestamp)
time.sleep(15 * 60)
def limit_handled(cursor):
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
time.sleep(15 * 60)
def limit_handled(cursor):
"""
:param cursor: A cursor to iterate
:type cursor: tweepy.Cursor
"""
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
sleep(60)
def get_user_init():
global status
global path
global rev
try:
path = 'D:/Twitter/Depth-%d/%s' % (depth_num,str(user_id))
try:
os.makedirs(path)
except FileExistsError:
print("Folder already exist")
status="skip"
return
data = api.get_user(user_id)
if data['protected']==False:
get_user(data)
rev=1
else:
status="skip"
print("Protected")
except tweepy.RateLimitError:
countdown(960)
get_user_init()
except tweepy.TweepError as e:
if tweepy.TweepError is "[{'message': 'Over capacity', 'code': 130}]" or e[0]['code'] is 130:
countdown_te(600,e)
get_user_init()
elif tweepy.TweepError is "[{'message': 'User not found.', 'code':50}]" or e[0]['code'] is 50:
status="skip"
return
else:
print(e)
def get_id(sn):
try:
return(api.get_user(screen_name=sn)['id'])
except tweepy.RateLimitError:
countdown(960)
get_id(sn)
except tweepy.TweepError as e:
if tweepy.TweepError is "[{u'message': u'Over capacity', u'code': 130}]" or e is "[{u'message': u'Over capacity', u'code': 130}]":
countdown_te(600,e)
get_id(sn)
else:
print(e)
def limit_handler(cursor):
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
countdown(960)
except tweepy.TweepError as e:
if tweepy.TweepError is "[{u'message': u'Over capacity', u'code': 130}]" or e is "[{u'message': u'Over capacity', u'code': 130}]":
countdown_te(600,e)
else:
print(e)
py_twitter_scrape.py 文件源码
项目:social-fork-analysis
作者: ethereumproject
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def limit_handled(cursor):
while True:
try:
yield cursor.next()
except tweepy.RateLimitError as error:
print('RateLimitError')
print(error)
time.sleep(60 * 5)
except tweepy.TweepError as error:
print('tweepy.TweepError')
print(error)
time.sleep(60 * 5)
network_builder.py 文件源码
项目:twitter-social-affiliation-network
作者: zacharykstine
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
def limit_handled(cursor):
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
print("\nRateLimitError... sleeping...\n")
time.sleep(15 * 60)
except tweepy.TweepError as e:
if e.message.split()[-1] == '429':
print("\nTweepError 429... sleeping...\n")
time.sleep(15 * 60)
else:
print(e.message)
def rate_limited(cursor, sleeping_time=(15*60+5)):
while True:
try:
yield next(cursor)
except tweepy.RateLimitError:
print("Sleeping...", file=sys.stderr)
time.sleep(sleeping_time)
def ratelimit_backoff(fxn, *args, **kwargs):
for _ in range(3):
try:
return fxn(*args, **kwargs)
except tweepy.RateLimitError:
yield gen.sleep(5 * 60)
raise tweepy.RateLimitError()
def get_friends(self):
for friend_id in self.api.friends_ids(self.user.id):
try:
yield self.api.get_user(friend_id)
except tweepy.RateLimitError:
logging.info("[%s] Timeout Reached, Sleep for 15 minutes before restart" % self.process_name)
time.sleep(15 * 60)
logging.info("[%s] Waking up. Try again" % self.process_name)
except StopIteration:
logging.info("[%s] Stop Iteration, process complete" % self.process_name)
break
except Exception as e:
logging.info("[%s] Generic Error, restart in 60 seconds: %s" % (self.process_name, e))
time.sleep(60)
def get_followers(self):
while True:
try:
yield self.user_cursor.pages().next()
except tweepy.RateLimitError:
logging.info("[%s] Timeout Reached, Sleep for 15 minutes before restart" % self.process_name)
time.sleep(15 * 60)
logging.info("[%s] Waking up. Try again" % self.process_name)
except StopIteration:
logging.info("[%s] Stop Iteration, process complete" % self.process_name)
break
except Exception as e:
logging.info("[%s] Generic Error, restart in 60 seconds: %s" % (self.process_name, e))
time.sleep(60)
def create_api():
if APIHandler._api is None:
try:
with open(os.path.join(os.path.dirname(__file__), '../api_keys/twitter_keys'), 'r') as f:
keys = f.readlines()
except FileNotFoundError:
raise NoSecretFileException("File containing twitter OAuth keys doesn't exists. "
"Please create a file named twitter_keys "
"containing your OAuth credentials "
"in ultron/ultron/api_keys directory.")
keys = [key.strip() for key in keys]
if len(keys) != 4:
raise InvalidAPIException('Twitter requires 4 keys. But, '
+ str(len(keys)) +
' are present. Please ensure '
'that keys are placed in same order '
'as mentioned in README')
else:
consumer_key = keys[0]
consumer_secret = keys[1]
access_token = keys[2]
access_token_secret = keys[3]
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
APIHandler._api = tweepy.API(auth)
try:
APIHandler._api.verify_credentials()
except RateLimitError:
raise MaximumLimitCrossedException('Rate limit for API is reached. '
'Please wait for 15 minutes.')
except TweepError:
raise InvalidAPIException('Invalid or expired token')
return APIHandler._api
def limit_handled (self, cursor):
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
time.sleep(15 * 60)
def limit_handled (self, cursor):
while True:
try:
yield cursor.next()
except tweepy.RateLimitError:
time.sleep(15 * 60)