Skip to content

Instantly share code, notes, and snippets.

@Sonictherocketman
Created March 10, 2023 03:26
Show Gist options
  • Save Sonictherocketman/7951ec710b7be1f675e2e95cbcc5220e to your computer and use it in GitHub Desktop.
Save Sonictherocketman/7951ec710b7be1f675e2e95cbcc5220e to your computer and use it in GitHub Desktop.
A script to automatically cross-post from Pine.blog to Mastodon.
#! /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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment