Mastodon 同步到 Blinko 脚本

利用 Mastodon 的 Webhook 可以主动同步 Status 到 Blinko（或其他平台），而不是利用 RSS 或 Crontab 被动式同步。

已测试版本

• Blinko: v0.8.8
• Mastodon: v4.3.1
• Sink: v0.1.4

安装工具

请安装工具，若有报错，请安装其他对应工具

• sudo apt install jq
• sudo apt install lynx

配置

以下为 Shell Script 脚本内容，请注意替换：

• BLINKO_HOST=""
• BLINKO_ACCESS_TOKEN=""
• BLINKO_TYPE=""
• BLINKO_SHARE=""
• MASTODON_INSTANCE=""
• MASTODON_ID=""
• SKIP_MASTODON_REPLY=
• SKIP_MASTODON_REBLOG=
• HOME_DIR=~
• FILE_PATH=$HOME_DIR/.latest_blinko_id.json
• AI_DIFF=true
• AI_API="https://api.deepseek.com"
• AI_AUTHORIZATION=""
• AI_MODEL
• SINK_ENABLE
• SINK_HOST="https://s.e5n.cc"
• SINK_NUXT_SITE_TOKEN=""
• S3_ENABLE

查找 ID： https://INSTANCE/api/v1/accounts/lookup?acct=USERNAME

JSON 数据文件内容

.latest_blinko_id.json

{
  "latest_blinko_id": "",
  "latest_mastodon_id": "",
  "bind": []
}

脚本内容

mastodon_sync_to_blinko.sh

#!/bin/bash

sleep 5

# Version: 2024.11.23

# 已测试版本:
# Blinko: v0.8.6
# Mastodon: v4.3.1
# Sink: v0.1.4

# ======================================================
# 配置开始

# Blinko Host
BLINKO_HOST="https://blinko.eallion.com/"

# Blinko Access Token
BLINKO_ACCESS_TOKEN="eyJh****"

# 发布 Blinko 的类型 ('type=0' 为闪念，'type=1' = 笔记) 二选一
BLINKO_TYPE=1

# 发布 Blinko 的可见性，即 isShare ('true' 为公开，'false' = 不公开) 二选一
BLINKO_SHARE=true

# Mastodon Instance
MASTODON_INSTANCE="https://e5n.cc/"

# Mastodon ID, Find ID: https://INSTANCE/api/v1/accounts/lookup?acct=USERNAME
MASTODON_ID="111136231674527355"

# 跳过回复和转嘟
SKIP_MASTODON_REPLY=true
SKIP_MASTODON_REBLOG=true

# 获取当前用户的家目录路径及保存 ID 的文件，保持默认，不用更改
HOME_DIR=~
FILE_PATH=$HOME_DIR/.mastodon_blinko_id.json

# AI 比较文本相似度，兼容 OpenAI 格式的模型都可以
AI_DIFF=false
AI_API="https://api.deepseek.com"
AI_TOKEN="sk-****"
AI_MODEL="deepseek-chat"

# Deploy Sink: https://github.com/ccbikai/Sink
SINK_ENABLE=false
SINK_HOST="https://s.e5n.cc"
SINK_NUXT_SITE_TOKEN="SINK-****"

# 上传 Statuses 到 s3 对象存储
# 未配置好最下面的 aliyun cli 或者 coscmd，请勿打开此设置
S3_ENABLE=false

# 配置结束
# ======================================================

# 以下内容不用更改

# 检查 ID 文件是否存在
if [ ! -f "$FILE_PATH" ]; then
  # 如果文件不存在，则创建文件并写入 JSON 数据
  echo '
{
  "latest_blinko_id": "0",
  "latest_mastodon_id": "0",
  "bind": []
}
' > "$FILE_PATH"
  echo "Data file created: $FILE_PATH"
else
  # 如果文件存在，则跳过并进行后续步骤
  echo "Local data exist, skipping..."
fi

# 拼接 Blinko API 和 Token
if [[ "$BLINKO_HOST" != */ ]]; then
  BLINKO_HOST="$BLINKO_HOST/"
fi
BLINKO_API_HOST="${BLINKO_HOST}api/v1/note/upsert"

BLINKO_URL="${BLINKO_HOST}api/v1/note/public-list"

# Mastodon 的 API
if [[ "$MASTODON_INSTANCE" != */ ]]; then
  MASTODON_INSTANCE="$MASTODON_INSTANCE/"
fi
MASTODON_CONTENT_URL="${MASTODON_INSTANCE}api/v1/accounts/${MASTODON_ID}/statuses?limit=1&exclude_replies=${SKIP_MASTODON_REPLY}&exclude_reblogs=${SKIP_MASTODON_REBLOG}"

# 前置判断是否为回复嘟文，减少 AI Token 开支
LATEST_CONTENT_URL="${MASTODON_INSTANCE}api/v1/accounts/${MASTODON_ID}/statuses?limit=1"
LATEST_CONTENT_RESPONSE=$(curl -s "$LATEST_CONTENT_URL")
IS_REPLY=$(echo "$LATEST_CONTENT_RESPONSE" | jq -r '.[0].in_reply_to_id')
IS_REBLOG=$(echo "$LATEST_CONTENT_RESPONSE" | jq -r '.[0].reblog')
# 检查 IS_REPLY 是否为 null
if [ "$SKIP_MASTODON_REPLY" == true ] && [ "$IS_REPLY" != "null" ]; then
  echo "Latest status is reply, exiting..."
  echo "Skipped: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
  echo "============================="
  exit 0
fi
# 前置判断是否为转载，减少 AI Token 开支
if [ "$SKIP_MASTODON_REBLOG" == true ] && [ "$IS_REBLOG" != "null" ]; then
  echo "Latest status is reblog, exiting..."
  echo "Skipped: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
  echo "============================="
  exit 0
fi

# Mastodon 最新 Status 的 ID
LATEST_MASTODON_ID=$(curl --connect-timeout 60 -s $MASTODON_CONTENT_URL | jq -r '.[0].id')

# Blinko 获取最新的 Blinko ID
LATEST_BLINKO_ID=$(curl --connect-timeout 60 -s -X 'POST' $BLINKO_URL -H 'accept: application/json' -H 'Content-Type: application/json' -d '{ "page": 1, "size": 1 }' | jq -r '.[0].id')

# 定义 LOCAL_BLINKO_ID 变量
LOCAL_BLINKO_ID=$(cat "$FILE_PATH" | jq -r '.latest_blinko_id')
LOCAL_MASTODON_ID=$(cat "$FILE_PATH" | jq -r '.latest_mastodon_id')

# Webhook 触发时，判断 Mastodon 最新 ID 是否为暂存 ID，防止重复同步
if [ "$LATEST_MASTODON_ID" == "$LOCAL_MASTODON_ID" ]; then
  echo "Mastodon no updated, skipping..."
  echo "Skipped: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
  echo "============================="
  exit 0
fi

CONTENT=$(curl --connect-timeout 60 -s $MASTODON_CONTENT_URL | jq -r '.[0]')

TEXT=$(echo "$CONTENT" | jq -r '.content')

# 解码 Unicode 转义序列
TEXT=$(echo "$TEXT" | sed 's/\\u[0-9a-fA-F]\{4\}/\\x/g')

# 去除外层引号
TEXT=$(echo "$TEXT" | sed 's/^"//;s/"$//')

# 处理 <span> 标签（去除）
TEXT=$(echo "$TEXT" | sed -E 's/<span[^>]*>//g;s/<\/span>//g')

# 处理 <a> 标签，根据 class 属性进行不同的处理
# 如果 class 包含 hashtag，保留标签内的文字内容
TEXT=$(echo "$TEXT" | sed -E 's/<a[^>]*class="[^"]*hashtag[^"]*"[^>]*>([^<]*)<\/a>/\1/g')
# 如果 class 不包含 hashtag，提取 href 内容
TEXT=$(echo "$TEXT" | sed -E 's/<a[^>]*href="([^"]+)"[^>]*>([^<]*)<\/a>/\1/g')

# 去除 <p> 标签
TEXT=$(echo "$TEXT" | sed 's/<p[^>]*>//g;s/<\/p>//g')

# 处理 <br /> 标签，替换为换行符
TEXT=$(echo "$TEXT" | sed 's/<br \/>/\n/g')

# 处理 blockquote，替换为 Markdown 的 `>`
TEXT=$(echo "$TEXT" | sed 's/<blockquote>/>/g;s/<\/blockquote>//g')

# 替换 &gt;
TEXT=$(echo "$TEXT" | sed 's/\&gt\;/>/g')

MEDIA=$(echo $CONTENT | jq -r '.media_attachments')
# 判断 Media 的内容
if [ "$MEDIA" != "null" ]; then
  MEDIAS=$(echo $CONTENT | jq -r '.media_attachments[] | select(.type=="image") | .url')
  # 拼接图片
  images=""
  for url in $MEDIAS; do
    images="$images![image]($url)\n"
  done
  TEXT=$(echo "$TEXT\n$images" | sed 's/\\n*$//')
else
  # 普通内容
  TEXT=$(echo "$TEXT" | sed 's/\\n*$//')
fi

# 判断内容是否为空
if [ -z "$TEXT" ] || [ "$TEXT" == "\\n" ]; then
  echo "Content is empty, skipping..."
  echo "Skipped: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
  echo "============================="
  exit 0
fi

# 双引号转义
TEXT=$(echo "$TEXT" | sed 's/"/\\"/g')

# Webhook 触发时，判断 Blinko 最新 ID 是否为暂存 ID
# 当 Blinko 单方面有更新后，验证 Mastodon 和 Blinko 的 ID 绑定关系（Todo）
# if [ "$LATEST_BLINKO_ID" == "$LOCAL_BLINKO_ID" ]; then
#  echo "Blinko no updated, skipping..."
#  echo "Skipped: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
#  echo "============================="
# exit 0
# fi

# 利用 Deepseek 对比 Mastodon 和 Blinko 的相似度
CONTENT_BLINKO=$(curl --connect-timeout 60 -s -X 'POST' $BLINKO_URL -H 'accept: application/json' -H 'Content-Type: application/json' -d '{ "page": 1, "size": 1 }' | jq -r '.[0].content')
CONTENT_MASTODON=$TEXT

if [[ "$AI_DIFF" == true ]]; then
    REQUEST_BODY=$(cat <<EOF
        {
          "messages": [
            {
              "content": "你是一个比较文本相似度的助手",
              "role": "system"
            },
            {
              "content": "比较文本1:“ --- $CONTENT_BLINKO --- ”和文本2:“ --- $CONTENT_MASTODON --- ”的相似度，超过65%的相似度就判定为相似，如果相似就回答数字1，如果不相似就回答数字0，除了数字1或者数字0不能回答其他任何内容。",
              "role": "user"
            }
          ],
          "model": "$AI_MODEL",
          "frequency_penalty": 0,
          "max_tokens": 2048,
          "presence_penalty": 0,
          "stop": null,
          "stream": false,
          "temperature": 1,
          "top_p": 1,
          "logprobs": false,
          "top_logprobs": null
        }
EOF
    )

    AI_RESPONSE=$(curl -s -L -X POST "$AI_API/chat/completions" \
    -H 'Content-Type: application/json' \
    -H 'Accept: application/json' \
    -H "Authorization: Bearer $AI_TOKEN" \
    --data-raw "$REQUEST_BODY")

    AI_DIFF_RESULT=$(echo "$AI_RESPONSE" | jq -r '.choices[0].message.content')
    if [ "$AI_DIFF_RESULT" == 1 ]; then
      echo "[AI] Content is duplicate, skipping..."
      echo "Skipped: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
      echo "============================="
      exit 0
    fi
else
    # 对比 Matodon 和 Blinko 的 Content 内容的 MD5 值（不一定精确）

    # 获取最新 Blinko 的 MD5
    LATEST_BLINKO_MD5=$(echo $CONTENT_BLINKO | tr -d '"' | md5sum | cut -d' ' -f1)
    # 获取最新 Mastodon 的 MD5
    LATEST_TEXT_MD5=$(echo $TEXT | tr -d '"' | md5sum | cut -d' ' -f1)

    # 通过 MD5 判断内容是否重复
    if [ "$LATEST_TEXT_MD5" == "$LATEST_BLINKO_MD5" ]; then
      echo "[MD5] Content is duplicate, skipping..."
      echo "Skipped: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
      echo "============================="
      exit 0
    fi
fi

# 替换 NeoDB 的评分 Emoji
TEXT=$(echo "$TEXT" | sed "s/:star_empty:/🌑/g; s/:star_half:/🌗/g; s/:star_solid:/🌕/g")

# 去掉最末尾的空行
TEXT=$(echo "$TEXT" | sed 's/\\n$//')

# 发布 Blinko 并获取返回的 JSON 数据
BLINKO_RESPONSE=$(curl --request POST \
  --url $BLINKO_API_HOST \
  --header "Authorization: Bearer $BLINKO_ACCESS_TOKEN" \
  --header "Content-Type: application/json" \
  --data "{
  \"content\": \"$TEXT\",
  \"type\": $BLINKO_TYPE,
  \"isShare\": $BLINKO_SHARE
}")

# 从返回的 JSON 数据中提取 Blinko 的 id 值
NEW_BLINKO_ID=$(echo "$BLINKO_RESPONSE" | jq -r '.id')

# 更新 JSON 文件中的 latest_blinko_id 的值
jq ".latest_blinko_id = \"$NEW_BLINKO_ID\"" "$FILE_PATH" > "${FILE_PATH}.tmp" && mv "${FILE_PATH}.tmp" "$FILE_PATH"

# 更新 JSON 文件中的 latest_mastodon_id 的值
jq ".latest_mastodon_id = \"$LATEST_MASTODON_ID\"" "$FILE_PATH" > "${FILE_PATH}.tmp" && mv "${FILE_PATH}.tmp" "$FILE_PATH"

# 更新 Mastodon 和 Blinko 的 ID 的绑定关系，并确保 "bind" 中的数组保留唯一键，键也只有唯一值
jq ".bind += [{\"$LATEST_MASTODON_ID\": \"$NEW_BLINKO_ID\"}] | .bind = (.bind | unique)" "$FILE_PATH" > "${FILE_PATH}.tmp" && mv "${FILE_PATH}.tmp" "$FILE_PATH"

# POST 到 Sink
if [[ "$SINK_ENABLE" == true ]]; then
  SINK_URL="${MASTODON_INSTANCE}@eallion/${LATEST_MASTODON_ID}"
  # B + id 以防冲突
  SINK_SLUG="B${NEW_BLINKO_ID}"

  curl -s -X POST \
      -H "authorization: Bearer ${SINK_NUXT_SITE_TOKEN}" \
      -H "content-type: application/json" \
      -d "{\"url\": \"${SINK_URL}\", \"slug\": \"${SINK_SLUG}\"}" \
      "${SINK_HOST}/api/link/create"
fi

echo "Sync Mastodon to Blinko Successful!"
echo "Done: $(TZ=UTC-8 date +"%Y-%m-%d %T")"
echo "============================="