implemented liked_tweets end to end, as /twitter/likes.html

extensions/twitter_v2_facade/content_source.py

@@ -608,6 +608,67 @@ def get_bookmarks_feed (user_id, pagination_token=None, max_results=10, me=None)
     
     return collection_page
 
+
+def get_liked_tweets_feed (user_id, pagination_token=None, max_results=10, me=None):
+    
+    if not me:
+        me = g.get('me') or request.args.get('me')
+        
+    
+    print(f'get_liked_tweets_feed. me={me}')
+    
+    twitter_user = session.get( me )
+    
+    if not twitter_user:
+        return None
+        
+    token = twitter_user['access_token']
+    
+    tweet_source = ApiV2TweetSource(token)
+    response_tweets = tweet_source.get_liked_tweets(user_id,
+                        pagination_token = pagination_token,
+                        return_dataclass=True,
+                        max_results=max_results
+                        )
+    
+    #print(response_json)
+    
+    cache_tweets_response(response_tweets, 'liked_tweets', user_id, user_id=user_id, pagination_token=pagination_token)
+    
+    if response_tweets.data:
+        includes = response_tweets.includes
+        tweets = list(map(lambda t: tweet_model_dc_vm(includes, t, me), response_tweets.data))
+        next_token = response_tweets.meta.next_token
+    else:
+        print('no tweet data:')
+        print(response_tweets)
+        tweets = []
+        next_token = None
+    
+    query = {}
+    
+    if next_token:
+        query = {
+            **query,
+
+        }
+    
+    user = {
+            'id': user_id
+        }
+    
+    ts = int(time.time() * 1000)
+    with open(f'{DATA_DIR}/cache/liked_tweets_{user_id}_{ts}_{pagination_token}.json', 'wt') as f:
+        f.write(json.dumps(cleandict(asdict(response_tweets))))
+    
+    collection_page = CollectionPage(
+        id = user_id, # FIXME this should perhaps be the unresolved id
+        items = tweets,
+        next_token = next_token
+    )
+    
+    return collection_page
+
 def get_user_feed (user_id, me=None, **twitter_kwargs):
     
     if not me and 'me' in g:
@@ -1014,6 +1075,7 @@ def register_content_sources ():
     register_content_source('twitter:tweet:', get_tweet_item, id_pattern='(?P<tweet_id>\d+)')
     register_content_source('twitter:tweet:', get_tweet_embed, id_pattern='(?P<tweet_id>\d+)')
     register_content_source('twitter:bookmarks:', get_bookmarks_feed, id_pattern='(?P<user_id>\d+)')
+    register_content_source('twitter:liked_tweets:', get_liked_tweets_feed, id_pattern='(?P<user_id>\d+)')
     register_content_source('twitter:feed:user:', get_user_feed, id_pattern='(?P<user_id>\d+)')
     register_content_source('twitter:user:', get_user, id_pattern='(?P<user_id>\d+)')
     register_content_source('twitter:users', get_users, id_pattern='')

extensions/twitter_v2_facade/facade.py

@@ -776,6 +776,41 @@ def get_bookmarks2_html ():
     else:
         return render_template(f'tweet-collection{theme_variant}.html', user = user, tweets = tweets, query = query)
         
+
+@twitter_app.route('/likes.html', methods=['GET'])
+def get_likes_html ():
+    
+    user_id = g.twitter_user['id']
+    token = g.twitter_user['access_token']
+    
+    pagination_token = request.args.get('pagination_token')
+    max_results = int(request.args.get('limit', 10))
+    
+    collection_page = get_content(f'twitter:liked_tweets:{user_id}', pagination_token=pagination_token, max_results=max_results)
+    tweets = collection_page.items
+    
+    next_token = collection_page.next_token
+    
+    query = {}
+    
+    if next_token:
+        query = {
+            **query,
+            
+            'next_data_url': url_for('.get_likes_html', user_id=user_id, pagination_token=next_token, limit=max_results),
+            'next_page_url': url_for('.get_likes_html', user_id=user_id, pagination_token=next_token, limit=max_results)
+        }
+    
+    user = {
+            'id': user_id
+        }
+    
+    
+    if 'HX-Request' in request.headers:
+        return render_template(f'partial/tweets-timeline{theme_variant}.html', user = user, tweets = tweets, query = query)
+    else:
+        return render_template(f'tweet-collection{theme_variant}.html', user = user, tweets = tweets, query = query)
+
 #@twitter_app.route('/bookmarks.html', methods=['GET'])
 def get_bookmarks_old_html ():
     

extensions/twitter_v2_facade/view_model.py

@@ -106,29 +106,7 @@ def tweet_model_dc_vm (includes: TweetExpansions, tweet: Tweet, me, my_url_for=u
         is_bookmarked = cache_db.execute('select count(*) from tweet t, query q where q.rowid = t.query_id and q.query_type=? and t.id=? and q.auth_user_id=?', ['bookmarks', tweet.id, auth_user_id]).fetchone()[0] and True
         cache_db.close()
     
-    if my_g.get('twitter_user'):
-        actions.update(
-            retweet = FeedItemAction('twitter_v2_facade.post_tweet_retweet', {'tweet_id': tweet.id})
-            )
-        setattr(actions['retweet'], 'url', my_url_for(actions['retweet'].route, actions['retweet'].route_params))
-        if is_bookmarked:
-            actions.update(
-                delete_bookmark = FeedItemAction('twitter_v2_facade.delete_tweet_bookmark', {'tweet_id': tweet.id})
-                )
-            setattr(actions['delete_bookmark'], 'url', my_url_for(actions['delete_bookmark'].route, **actions['delete_bookmark'].route_params))
-        else:
-            actions.update(
-                bookmark = FeedItemAction('twitter_v2_facade.post_tweet_bookmark', {'tweet_id': tweet.id})
-                )
-            setattr(actions['bookmark'], 'url', my_url_for(actions['bookmark'].route, **actions['bookmark'].route_params))
-    
-    if my_g.get('twitter_live_enabled'):
-        actions.update(
-            view_activity = FeedItemAction('twitter_v2_live_facade.get_tweet_activity_html', {'tweet_id': tweet.id})
-            )
-        setattr(actions['view_activity'], 'url', my_url_for(actions['view_activity'].route, **actions['view_activity'].route_params))
-    
-    
+
 
     t = FeedItem(
         id = tweet.id,

lib/twitter_v2/api.py

@@ -662,6 +662,15 @@ class ApiV2TweetSource:
         
         return self.get_timeline(path, 
             max_results=max_results, pagination_token=pagination_token, since_id=since_id, return_dataclass=return_dataclass)
+            
+    def get_liked_tweets (self, user_id,
+                          max_results = 10, pagination_token = None, since_id = None,
+                          return_dataclass=False):
+        path = "users/{}/liked_tweets".format(user_id)
+        
+        return self.get_timeline(path, 
+            max_results=max_results, pagination_token=pagination_token, since_id=since_id, return_dataclass=return_dataclass)
+    
     
     def get_media_tweets (self,
                    author_id = None,