feat: get liked tweets from a specific time

bot/services/liked_tweet.py

@@ -33,6 +33,28 @@ def get_liked_tweets(user_id: str) -> list[tweepy.Tweet]:
     return all_tweets
 
 
+def get_liked_tweets_since(user_id: str, since: int | None = None):
+    all_tweets: list[tweepy.Tweet] = []
+    next_token = None
+
+    for _ in count(1):
+        tweets, tweets_meta = fetch_liked_tweets(user_id, next_token)
+        all_tweets += tweets
+
+        if since and any(
+            tweet.created_at.timestamp() * 1000 < since for tweet in tweets
+        ):
+            break
+
+        if "next_token" not in tweets_meta:
+            # when we don't have "next_token" in meta object, there is no more data
+            break
+        else:
+            next_token = tweets_meta["next_token"]
+
+    return all_tweets
+
+
 def get_likers_of_tweet(tweet_id: str) -> list[tweepy.User]:
     all_liker_users: list[tweepy.User] = []
     next_token = None
@@ -57,3 +79,16 @@ def get_likers_of_tweet(tweet_id: str) -> list[tweepy.User]:
             next_token = users_meta["next_token"]
 
     return all_liker_users
+
+
+def fetch_liked_tweets(user_id, next_token):
+    max_results = FetchConfigs.max_tweet_results
+    tweets, tweets_meta = TwitterClient.client.get_liked_tweets(
+        id=user_id,
+        tweet_fields=FetchConfigs.tweet_fields,
+        max_results=max_results,
+        pagination_token=next_token,
+    )
+
+    tweets_list = tweets.data or []
+    return tweets_list, tweets_meta

bot/twitter.py

@@ -10,7 +10,7 @@
     save_user_likes_neo4j,
     save_user_profile_neo4j,
 )
-from bot.services.liked_tweet import get_liked_tweets, get_likers_of_tweet
+from bot.services.liked_tweet import get_liked_tweets_since, get_likers_of_tweet
 from bot.services.quote_tweet import get_quotes_of_tweet
 from bot.services.reply_tweet import (
     get_all_replies_of_tweet,
@@ -19,6 +19,7 @@
 from bot.services.retweet_tweet import get_retweets_of_tweet
 from bot.services.user_info import get_twitter_user, get_twitter_users
 from bot.services.user_tweet import get_mentioned_tweets_by_username, get_user_tweets
+from bot.utils.get_epoch import get_x_days_ago_UTC_timestamp
 
 
 def extract_and_save_tweets(
@@ -170,6 +171,8 @@ def extract_and_save_liker_users(user_id: str):
 
 
 def extract_and_save_liked_tweets(user_id: str):
-    liked_tweets = get_liked_tweets(user_id=user_id)
+    liked_tweets = get_liked_tweets_since(
+        user_id=user_id, since=get_x_days_ago_UTC_timestamp(7)
+    )
     save_user_likes_neo4j(user_id=user_id, tweets_liked=liked_tweets)
     save_tweets_in_neo4j(liked_tweets)