TwitterToBluesky
仕組み
Zapierを使って「新しいTwitterの投稿を取得し、PythonのコードでBlueskyへツイート」の流れを組む Twitter APIはよく知らないけど色々辛そうなのであんまり触りたくなかった
せっかくZapierがラップしてくれてるのでありがたく使わせてもらう
現状できていること
ツイートされた文字列をBlueskyに投稿
現状できていないこと
リンクをクリック可能な青い形式で投稿
なぜかクリックできない黒文字で投稿されてしまう
多分richtextのlinkとして渡す必要がありそう?
本来のリンクを投稿
本来のリンクではなく、t.co/~~のTwitterによる省略リンクがBlueskyに投稿されてしまう
画像や動画を投稿
t.co/~~のリンクがBlueskyで投稿されてしまう
やり方
Zapierで、以下のZapを組む
https://gyazo.com/9693e542c4aaa95d1cc599e67e7be95a
https://gyazo.com/cbcaea3d03a2e3d53967e75f0552979a https://gyazo.com/b2a5ff290cd008597c2ace2931740bd7
ActionのPythonコードは以下
ATP_USERNAMEとATP_PASSWORDに、自分のユーザー名とパスワードを入れる
あんまり直にパスワードは書きたくないが、Zapierで環境変数を使うには有料プランに課金しないといけないので仕方ないblu3mo.icon
意図的に公開しない限り、コードはprivateに保たれるという認識blu3mo.icon
違ったら教えてください🙇
(ここに入力した結果パスワードを漏らしても責任は負いません)
code: zapier.py
import requests
import datetime
#
#####
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)
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):
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) # to test locally
main(input_data)
-.icon
以下調べた時の参考メモ
embed linkを出したい
技術的には出来たけど、zapierで1秒以上かかるのでダメだった
リンクを青くしたい
Skylightが使っているentitiesはdeprecated
代わりにrichtextがある
なるほど、段々読み方がわかってきたblu3mo.icon
type:refとかunionだと、ref/refsにある別の型を参照できるのね
pythonでポストしている
ATP_HOST + "/xrpc/com.atproto.repo.createRecord"に普通にpost requestを投げているな
xrpcはhttpベースの通信プロトコルだから、普通にpostを投げれば良いのねblu3mo.icon