Agile611/post2bsky
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fet

Browse Source
This commit is contained in:
Guillem Hernandez Sola
2026-05-08 14:13:32 +02:00
parent e748c05156
commit fc3dbd4268
1 changed files with 82 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
bsky_post.py
View File

@@ -1,27 +1,27 @@
#!/usr/bin/env python3
"""
bsky_post.py — Post text + optional image or video to Bluesky.
bsky_post.py — Post text + optional image or video to Bluesky/federated PDS.
Usage examples:
python3 bsky_post.py "DIVENDRES!!!!" --video friday.mp4 --username you --password app-pass
python3 bsky_post.py "DIVENDRES!!!!" --video friday.mp4 --username you --password app-pass --service https://eurosky.social
python3 bsky_post.py "Dijous!!!!" --image thursday.jpg --username you --password app-pass
python3 bsky_post.py "Bon dia!" --username you --password app-pass
python3 bsky_post.py "Bon dia!" --image photo.jpg --lang ca --username you --password app-pass
python3 bsky_post.py "Long video!" --video clip.mp4 --username you --password app-pass
python3 bsky_post.py "Long video!" --video clip.mp4 --username you --password app-pass --allow-pds-video-fallback
"""
import argparse
import logging
import mimetypes
import os
import sys
import time
import random
import re
import requests
import sys
import time
from dataclasses import dataclass
from urllib.parse import urlparse
import requests
from atproto import Client, client_utils, models
@@ -271,24 +271,19 @@ def login_with_backoff(
# ============================================================
# Utility
# ============================================================
def is_official_bsky_pds(service_url: str) -> bool:
try:
host = (urlparse(service_url).hostname or "").lower()
return host in {"bsky.social", "bsky.app"} or host.endswith(".bsky.network")
except Exception:
return False
def detect_mime_type(path: str) -> str:
mime, _ = mimetypes.guess_type(path)
if mime:
return mime
ext = os.path.splitext(path)[1].lower()
fallbacks = {
".jpg": "image/jpeg", ".jpeg": "image/jpeg",
".png": "image/png", ".gif": "image/gif",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".webp": "image/webp",
".mp4": "video/mp4", ".mov": "video/quicktime",
".mp4": "video/mp4",
".mov": "video/quicktime",
".webm": "video/webm",
}
return fallbacks.get(ext, "application/octet-stream")
@@ -306,10 +301,16 @@ def wait_with_heartbeat(total_seconds: float, label: str = "indexing") -> None:
remaining -= step
if remaining > 0:
logging.info(f" ...still waiting ({remaining:.0f}s remaining)...")
logging.info("✅ Wait complete.")
def pds_did_from_service_url(service_url: str) -> str:
host = (urlparse(service_url).hostname or "").lower()
if not host:
raise ValueError(f"Invalid --service URL: {service_url}")
return f"did:web:{host}"
# ============================================================
# Media upload — Image
# ============================================================
@@ -327,10 +328,7 @@ def upload_image(
response = client.upload_blob(data)
logging.info("✅ Image uploaded successfully.")
return models.AppBskyEmbedImages.Image(
image=response.blob,
alt=alt_text,
)
return models.AppBskyEmbedImages.Image(image=response.blob, alt=alt_text)
except Exception as e:
logging.error(f"❌ Failed to upload image: {repr(e)}")
@@ -338,7 +336,7 @@ def upload_image(
# ============================================================
# Media upload — Video (PDS-direct fallback only)
# Media upload — Video (PDS direct fallback only)
# ============================================================
def upload_video_via_pds(
client: Client,
@@ -347,9 +345,8 @@ def upload_video_via_pds(
settle_delay_seconds: float = 30.0,
) -> models.AppBskyEmbedVideo.Main | None:
"""
Direct upload to home PDS using upload_blob.
This can produce posts where blob exists but AppView playback is unreliable.
Use only as explicit fallback.
Direct upload to home PDS via upload_blob.
Kept only as optional fallback; playback can be unreliable in clients.
"""
try:
if not os.path.exists(video_path):
@@ -360,20 +357,15 @@ def upload_video_via_pds(
video_bytes = f.read()
size_mb = len(video_bytes) / (1024 * 1024)
logging.warning(
f"🎬 [PDS-direct fallback] Uploading to home PDS: {video_path} ({size_mb:.2f} MB)"
)
logging.warning(f"🎬 [PDS-direct fallback] Uploading: {video_path} ({size_mb:.2f} MB)")
response = client.upload_blob(video_bytes)
blob = response.blob
logging.warning("⚠️ [PDS-direct fallback] Video blob uploaded.")
logging.warning("⚠️ [PDS-direct fallback] Blob uploaded. Waiting for indexing...")
wait_with_heartbeat(settle_delay_seconds, label="PDS/AppView indexing the video blob")
wait_with_heartbeat(settle_delay_seconds, label="PDS/AppView indexing")
return models.AppBskyEmbedVideo.Main(
video=blob,
alt=alt_text,
)
return models.AppBskyEmbedVideo.Main(video=blob, alt=alt_text)
except Exception as e:
logging.error(f"❌ PDS-direct video upload failed: {repr(e)}")
@@ -381,7 +373,7 @@ def upload_video_via_pds(
# ============================================================
# Media upload — Video (video.bsky.app primary path)
# Media upload — Video (video.bsky.app primary)
# ============================================================
def _extract_service_auth_token(upload_auth) -> str | None:
token = getattr(upload_auth, "token", None)
@@ -395,11 +387,15 @@ def _extract_service_auth_token(upload_auth) -> str | None:
def upload_video_via_bsky_service(
client: Client,
video_path: str,
service_url: str,
alt_text: str = "",
) -> models.AppBskyEmbedVideo.Main | None:
"""
Upload a video via centralized video.bsky.app service.
This is the reliable playback path for Bluesky clients.
Upload via centralized video.bsky.app service.
IMPORTANT FIX:
getServiceAuth(aud=...) must use the user's PDS DID (e.g. did:web:eurosky.social),
not did:web:video.bsky.app.
"""
try:
if not os.path.exists(video_path):
@@ -413,26 +409,28 @@ def upload_video_via_bsky_service(
logging.info(f"🎬 [video.bsky.app] Uploading: {video_path} ({size_mb:.2f} MB)")
VIDEO_HOST = "https://video.bsky.app"
VIDEO_DID = "did:web:video.bsky.app"
pds_did = pds_did_from_service_url(service_url)
# Robust params typing across atproto versions
# Robust for different atproto versions
try:
params = models.ComAtprotoServerGetServiceAuth.Params(
aud=VIDEO_DID,
aud=pds_did, # <-- critical fix
lxm="app.bsky.video.uploadVideo",
exp=int(time.time()) + 60 * 30,
)
upload_auth = client.com.atproto.server.get_service_auth(params)
except Exception:
upload_auth = client.com.atproto.server.get_service_auth({
"aud": VIDEO_DID,
"lxm": "app.bsky.video.uploadVideo",
"exp": int(time.time()) + 60 * 30,
})
upload_auth = client.com.atproto.server.get_service_auth(
{
"aud": pds_did, # <-- critical fix
"lxm": "app.bsky.video.uploadVideo",
"exp": int(time.time()) + 60 * 30,
}
)
token = _extract_service_auth_token(upload_auth)
if not token:
logging.error("❌ Failed to get service auth token for video.bsky.app.")
logging.error("❌ Failed to extract service auth token.")
return None
user_did = client.me.did
@@ -447,30 +445,23 @@ def upload_video_via_bsky_service(
upload_resp = requests.post(upload_url, headers=headers, data=video_bytes, timeout=180)
if upload_resp.status_code != 200:
logging.error(
f"❌ video.bsky.app upload failed: "
f"{upload_resp.status_code} - {upload_resp.text}"
)
logging.error(f"❌ video.bsky.app upload failed: {upload_resp.status_code} - {upload_resp.text}")
return None
payload = upload_resp.json()
job_id = payload.get("jobId")
body = upload_resp.json()
job_id = body.get("jobId")
if not job_id:
logging.error(f"❌ No jobId returned from video service. Response: {payload}")
logging.error(f"❌ No jobId returned from video service. Response: {body}")
return None
logging.info(f"⏳ Job {job_id} accepted — polling status...")
status_url = f"{VIDEO_HOST}/xrpc/app.bsky.video.getJobStatus"
deadline = time.time() + 600 # 10 min for big videos
deadline = time.time() + 600
while time.time() < deadline:
status_resp = requests.get(status_url, params={"jobId": job_id}, timeout=30)
if status_resp.status_code != 200:
logging.error(
f"❌ Job status check failed: "
f"{status_resp.status_code} - {status_resp.text}"
)
logging.error(f"❌ Job status check failed: {status_resp.status_code} - {status_resp.text}")
return None
status_json = status_resp.json()
@@ -483,15 +474,10 @@ def upload_video_via_bsky_service(
logging.error(f"❌ No blob in completed job status: {status_json}")
return None
# Small propagation cushion
wait_with_heartbeat(8, label="CDN propagation")
blob_ref = models.BlobRef.from_dict(blob_dict)
logging.info("✅ Video processed and blob returned.")
return models.AppBskyEmbedVideo.Main(
video=blob_ref,
alt=alt_text,
)
logging.info("✅ Video processed successfully.")
return models.AppBskyEmbedVideo.Main(video=blob_ref, alt=alt_text)
if state == "JOB_STATE_FAILED":
logging.error(f"❌ Video processing failed: {job_status}")
@@ -509,7 +495,7 @@ def upload_video_via_bsky_service(
# ============================================================
# Media upload — Video dispatcher
# Video dispatcher
# ============================================================
def upload_video_smart(
client: Client,
@@ -519,38 +505,34 @@ def upload_video_smart(
settle_delay_seconds: float = 30.0,
allow_pds_video_fallback: bool = False,
) -> models.AppBskyEmbedVideo.Main | None:
"""
Reliable policy:
1) Always try video.bsky.app first (best playback compatibility).
2) Optional direct-PDS fallback only if explicitly enabled.
"""
if is_official_bsky_pds(service_url):
logging.info(f"🌐 PDS appears official ({service_url}). Using video.bsky.app.")
else:
logging.info(
f"🌍 PDS is self-hosted/federated ({service_url}). "
"Still trying video.bsky.app first for client playback reliability."
)
logging.info(
f"🌍 PDS ({service_url}). Trying video.bsky.app first for playback reliability."
)
embed = upload_video_via_bsky_service(client, video_path, alt_text=alt_text)
embed = upload_video_via_bsky_service(
client=client,
video_path=video_path,
service_url=service_url,
alt_text=alt_text,
)
if embed:
return embed
if allow_pds_video_fallback:
logging.warning(
"⚠️ video.bsky.app upload failed; trying direct PDS fallback "
"(may produce unplayable videos in some clients)."
"⚠️ video.bsky.app failed; trying direct PDS fallback "
"(may be unplayable in some clients)."
)
return upload_video_via_pds(
client,
video_path,
client=client,
video_path=video_path,
alt_text=alt_text,
settle_delay_seconds=settle_delay_seconds,
)
logging.error(
"❌ video.bsky.app upload failed. Not posting unreliable direct-PDS video "
"(enable --allow-pds-video-fallback to override)."
"❌ video.bsky.app failed. Not posting unreliable direct-PDS video. "
"Use --allow-pds-video-fallback to override."
)
return None
@@ -576,8 +558,8 @@ def post_to_bsky(
logging.info(f"🎬 Preparing video upload: {video_path}")
video_embed = upload_video_smart(
client,
video_path,
client=client,
video_path=video_path,
service_url=service_url,
alt_text=alt_text,
settle_delay_seconds=video_settle_delay,
@@ -589,11 +571,7 @@ def post_to_bsky(
return False
logging.info("🚀 Sending video post...")
result = client.send_post(
text=rich_text,
embed=video_embed,
langs=langs,
)
result = client.send_post(text=rich_text, embed=video_embed, langs=langs)
elif image_path:
image = upload_image(client, image_path, alt_text=alt_text)
@@ -625,7 +603,7 @@ def main():
setup_logging()
parser = argparse.ArgumentParser(
description="Post text + optional image or video to Bluesky."
description="Post text + optional image or video to Bluesky/federated PDS."
)
parser.add_argument("text", help="Post text content")
parser.add_argument("--username", required=True, help="Bluesky handle or email")
@@ -639,12 +617,12 @@ def main():
"--video-settle-delay",
type=float,
default=30.0,
help="Seconds to wait after direct-PDS video fallback upload before posting (default: 30)",
help="Seconds to wait after direct-PDS fallback upload before posting (default: 30).",
)
parser.add_argument(
"--allow-pds-video-fallback",
action="store_true",
help="Allow direct-PDS fallback if video.bsky.app upload fails (less reliable playback).",
help="Allow direct-PDS fallback if video.bsky.app fails (less reliable playback).",
)
args = parser.parse_args()
@@ -655,10 +633,10 @@ def main():
client = Client(base_url=args.service)
success = login_with_backoff(
client,
args.username,
args.password,
args.service,
client=client,
username=args.username,
password=args.password,
service_url=args.service,
max_attempts=RetryConfig.login_max_attempts,
base_delay=RetryConfig.login_base_delay_seconds,
max_delay=RetryConfig.login_max_delay_seconds,
@@ -668,8 +646,9 @@ def main():
sys.exit(1)
langs = [l.strip() for l in args.lang.split(",") if l.strip()]
post_success = post_to_bsky(
client,
client=client,
text=args.text,
langs=langs,
image_path=args.image,