Posting into BlueSky, Nostr and Threads from Python
At the beginning of this year, I wrote about how I was starting to play around with automating syndication of my content into various social networks in order to better pursue the approach known as POSSE. That earlier post provides quite a long explanation of why I prefer to write here rather than elsewhere, so I won't revisit that now.
The underlying concept, though, is simple: I publish content on www.bentasker.co.uk, something picks up on the change in my rss feed and then posts it into social networks to help increase discoverability.
In this post, I want to write about how I've implemented support for automatic posting into Nostr, Threads and BlueSky.
Notes
- In this post, I'm only really going to deal in code snippets and examples, however, for those who want it, there's a full copy of my eventual script here.
- If you're only here for the infosec/misinformation concern, it's below.
The RSS Fetcher
The logic used to fetch and iterate through my RSS feed is exactly the same as that described in Writing a Simple RSS To Mastodon Bot (in fact, it's a fork of the repo containing that code).
The bit that matters, is that it uses feedparser to process the feed and build a dict containing various elements taken from the feed item
import feedparser
d = feedparser.parse(feed['FEED_URL'])
# Iterate over entries
for entry in d.entries:
# Load the summary
soup = BeautifulSoup(entry.summary, 'html.parser')
# See whether there's an image
thumb = None
img = soup.find("img")
if img:
thumb = img["src"]
en = {}
en['title'] = entry.title
en['link'] = entry.link
en['author'] = False
en['tags'] = []
en['description'] = soup.get_text()
en['thumb'] = thumb
if hasattr(entry, "tags"):
# Iterate over tags and add them
[en['tags'].append(x['term']) for x in entry.tags]
if INCLUDE_AUTHOR == "True" and hasattr(entry, "author"):
en['author'] = entry.author
The dict en is then passed onwards to the social-network specific functions.
For the examples given later in this post, en looks like this (though it actually has more tags and a longer description)
{
'title': 'Writing data from a Bip 3 Smartwatch into InfluxDB',
'link': 'https://www.bentasker.co.uk/posts/blog/software-development/extracting-data-from-zepp-app-for-local-storage-in-influxdb.html',
'author': 'Ben Tasker',
'tags': [
'android',
'blog',
'data'
],
'description': ' I\'ve done a fair bit of playing around with my Watchy since I bought it a couple of months back and, generally, I really like it.\nUnfortunately, as cool as it is, it\'s proven',
'thumb': 'https://www.bentasker.co.uk/images/Logos/matrix.jpg'
}
Threads
The name of Meta's Threads might evoke images of societal breakdown, but being tightly coupled with Instagram means that it had a huge userbase from day 1 (though there are reports that it's declined significantly).
Important note: My Threads account was permanently disabled not long after setting this live (though comments on Github imply others weren't so unlucky).
Meta have also since issued a cease-and-desist against the library's developer
Meta are making it very clear that they don't want you posting into Threads unless it's via their privacy-sucking app.
The following information is preserved for posterity:
They've not yet officially announced a public API, but the app has been reverse engineered to create a python wrapper which can be used to post into Threads.
pip install threads-net
The module makes it incredibly easy to syndicate content into the social network:
from threads import Threads
def build_hashtagless_text_post(entry):
''' Build a string detailing the post
Don't include hashtags - the output of this function
is intended for platforms that don't (currently) use them
Response doesn't include the link (because services like threads
expects them to be provided seperately)
'''
toot_str = ''
if "blog" in entry['tags']:
toot_str += "New Blog: "
elif "documentation" in entry['tags']:
toot_str += "New Documentation: "
toot_str += f"{entry['title']}\n"
return toot_str
def create_Threads_Post(entry):
''' Create a threads posting
'''
# Build a message
toot_str = build_hashtagless_text_post(entry)
try:
threads = Threads(username=THREADS_USER, password=THREADS_PASS)
created_thread = threads.private_api.create_thread(
caption=toot_str,
url=entry['link']
)
return True
except Exception as e:
print(f"Submitting to Threads failed: {e}")
return False
create_Threads_Post(en)
When submitted, a text post is created with an attached link, resulting in a rich preview being displayed:
It really couldn't be more straightforward.
Nostr
Nostr is a decentralised social media protocol and rather than user-chosen usernames, relies on public key cryptography for identifiers (and authentication).
For example, my "username" is npub1tfwl34ldh49ydrx86mg0sdl4es9eaw0s57hcrfdcxneqfrrqu80qydy2p4 and my profile can be viewed via any Nostr app, for example, via Coracle
The use of pki doesn't make for a particularly memorable username, but essentially you have a private key that is then used to sign and publish event objects (i.e. posts).
These are sent via relays - essentially independent but interconnected servers - and anyone using a relay that your post has been published into (directly or indirectly) can see your post. The core protocol can be extended via optional nips, which add support for things like nicknames and contact lists.
There are quite a few libraries available to interact with Nostr - some work well, other's don't seem to work at all.
I added support to my script by using pynostr:
pip install pynostr[websocket-client]
The code driving submission into Nostr looks like this
from pynostr.event import Event
from pynostr.relay_manager import RelayManager
from pynostr.key import PrivateKey
# Replace with your private key
NOSTR_PRIVATE_KEY="nsec12345"
# Define the relays to publish into
NOSTR_RELAYS=[
"wss://relayable.org",
"wss://relay.damus.io",
"wss://nostr.easydns.ca",
"wss://nostrrelay.com",
"wss://relay.snort.social",
"wss://relay.nsecbunker.com"
]
# Any tags that we shouldn't include in posts
SKIP_TAGS = []
def build_nostr_entry(entry):
''' Take the entry dict and build a string to post to nostr
Nostr supports hashtags, so include them
The result is something like
New #Blog: Lorem Ipsum
https://example.com/some-post
#placeholder #CommentInfo #blah
'''
skip_tags = SKIP_TAGS
skip_tags.append("blog")
skip_tags.append("documentation")
toot_str = ''
if "blog" in entry['tags']:
toot_str += "New #Blog: "
elif "documentation" in entry['tags']:
toot_str += "New #Documentation: "
toot_str += f"{entry['title']}\n"
toot_str += f"\n\n{entry['link']}\n\n"
# Tags to hashtags
if len(entry['tags']) > 0:
for tag in entry['tags']:
if tag in skip_tags:
# Skip the tag
continue
toot_str += f'#{tag.replace(" ", "")} '
return toot_str
def create_Nostr_event(entry):
''' Publish into Nostr
'''
try:
out_str = build_nostr_entry(entry)
# Create and sign the event
event = Event(out_str)
event.sign(NOSTR_PK.hex())
# Publish into the relays
NOSTR_RELAY.publish_event(event)
NOSTR_RELAY.run_sync()
return True
except Exception as e:
print(f"Submitting to Nostr failed: {e}")
return False
# Set up the client
DO_NOSTR = False
if NOSTR_PRIVATE_KEY:
DO_NOSTR = True
# Set up the relay connections
NOSTR_RELAY = RelayManager()
NOSTR_PK = PrivateKey.from_nsec(NOSTR_PRIVATE_KEY)
for relay in NOSTR_RELAYS:
NOSTR_RELAY.add_relay(relay)
# Allow time for connections to open
time.sleep(1.25)
create_Nostr_event(en)
# When done, close the connections
NOSTR_RELAY.close_connections()
Definitely a little more complex than publishing into Threads but, then, we're also building a longer message and including hashtags.
The result, viewed in Coracle, looks quite nice:
Obviously, results will vary based on the Nostr client that the user is using.
Bluesky
Until Threads launched, it looked a lot like those fleeing Twitter might all end up on Bluesky, with its growth only slowed by the need to first be invited.
Automated posting into Bluesky is achieved by using the AT Protocol and again there's a python library that can be used:
pip install atproto
Fair warning: of the social networks covered in this post, Bluesky is by far the biggest pain in the arse to get right.
Things start off simple enough:
import requests
import re
from atproto import Client, models
from datetime import datetime
BSKY_USER="me@example.com"
BSKY_PASS="abcdefg"
# Set the client up
BSKY_CLIENT = Client()
profile = BSKY_CLIENT.login(BSKY_USER, BSKY_PASS)
Bluesky doesn't currently do hashtags, so we can build a post using the build_hashtagless_text_post() function that was used for Threads earlier
def build_hashtagless_text_post(entry):
''' Build a string detailing the post
Don't include hashtags - the output of this function
is intended for platforms that don't (currently) use them
Response doesn't include the link (because services like threads
expects them to be provided seperately)
'''
toot_str = ''
if "blog" in entry['tags']:
toot_str += "New Blog: "
elif "documentation" in entry['tags']:
toot_str += "New Documentation: "
toot_str += f"{entry['title']}\n"
return toot_str
So far, so simple, but this is where the headaches start...
If we were to append the link and submit the post, it'd appear, but wouldn't deliver quite the result that we want:
It really doesn't scream "eyecatching" or "interesting content", does it?
The problem is that, unlike all the other social networks, Bluesky puts the onus on the sender to generate and attach the rich preview (worse than that, in fact, for the link to actually be functional, it also needs a facet generated for it - big shout-out to GanWeaving for having shared the code which helped me figure this out)).
Implementing all of this means that things suddenly get quite a bit more complex
URL_PATTERN = re.compile(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+')
def generate_facets_from_links_in_text(text):
''' Based on logic in
https://github.com/GanWeaving/social-cross-post/blob/main/helpers.py
Generate atproto facets for each URL in the text
'''
facets = []
for match in URL_PATTERN.finditer(text):
facets.append(gen_link(*match.span(), match.group(0)))
return facets
def gen_link(start, end, uri):
''' Return a dict defining start + end character along
with the type of the facet and where it should link to.
We're literally saying "characters 4-12" are a link which
should point to foo
'''
return {
"index": {
"byteStart": start,
"byteEnd": end
},
"features": [{
"$type": "app.bsky.richtext.facet#link",
"uri": uri
}]
}
def create_Bluesky_Post(entry):
''' Post into Bluesky
'''
try:
text = build_hashtagless_text_post(entry)
# Append the link
text += f"\n{entry['link']}"
# Identify links and generate AT facets so that they act as links
facets = generate_facets_from_links_in_text(text)
# Create a short description for the preview card
short_desc = ' '.join(entry['description'].split(" ")[0:25]).lstrip()
thumb = None
# See whether there's a thumbnail defined
if entry['thumb']:
# fetch it
response = requests.get(entry['thumb'])
img_data = response.content
# Upload the image
upload = BSKY_CLIENT.com.atproto.repo.upload_blob(img_data)
thumb = upload.blob
# Create a link card embed
embed_external = models.AppBskyEmbedExternal.Main(
external=models.AppBskyEmbedExternal.External(
title=entry['title'],
description=short_desc,
uri=entry['link'],
thumb=thumb
)
)
# Submit the post
BSKY_CLIENT.com.atproto.repo.create_record(
models.ComAtprotoRepoCreateRecord.Data(
repo=BSKY_CLIENT.me.did,
collection='app.bsky.feed.post',
record=models.AppBskyFeedPost.Main(
createdAt=datetime.now().isoformat(),
text=text,
embed=embed_external,
facets=facets
),
)
)
return True
except Exception as e:
print(f"Failed to post to Bluesky: {e}")
return False
Once run, the resulting post looks much better, being far more along the lines that we'd expect
The ability to create custom link preview cards (and, indeed, custom links), raises some security concerns which I've written about seperately
Scheduling and Automation
For my purposes, all of this is rolled together in a container wrapped around a single python script.
Runs are then scheduled via a Kubernetes CronJob so that the script checks my feed for changes every 15 minutes:
apiVersion: batch/v1
kind: CronJob
metadata:
name: posse-publishing
spec:
schedule: "*/15 * * * *"
failedJobsHistoryLimit: 5
successfulJobsHistoryLimit: 5
jobTemplate:
spec:
template:
spec:
restartPolicy: Never
volumes:
- name: hashdir
persistentVolumeClaim:
claimName: hashstore-claim
containers:
- name: posse-bot-container
image: bentasker12/posse-publishing:0.4
imagePullPolicy: IfNotPresent
volumeMounts:
- mountPath: /hashes
name: hashdir
readOnly: false
env:
- name: DRY_RUN
value: "N"
- name: "FEED_URL"
value: "https://www.bentasker.co.uk/rss.xml"
- name: "HASH_DIR"
value: "/hashes"
- name: "TRACKING_MODE"
value: "PERURL"
- name: "INCLUDE_AUTHOR"
value: "False"
- name: "NOSTR_RELAYS"
value: "wss://relayable.org,wss://relay.damus.io,wss://nostr.easydns.ca,wss://nostrrelay.com,wss://relay.snort.social,wss://relay.nsecbunker.com"
- name: BSKY_USER
valueFrom:
secretKeyRef:
name: bluesky
key: user
- name: BSKY_PASS
valueFrom:
secretKeyRef:
name: bluesky
key: pass
- name: THREADS_USER
valueFrom:
secretKeyRef:
name: threads
key: username
- name: THREADS_PASS
valueFrom:
secretKeyRef:
name: threads
key: password
- name: NOSTR_PK
valueFrom:
secretKeyRef:
name: nostr
key: nsec
There's a fuller example of this in my Article scripts repo.
If you want to run the container without Kubernetes, you can also just use docker:
mkdir hashes
docker run --rm \
-v $PWD/hashdir:/hashdir/ \
-e FEED_URL="https://www.bentasker.co.uk/rss.xml" \
-e HASH_DIR="/hashes" \
-e INCLUDE_AUTHOR="False" \
-e NOSTR_RELAYS="wss://relayable.org,wss://relay.damus.io" \
-e NOSTR_PK="<my key>" \
-e BSKY_USER="<my user>" \
-e BSKY_PASS="<my pass>" \
-e THREADS_USER="<my user>" \
-e THREADS_PASS="<my pass>" \
bentasker12/posse-publishing:0.4
Similarly, if you just want to run the script without containerisation that's an option too (there's a copy in my Article scripts repo).
Scheduling is pretty much just a case of adding a cron job, at job, or whatever suits you.
Conclusion
Although they don't all officially provide an API yet, using Python to publish content into these three social networks is perfectly possible.
Somewhat ironically, the network that hasn't got an official public API (Threads) is, by far, the easiest to publish into.
Nostr is a little more complex, but still pretty straightforward.
Bluesky is a full on PITA, having taken an approach that not only increases complexity but means that the network is likely ripe for disinformation tactics (in turn, only made worse by the network's moderation issues).
Official or not, the availability of publishing APIs means that I've been able to add these three networks to my POSSE set.
Now, when I publish a post, it will generally be syndicated to each of the following
- Mastodon
- Nostr
- Bluesky
- Threads
As well as continuing to be available via good old RSS.
