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

spinning things back up slowly

Browse Source
This commit is contained in:
muskit
2023-08-14 02:47:29 -07:00
parent 9ab486d1bc
commit 8c7ba26300
5 changed files with 93 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/TEST_TWEETY.py
+72
View File
@@ -0,0 +1,72 @@
from time import sleep
from datetime import datetime, timedelta
from dotenv import dotenv_values
import pytz
from tweety import Twitter
from tweety.types import *
creds = dotenv_values()
app = Twitter("session")
app.sign_in(creds["username"], creds["password"])
def url(t: Tweet):
return f'https://twitter.com/{t.author.username}/status/{t.id}'
def print_tweets(tweets: list):
print(f'{len(tweets)} tweets:')
for t in tweets:
if isinstance(t, Tweet):
print(f'{t.date} : {url(t)} : RT? {t.is_retweet}')
elif isinstance(t, TweetThread):
print('-----------TTd----------')
print_tweets(t.tweets)
print('-----------end----------')
def get_tweets_from_user(uid: int | str, since: datetime = None) -> list:
reached_backdate = False
tweets: [Tweet] = []
if since == None:
since = datetime.utcnow().replace(tzinfo=pytz.utc) - timedelta(days=7)
print(f'Grabbing tweets since 7 days ago ({since.date()})')
if isinstance(uid, str):
name = uid
uid = app._get_user_id(uid)
print(f"{name} = {uid}")
def add_tweet(tweet: Tweet):
nonlocal reached_backdate
try:
if tweet.is_retweet or tweet.author.id == uid:
tweets.append(tweet)
if not reached_backdate and tweet.date <= since:
print("reached backdate")
reached_backdate = True
except AttributeError:
print("skipping malformed tweet: {tweet}")
return
uts = app.get_tweets(uid, replies=True)
while not reached_backdate:
cur_page = uts.tweets
print(f'obtained {len(cur_page)} tweets')
if len(cur_page) == 0: break
for e in cur_page:
if isinstance(e, Tweet):
add_tweet(e)
elif isinstance(e, TweetThread):
for t in e.tweets:
add_tweet(t)
uts.get_next_page()
tweets.sort(key=lambda t: t.id)
return tweets
tweets = get_tweets_from_user("ninakosaka", since=datetime(2023, 7, 1))
print_tweets(tweets)
src/talent_lists.py
+2
View File
@@ -47,6 +47,8 @@ def is_niji(id: int) -> bool:
def is_holo(id: int) -> bool:
return id in holo_en or id in holo_id
# For filtered stream
# DEPRECATED: thx elon
def get_twitter_rules():
global talents
rules = list()
src/talenttweet.py
+17 -77
View File
@@ -42,72 +42,26 @@ class TalentTweet:
tweet_id=tweet_id, author_id=author_id,
date_time=date_time, mrq=(mentions, reply_to, quote_retweeted)
)
@staticmethod
async def create_from_twint_tweet(tweet):
# MRQ
mentions = set()
reply_to = None
quoted_id = None
# reply_to/mentions
is_reply = tweet.id != int(tweet.conversation_id)
mentions = set([x['id'] for x in tweet.mentions])
if is_reply and len(tweet.reply_to) > 0:
reply_to = tweet.reply_to[0]['id'] # FIXME: QRT = is_reply and len(tweet.reply_to) == 0?
reply_others = [x['id'] for x in tweet.reply_to[1:]]
mentions.update(reply_others)
try: mentions.remove(reply_to)
except: pass
# qrt
if type(tweet.quote_url) == str:
# print(f'url: {tweet.quote_url} ({type(tweet.quote_url)})')
quote_tokens = tweet.quote_url.split('/')
if len(quote_tokens) >= 2:
quoted_username = quote_tokens[-2]
quoted_id = util.get_user_id_local(quoted_username)
if quoted_id == -1:
quoted_id = util.get_user_id_online(quoted_username)
# Serialized one-liner format:
# {tweet} {author} {time in seconds since epoch} m {mention_set} r {reply_to_author} q {quote_retweet_author}
def serialize(self):
s = f'{self.tweet_id} {self.author_id} {self.date_time.timestamp()} '
# NOTE: strptime doesn't attach timezone info.
# tweet's datetime will be in local time
date_time = datetime.datetime.strptime(tweet.datetime, '%Y-%m-%d %H:%M:%S %Z')
LOCAL_TIMEZONE = datetime.datetime.now().astimezone().tzinfo
date_time = date_time.replace(tzinfo=LOCAL_TIMEZONE) # attach system local timezone
return TalentTweet(tweet_id=tweet.id, author_id=tweet.user_id, date_time=date_time, mrq=(mentions, reply_to, quoted_id))
@staticmethod
def create_from_v2api_response(resp):
tweet = resp.data
if tweet is None: return None
if None not in [self.rt_target, self.rt_author_id]:
s += f'rt {self.rt_target} {self.rt_author_id}'
return s[:-1] # stop here since retweets can't have other info
mrq = twapi.TwAPI.get_mrq(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,
rt_target=rt_target,
rt_author_id=rt_author_id
)
@staticmethod
async def create_from_id(id):
resp = await twapi.TwAPI.instance.get_tweet_response(id)
return TalentTweet.create_from_v2api_response(resp)
if len(self.mentions) > 0:
s += 'm '
for id in self.mentions:
s += f'{id} '
if self.reply_to:
s += f'r {self.reply_to} '
if self.quote_retweeted:
s += f'q {self.quote_retweeted} '
return s[:-1]
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
@@ -140,20 +94,6 @@ class TalentTweet:
f'{self.serialize()}\n'
f'======================================================'
)
# Serialized one-liner format:
# {tweet} {author} {time in seconds since epoch} m {mention_set} r {reply_to_author} q {quote_retweet_author}
def serialize(self):
s = f'{self.tweet_id} {self.author_id} {self.date_time.timestamp()} '
if len(self.mentions) > 0:
s += 'm '
for id in self.mentions:
s += f'{id} '
if self.reply_to:
s += f'r {self.reply_to} '
if self.quote_retweeted:
s += f'q {self.quote_retweeted} '
return s[:-1]
def is_cross_company(self):
for other_id in self.all_parties: