diff --git a/.gitignore b/.gitignore index 58ba0b2..ac961e8 100644 --- a/.gitignore +++ b/.gitignore @@ -143,5 +143,5 @@ cython_debug/ .vscode # project-specific (secret.ini: can't ignore existing file?) -secrets.ini *.png +session.json \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index f8c83f2..e949f7b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,6 @@ python-dotenv nest-asyncio pytz -Scweet +tweety-ns tweet-capture opencv-python \ No newline at end of file diff --git a/src/TEST_TWEETY.py b/src/TEST_TWEETY.py new file mode 100644 index 0000000..1bce124 --- /dev/null +++ b/src/TEST_TWEETY.py @@ -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) \ No newline at end of file diff --git a/src/talent_lists.py b/src/talent_lists.py index 02ee2d2..51b945b 100644 --- a/src/talent_lists.py +++ b/src/talent_lists.py @@ -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() diff --git a/src/talenttweet.py b/src/talenttweet.py index 0df9631..ab91737 100644 --- a/src/talenttweet.py +++ b/src/talenttweet.py @@ -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: