Browse Source

user_timeline dump, SOCKS support, list_actions

This commit involves a lot of refactoring that should've been moved to
a separate commit.  Some PseudBot methods that don't need to use
PseudBot attributes have been moved to the newly-created `util.py`.

The real_main function and its associated paraphernalia have been moved
to the also newly-created `cli.py`.

All PseudBot internal methods have been prepended with an underscore
(_).

A new method called list_actions has been added.  It generates a list
of CLI-callable methods and prints each and its accompanying docstring
to the terminal.

SOCKS proxy support has been added to Pseudbot.  The CLI now has a new
`-p` flag that supports fully-qualified SOCKS proxy urls.  For example:
    $ pseudbot run_bot -p socks5://username:password@host:port/

A user timeline dumper action named `user_timeline` has also been added
to PseudBot.  Here's an example of it in action:
    $ pseudbot user_timeline -s CheckmarkLs
pull/1/head
Albert Sanchez 8 months ago
parent
commit
d7d09d8f3f
  1. 71
      README.md
  2. 1
      img/live-demo-badge.svg
  3. 2
      pseudbot/__init__.py
  4. 218
      pseudbot/bot.py
  5. 107
      pseudbot/cli.py
  6. 2
      pseudbot/exceptions.py
  7. 34
      pseudbot/util.py
  8. 1
      requirements.txt

71
README.md

@ -1,6 +1,10 @@
<p align="center">
<img alt="pseudbot logo" src="./img/pseudbot-icon-1000.png" width=223 height=223>
<h1 align="center">Pseudbot</h1>
<h1 align="center">Pseudbot
<a href="https://twitter.com/pseudbot">
<img alt="Live Demo" src="./img/live-demo-badge.svg">
</a>
</h1>
</p>
<p align="center">
@ -170,3 +174,68 @@ Finished chain with 1452754226820571148
The next time you restart your bot, it will resume where it left off using the
`last_id` file.
## Getting help
### Basic help
Pseudbot has a basic help that can be called with `-h`:
```
$ pseudbot -h
usage: /home/anon/src/pseudbot/.venv/bin/pseudbot [-h] [-i REPLY_TO_ID] [-s SCREEN_NAME]
[-c CFG_JSON] [-p PROXY_URL]
action
positional arguments:
action Method to call. Use list_actions to see more information about which
actions are available.
optional arguments:
-h, --help show this help message and exit
-i REPLY_TO_ID, --reply-to-id REPLY_TO_ID
ID to reply to. Has no affect unless "action" is meant to be directed
at a specific ID.
-s SCREEN_NAME, --screen-name SCREEN_NAME
User screen name to run action on. Has no affect unless "action" is
meant to be directed at a specific user's screen name
-c CFG_JSON, --cfg-json CFG_JSON
JSON file with Twitter secrets
-p PROXY_URL, --proxy-url PROXY_URL
Use Twitter API through a SOCKS proxy.
```
### Actions
To get a list of actions that Pseudbot can perform, run `pseudbot` with
`list_actions`:
```
$ pseudbot list_actions
dump_all_mentions:
Dump all times your bot has been mentioned.
dump_mentions:
Dump all mentions since last_id.
Override with -i on the command line.
dump_tweet:
Dump the JSON data dictionary of a specific tweet.
If called from the CLI, requires -i to be set.
hello:
Tweet "Hello pseudbot" with a timestamp.
list_actions:
List actions that Pseudbot can run.
pasta_tweet:
Insert a copy pasta in a Tweet chain manually starting from a specific
tweet ID. Requires -i to be set if calling from the CLI.
run_bot:
Start Pseudbot in its main mode: mention listener mode.
timeline:
Get and dump your Pseudbot account's home timeline.
user_timeline:
Get all tweets from a user's timeline.
Requires target_screen_name to be set
(set by -s if using the CLI).
```

1
img/live-demo-badge.svg

@ -0,0 +1 @@
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="63" height="20" role="img" aria-label="live demo"><title>live demo</title><linearGradient id="s" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="r"><rect width="63" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#r)"><rect width="0" height="20" fill="#97ca00"/><rect x="0" width="63" height="20" fill="#97ca00"/><rect width="63" height="20" fill="url(#s)"/></g><g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" text-rendering="geometricPrecision" font-size="110"><text aria-hidden="true" x="315" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="530">live demo</text><text x="315" y="140" transform="scale(.1)" fill="#fff" textLength="530">live demo</text></g></svg>

After

Width:  |  Height:  |  Size: 923 B

2
pseudbot/__init__.py

@ -1,6 +1,6 @@
import typing
from .bot import main as real_main
from .cli import main as real_main
def main(args: [str], name: str) -> int:

218
pseudbot/bot.py

@ -1,67 +1,50 @@
import argparse
import inspect
import json as j
import random
from sys import stderr
from textwrap import indent
from time import sleep, time
import tweepy as t
from tweepy.errors import Forbidden, TooManyRequests
import typing
from .exceptions import *
from .pastas import PASTAS
def parse_args(args: [str], name: str):
parser = argparse.ArgumentParser(prog=name)
parser.add_argument(
"-i",
"--reply-to-id",
type=int,
default=None,
help="ID to reply to, "
+ 'has no affect unless "action" manually directs replies.',
)
parser.add_argument(
"-c",
"--cfg-json",
type=argparse.FileType("r"),
default="pseud.json",
help="JSON file with Twitter secrets",
)
parser.add_argument(
"action",
type=str,
default="timeline",
help="Method to call",
)
return parser.parse_args(args=args)
def get_timestamp_s() -> str:
return "🕑" + str(int(time()))
from .util import get_timestamp_s, jdump, log_t_by_sname, surl_prefix
class PseudBot:
last_stat = None
def __init__(
self, tcfg: dict, custom_welcome: str = None, last_id: int = None
self,
tcfg: dict,
custom_welcome: str = None,
last_id: int = None,
target_screen_name: str = None,
proxy_url: str = None,
quiet: bool = False,
):
tauth = t.OAuthHandler(tcfg["consumer"], tcfg["consumer_secret"])
tauth.set_access_token(tcfg["tok"], tcfg["tok_secret"])
self.tapi = t.API(tauth)
if custom_welcome is not None:
welcome_tweet = custom_welcome
if proxy_url is not None:
self.tapi = t.API(tauth, proxy=proxy_url)
else:
welcome_tweet = "Powered on at " + str(int(time()))
self.tapi = t.API(tauth)
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)
if quiet is False:
if custom_welcome is not None:
welcome_tweet = custom_welcome
else:
welcome_tweet = "Powered on at " + str(int(time()))
self.target_screen_name = target_screen_name
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)
if last_id is None:
idr = open("last_id", mode="r")
@ -71,33 +54,83 @@ class PseudBot:
else:
self.last_id = last_id
def list_actions(self):
"""
List actions that Pseudbot can run.
"""
actions = [
meth_name
for meth_name in dir(self)
if callable(getattr(self, meth_name))
and not meth_name.startswith("_")
]
for action_name in actions:
print(action_name + ":", end="")
docstr = getattr(self, action_name).__doc__
print(docstr)
if docstr is None:
print()
def _log_tweet(self, msg, tstat) -> None:
print('[TWEET]: "{}" ({})'.format(msg, self.url_prefix + str(tstat.id)))
def _jdump(self, itms, echo: bool = False):
dfname = (
str(inspect.stack()[1][3]) + "." + str(int(time())) + ".dump.json"
def _log_tweet_by_sname(self, tweet):
print(
'[@{}]: "{}" ({})'.format(
tweet.user.screen_name,
tweet.text,
surl_prefix(tweet.user.screen_name) + str(tweet.id),
)
)
df = open(dfname, mode="w")
pretty = j.dumps(itms, sort_keys=True, indent=2)
if echo is True:
print(pretty)
df.write(pretty)
def _check_sname_exists(self):
if self.target_screen_name is None:
raise PseudBotNoTargetScreenName
df.close()
def user_timeline(self):
"""
Get all tweets from a user's timeline.
Requires ``target_screen_name`` to be set
(set by ``-s`` if using the CLI).
"""
self._check_sname_exists()
tweets_j = []
rq_n = 0
for tweet in t.Cursor(
self.tapi.user_timeline,
screen_name=self.target_screen_name,
since_id=1,
).items():
log_t_by_sname(tweet)
tweets_j.append(tweet._json)
sleep(0.2)
rq_n += 1
if rq_n % 899 == 0:
print(
"[WARN]: At standard v1.1 API limit! "
+ "Sleeping for 15 minutes...",
file=stderr,
)
sleep(901)
jdump(tweets_j)
def timeline(self):
"""
Get and dump recent tweets from your Pseudbot account's home timeline.
"""
home_tl = self.tapi.home_timeline()
jsons = []
for tweet in home_tl:
jsons.append(tweet._json)
self._jdump(jsons, echo=True)
jdump(jsons, echo=True)
def _tweet_pasta(self, id_reply_to: int, pasta: [str]):
"""
In this house we stan recursion.
Recursively tweet an entire pasta, noodle by noodle::
In this house we stan recursion.
"""
_stat = self.last_stat
try:
@ -116,20 +149,33 @@ class PseudBot:
return self.last_stat
def hello(self):
"""
Tweet "Hello pseudbot" with a timestamp.
"""
hello_msg = get_timestamp_s() + ": Hello pseudbot"
hello_stat = self.tapi.update_status(hello_msg)
self._jdump(hello_stat._json)
jdump(hello_stat._json)
self._log_tweet(hello_msg, hello_stat)
def write_last_id(self):
def _write_last_id(self):
"""
Write the tweet ID of the last mention responded to.
"""
idw = open("last_id", mode="w")
idw.write(str(self.last_id))
idw.close()
def dump_all_mentions(self):
"""
Dump all times your bot has been mentioned.
"""
self.dump_mentions(start_id=1)
def dump_mentions(self, start_id: int = None):
"""
Dump all mentions since ``last_id``.
Override with ``-i`` on the command line.
"""
if start_id is None:
start_id = self.last_id
@ -150,17 +196,26 @@ class PseudBot:
self.last_id = max(tweet.id, self.last_id)
sleep(2)
self._jdump(tweets_j)
jdump(tweets_j)
def dump_tweet(self):
"""
Dump the JSON data dictionary of a specific tweet.
If called from the CLI, requires ``-i`` to be set.
"""
tweets = self.tapi.lookup_statuses([self.last_id])
jtweets = []
for tweet in tweets:
log_t_by_sname(tweet)
jtweets.append(tweet._json)
self._jdump(jtweets)
jdump(jtweets)
def pasta_tweet(self):
"""
Insert a copy pasta in a Tweet chain manually starting from a specific
tweet ID. Requires ``-i`` to be set if calling from the CLI.
"""
pasta = []
while len(pasta) < 1:
pasta = random.choice(PASTAS)
@ -171,6 +226,9 @@ class PseudBot:
self._send_pasta_chain(tweet)
def _send_pasta_chain(self, tweet):
"""
Send a copypasta chain.
"""
pasta = []
while len(pasta) < 1:
pasta = random.choice(PASTAS)
@ -198,7 +256,11 @@ class PseudBot:
pasta[0] = "@" + tweet.user.screen_name + " " + pasta[0]
self.last_stat = self._tweet_pasta(tweet.id, pasta)
def reply_mentions(self):
def _reply_mentions(self):
"""
Check mentions since ``last_id`` and reply in each thread with a
copypasta chain.
"""
for tweet in t.Cursor(
self.tapi.mentions_timeline, since_id=self.last_id
).items():
@ -213,13 +275,16 @@ class PseudBot:
print("Finished chain with {}".format(self.last_stat.id))
sleep(10)
self.write_last_id()
self._write_last_id()
def run_bot(self):
"""
Start Pseudbot in its main mode: mention listener mode.
"""
try:
while True:
try:
self.reply_mentions()
self._reply_mentions()
sleep(120)
except TooManyRequests:
cooldown = 1000
@ -235,32 +300,3 @@ class PseudBot:
"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):
return getattr(pb, methname)()
def main(args: [str], name: str) -> int:
opts = parse_args(args=args, name=name)
if opts.action == "run_bot":
pb = PseudBot(j.loads(opts.cfg_json.read()))
elif opts.action in ("pasta_tweet", "dump_tweet"):
if opts.reply_to_id is not None:
pb = PseudBot(
j.loads(opts.cfg_json.read()),
custom_welcome=get_timestamp_s()
+ ': Running method: "{}"'.format(opts.action),
last_id=opts.reply_to_id,
)
else:
print("[ERROR]: Must specify tweet ID to reply to!", file=stderr)
exit(1)
else:
pb = PseudBot(
j.loads(opts.cfg_json.read()),
custom_welcome=get_timestamp_s()
+ ': Running method: "{}"'.format(opts.action),
)
callm(pb, opts.action)

107
pseudbot/cli.py

@ -0,0 +1,107 @@
import argparse
import json as j
from sys import stderr
import typing
from .bot import PseudBot
from .util import get_timestamp_s
def parse_args(args: [str], name: str):
parser = argparse.ArgumentParser(prog=name)
parser.add_argument(
"-i",
"--reply-to-id",
type=int,
default=None,
help="ID to reply to. "
+ 'Has no affect unless "action" is meant to be directed at a specific '
+ "ID.",
)
parser.add_argument(
"-s",
"--screen-name",
type=str,
default=None,
help="User screen name to run action on. "
+ 'Has no affect unless "action" is meant to be directed at a '
+ "specific user's screen name",
)
parser.add_argument(
"-c",
"--cfg-json",
type=argparse.FileType("r"),
default="pseud.json",
help="JSON file with Twitter secrets",
)
parser.add_argument(
"-p",
"--proxy-url",
type=str,
default=None,
help="Use Twitter API through a SOCKS proxy.",
)
parser.add_argument(
"action",
type=str,
default="timeline",
help="Method to call. "
+ "Use list_actions to see more information about which actions "
+ "are available.",
)
return parser.parse_args(args=args)
def callm(pb: PseudBot, methname: str):
return getattr(pb, methname)()
def main(args: [str], name: str) -> int:
opts = parse_args(args=args, name=name)
custom_welcome = get_timestamp_s() + ': Running method: "{}"'.format(
opts.action
)
tcfg = j.loads(opts.cfg_json.read())
if opts.action == "run_bot":
pb = PseudBot(tcfg=tcfg, proxy_url=opts.proxy_url)
elif opts.action == "list_actions":
pb = PseudBot(tcfg=tcfg, quiet=True)
elif opts.action in ("pasta_tweet", "dump_tweet"):
if opts.reply_to_id is not None:
pb = PseudBot(
tcfg=tcfg,
custom_welcome=custom_welcome,
last_id=opts.reply_to_id,
proxy_url=opts.proxy_url,
)
else:
print(
"[ERROR]: Must specify tweet ID with -i",
file=stderr,
)
exit(1)
elif opts.action in ("user_timeline",):
print(opts.action)
if opts.screen_name is not None:
pb = PseudBot(
tcfg=tcfg,
custom_welcome=custom_welcome,
target_screen_name=opts.screen_name,
proxy_url=opts.proxy_url,
)
else:
print(
"[ERROR]: Must specify screen name with -s",
file=stderr,
)
exit(1)
else:
pb = PseudBot(
tcfg=tcfg,
custom_welcome=custom_welcome,
proxy_url=opts.proxy_url,
)
callm(pb, opts.action)

2
pseudbot/exceptions.py

@ -0,0 +1,2 @@
class PseudBotNoTargetScreenName(Exception):
pass

34
pseudbot/util.py

@ -0,0 +1,34 @@
import inspect
import json as j
from time import time
import typing
def get_timestamp_s() -> str:
return "🕑" + str(int(time()))
def surl_prefix(screen_name: str):
return "https://twitter.com/" + screen_name + "/status/"
def jdump(itms, echo: bool = False):
dfname = str(inspect.stack()[1][3]) + "." + str(int(time())) + ".dump.json"
df = open(dfname, mode="w")
pretty = j.dumps(itms, sort_keys=True, indent=2)
if echo is True:
print(pretty)
df.write(pretty)
df.close()
def log_t_by_sname(tweet):
print(
'[@{}]: "{}" ({})'.format(
tweet.user.screen_name,
tweet.text,
surl_prefix(tweet.user.screen_name) + str(tweet.id),
)
)

1
requirements.txt

@ -1 +1,2 @@
tweepy
requests[socks]

Loading…
Cancel
Save