forgefed/deprecated/5. extending pagure database model/activitypub.py

# -*- coding: utf-8 -*-

"""
    (c) 2020 - Copyright ...
    
    Authors:
        zPlus <zplus@peers.community>
"""

import datetime
import logging
import pyld
import random
import requests
import string

from . import APP_URL
from . import model
from . import tasks
from . import settings

log = logging.getLogger(__name__)

# List of HTTP headers used by ActivityPub.
# https://www.w3.org/TR/activitypub/#server-to-server-interactions
default_header  = 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'
optional_header = 'application/activity+json'
headers         = [ default_header, optional_header ]

# Headers to use when GETting/POSTing remote objects
REQUEST_HEADERS = { 'Accept':       default_header,
                    'Content-Type': default_header}

# The JSON-LD context to use with ActivityPub activities.
jsonld_context = [
    'https://www.w3.org/ns/activitystreams',
    'https://w3id.org/security/v1',
    'https://forgefed.peers.community/ns' ]

# Fetch an ActivityPub object
def fetch(uri):
    response = requests.get(uri, headers=REQUEST_HEADERS)
    
    if response.status_code != 200:
        log.info('[{}] Error while retrieving remote object: {}'.format(response.status_code, uri))
        return None
    
    # The remote server is expected to serve a JSON-LD document.
    object = response.json()
    
    # Because JSON-LD can represent the same graph in several different
    # ways, we should normalize the JSONLD object before passing it to the
    # actor for processing. This simplifies working with the object.
    # Normalization could mean "flattening" or "compaction" of the JSONLD
    # document.
    # However, this step is left out for now and not implemented unless
    # needed because the ActivityStream specs already specifies that
    # objects should be served in compact form:
    # https://www.w3.org/TR/social-web-protocols/#content-representation
    # 
    # object = response.json()
    # return normalize(object)
    
    return Document(object)

# Cache a copy of the JSON-LD context such that we can work with activities
# without sending HTTP requests all the time.
cached_jsonld_context = []
for context in jsonld_context:
    cached_jsonld_context.append(requests.get(context, headers=REQUEST_HEADERS).json())

def format_datetime(dt):
    """
    This function is used to format a datetime object into a string that is
    suitable for publishing in Activities.
    """
    
    return dt.replace(microsecond=0) \
             .replace(tzinfo=datetime.timezone.utc) \
             .isoformat()

def new_activity_id(length=32):
    """
    Generate a random string suitable for using as part of an Activity ID.
    """
    
    symbols = string.ascii_lowercase + string.digits
    return ''.join([ random.choice(symbols) for i in range(length) ])

class Document(dict):
    
    def __init__(self, *args, **kwargs):
      super().__init__(*args, **kwargs)
    
    def match(self, pattern):
        """
        Check if this Document matches another dictionary. If this Document
        contains all the keys in :pattern:, and they have the same value, then
        the function returns True. Otherwise return False.
        To only check for existence of a key, not its value, use the empty
        dictionary like this: pattern={"key": {}}
        
        :param pattern: The Document (or Python dictionary) to match this Document against.
        """
        
        for key, value in pattern.items():
            if key not in self:
                return False
            
            # If the pattern is a dict, then recourse
            if isinstance(value, dict):
                if not Document(self[key] if isinstance(self[key], dict)
                                          else {}
                               ).match(value):
                    return False
            else:
                if self[key] != value:
                    return False
            
        return True
    
    def first(self, property):
        """
        Return value if it's a scalar, otherwise return the first element if it
        is an array.
        """
        
        if isinstance(self[property], list):
            return self[property][0]
        
        if isinstance(self[property], dict):
            return None
        
        return self[property]
    
    def node(self, property):
        """
        Return the value if it's an object. If it's a string, try to fetch a
        URI.
        """
        
        if isinstance(self[property], dict):
            return Document(self[property])
        
        if isinstance(self[property], str):
            self[property] = fetch(self[property])
            return self.node(property)
        
        return None

class Activity(Document):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        self.update(kwargs)
        
        self.update({
            '@context': jsonld_context
        })
    
    def distribute(self):
        tasks.delivery.distribute.delay(self)

class Follow(Activity):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        self.update({ 'type': 'Follow' })

class Accept(Activity):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        self.update({ 'type': 'Accept' })