Browse Source

basic bot functionality

The bot mostly works.  Not exactly happy with the code thus far,
may rewrite in the future.  I also am not using the tweepy 2.0 API...

run the bot with:
  pseudbot -a run_bot pseud_cfg.json
canonical
Andrew Rogers 7 months ago
parent
commit
a1527e4f61
  1. 3
      .gitignore
  2. 85
      pseudbot/bot.py

3
.gitignore vendored

@ -122,5 +122,8 @@ pseud.json
# Pseudbot JSON dumps
*.dump.json
# Pseudbot last id file
last_id
# The editor that people seem to like for some reason
.vscode

85
pseudbot/bot.py

@ -1,7 +1,10 @@
import argparse
import inspect
import json as j
import random
from time import sleep, time
import tweepy as t
from tweepy.errors import Forbidden, TooManyRequests
import typing
from .pastas import PASTAS
@ -27,11 +30,22 @@ def parse_args(args: [str], name: str):
class PseudBot:
last_stat = None
def __init__(self, tcfg: dict):
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())
self.wstatus = self.tapi.update_status(welcome_tweet)
self.screen_name = self.wstatus.user.screen_name
idr = open("last_id", mode="r")
self.last_id = int(idr.read())
idr.close()
sleep(0.5)
def jdump(self, itms, echo: bool = False):
dfname = str(inspect.stack()[1][3]) + ".dump.json"
df = open(dfname, mode="w")
@ -51,13 +65,82 @@ class PseudBot:
self.jdump(jsons, echo=True)
def tweet_pasta(self, id_reply_to: int, pasta: [str]):
"""
In this house we stan recursion.
"""
_stat = self.last_stat
try:
self.last_stat = self.tapi.update_status(
pasta.pop(0), in_reply_to_status_id=id_reply_to
)
except Forbidden:
return _stat
# print(status)
if len(pasta) > 0:
pasta[0] = "@" + self.last_stat.user.screen_name + " " + pasta[0]
sleep(2)
return self.tweet_pasta(self.last_stat.id, pasta)
else:
return self.last_stat
def hello(self):
self.jdump(
self.tapi.update_status(
"pseudbot is still under construction..."
str(time()) + ": pseudbot is still under construction..."
)._json
)
def reply_test(self):
pasta = random.choice(PASTAS)
pasta[0] = "@bustin4201 " + pasta[0]
print(self.tweet_pasta(1451688288591417348, pasta))
def write_last_id(self):
idw = open("last_id", mode="w")
idw.write(str(self.last_id))
idw.close()
def reply_mentions(self):
for tweet in t.Cursor(
self.tapi.mentions_timeline, since_id=self.last_id
).items():
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 = []
while len(pasta) < 1:
pasta = random.choice(PASTAS)
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)
else:
pasta[0] = "@" + tweet.user.screen_name + " " + pasta[0]
self.last_stat = self.tweet_pasta(tweet.id, pasta)
if self.last_stat is not None:
print("Finished chain with {}".format(self.last_stat.id))
sleep(10)
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)
def callm(pb: PseudBot, methname: str):
return getattr(pb, methname)()

Loading…
Cancel
Save