pelican-plugin-activitypub/pelican/plugins/activitypub/activitypub.py

import datetime
import logging
import json
import os
import urllib.parse
import pelican.writers

from pelican import signals

log = logging.getLogger(__name__)


__version__ = '0.1.2'

pagination = 25


def ap_article(generator: pelican.ArticlesGenerator, writer: pelican.writers.Writer):

    now = datetime.datetime.utcnow()

    author = generator.settings['AUTHOR']
    domain = urllib.parse.urlparse(generator.settings['SITEURL']).netloc

    wkhmpath = os.path.join(writer.output_path, '.well-known/host-meta')
    wknipath = os.path.join(writer.output_path, '.well-known/nodeinfo')
    nipath = os.path.join(writer.output_path, 'activitypub/nodeinfo')
    wfpath = os.path.join(writer.output_path, '.well-known/webfinger')
    awfpath = os.path.join(writer.output_path, '.well-known/_webfinger')
    authorpath = os.path.join(writer.output_path, 'activitypub/users')
    os.makedirs(os.path.join(writer.output_path, '.well-known'), exist_ok=True)
    os.makedirs(awfpath, exist_ok=True)
    os.makedirs(os.path.join(writer.output_path, 'activitypub/users'), exist_ok=True)
    os.makedirs(os.path.join(writer.output_path, 'activitypub/posts'), exist_ok=True)
    os.makedirs(os.path.join(writer.output_path, 'activitypub/tags'), exist_ok=True)
    os.makedirs(os.path.join(writer.output_path, 'activitypub/collections/inbox'), exist_ok=True)
    os.makedirs(os.path.join(writer.output_path, 'activitypub/collections/outbox'), exist_ok=True)
    for author, _ in generator.authors:
        os.makedirs(os.path.join(writer.output_path, 'activitypub/collections/outbox_page', author.slug), exist_ok=True)
    os.makedirs(os.path.join(writer.output_path, 'activitypub/collections/following'), exist_ok=True)
    os.makedirs(os.path.join(writer.output_path, 'activitypub/collections/followers'), exist_ok=True)
    wknodeinfo = {
        'links': [
            {
                'href': os.path.join(generator.settings['SITEURL'], 'activitypub/nodeinfo'),
                'rel': 'http://nodeinfo.diaspora.software/ns/schema/2.0'
            }
        ]
    }
    nodeinfo = {
        'version': '2.0',
        'software': {
            'name': 'pelican-activitypub',
            'version': __version__
        },
        'protocols': ['activitypub'],
        'services': {
            'inbound': [],
            'outbound': ['atom1.0', 'rss2.0']
        },
        'openRegistrations': False,
        'usage': {
            'users': {
                'total': len(generator.authors),
            },
            'localPosts': len(generator.articles)
        },
        'metadata': {
            'nodeName': generator.settings['SITENAME'],
        }
    }
    wfurl = os.path.join(generator.settings['SITEURL'], '.well-known/webfinger?resource={uri}')
    hostmeta = f'<?xml version="1.0" encoding="UTF-8"?><XRD xmlns="http://docs.oasis-open.org/ns/xri/xrd-1.0"><Link rel="lrdd" template="{wfurl}" type="application/xrd+xml" /></XRD>'
    webfinger = {
        'subject': f'acct:{author}@{domain}',
        'aliases': [
            os.path.join(generator.settings['SITEURL'], 'author', author.slug + '.html'),
            os.path.join(generator.settings['SITEURL'], 'activitypub/users/', author.slug)
        ],
        'links': [
            {
                'rel': 'http://webfinger.net/rel/profile-page',
                'type': 'text/html',
                'href': os.path.join(generator.settings['SITEURL'], 'author', author.slug + '.html')
            },
            {
                'rel': 'self',
                'type': 'application/activity+json',
                'href': os.path.join(generator.settings['SITEURL'], 'activitypub/users', author.slug)
            }
        ]
    }
    with open(wkhmpath, 'w') as hf:
        hf.write(hostmeta)
    with open(wknipath, 'w') as nf:
        json.dump(wknodeinfo, nf)
    with open(nipath, 'w') as nf:
        json.dump(nodeinfo, nf)
    with open(wfpath, 'w') as wf:
        json.dump(webfinger, wf)

    for t in generator.tags:
        url = os.path.join(generator.settings['SITEURL'], 'activitypub/tags', t.slug)
        path = os.path.join(writer.output_path, 'activitypub/tags', t.slug)
        articles = []
        for article in generator.articles:
            if t.name not in article.metadata.get('tags', []):
                continue
            articles.append(
                os.path.join(generator.settings['SITEURL'], 'activitypub/posts', article.slug)
            )
        tag = {
            '@context': ['https://www.w3.org/ns/activitystreams'],
            'id': url,
            'type': 'OrderedCollection',
            'totalItems': len(articles),
            'orderedItems': articles
        }
        with open(path, 'w') as f:
            json.dump(tag, f)

    articlemap = {}
    for article in generator.articles:
        aurl = os.path.join(generator.settings['SITEURL'], 'activitypub/posts', article.slug)
        apath = os.path.join(writer.output_path, 'activitypub/posts', article.slug)
        tags = []
        for tag in article.metadata.get('tags', []):
            tags.append({
                'type': 'Hashtag',
                'name': '#' + tag.slug,
                'href': os.path.join(generator.settings['SITEURL'], 'activitypub/tags', tag.slug)
            })
        replyto = None
        if 'series' in article.metadata and 'seriesindex' in article.metadata:
            series = article.metadata['series']
            sindex = int(article.metadata['seriesindex'])
            for sa in generator.articles:
                if 'series' in sa.metadata and 'seriesindex' in sa.metadata \
                        and sa.metadata['series'] == series and int(sa.metadata['seriesindex']) == sindex - 1:
                    replyto = os.path.join(generator.settings['SITEURL'], 'activitypub/posts', sa.slug)
                    break
        cc = [os.path.join(generator.settings['SITEURL'], 'activitypub/collections/followers', article.author.slug)]
        aa = generator.settings.get('ACTIVITYPUB_AUTHORS', {}).get(author.name, {})
        if 'movedTo' in aa:
            cc.append(aa['movedTo'])
            tags.append({
                'type': 'Mention',
                'href': aa['movedTo'],
                'name': aa['movedToName']
            })
        cmap = {}
        tmap = {}
        for lang in article.translations + [article]:
            tmap[lang.lang] = lang.title
            cmap[lang.lang] = lang.content
        articlemap[article.slug] = {
            '@context': ['https://www.w3.org/ns/activitystreams'],
            'type': 'Article',
            'id': aurl,
            'published': article.date.isoformat(timespec='seconds'),
            'inReplyTo': replyto,
            'url': os.path.join(generator.settings['SITEURL'], article.url),
            'attributedTo': os.path.join(generator.settings['SITEURL'], 'activitypub/users', article.author.slug),
            'to': ['https://www.w3.org/ns/activitystreams#Public'],
            'cc': cc,
            'name': article.title,
            'nameMap': tmap,
            'content': article.content,
            'contentMap': cmap,
            'summary': None,
            'attachment': [],
            'tag': tags
        }
        with open(apath, 'w') as f:
            json.dump(articlemap[article.slug], f)

    for author, articles in generator.authors:
        aa = generator.settings.get('ACTIVITYPUB_AUTHORS', {}).get(author.name, {})
        url = os.path.join(generator.settings['SITEURL'], 'activitypub/users', author.slug)
        wwwurl = os.path.join(generator.settings['SITEURL'], 'author', author.slug + '.html')
        inboxurl = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/inbox', author.slug)
        inboxpath = os.path.join(writer.output_path, 'activitypub/collections/inbox', author.slug)
        outboxurl = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox', author.slug)
        outboxpath = os.path.join(writer.output_path, 'activitypub/collections/outbox', author.slug)
        followersurl = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/followers', author.slug)
        followerspath = os.path.join(writer.output_path, 'activitypub/collections/followers', author.slug)
        followingurl = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/following', author.slug)
        followingpath = os.path.join(writer.output_path, 'activitypub/collections/following', author.slug)

        creates = []
        for article in articles:
            art = articlemap[article.slug]
            creates.append({
                '@context': ['https://www.w3.org/ns/activitystreams'],
                'type': 'Create',
                'id': os.path.join(generator.settings['SITEURL'], 'activitypub/posts', article.slug),
                'actor': url,
                'published': article.date.isoformat(timespec='seconds'),
                'to': art['to'],
                'cc': art['cc'],
                'object': art
            })

        if len(articles) == 0:
            published = now.isoformat(timespec='seconds') + 'Z'
        else:
            published = min([x.date for x in articles]).isoformat(timespec='seconds')
        a = {
            '@context': [
                'https://www.w3.org/ns/activitystreams',
                {
                    'schema': 'http://schema.org#',
                    'toot': 'http://joinmastodon.org/ns#',
                    'PropertyValue': 'schema:PropertyValue',
                    'value': 'schema:value',
                    'alsoKnownAs': {
                        '@id': 'as:alsoKnownAs',
                        '@type': '@id'
                    },
                    'movedTo': {
                        '@id': 'as:movedTo',
                        '@type': '@id'
                    },
                    'discoverable': 'toot:discoverable',
                }
            ],
            'type': 'Person',
            'id': url,
            'inbox': inboxurl,
            'outbox': outboxurl,
            'following': followingurl,
            'followers': followersurl,
            'preferredUsername': author.name,
            'url': wwwurl,
            'name': author.slug,
            'discoverable': True,
            'manuallyApprovesFollowers': True,
            'published': published,
            'updated': now.isoformat(timespec='seconds') + 'Z',
            'tag': [],
            'attachment': [],
            'endpoints': {
                'sharedInbox': inboxurl
            },
            'icon': {},
            'image': {}
        }
        a.update({k: v for k, v in aa.items() if not k.startswith('_')})
        inbox = {
            '@context': ['https://www.w3.org/ns/activitystreams'],
            'type': 'OrderedCollection',
            'id': inboxurl,
            'totalItems': 0,
            'orderedItems': []
        }
        maxpage = len(creates) // pagination
        outbox = {
            '@context': ['https://www.w3.org/ns/activitystreams'],
            'type': 'OrderedCollection',
            'id': outboxurl,
            'totalItems': len(creates),
            'first': os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page/', author.slug, str(0)),
            'last': os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page/', author.slug, str(maxpage))
        }
        following = {
            '@context': ['https://www.w3.org/ns/activitystreams'],
            'type': 'OrderedCollection',
            'id': followingurl,
            'totalItems': 0,
            'orderedItems': []
        }
        followers = {
            '@context': ['https://www.w3.org/ns/activitystreams'],
            'type': 'OrderedCollection',
            'id': followersurl,
            'totalItems': 0,
            'orderedItems': []
        }
        for i in range(0, len(creates), pagination):
            ipage = i // pagination
            outpageurl = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage))
            outpagepath = os.path.join(writer.output_path, 'activitypub/collections/outbox_page', author.slug, str(ipage))
            page = {
                '@context': ['https://www.w3.org/ns/activitystreams'],
                'type': 'OrderedCollectionPage',
                'id': outpageurl,
                'totalItems': len(creates),
                'partOf': outboxurl,
                'orderedItems': creates[i:i+pagination]
            }
            if ipage > 0:
                page['prev'] = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage-1))
            if ipage < maxpage:
                page['next'] = os.path.join(generator.settings['SITEURL'], 'activitypub/collections/outbox_page', author.slug, str(ipage+1))
            with open(outpagepath, 'w') as f:
                json.dump(page, f)
        author_webfinger = {
            'subject': f'acct:{author.name}@{domain}',
            'aliases': [
                os.path.join(generator.settings['SITEURL'], 'author', author.name + '.html'),
                os.path.join(generator.settings['SITEURL'], 'activitypub/users/', author.name)
            ],
            'links': [
                {
                    'rel': 'http://webfinger.net/rel/profile-page',
                    'type': 'text/html',
                    'href': os.path.join(generator.settings['SITEURL'], 'author', author.name + '.html')
                },
                {
                    'rel': 'self',
                    'type': 'application/activity+json',
                    'href': os.path.join(generator.settings['SITEURL'], 'activitypub/users', author.name)
                }
            ]
        }
        with open(os.path.join(awfpath, author.name), 'w') as f:
            json.dump(author_webfinger, f)
        with open(os.path.join(authorpath, author.name), 'w') as f:
            json.dump(a, f)
        with open(inboxpath, 'w') as f:
            json.dump(inbox, f)
        with open(outboxpath, 'w') as f:
            json.dump(outbox, f)
        with open(followingpath, 'w') as f:
            json.dump(following, f)
        with open(followerspath, 'w') as f:
            json.dump(followers, f)


def register():
    signals.article_writer_finalized.connect(ap_article)