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

fully functional?!

Browse Source
This commit is contained in:
msk
2022-09-27 22:04:26 -07:00
parent 609e034063
commit 978ba92901
7 changed files with 120 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -3
View File
@@ -142,8 +142,8 @@ cython_debug/
# VS Code files
.vscode
# project-specific
/secrets.ini
/queue.txt
# project-specific (secret.ini: can't ignore existing file?)
secrets.ini
queue.txt
/img.png
/src/img.png
src/listen.py
+30 -3
View File
@@ -1,7 +1,34 @@
## The bot's listen mode
# Continuously listen for cross-company interactions.
import twapi
import tweepy
from talenttweet import TalentTweet
async def run():
pass
from twapi import TwAPI
import api_secrets
import talent_lists as tl
async def on_response(resp):
print(resp)
print(resp.data)
ttweet = TalentTweet.create_from_v2api_response(resp)
await TwAPI.instance.post_ttweet(ttweet)
def run():
sc = tweepy.StreamingClient(api_secrets.bearer_token())
# clear rules
rules_resp = sc.get_rules()
if rules_resp.data:
sc.delete_rules(rules_resp.data)
# create new rules
for rule in tl.get_twitter_rules():
sc.add_rules(tweepy.StreamRule(rule))
sc.on_response=on_response
sc.filter(
expansions=TwAPI.TWEET_EXPANSIONS,
media_fields=TwAPI.TWEET_MEDIA_FIELDS,
tweet_fields=TwAPI.TWEET_FIELDS
)
src/main.py
+1
View File
@@ -51,6 +51,7 @@ async def async_main():
case 'c' | 'catchup':
print('RUNNING IN CATCH-UP MODE\n')
await catchup.run(PROGRAM_ARGS)
await listen.run()
case 'd' | 'delete-all':
print('WARNING: SELF-DESTRUCT MODE')
await self_destruct()
src/talent_lists.py
+15 -1
View File
@@ -17,7 +17,7 @@ def __create_dict(file, _dict):
if len(words) == 2 and line[0] != '#':
name, id = line.split()
talents[int(id)] = name
# name = util.get_username_online(id) # attempt to get updated name
name = util.get_username_online(id, default=name) # attempt to get updated name
talents[int(id)] = name
_dict[int(id)] = name
def init():
@@ -38,3 +38,17 @@ def init():
test_talents = holo_en
def get_twitter_rules():
global talents
rules = list()
names = list(talents.values())
curr_rule = f'from:{names[0]}'
for name in list(talents.values())[1:]:
test_rule = curr_rule + f' OR from:{name}'
if len(test_rule) > 512:
rules.append(curr_rule)
curr_rule = f'from:{name}'
else:
curr_rule = test_rule
return rules
src/talenttweet.py
+22 -5
View File
@@ -72,26 +72,43 @@ class TalentTweet:
date_time = datetime.datetime.strptime(tweet.datetime, '%Y-%m-%d %H:%M:%S %Z')
return TalentTweet(tweet_id=tweet.id, author_id=tweet.user_id, date_time=date_time, mrq=(mentions, reply_to, quoted_id))
@staticmethod
async def create_from_id(id):
resp = await TwAPI.instance.get_tweet_response(id)
def create_from_v2api_response(resp):
tweet = resp.data
mrq = TwAPI.get_mrq(tweet, resp)
rt_target = None
rt_author_id = None
# check if is RT
if tweet.referenced_tweets is not None and len(tweet.referenced_tweets) > 0:
for ref in tweet.referenced_tweets:
if ref.type == 'retweeted':
rt_target = ref.id
for incl_tweet in resp.includes['tweets']:
if incl_tweet.id == ref.id:
rt_author_id = incl_tweet.author_id
return TalentTweet(
tweet_id=tweet.id,
author_id=tweet.author_id,
date_time=tweet.created_at,
mrq=mrq
mrq=mrq,
rt_target=rt_target,
rt_author_id=rt_author_id
)
def __init__(self, tweet_id: int, author_id: int,date_time: datetime.datetime, mrq: tuple):
@staticmethod
async def create_from_id(id):
resp = await TwAPI.instance.get_tweet_response(id)
return TalentTweet.create_from_v2api_response(resp)
def __init__(self, tweet_id: int, author_id: int, date_time: datetime.datetime, mrq: tuple, rt_target: int=None, rt_author_id: int=None):
self.tweet_id, self.author_id = tweet_id, author_id
self.date_time = date_time
self.mentions = tuple(int(x) for x in mrq[0])
self.reply_to = int(mrq[1]) if mrq[1] is not None else None
self.quote_retweeted = int(mrq[2]) if mrq[2] is not None else None
self.rt_target, self.rt_author_id = rt_target, rt_author_id
# all users involved, except for the author
self.all_parties = {self.reply_to, self.quote_retweeted}
src/twapi.py
+15 -10
View File
@@ -3,7 +3,6 @@ import datetime
import traceback
import tweepy
import twint
import api_secrets
import talenttweet as tt
@@ -13,7 +12,7 @@ class TwAPI:
tweets_fetched = 0
instance = None
TWEET_MEDIA_FIELDS = ['url']
TWEET_FIELDS = ['created_at', 'in_reply_to_user_id']
TWEET_FIELDS = ['created_at', 'in_reply_to_user_id', 'referenced_tweets']
TWEET_EXPANSIONS = ['entities.mentions.username', 'referenced_tweets.id.author_id']
# Returns a tuple of user IDs:(reply_to, qrt, {mentions})
@@ -171,10 +170,12 @@ class TwAPI:
REPLY = '{0} replied to {1}!\n'
QUOTE_TWEET = '{0} quote tweeted {1}!\n'
MENTION = '{0} tweeted!\n'
TWEET = '{0} tweeted!\n'
RETWEET = '{0} retweeted {1}!\n'
def create_text():
author_username = f'@/{util.get_username_online(ttweet.author_id)}'
author_username = f'@/{util.get_username_local(ttweet.author_id)}'
mention_ids = set()
ret = str()
if is_catchup:
# ret += '[catch-up tweet]\n'
@@ -182,8 +183,11 @@ class TwAPI:
pass
# Tweet types
if ttweet.reply_to is not None: # reply (w/ qrt; push it into mentions)
reply_username = f'@/{util.get_username_online(ttweet.reply_to)}'
if ttweet.rt_target is not None: # standalone tweet
ret += RETWEET.format(author_username, f'@/{util.get_username(ttweet.rt_author_id)}')
mention_ids.clear()
elif ttweet.reply_to is not None: # reply (w/ qrt; push it into mentions)
reply_username = f'@/{util.get_username_local(ttweet.reply_to)}'
ret += REPLY.format(author_username, reply_username)
mention_ids = set(ttweet.mentions)
@@ -191,16 +195,16 @@ class TwAPI:
try: mention_ids.remove(None)
except: pass
elif ttweet.quote_retweeted is not None: # standalone qrt
quoted_username = f'@/{util.get_username_online(ttweet.quote_retweeted)}'
quoted_username = f'@/{util.get_username_local(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)
ret += TWEET.format(author_username)
else:
raise ValueError(f'TalentTweet {ttweet.tweet_id} has insufficient other parties')
# mention line
if len(mention_ids) > 0:
mention_usernames = [f'@/{util.get_username_online(x)}' for x in mention_ids]
mention_usernames = [f'@/{util.get_username_local(x)}' for x in mention_ids]
ret += (
'mentioning '
f'{" ".join(mention_usernames)}\n'
@@ -210,11 +214,12 @@ class TwAPI:
img_media_id_task = asyncio.create_task(self.get_ttweet_image_media_id(ttweet))
text = create_text()
media_id = await img_media_id_task
try:
print('posting main tweet')
twt_resp = await self.post_tweet(text)
twt_id = twt_resp.data['id']
print('waiting on reply img')
media_id = await img_media_id_task
print('posting reply tweet')
await self.post_tweet(reply_to_tweet=twt_id, media_id=media_id,)
print('successfully posted ttweet!')
src/util.py
+34 -13
View File
@@ -1,10 +1,12 @@
## Shared utility functions.
import datetime
import os
import traceback
import datetime
import pytz
import twint
import twapi
from tweetcapture import TweetCapture
import talent_lists
@@ -60,21 +62,40 @@ def ttweet_to_url(ttweet):
username = get_username_online(ttweet.author_id)
return f'https://twitter.com/{username}/status/{ttweet.tweet_id}'
def get_username_local(user_id):
return talent_lists.talents.get(user_id, f'#{id}')
def get_username_local(id):
return talent_lists.talents.get(id, f'{id}')
def get_username_online(user_id):
c = twint.Config()
c.User_id = user_id
c.Store_object = True
c.Hide_output = True
# twint
# May not work with short user IDs (ie. 1354241437)
# def get_username_online(id, default=None):
# c = twint.Config()
# c.User_id = id
# c.Store_object = True
# c.Hide_output = True
# try:
# twint.output.users_list.clear()
# twint.run.Lookup(c)
# user = twint.output.users_list[0]
# return user.username
# except:
# return str(default) if default is not None else f'{id}'
# API v2 (tweepy)
# Short user IDs (ie. 1354241437) apparently don't work with twint
def get_username_online(id, default=None):
try:
twint.output.users_list.clear()
twint.run.Lookup(c)
user = twint.output.users_list[0]
return user.username
resp = twapi.TwAPI.instance.client.get_user(id=id)
return resp.data.username
except:
return f'#{user_id}'
print(f'Unhandled error retrieving username for {id}!')
traceback.print_exc()
return str(default) if default is not None else f'{id}'
## Attempt to pull username from local; pull from online if doesn't exist.
def get_username(id):
ret = talent_lists.talents.get(id, None)
if ret == None:
return get_username_online(id)
def get_user_id_local(username) -> int:
talent_usernames = list(talent_lists.talents.values())