A script to automatically cross-post from Pine.blog to Mastodon.

pine-to-mastodon.py

#! /usr/bin/env python3
import argparse
from bs4 import BeautifulSoup
from contextlib import contextmanager
from datetime import datetime
import markdown
import json
import httpx
import sys
import uuid
import os

# Post to mastodon

PINE_URL = "https://pine.blog/api/users/drafts?show_all=true"
PINE_API_KEY = ""
MASTODON_URL = "https://mastodon.social/api/v1/statuses"
MASTODON_API_KEY = ""
ID_FILE = "postids.json"
PIDFILE = "p2m.pid"
DEFAULT_CACHE = {
    'ids': [],
}


@contextmanager
def pidguard():
    id = uuid.uuid4().hex
    with open(PIDFILE, 'w+') as f:
        pid = f.read()
        if pid and pid != id:
            raise Exception('Refusing to execute. Already running.')
        f.seek(0)
        f.write(id)

    try:
        yield
    finally:
        os.remove(PIDFILE)


def get_recent_published_posts():
    response = httpx.get(
        PINE_URL,
        headers={
            'Authorization': f'Token {PINE_API_KEY}',
        }
    )
    return (
        post for post in response.json()['results']
        if post['published']
    )


def post_to_mastodon(id, text):
    response = httpx.post(
        MASTODON_URL,
        headers={
            'Authorization': f'Bearer {MASTODON_API_KEY}',
            'Idempotency-Key': id,
        },
        json={
            'status': text,
        }
    )
    response.raise_for_status()


def build_post(post):
    html = markdown.markdown(post['contents'])
    return (
        strip_tags(html).strip()
        + ' '
        + ' '.join([link for link in get_links(html)])
    )

def strip_tags(html):
    return BeautifulSoup(
        html,
        features='html.parser',
    ).get_text()


def get_links(html):
    soup = BeautifulSoup(html, 'html.parser')
    links = [
        a['href']
        for a in soup.find_all('a')
    ]
    images = [
        img['src']
        for img in soup.find_all('img')
    ]
    return set(images + links)


def get_cache():
    try:
        with open(ID_FILE, 'r') as f:
            found_cache = json.load(f)
            assert all(key in DEFAULT_CACHE for key in found_cache)
            return found_cache
    except FileNotFoundError:
        return dict(DEFAULT_CACHE)
    except AssertionError as e:
        print('Invalid cache. Please remedy and retry.')
        raise e


def set_cache(cache):
    with open(ID_FILE, 'w') as f:
        return json.dump(cache, f)


def main(max_items_to_post=None, dry_run=False):
    with pidguard():
        cache = get_cache()
        posts = get_recent_published_posts()
        new_posts = [
            post for post in posts
            if post['id'] not in cache['ids']
        ]
        for i, post in enumerate(new_posts):
            if max_items_to_post and i >= max_items_to_post:
                break

            try:
                if dry_run:
                    print(f'Dry Run - Marking {post["id"]} as posted...')
                else:
                    print(f'Posting {post["id"]} to Mastodon...')
                    post_to_mastodon(post['id'], build_post(post))
            except Exception as e:
                print(f'Failed to post {post["id"]} to mastodon. {e}')
            finally:
                cache['ids'].append(post['id'])

        set_cache(cache)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        'Injest a feed from Pine.blog and post new statuses to Mastodon.'
    )
    parser.add_argument(
        '-d', '--dry-run',
        action='store_true',
        help=(
            'A dry run will mark all unposted items as posted but will '
            'not actually post anything.'
        ),
    )
    parser.add_argument(
        '-m', '--max-items',
        type=int,
        default=None,
        help=(
            'A dry run will mark all unposted items as posted but will '
            'not actually post anything.'
        ),
    )
    args = parser.parse_args()

    main(max_items_to_post=args.max_items, dry_run=args.dry_run)