hogumathi-app/mastodon_source.py

from typing import List
from dataclasses import asdict, replace
from dacite import from_dict

import json

import requests

from mastodon_v2_types import Status, Conversation, Context

# https://github.com/mastodon/documentation/blob/master/content/en/api/guidelines.md#paginating-through-api-responses-pagination
class MastodonAPSource:
    def __init__ (self, endpoint, token):
        self.endpoint = endpoint
        self.token = token
        super().__init__()
        
    def get_status_collection (self, path,
        params = {}, return_dataclasses = False):
        
        url = self.endpoint + path
        params = {
            **params
        }
        
        headers = {}
        if self.token:
            headers.update({
               'Authorization': f'Bearer {self.token}'
            })
        
        response = requests.get(url, params=params, headers=headers)
        
        print(url)
        #print(response.status_code)
        #print(response.text)
        
        response_json = json.loads(response.text)
        
        try:
            print('trying to force timeline response to Status type')
            typed_resp = list(map(lambda s: from_dict(data=s, data_class=Status), response_json))
        except:
            raise 'error forcing status to type'
        
        if return_dataclasses:
            return typed_resp
        else:
            response_json = list(map(lambda s: asdict(s), typed_resp))
            
        return response_json

    def get_timeline (self, path = "home", params = {}, return_dataclasses=False):
        
        collection_path = "/api/v1/timelines/" + path
        params = {
            **params
        }
        
        return self.get_status_collection(collection_path, params = params, return_dataclasses=return_dataclasses)
        
    
    # nice parallel is validation
    # https://github.com/moko256/twitlatte/blob/master/component_client_mastodon/src/main/java/com/github/moko256/latte/client/mastodon/MastodonApiClientImpl.kt
    
    def get_mentions_timeline (self):
        # /api/v1/notifications?exclude_types=follow,favourite,reblog,poll,follow_request
        return
        
    def get_favorited_timeline (self):
        # /api/v1/notifications?exclude_types=follow,reblog,mention,poll,follow_request
        return
        
    def get_conversations (self,
            max_id = None, since_id = None, min_id = None, limit = None
            ) -> List[Conversation]:
        """
        Use get_status_context on the last_status for a conversation.
        """
        
        url = f'{self.instance}/api/v1/conversations'
        # https://docs.joinmastodon.org/methods/conversations/
        # Use HTTP Link header for pagination.
        headers = {
            'Authorization': f'Bearer {self.token}'
        }
        params = cleandict({
            'max_id': max_id,
            'since_id': since_id,
            'min_id': min_id,
            'limit': limit
        })
        
        resp = requests.get(url, params=params, headers=headers)
        response_json = json.loads(resp.text)
        conversations = list(map(lambda c: from_dict(data=c, data_class=Conversation), response_json))
        
        return conversations
    
    def get_status_context (self, status_id) -> Context:
        """
        Returns all parents and direct children.
        """
        url = f'{self.instance}/api/v1/statuses/{status_id}/context'
        headers = {
            'Authorization': f'Bearer {self.token}'
        }
        
        resp = requests.get(url, headers=headers)
        response_json = json.loads(resp.text)
        context = from_dict(data=response_json, data_class=Context)
        
        return context
    
    def get_bookmarks (self, 
        max_id=None, return_dataclasses=False):
        
        collection_path = '/api/v1/bookmarks'
        params = {}
        
        if max_id:
            params['max_id'] = max_id
        
        return self.get_status_collection(collection_path, params = params, return_dataclasses=return_dataclasses)
        
    def get_favorites_timeline (self):
        # /account/favourites
        return
    
    
    def get_user_statuses (self, account_id, 
        max_id = None, return_dataclasses=False):
        # /api/v2/search?type=statuses&account_id=
        # /api/v1/accounts/:id/statuses
        
        collection_path = f'/api/v1/accounts/{account_id}/statuses'
        
        params = {}
        
        if max_id:
            params['max_id'] = max_id
        
        return self.get_status_collection(collection_path, params = params, return_dataclasses=return_dataclasses)
    
    def get_statuses (self, ids):
        # /api/v1/statuses/:id
        return
        
    def get_profile (self, username):
        # /api/v1/accounts/search
        return self.search('@' + username, result_type='accounts')
        
    def search (q, result_type = None, following = False):
        # /api/v2/search
        return
        
def cleandict(d):
    if isinstance(d, dict):
        return {k: cleandict(v) for k, v in d.items() if v is not None}
    elif isinstance(d, list):
        return [cleandict(v) for v in d]
    else:
        return d