- I want to transfer TwitterToBluesky (blu3mo).
- Procedure:
- Use Zapier to create a flow that retrieves new Twitter posts and tweets them to Bluesky using Python code.
- I don’t know much about the Twitter API, so I didn’t want to touch it too much.
- Since Zapier is handling it for me, I’m grateful to use it.
- What is currently working:
- Posting the tweeted text to Bluesky.
- What is currently not working:
- Posting links in a clickable blue format.
- Posting the original link.
- Instead of the original link, the shortened link by Twitter (t.co/~~) is being posted to Bluesky.
- Posting images or videos.
- The link t.co/~~ is being posted to Bluesky.
- How to do it:
- Create the following Zap in Zapier:
- The Python code for the Action is as follows:
- Put your username and password in
ATP_USERNAME and ATP_PASSWORD.
- I don’t want to write the password directly, but I have to because using environment variables in Zapier requires a paid plan.
- As long as you don’t intentionally make it public, the code will be kept private.
- Please let me know if I’m wrong. 🙇
- (I am not responsible if you leak your password by entering it here)
zapier.py
import requests
import datetime
#
#####
ATP_HOST = "https://bsky.social" # Change host if you're using other PDS
ATP_USERNAME = "" # Your username, without @
ATP_PASSWORD = "" # Your password
######
def fetch_external_embed(uri):
try:
response = requests.get(uri)
if response.status_code == 200:
html_content = response.text
title_match = re.search(r'<title>(.+?)</title>', html_content, re.IGNORECASE | re.DOTALL)
title = title_match.group(1) if title_match else ""
description_match = re.search(r'<meta[^>]+name=["\']description["\'][^>]+content=["\'](.*?)["\']', html_content, re.IGNORECASE)
description = description_match.group(1) if description_match else ""
return {
"uri": uri,
"title": title,
"description": description
}
else:
print("Error fetching the website")
return None
except Exception as e:
print(f"Error: {e}")
return None
def find_uri_position(text):
pattern = r'(https?://\S+)'
match = re.search(pattern, text)
if match:
uri = match.group(0)
start_position = len(text[:text.index(uri)].encode('utf-8'))
end_position = start_position + len(uri.encode('utf-8')) - 1
return (uri, start_position, end_position)
else:
return None
def login(username, password):
data = {"identifier": username, "password": password}
resp = requests.post(
ATP_HOST + "/xrpc/com.atproto.server.createSession",
json=data
)
atp_auth_token = resp.json().get('accessJwt')
if atp_auth_token == None:
raise ValueError("No access token, is your password wrong?")
did = resp.json().get("did")
return atp_auth_token, did
def post_text(text, atp_auth_token, did, timestamp=None):
if not timestamp:
timestamp = datetime.datetime.now(datetime.timezone.utc)
timestamp = timestamp.isoformat().replace('+00:00', 'Z')
headers = {"Authorization": "Bearer " + atp_auth_token}
found_uri = find_uri_position(text)
if found_uri:
uri, start_position, end_position = found_uri
facets = [
{
"index": {
"byteStart": start_position,
"byteEnd": end_position + 1
},
"features": [
{
"$type": "app.bsky.richtext.facet#link",
"uri": uri
}
]
},
]
# taking over 1 sec, so this cannot be used in Zapier
# embed = {
# "$type": "app.bsky.embed.external",
# "external": fetch_external_embed(uri)
# }
data = {
"collection": "app.bsky.feed.post",
"$type": "app.bsky.feed.post",
"repo": "{}".format(did),
"record": {
"$type": "app.bsky.feed.post",
"createdAt": timestamp,
"text": text,
"facets": facets,
# "embed": embed
}
}
resp = requests.post(
ATP_HOST + "/xrpc/com.atproto.repo.createRecord",
json=data,
headers=headers
)
return resp
def main(input_data):
if (input_data['TWEET_TEXT'][0] == '@'):
print(input_data['TWEET_TEXT'])
print("Tweet starting with @, not posting to ATP")
return []
atp_auth_token, did = login(ATP_USERNAME, ATP_PASSWORD)
print(atp_auth_token, did)
post_resp = post_text(input_data['TWEET_TEXT'], atp_auth_token, did)
return[post_resp.json()]
# to test locally# input_data = {"TWEET_TEXT": "post test https://t.co"}
main(input_data)
---
Below are some notes I found when researching:
- I want to embed a link.
- Technically, it's possible, but it doesn't work with Zapier because it takes more than 1 second.
- [https://gist.github.com/blu3mo/ecb735b16b12395166eec452c5816fb3](https://gist.github.com/blu3mo/ecb735b16b12395166eec452c5816fb3)
- I want to make the links blue.
- [https://github.com/penpenpng/skylight/blob/650330bddbbfbd1c165010e881ef1382f713aa60/src/lib/bsky.ts#L9](https://github.com/penpenpng/skylight/blob/650330bddbbfbd1c165010e881ef1382f713aa60/src/lib/bsky.ts#L9)
- The "entities" used by Skylight is deprecated.
- Instead, there is "richtext".
- [https://atproto.com/lexicons/app-bsky-feed](https://atproto.com/lexicons/app-bsky-feed)
- I see, I'm starting to understand how it works (blu3mo).
- With "type:ref" or "union", you can refer to other types in "ref/refs".
- [https://github.com/ianklatzco/atprototools/blob/4674370b7a5766b5106eb5e7a433c3fae817a6c1/atprototools/__init__.py#L27](https://github.com/ianklatzco/atprototools/blob/4674370b7a5766b5106eb5e7a433c3fae817a6c1/atprototools/__init__.py#L27)
- It's posting in Python.
- It's simply sending a post request to `ATP_HOST + "/xrpc/com.atproto.repo.createRecord"`.
- [[xrpc]] is an HTTP-based communication protocol, so you can just send a regular post request (blu3mo).
- [https://atproto.com/specs/xrpc](https://atproto.com/specs/xrpc)