muskit/NijiHolo_EN_ID_Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wrap up catch-up mode, add tweets nuker

Browse Source
This commit is contained in:
muskit
2022-09-27 15:09:09 -07:00
committed by msk
parent 07e7e28dcb
commit 4f53bda1f8
6 changed files with 154 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/catchup.py
+23 -10
View File
@@ -62,10 +62,12 @@ def get_finished_user_timestamps(queue_file):
results = dict()
for line in queue_file:
tokens = line.split()
if len(tokens) != 3 or tokens[0][0] != '#':
if len(tokens) == 0: continue
if tokens[0][0] != '#':
print(f'{line} is our stopper!')
# reached end of accounts list
break
if tokens[2] != '-1':
results[int(tokens[1])] = float(tokens[2])
return results
@@ -86,11 +88,13 @@ def get_user_timestamps_str(queue_file):
async def get_cross_talent_tweets(queue_path):
finished_user_timestamps = dict()
ttweets_dict = dict()
posted_ttweets = set() # TODO: don't add TTweet to ttweets_dict if its id exists in posted_ttweets
# Populate structures with existing data from queue.txt
try:
with open(queue_path, 'r') as f:
finished_user_timestamps = get_finished_user_timestamps(f)
finished_user_timestamps = get_finished_user_timestamps(f)
print(finished_user_timestamps)
# Get existing queued TalentTweets
for line in f:
@@ -141,20 +145,23 @@ async def get_cross_talent_tweets(queue_path):
return ttweets_dict
async def process_queue(ttweets_dict: dict):
async def process_queue(ttweets_dict: dict) -> int:
global PROGRAM_ARGS
ttweets_posted = 0
if len(ttweets_dict) == 0: return
if len(ttweets_dict) == 0: return ttweets_posted
if PROGRAM_ARGS.announce_catchup:
TwAPI.instance.post_tweet(text=f'Starting to catch-up through {len(ttweets_dict)} logged tweets.')
TwAPI.instance.post_tweet(text=f'Starting to catch up through {len(ttweets_dict)} logged tweets.')
try:
while len(ttweets_dict) > 0:
key = list(ttweets_dict.keys())[0]
ttweet = ttweets_dict[key]
await TwAPI.instance.post_ttweet(ttweet)
if await TwAPI.instance.post_ttweet(ttweet, is_catchup=True):
ttweets_posted += 1
ttweets_dict.pop(key)
# TODO: add ttweet.tweet_id to some success list
except:
print('Unhandled error occurred while posting tweets from queue.')
traceback.print_exc()
@@ -170,10 +177,16 @@ async def process_queue(ttweets_dict: dict):
for ttweet in ttweets_dict.values():
f.write(f'{ttweet.serialize()}\n')
return ttweets_posted
async def run(program_args):
global PROGRAM_ARGS
PROGRAM_ARGS = program_args
queue_path = get_queue_path()
ttweets_dict = await get_cross_talent_tweets(queue_path)
print(f'got {len(ttweets_dict)} tweets')
await process_queue(ttweets_dict)
while True:
ttweets_dict = await get_cross_talent_tweets(queue_path)
print(f'found {len(ttweets_dict)} cross-company tweets')
if await process_queue(ttweets_dict) == 0:
print('Posted no new tweets; we\'re caught up!')
break
# TODO: go to listen mode
src/main.py
+18 -4
View File
@@ -2,6 +2,7 @@ import sys
import asyncio
import argparse
from argparse import RawTextHelpFormatter
import code
import nest_asyncio
@@ -14,8 +15,9 @@ from twapi import TwAPI
PROGRAM_ARGS = None
MODES_HELP_STR = '''mode to run the bot at:
l,listen: listen for new tweets from all accounts; will not terminate unless error occurs
c,catchup: scan all tweets from all accounts; will terminate when done'''
l,listen: listen for new tweets from all accounts; will not terminate unless error occurs
c,catchup: scan all tweets from all accounts; will terminate when done
d,delete-all: delete all tweets on account provided by secrets.ini; make sure the function is uncommented in twapi.py'''
def init_argparse():
p = argparse.ArgumentParser(description='Twitter bot that follows interactions between Nijisanji EN/ID and hololive EN/ID members.', formatter_class=RawTextHelpFormatter)
@@ -23,12 +25,21 @@ def init_argparse():
help=MODES_HELP_STR)
p.add_argument('--show-tokens', action='store_true', help='[DO NOT USE IN PUBLIC SETTING] print stored tokens from secrets.ini')
p.add_argument('--announce-catchup', action='store_true', help='In catch-up mode, post a tweet announcing catch-up mode.')
p.add_argument('--no-delay', action='store_true', help='In self-destruct mode, clear tweets without safety waiting.')
return p
def command_line():
# TODO: implement command line mode for manually controlling the bot
print('Shell coming soon. For now, here\'s a Python interpretor.')
code.interact(local=globals())
pass
async def self_destruct():
if not PROGRAM_ARGS.no_delay:
print('\033[31;6m-----DELETING ALL TWEETS IN 10 SECONDS!! PRESS CTRL+C TO CANCEL.-----\033[0m')
await asyncio.sleep(10)
await TwAPI.instance.nuke_tweets()
async def async_main():
global PROGRAM_ARGS
@@ -40,9 +51,12 @@ async def async_main():
case 'c' | 'catchup':
print('RUNNING IN CATCH-UP MODE\n')
await catchup.run(PROGRAM_ARGS)
case _:
case 'd' | 'delete-all':
print('WARNING: SELF-DESTRUCT MODE')
await self_destruct()
case 'cmd':
command_line()
#TODO: remove message
case _:
print('\ninvalid mode. run with no arguments or "-h" for help page, including mode list.')
return
src/talent_lists.py
+1 -3
View File
@@ -36,7 +36,5 @@ def init():
# nijiexID
__create_dict(f'{util.get_project_dir()}/lists/nijiexid.txt', niji_exid)
test_talents = {
1390637197167038464: 'PomuRainpuff'
}
test_talents = holo_en
src/talenttweet.py
+1
View File
@@ -5,6 +5,7 @@ import pytz
from twapi import *
import talent_lists
import util
class TalentTweet:
@staticmethod
src/twapi.py
+110 -15
View File
@@ -1,8 +1,9 @@
import asyncio
import datetime
import traceback
import tweepy
from tweetcapture import TweetCapture
import twint
import api_secrets
import talenttweet as tt
@@ -67,6 +68,72 @@ class TwAPI:
access_token=api_secrets.access_token(), access_token_secret=api_secrets.access_secret()
)
)
self.me = self.client.get_me().data
print(f'Assuming the account of @{self.me.data["username"]} ({self.me["id"]})')
## ---[COMMENT OUT WHEN NOT IN USE]---
async def nuke_tweets(self):
async def delete_tweet(id):
try:
self.client.delete_tweet(id)
except tweepy.TooManyRequests as e:
wait_for = float(e.response.headers["x-rate-limit-reset"]) - datetime.datetime.now().timestamp() + 1
print(f'\thit rate limit deleting {id}, retrying in {wait_for} seconds...')
await asyncio.sleep(wait_for)
print('continuing...')
await delete_tweet(id)
print(f'Retrieving all of {self.me["username"]}\'s tweets...')
tweets = self.get_all_tweet_ids_from_user(self.me['id'])
print(f'Retrieved {len(tweets)} tweets.')
if not len(tweets) > 0:
print('No tweets obtained. Make sure the profile is public.')
return
print(f'Deleting {len(tweets)} tweets...')
deleted_count = 0
try:
for tweet in tweets:
print(f'deleted {deleted_count}/{len(tweets)}')
await delete_tweet(tweet.id)
await asyncio.sleep(0.5)
deleted_count += 1
except:
print('Unhandled error occurred while trying to delete tweets.')
traceback.print_exc()
print('Try running again.')
else:
print('Saul Gone')
def get_all_tweet_ids_from_user(self, user_id):
next_page_token = None
tokens_retrieved = 0
tweets_retrieved = 0
tweets = list()
while True:
print(f'Retrieved {tokens_retrieved} tokens so far...')
resp = self.client.get_users_tweets(
user_id, max_results=100, pagination_token=next_page_token,
media_fields=TwAPI.TWEET_MEDIA_FIELDS,
tweet_fields=TwAPI.TWEET_FIELDS,
expansions=TwAPI.TWEET_EXPANSIONS
)
for tweet in resp.data:
tweets.append(tweet)
# update counters and pagination token
tweets_retrieved += resp.meta['result_count']
try:
next_page_token = resp.meta['next_token']
tokens_retrieved += 1
except KeyError:
print("next_token wasn't provided; we've reached the end!")
break # reached end of user's tweets
print(f'Retrieved {tweets_retrieved} tweets using {tokens_retrieved} tokens.')
return tweets
async def get_tweet_response(self, id, attempt = 0):
try:
@@ -84,7 +151,7 @@ class TwAPI:
await asyncio.sleep(wait_for)
return await self.get_tweet_response(id, attempt=attempt+1)
async def post_tweet(self, text, media_id=None, reply_to_tweet: int=None):
async def post_tweet(self, text='', media_id=None, reply_to_tweet: int=None):
try:
tweet = self.client.create_tweet(text=text, media_ids=None if media_id == None else [media_id], in_reply_to_tweet_id=reply_to_tweet)
return tweet
@@ -92,42 +159,70 @@ class TwAPI:
wait_for = float(e.response.headers["x-rate-limit-reset"]) - datetime.datetime.now().timestamp() + 1
print(f'\thit rate limit -- attempting to create Tweet again in {wait_for} seconds...')
await asyncio.sleep(wait_for)
return await self.post_tweet(text=text, media_ids=[media_id])
return await self.post_tweet(text=text, media_id=media_id, reply_to_tweet=reply_to_tweet)
async def get_ttweet_image_media_id(self, ttweet):
img = await util.create_ttweet_image(ttweet)
media = self.api.media_upload(img)
return media.media_id
async def post_ttweet(self, ttweet: tt.TalentTweet):
async def post_ttweet(self, ttweet: tt.TalentTweet, is_catchup=False):
print(f'------{ttweet.tweet_id} ({util.get_username_local(ttweet.author_id)})------')
REPLY = '{0} replied to {1}!\n'
MENTION = '{0} mentioned {1}!\n'
QUOTE_TWEET = '{0} quote tweeted {1}!\n'
MENTION = '{0} tweeted with '
def create_text():
if ttweet.reply_to is not None:
author_username = f'@/{util.get_username_online(ttweet.author_id)}'
author_username = f'@/{util.get_username_online(ttweet.author_id)}'
ret = str()
if is_catchup:
# ret += '[catch-up tweet]\n'
pass
ret += f'{ttweet.get_datetime_str()}\n'
if ttweet.reply_to is not None: # reply (w/ qrt; push it into mentions)
reply_username = f'@/{util.get_username_online(ttweet.reply_to)}'
ret += REPLY.format(author_username, reply_username)
mention_ids = set(ttweet.mentions)
mention_ids.add(ttweet.quote_retweeted)
try: mention_ids.remove(None)
except: pass
mention_usernames = [f'@/{util.get_username_online(x)}' for x in mention_ids]
elif ttweet.quote_retweeted is not None: # standalone qrt
quoted_username = f'@/{util.get_username_online(ttweet.quote_retweeted)}'
ret += QUOTE_TWEET.format(author_username, quoted_username)
elif len(ttweet.mentions) > 0: # standalone tweet w/ mentions
ret += MENTION.format(author_username)
else:
raise ValueError(f'TalentTweet {ttweet.tweet_id} has insufficient other parties')
ret = REPLY.format(author_username, reply_username)
# mention line
if len(mention_ids) > 0:
mention_usernames = [f'@/{util.get_username_online(x)}' for x in mention_ids]
ret += (
'mentions '
f'{" ".join(mention_usernames)}'
f'\n{util.ttweet_to_url(ttweet)}'
f'{" ".join(mention_usernames)}\n'
)
return ret
ret += f'\n{util.ttweet_to_url(ttweet)}'
return ret
img_media_id_task = asyncio.create_task(self.get_ttweet_image_media_id(ttweet))
text = create_text()
media_id = await img_media_id_task
twt_resp = await self.post_tweet(text)
twt_id = twt_resp.data['id']
await self.post_tweet(text='Image backup', reply_to_tweet=twt_id, media_id=media_id,)
try:
print('posting main tweet')
twt_resp = await self.post_tweet(text)
twt_id = twt_resp.data['id']
print('posting reply tweet')
await self.post_tweet(reply_to_tweet=twt_id, media_id=media_id,)
print('successfully posted ttweet!')
return True
except tweepy.Forbidden as e:
if 'duplicate content' in e.api_messages[0]:
print('Twitter says the TalentTweet is a duplicate; skipping error-free...')
return False
else:
raise e
src/util.py
+1 -1
View File
@@ -36,7 +36,7 @@ def get_key_from_value(d, val):
async def create_ttweet_image(ttweet):
tc = TweetCapture()
filename = 'img.png'
filename = f'{get_project_dir()}/img.png'
url = ttweet_to_url(ttweet)
img = None