Browse Source

improved logging of tweets, start/shutdown msgs

New logging features have been added.  Tweets and their URLs are now printed to
the console's log.  Added new keyboard interrupt signal handler that tweets out
a shutdown message with a timestamp.
canonical
Andrew Rogers 7 months ago
parent
commit
cba10a910d
  1. 66
      pseudbot/bot.py

66
pseudbot/bot.py

@ -32,20 +32,33 @@ def parse_args(args: [str], name: str):
class PseudBot:
last_stat = None
def __init__(self, tcfg: dict):
def __init__(self, tcfg: dict, custom_welcome: str = None):
tauth = t.OAuthHandler(tcfg["consumer"], tcfg["consumer_secret"])
tauth.set_access_token(tcfg["tok"], tcfg["tok_secret"])
self.tapi = t.API(tauth)
welcome_tweet = "Testing at " + str(time())
if custom_welcome is not None:
welcome_tweet = custom_welcome
else:
welcome_tweet = "Powered on at " + str(int(time()))
self.wstatus = self.tapi.update_status(welcome_tweet)
self.screen_name = self.wstatus.user.screen_name
self.url_prefix = "https://twitter.com/" + self.screen_name + "/status/"
self._log_tweet(welcome_tweet, self.wstatus)
idr = open("last_id", mode="r")
self.last_id = int(idr.read())
idr.close()
sleep(0.5)
def _log_tweet(self, msg, tstat) -> None:
print(
'[INFO]: Tweeted "{}" ({})'.format(
msg, self.url_prefix + str(tstat.id)
)
)
def jdump(self, itms, echo: bool = False):
dfname = str(inspect.stack()[1][3]) + ".dump.json"
df = open(dfname, mode="w")
@ -71,9 +84,11 @@ class PseudBot:
"""
_stat = self.last_stat
try:
noodle = pasta.pop(0)
self.last_stat = self.tapi.update_status(
pasta.pop(0), in_reply_to_status_id=id_reply_to
noodle, in_reply_to_status_id=id_reply_to
)
self._log_tweet(noodle, self.last_stat)
except Forbidden:
return _stat
# print(status)
@ -87,7 +102,9 @@ class PseudBot:
def hello(self):
self.jdump(
self.tapi.update_status(
str(time()) + ": pseudbot is still under construction..."
# str(time()) + ": pseudbot is still under construction..."
str(time())
+ ": Hello pseudbot"
)._json
)
@ -108,10 +125,6 @@ class PseudBot:
if tweet.user.screen_name == self.screen_name:
continue
# self.jdump(tweet._json)
# print("Mentioned in: " + str(tweet.id))
# continue
self.last_id = max(tweet.id, self.last_id)
pasta = []
@ -120,7 +133,9 @@ class PseudBot:
if tweet.in_reply_to_status_id is not None:
pasta[0] = "@" + tweet.in_reply_to_screen_name + " " + pasta[0]
self.last_stat = self.tweet_pasta(tweet.in_reply_to_status_id, pasta)
self.last_stat = self.tweet_pasta(
tweet.in_reply_to_status_id, pasta
)
else:
pasta[0] = "@" + tweet.user.screen_name + " " + pasta[0]
self.last_stat = self.tweet_pasta(tweet.id, pasta)
@ -132,14 +147,23 @@ class PseudBot:
self.write_last_id()
def run_bot(self):
while True:
try:
self.reply_mentions()
sleep(120)
except TooManyRequests:
cooldown = 1000
print("[WARN]: Rate limited, cooling down for {} seconds...".format(cooldown))
sleep(cooldown)
try:
while True:
try:
self.reply_mentions()
sleep(120)
except TooManyRequests:
cooldown = 1000
print(
"[WARN]: Rate limited, cooling down for {} seconds...".format(
cooldown
)
)
sleep(cooldown)
except KeyboardInterrupt:
print()
shutdown_msg = "Shut down for maintenance at " + str(int(time()))
self._log_tweet(shutdown_msg, self.tapi.update_status(shutdown_msg))
def callm(pb: PseudBot, methname: str):
@ -149,5 +173,11 @@ def callm(pb: PseudBot, methname: str):
def main(args: [str], name: str) -> int:
opts = parse_args(args=args, name=name)
pb = PseudBot(j.loads(opts.cfg_json.read()))
if opts.action == "run_bot":
pb = PseudBot(j.loads(opts.cfg_json.read()))
else:
pb = PseudBot(
j.loads(opts.cfg_json.read()),
custom_welcome='Running method: "{}"'.format(opts.action),
)
callm(pb, opts.action)

Loading…
Cancel
Save