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

fff

Browse Source
This commit is contained in:
Guillem Hernandez Sola
2026-05-08 11:52:23 +02:00
parent 337a59039a
commit 1224cdf9c7
1 changed files with 64 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
bsky_post.py
View File

@@ -7,15 +7,7 @@ Usage examples:
python3 bsky_post.py "Dijous!!!!" --image thursday.jpg
python3 bsky_post.py "Bon dia!"
python3 bsky_post.py "Bon dia!" --image photo.jpg --lang ca
Notes:
* Self-hosted / federated PDSes (e.g. eurosky.social) upload videos
DIRECTLY to the user's PDS as a generic blob — no video.bsky.app
service auth is required.
* The official Bluesky network (bsky.social, *.bsky.network) uses the
centralized video.bsky.app pipeline with a service-auth token scoped
via `lxm` to `app.bsky.video.uploadVideo`.
* The script auto-detects which path to use based on --service.
python3 bsky_post.py "Long video!" --video clip.mp4 --video-settle-delay 25
"""
import argparse
@@ -153,9 +145,6 @@ def is_transient_error(error_obj) -> bool:
def get_rate_limit_wait_seconds(error_obj, default_delay: float, max_delay: float) -> float:
"""
Parse common rate-limit headers and return a bounded wait time in seconds.
"""
try:
now_ts = int(time.time())
@@ -341,6 +330,29 @@ def upload_image(
return None
# ============================================================
# Helpers — settle delay
# ============================================================
def wait_with_heartbeat(total_seconds: float, label: str = "indexing") -> None:
"""
Sleep `total_seconds` while logging a heartbeat every 5s so the operator
can see we're still alive and how much time is left.
"""
if total_seconds <= 0:
return
logging.info(f"⏳ Waiting {total_seconds:.0f}s for PDS {label}...")
remaining = total_seconds
while remaining > 0:
step = min(5.0, remaining)
time.sleep(step)
remaining -= step
if remaining > 0:
logging.info(f" ...still waiting ({remaining:.0f}s remaining)...")
logging.info("✅ Settle delay complete.")
# ============================================================
# Media upload — Video (PDS-direct)
# ============================================================
@@ -348,6 +360,7 @@ def upload_video_via_pds(
client: Client,
video_path: str,
alt_text: str = "",
settle_delay_seconds: float = 15.0,
) -> models.AppBskyEmbedVideo.Main | None:
"""
Upload a video as a generic blob directly to the user's PDS.
@@ -355,6 +368,12 @@ def upload_video_via_pds(
Used for self-hosted / federated PDSes (e.g. eurosky.social) that don't
proxy to the centralized video.bsky.app service. The PDS stores the bytes
and returns a blob ref we can embed directly.
A settle delay is applied AFTER the upload returns, because some PDSes
(notably non-Bluesky-operated ones) index the blob asynchronously. If we
publish the post immediately, the AppView can render "video not found"
until indexing catches up — the delay ensures the post references a blob
that's already retrievable.
"""
try:
if not os.path.exists(video_path):
@@ -365,11 +384,16 @@ def upload_video_via_pds(
video_bytes = f.read()
size_mb = len(video_bytes) / (1024 * 1024)
logging.info(f"🎬 [PDS-direct] Uploading video to home PDS: {video_path} ({size_mb:.2f} MB)")
logging.info(
f"🎬 [PDS-direct] Uploading video to home PDS: {video_path} ({size_mb:.2f} MB)"
)
response = client.upload_blob(video_bytes)
blob = response.blob
logging.info(f"✅ [PDS-direct] Video blob uploaded successfully.")
logging.info("✅ [PDS-direct] Video blob uploaded successfully.")
# Give PDS / AppView time to index the blob before we reference it.
wait_with_heartbeat(settle_delay_seconds, label="to index the video blob")
return models.AppBskyEmbedVideo.Main(
video=blob,
@@ -402,12 +426,13 @@ def upload_video_via_bsky_service(
video_bytes = f.read()
size_mb = len(video_bytes) / (1024 * 1024)
logging.info(f"🎬 [video.bsky.app] Uploading via shared service: {video_path} ({size_mb:.2f} MB)")
logging.info(
f"🎬 [video.bsky.app] Uploading via shared service: {video_path} ({size_mb:.2f} MB)"
)
VIDEO_HOST = "https://video.bsky.app"
VIDEO_DID = "did:web:video.bsky.app"
# --- Service auth token (lxm-scoped → 30 min OK) ---
upload_auth = client.com.atproto.server.get_service_auth({
"aud": VIDEO_DID,
"lxm": "app.bsky.video.uploadVideo",
@@ -439,9 +464,8 @@ def upload_video_via_bsky_service(
logging.info(f"⏳ Job {job_id} accepted — polling status...")
# getJobStatus is unauthenticated on video.bsky.app
status_url = f"{VIDEO_HOST}/xrpc/app.bsky.video.getJobStatus"
deadline = time.time() + 300 # 5-minute ceiling
deadline = time.time() + 300
while time.time() < deadline:
status_resp = requests.get(status_url, params={"jobId": job_id}, timeout=30)
@@ -493,14 +517,12 @@ def upload_video_smart(
video_path: str,
service_url: str,
alt_text: str = "",
settle_delay_seconds: float = 15.0,
) -> models.AppBskyEmbedVideo.Main | None:
"""
Smart dispatcher: pick PDS-direct or video.bsky.app based on the PDS.
* Self-hosted / federated PDSes (eurosky.social etc.) → direct PDS upload
(no service auth needed; PDS stores video itself).
* Official Bluesky network (bsky.social, *.bsky.network) → video.bsky.app
with lxm-scoped service auth, falling back to PDS-direct on failure.
Smart dispatcher:
* Self-hosted / federated PDSes → direct PDS upload (with settle delay)
* Official Bluesky network → video.bsky.app, fallback to PDS-direct
"""
if is_official_bsky_pds(service_url):
logging.info(f"🌐 Detected official Bluesky PDS ({service_url}) — using video.bsky.app")
@@ -511,10 +533,16 @@ def upload_video_smart(
logging.warning(
"⚠️ video.bsky.app upload failed; falling back to direct PDS upload."
)
return upload_video_via_pds(client, video_path, alt_text=alt_text)
return upload_video_via_pds(
client, video_path, alt_text=alt_text,
settle_delay_seconds=settle_delay_seconds,
)
logging.info(f"🌍 Detected self-hosted/federated PDS ({service_url}) — using direct upload")
return upload_video_via_pds(client, video_path, alt_text=alt_text)
return upload_video_via_pds(
client, video_path, alt_text=alt_text,
settle_delay_seconds=settle_delay_seconds,
)
# ============================================================
@@ -528,11 +556,11 @@ def post_to_bsky(
video_path: str | None = None,
alt_text: str = "",
service_url: str = "https://bsky.social",
video_settle_delay: float = 15.0,
) -> bool:
rich_text = make_rich(text)
try:
# --- VIDEO POSTING ---
if video_path:
logging.info(f"🎬 Preparing video upload: {video_path}")
@@ -541,6 +569,7 @@ def post_to_bsky(
video_path,
service_url=service_url,
alt_text=alt_text,
settle_delay_seconds=video_settle_delay,
)
if not video_embed:
@@ -554,7 +583,6 @@ def post_to_bsky(
langs=langs,
)
# --- IMAGE POSTING ---
elif image_path:
image = upload_image(client, image_path, alt_text=alt_text)
if not image:
@@ -565,7 +593,6 @@ def post_to_bsky(
logging.info("🚀 Sending image post...")
result = client.send_post(text=rich_text, embed=embed, langs=langs)
# --- TEXT ONLY POSTING ---
else:
logging.info("🚀 Sending text post...")
result = client.send_post(text=rich_text, langs=langs)
@@ -596,6 +623,12 @@ def main():
parser.add_argument("--image", default=None, help="Path to image file to attach")
parser.add_argument("--video", default=None, help="Path to video file to attach")
parser.add_argument("--alt", default="", help="Alt text for media")
parser.add_argument(
"--video-settle-delay",
type=float,
default=15.0,
help="Seconds to wait after PDS-direct video upload before posting (default: 15)",
)
args = parser.parse_args()
@@ -622,6 +655,7 @@ def main():
video_path=args.video,
alt_text=args.alt,
service_url=args.service,
video_settle_delay=args.video_settle_delay,
)
if not post_success:
@@ -629,4 +663,4 @@ def main():
if __name__ == "__main__":
main()
main()