hogumathi-app/extensions/twitter_v2_facade/content_source.py

"""
This translates from the Tweet Source and Twitter v2 types

Into ViewModel types such as FeedItem

And the rest of the Taxonomy.

"""

from dataclasses import asdict
import os
from flask import session, g, request
import time
import json

from twitter_v2.api import ApiV2TweetSource, TwitterApiV2SocialGraph, ApiV2ConversationSource

from hogumathi_app.view_model import CollectionPage, cleandict
from .view_model import tweet_model_dc_vm, user_model_dc

from hogumathi_app.content_system import register_content_source, get_content, register_hook

DATA_DIR='.data'

def get_tweet_item (tweet_id, me=None):
    
    
    if me:
        twitter_user = session.get(me)
        token = twitter_user['access_token']
    else:
        token = os.environ.get('BEARER_TOKEN')
    
    tweet_source = ApiV2TweetSource(token)
    tweets_response = tweet_source.get_tweet(tweet_id, return_dataclass=True)

    
    #print(response_json)
    if tweets_response.errors:
        # types:
        # https://api.twitter.com/2/problems/not-authorized-for-resource (blocked or suspended)
        # https://api.twitter.com/2/problems/resource-not-found (deleted)
        #print(response_json.get('errors'))
        for err in tweets_response.errors:
            if not 'type' in err:
                print('unknown error type: ' + str(err))
            elif err['type'] == 'https://api.twitter.com/2/problems/not-authorized-for-resource':
                print('blocked or suspended tweet: ' + err['value'])
            elif err['type'] == 'https://api.twitter.com/2/problems/resource-not-found':
                print('deleted tweet: ' + err['value'])
            else:
                print('unknown error')
            
            print(json.dumps(err, indent=2))
    
    
    includes = tweets_response.includes
    tweets = list(map(lambda t: tweet_model_dc_vm(includes, t, me), tweets_response.data))
    
    collection_page = CollectionPage(
        id = tweet_id,
        items = tweets,
        next_token = None # Fixme
        )
    
    return collection_page


def get_bookmarks_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_bookmarks_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_bookmarks(user_id,
                        pagination_token = pagination_token,
                        return_dataclass=True,
                        max_results=max_results
                        )
    
    #print(response_json)
    
    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
    
    query = {}
    
    if next_token:
        query = {
            **query,

        }
    
    user = {
            'id': user_id
        }
    
    ts = int(time.time() * 1000)
    with open(f'{DATA_DIR}/cache/bookmarks_{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, pagination_token=None, me=None, exclude_replies=False, exclude_retweets=True, format=None):
    
    if not me and 'me' in g:
        me = g.me
    
    if 'twitter_user' in g and g.twitter_user:
        token = g.twitter_user['access_token']
        # issue: retweets don't come back if we request non_public_metrics
        is_me = False and user_id == g.twitter_user['id']
    else:
        token = os.environ.get('BEARER_TOKEN')
        is_me = False
    
    
    tweet_source = ApiV2TweetSource(token)
    tweets_response = tweet_source.get_user_timeline(user_id,
                        exclude_replies = exclude_replies,
                        exclude_retweets = exclude_retweets,
                        pagination_token = pagination_token,
                        non_public_metrics = False,
                        return_dataclass=True)
    
    tweets = None
    if not tweets_response:
        print('no response_json')
        
    if tweets_response.meta.result_count == 0:
        print('no results')
    
    if not tweets_response.includes:
        print(tweets_response)
        print('no tweets_response.includes')
    
    if tweets_response.errors:
        print('profile get_user_timeline errors:')
        print(tweets_response.errors)
    
    ts = int(time.time() * 1000)
    with open(f'{DATA_DIR}/cache/tl_{user_id}_{ts}_{pagination_token}.json', 'wt') as f:
        f.write(json.dumps(cleandict(asdict(tweets_response))))
    
    if tweets_response.data:
        tweets = list(map(lambda t: tweet_model_dc_vm(tweets_response.includes, t, me), tweets_response.data))
    
    next_token = tweets_response.meta.next_token
    
    collection_page = CollectionPage(
        id = user_id,
        items = tweets,
        next_token = next_token
    )
    
    return collection_page

def get_tweets_collection (tweet_ids, pagination_token=None, max_results=None):
    """
    We might be able to have a generalizer in the content system as well...
    If a source exposes a get many interface then use it. We want to avoid many singular fetches.
    """
    return []
    
def get_user (user_id, me=None):
    
    if me:
        twitter_user = session.get(me)
        token = twitter_user['access_token']
    else:
        token = os.environ.get('BEARER_TOKEN')
    
    social_graph = TwitterApiV2SocialGraph(token)
    users_response = social_graph.get_user(user_id, return_dataclass=True)
    
    print(users_response)
    
    if not len(users_response.data):
        return
    
    user = user_model_dc(users_response.data[0])
    
    return user

def register_content_sources ():
    register_content_source('twitter:tweets', get_tweets_collection)
    register_content_source('twitter:tweet:', get_tweet_item, id_pattern='(?P<tweet_id>\d+)')
    register_content_source('twitter:bookmarks:', get_bookmarks_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+)')