Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_raise_api_exception(self):
message = "USER_REQUIRED: 'Please log in to do that.'"
with self.recorder.use_cassette(
"TestObjector.test_raise_api_exception"
):
submission = self.reddit.submission("4b536h")
with pytest.raises(APIException) as excinfo:
submission.mod.approve()
assert excinfo.value.error_type == "USER_REQUIRED"
assert str(excinfo.value) == message
def main():
### Login loop
while True:
try:
reddit = praw.Reddit(client_id = c_id, \
client_secret = c_secret, \
user_agent = u_a, \
username = usrnm, \
password = passwd) # Connect to Reddit
break
except praw.exceptions.APIException: # Catch Reddit API error. REVIEW PARAMS
print("Reddit API error. Try again.")
except praw.exceptions.ClientException: # Catch client login error
print("Client-side error. Try again.")
### Scraping loop
while True:
while True:
subs = get_subreddits(reddit)
master = create_dict(subs)
get_settings(subs,master)
confirm = print_settings(master)
if confirm == "y":
break
for sub,settings in master.items():
async def _send_reply(self, message):
"""Send a reply to the user."""
response = '{}{}'.format(message, BOT_FOOTER if not self.marked_as_spam else '')
try:
reply = self.item.reply(response)
if config.use_memory:
memory = UserMemory()
memory.add(self.item.author.name, self.item.submission.id, reply.id)
if not self.should_send_pointers():
commands = get_commands_footer(reply.id)
edit_response = '{}{}{}'.format(message, commands, BOT_FOOTER if not self.marked_as_spam else '')
reply.edit(edit_response)
return True
except APIException as e:
if e.error_type == 'DELETED_COMMENT':
logger.info('Username mention was deleted')
return False
raise e
if ver:
print("Previosuly processed.")
else:
url = submission.url
if url.split('.')[-1] in image_file_exts:
if ver:
print("Found supported image ("+str(url)+"): ")
perc = calc_perc(url,"online")
if perc == err:
time.sleep(itv)
continue
comment = "^^\(true\) Black pixel percentage: **"+str(round(perc[0],2))+"%** ^^^\("+str(perc[1])+"/"+str(perc[2])+"\)"+bot_msg
if not dbg:
try:
submission.reply(comment)
except praw.exceptions.APIException as e:
if "RATELIMIT" in str(e):
print("\n[ERROR] Reddit API ratelimit reached.")
if ver:
print("\n\n"+str(e))
exit(1)
if ver:
print("\t"+str(comment))
else:
print("[DEBUG] ["+str(time.strftime("%H:%M:%S"))+"]\n\t"+str(repr(comment))+"\n\tfor "+str(url))
else:
if ver:
print("No image or unsupported format. ("+str(url)+")")
processed_list.append(submission.id)
with open(log, "a") as f:
if item.id in seen_deque['inbox']:
logger.debug('[Inbox] Skip: seen item: t4_{}'.format(item.id))
continue
if item.created_utc < check_time['inbox']:
if item.created_utc < start_time:
logger.debug('[Inbox] Skip: item was submitted before bot started: t4_{}'.format(item.id))
else:
logger.debug('[Inbox] Skip: timestamp was supplanted: t4_{}'.format(item.id))
continue
check_time['inbox'] += control_checkpoint_progression(item.created_utc - check_time['inbox'])
seen_deque['inbox'].append(item.id)
process_inbox_item(item)
except (praw.exceptions.PRAWException, prawcore.exceptions.PrawcoreException) as e:
if isinstance(e, praw.exceptions.APIException):
if e.error_type == 'RATELIMIT':
logger.info('Exception: ratelimit exceeded: {}'.format(e.message))
time.sleep(11*60)
else:
logger.warning('Exception: unhandled PRAW APIException exception:', exc_info=True)
elif isinstance(e, prawcore.exceptions.ResponseException):
logger.info('Exception: ResponseException: {}'.format(e.response))
time.sleep(5)
elif isinstance(e, prawcore.exceptions.RequestException):
logger.info('Exception: RequestException: {}'.format(e.original_exception))
time.sleep(5)
else:
logger.warning('Exception: unhandled PRAW exception:', exc_info=True)
running = True
retries = 0
while running:
try:
if retries >= 60:
resetAll()
flushMsgs()
checkAndCreate()
updateThreads()
retries = 0
sleep(60)
except KeyboardInterrupt:
logger.warning("[MANUAL SHUTDOWN]")
print getTimestamp() + "[MANUAL SHUTDOWN]\n"
running = False
except praw.exceptions.APIException:
retries += 1
print getTimestamp() + "API error, check log file [retries = " + str(retries) + "]"
logger.exception("[API ERROR:]")
sleep(60)
except UnicodeDecodeError:
retries += 1
print getTimestamp() + "UnicodeDecodeError, check log file [retries = " + str(retries) + "]"
logger.exception("[UNICODE ERROR:]")
flushMsgs()
except UnicodeEncodeError:
retries += 1
print getTimestamp() + "UnicodeEncodeError, check log file [retries = " + str(retries) + "]"
logger.exception("[UNICODE ERROR:]")
flushMsgs()
except Exception:
retries += 1
def _attempt_reply(self, mention):
text_of_parent = get_text_of_parent(mention)
LOGGER.info("Text of parent is: " + text_of_parent)
emojipasta = self._emojipasta_generator.generate_emojipasta(text_of_parent)
LOGGER.info("Generated emojipasta: " + emojipasta)
try:
mention.reply(emojipasta)
except praw.exceptions.APIException as e:
LOGGER.warning("API exception: " + e.message)
if e.error_type == "RATELIMIT":
return True
return False
def reply_and_save(reply,submission,dry_run):
assert(type(reply) == type({}))
assert('original_reply' in reply)
if dry_run:
print('Would reply to post %s' % submission.id)
print('With comment:\n%s' % reply['original_reply'])
comment_id = 'fake_comment_id_for_test'
else:
try:
print('Replying to post %s' % submission.id)
comment = submission.reply(reply['original_reply'])
except praw.exceptions.APIException as e:
if 'ratelimit' in e.message.lower():
print('Hmm, I\'m being rate limited. Waiting 60 seconds')
time.sleep(60)
print('Waking up, trying again')
submission.reply(reply['original_reply'])
comment_id = comment.id
save_initial(reply,submission.id,comment_id,dry_run)
schedule_checks(submission.id,dry_run)
def __try_repeatedly_with_timeout(func):
start_time = time.time()
too_many_tries_exception = True
while too_many_tries_exception:
try:
too_many_tries_exception = False
func()
except Exception as e:
if not isinstance(e, praw.exceptions.APIException) or time.time() - start_time >= MAX_TIMEOUT:
RedditBotImpl.__send_pm_with_error_to_creator(traceback.format_exc())
Logger.log_error()
else:
too_many_tries_exception = True
Logger.log_error_stacktrace(str(e))
time.sleep(ATTEMPT_TIMEOUT)