hogumathi-app/tweet_source.py

import json
import requests
import sqlite3

class ArchiveTweetSource:
    """
    id, created_at, retweeted, favorited, retweet_count, favorite_count, full_text, in_reply_to_status_id_str, in_reply_to_user_id, in_reply_to_screen_nam
    """
    def __init__ (self, archive_path, db_path = "data/tweet.db", archive_user_id = None):
        self.archive_path = archive_path
        self.user_id = archive_user_id
        self.db_path = db_path
        return
    
    def get_db (self):
        db = sqlite3.connect(self.db_path)
        
        return db

    def get_user_timeline (self,
                         author_id = None, max_results = 10, since_id = None):
    
        if max_results == None:
            max_results = -1
            
        
        sql_params = []
        where_sql = []
        
        # if the ID is not stored as a number (eg. string) then this could be a problem
        if since_id:
            where_sql.append("id > ?")
            sql_params.append(since_id)
            
        #if author_id:
        #    where_sql.append("author_id = ?")
        #    sql_params.append(author_id)
        
        where_sql = " and ".join(where_sql)
        
        sql_cols = "id, created_at, retweeted, favorited, retweet_count, favorite_count, full_text, in_reply_to_status_id_str, in_reply_to_user_id, in_reply_to_screen_name"
        
        if author_id:
            sql_cols += ", '{}' as author_id".format(author_id)
        
        if where_sql:
            where_sql = "where {}".format(where_sql)
        
        sql = "select {} from tweet {} order by created_at asc limit ?".format(sql_cols, where_sql)
        sql_params.append(max_results)
        
        
        db = self.get_db()
        
        cur = db.cursor()
        cur.row_factory = sqlite3.Row
        
        print(sql)
        print(sql_params)
        
        results = list(map(dict, cur.execute(sql, sql_params).fetchall()))
        
        return results
    
    def get_tweet (self, id_):
        return self.get_tweets([id_])
    
    def get_tweets (self,
                    ids):
                    
        sql_params = []
        where_sql = []
        if since_id:
            ids_in_list_sql = "id in ({})".format( ','.join(['?'] * len(ids)))
            where_sql.append(ids_in_list_sql)
            sql_params += ids
        
        where_sql = " and ".join(where_sql)
        
        sql = "select * from tweet where {} limit ?".format(where_sql)
        
        db = self.get_db()
        
        cur = db.cursor()
        cur.row_factory = sqlite3.Row
        
        results = list(map(dict, cur.execute(sql, sql_params).fetchall()))
        
        return results
    
    def search_tweets (self,
                       query,
                       since_id = None,
                       max_results = 10,
                       sort_order = None
                       ):
        
        return

# https://developer.twitter.com/en/docs/twitter-api/v1/tweets/curate-a-collection/api-reference/get-collections-entries
# we can perhaps steal a token from the TweetDeck Console, otherwise we need to apply for Standard v1.1 / Elevated
class ApiV11TweetCollectionSource:
    def __init__ (self, token):
        self.token = token
        
    def create_collection (self, name):
        return
    
    def bulk_add_to_collection (self, collection_id, items):
        return
    
    def add_to_collection (self, collection_id, item):
        return
        
    def get_collection_tweets (self, collection_id):
        return

class TwitterApiV2SocialGraph:
    def __init__ (self, token):
        self.token = token
        
    def get_user (user_id, is_username=False):
        # GET /2/users/:id
        # GET /2/users/by/:username
        return
    
    def get_users (user_ids, are_usernames=False):
        # GET /2/users/by?usernames=
        # GET /2/users?ids=
        return
        
    def get_following (user_id, 
                        max_results = 10, pagination_token = None):
        # GET /2/users/:id/following
        return
        
    def get_followers (user_id,
                        max_results = 10, pagination_token = None):
        # GET /2/users/:id/followers
        return
        
    def follow_user (user_id, target_user_id):
        # POST /2/users/:id/following
        # {target_user_id}
        return
        
    def unfollow_user (user_id, target_user_id):
        # DELETE /2/users/:source_user_id/following/:target_user_id
        return

class ApiV2TweetSource:
    def __init__ (self, token):
        self.token = token
        
        
    def create_tweet (self, text, 
        reply_to_tweet_id = None, quote_tweet_id = None):
        
        url = "https://api.twitter.com/2/tweets"
        
        tweet = {
            'text': text
        }
        
        if reply_to_tweet_id:
            tweet['reply'] = {
                'in_reply_to_tweet_id': reply_to_tweet_id
            }
            
        if quote_tweet_id:
            tweet['quote_tweet_id'] = quote_tweet_id
        
        body = json.dumps(tweet)
        
        headers = {
            'Authorization': 'Bearer {}'.format(self.token),
            'Content-Type': 'application/json'
        }
        
        response = requests.post(url, data=body, headers=headers)
        result = json.loads(response.text)
        
        return result
    
    def retweet_tweet( self, user_id, tweet_id ):
        
        url = "https://api.twitter.com/2/users/{}/retweets".format(user_id)
        
        retweet = {
            'tweet_id': tweet_id
        }
        
        body = json.dumps(retweet)
        
        headers = {
            'Authorization': 'Bearer {}'.format(self.token),
            'Content-Type': 'application/json'
        }
        
        response = requests.post(url, data=body, headers=headers)
        result = json.loads(response.text)
        
        return result
        
    
    def get_home_timeline (self, user_id, variant = 'reverse_chronological', max_results = 10, pagination_token = None, since_id = None):
        """
        Get a user's timeline as viewed by the user themselves.
        """
        
        path = 'users/{}/timelines/{}'.format(user_id, variant)
        
        return self.get_timeline(path, 
            max_results=max_results, pagination_token=pagination_token, since_id=since_id) 
    
    def get_timeline (self, path,
        max_results = 10, pagination_token = None, since_id = None,
        non_public_metrics = False,
        exclude_replies=False):
        """
        Get any timeline, including custom curated timelines built by Tweet Deck / ApiV11.
        """
        
        token = self.token
        
        url = "https://api.twitter.com/2/{}".format(path)
        
        tweet_fields = ["created_at", "conversation_id",  "referenced_tweets", "text", "public_metrics", "entities", "attachments"]
        media_fields = ["alt_text", "type", "preview_image_url", "public_metrics", "url", "media_key", "duration_ms", "width", "height", "variants"]
        user_fields = ["created_at", "name", "username", "location", "profile_image_url", "verified"]
        expansions = ["entities.mentions.username",
                    "attachments.media_keys",
                    "author_id",
                    "referenced_tweets.id",
                    "referenced_tweets.id.author_id"]
        
        if non_public_metrics:
            tweet_fields.append("non_public_metrics")
            media_fields.append("non_public_metrics")
        
        params = {
            "expansions": ",".join(expansions),
            "media.fields": ",".join(media_fields),
            
            "tweet.fields": ",".join(tweet_fields),
            "user.fields": ",".join(user_fields),
            
            "max_results": max_results,
        }
        
        exclude = []
        
        if exclude_replies:
            exclude.append('replies')
            
        if len(exclude):
            params['exclude'] = exclude
        
        
        if pagination_token:
            params['pagination_token'] = pagination_token
            
        if since_id:
            params['since_id'] = since_id
        
        headers = {"Authorization": "Bearer {}".format(token)}
        
        #headers = {"Authorization": "access_token {}".format(access_token)}

        response = requests.get(url, params=params, headers=headers)
        response_json = json.loads(response.text)
        
        return response_json
    
    def get_mentions_timeline (self, user_id,
                                max_results = 10, pagination_token = None, since_id = None):
                                
        path = "users/{}/mentions".format(user_id)
        
        return self.get_timeline(path, 
            max_results=max_results, pagination_token=pagination_token, since_id=since_id)
    
    def get_user_timeline (self, user_id,
                          max_results = 10, pagination_token = None, since_id = None,
                          non_public_metrics=False,
                          exclude_replies=False):
        """
        Get a user's Tweets as viewed by another.
        """
        path = "users/{}/tweets".format(user_id)
        
        return self.get_timeline(path, 
            max_results=max_results, pagination_token=pagination_token, since_id=since_id,
            non_public_metrics = non_public_metrics,
            exclude_replies=exclude_replies)
    
    
    def get_tweet (self, id_, non_public_metrics = False):
        return self.get_tweets([id_], non_public_metrics = non_public_metrics)
    
    def get_tweets (self,
                    ids,
                    non_public_metrics = False):
                    
        token = self.token
        
        url = "https://api.twitter.com/2/tweets"
        
        tweet_fields = ["created_at", "conversation_id",  "referenced_tweets", "text", "public_metrics", "entities", "attachments"]
        media_fields = ["alt_text", "type", "preview_image_url", "public_metrics", "url", "media_key", "duration_ms", "width", "height", "variants"]
        user_fields = ["created_at", "name", "username", "location", "profile_image_url", "verified"]
        expansions = ["entities.mentions.username",
                    "attachments.media_keys",
                    "author_id",
                    "referenced_tweets.id",
                    "referenced_tweets.id.author_id"]
        
        if non_public_metrics:
            tweet_fields.append("non_public_metrics")
            media_fields.append("non_public_metrics")

        params = {
            "ids": ','.join(ids),
            "expansions": ",".join(expansions),
            "media.fields": ",".join(media_fields),
            
            "tweet.fields": ",".join(tweet_fields),
            "user.fields": ",".join(user_fields)
        }
        headers = {"Authorization": "Bearer {}".format(token)}
        
        response = requests.get(url, params=params, headers=headers)
        response_json = json.loads(response.text)
        
        return response_json
    
    def search_tweets (self,
                       query, 
                       pagination_token = None,
                       since_id = None,
                       max_results = 10,
                       sort_order = None,
                       non_public_metrics = False
                       ):
        
        token = self.token
        
        url = "https://api.twitter.com/2/tweets/search/recent"
        
        tweet_fields = ["created_at", "conversation_id",  "referenced_tweets", "text", "public_metrics", "entities", "attachments"]
        media_fields = ["alt_text", "type", "preview_image_url", "public_metrics", "url", "media_key", "duration_ms", "width", "height", "variants"]
        user_fields = ["created_at", "name", "username", "location", "profile_image_url", "verified"]
        expansions = ["entities.mentions.username",
                    "attachments.media_keys",
                    "author_id",
                    "referenced_tweets.id",
                    "referenced_tweets.id.author_id"]
        
        if non_public_metrics:
            tweet_fields.append("non_public_metrics")
            media_fields.append("non_public_metrics")
        
        params = {
            "expansions": ",".join(expansions),
            "media.fields": ",".join(media_fields),
            
            "tweet.fields": ",".join(tweet_fields),
            "user.fields": ",".join(user_fields),
            
            "query": query,
            "max_results": max_results,
        }
        
        if pagination_token:
            params['pagination_token'] = pagination_token
            
        if since_id:
            params['since_id'] = since_id
        
        if sort_order:
            params['sort_order'] = sort_order
        
        
        headers = {"Authorization": "Bearer {}".format(token)}
        
        response = requests.get(url, params=params, headers=headers)
        response_json = json.loads(response.text)
        
        return response_json
        
        
    
    def count_tweets (self,
                       query, 
                       since_id = None,
                       granularity = 'hour'
                       ):
        
        token = self.token
        
        url = "https://api.twitter.com/2/tweets/counts/recent"
        
        
        params = {
            "query": query
        }

        if since_id:
            params['since_id'] = since_id
        
        headers = {"Authorization": "Bearer {}".format(token)}
        
        response = requests.get(url, params=params, headers=headers)
        
        print(response.status_code)
        print(response.text)
        
        response_json = json.loads(response.text)
        
        return response_json

    #def get_conversation (self, tweet_id, pagination_token = None,
    # TODO

    def get_thread (self, tweet_id,
                       author_id = None,
                       pagination_token = None,
                       since_id = None,
                       max_results = 10,
                       sort_order = None
                       ):
        
        # FIXME author_id can be determined from a Tweet object
        query = "conversation_id:{}".format(tweet_id)
        if author_id:
            query += " from:{}".format(author_id)
            
        return self.search_tweets(query, 
            pagination_token = pagination_token, since_id = since_id, max_results = max_results, sort_order = sort_order)
    
    def get_bookmarks (self, user_id,
                          max_results = 10, pagination_token = None, since_id = None):
        path = "users/{}/bookmarks".format(user_id)
        
        return self.get_timeline(path, 
            max_results=max_results, pagination_token=pagination_token, since_id=since_id)
    
    def get_media_tweets (self,
                   author_id = None,
                   has_media = True,
                   has_links = None,
                   has_images = None,
                   has_videos = None,
                   pagination_token = None,
                   since_id = None,
                   max_results = 10,
                   sort_order = None
                   ):
        
        # FIXME author_id can be determined from a Tweet object
        
        query = ""
        if has_media != None:
            if not has_media:
                query += "-"
            query += "has:media "
                
        if has_links != None:
            if not has_links:
                query += " -"
            query += "has:links "
            
        if has_images != None:
            if not has_images:
                query += " -"
            query += "has:images "
            
        if has_videos != None:
            if not has_videos:
                query += " -"
            query += "has:videos "
            
        
        if author_id:
            query += "from:{} ".format(author_id)
            
        return self.search_tweets(query, 
            pagination_token = pagination_token, since_id = since_id, max_results = max_results, sort_order = sort_order)
    
        
    def get_retweets (self, tweet_id):
        # GET /2/tweets/:id/retweeted_by
        return 
        
    def get_quote_tweets( self, tweet_id):
        # GET /2/tweets/:id/quote_tweets
        return 
        
    def get_likes (self, tweet_id):
        # GET /2/tweets/:id/liking_users
        return 
        
    def get_liked_by (self, user_id):
        #  GET /2/users/:id/liked_tweets
        return 
    
    def get_list_tweets (self, list_id):
        # GET /2/lists/:id/tweets
        return