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

ff

Browse Source
This commit is contained in:
Guillem Hernandez Sola
2026-05-08 10:55:36 +02:00
parent d7f3e2792f
commit e6377757d3
1 changed files with 35 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
bsky_post.py
View File

@@ -322,68 +322,68 @@ def upload_image(
return None return None
def upload_video_and_wait( def upload_video_and_wait(
client: Client, client: Client,
video_data: bytes, video_data: bytes,
alt_text: str = "" alt_text: str = ""
) -> models.AppBskyEmbedVideo.Main | None: ) -> models.AppBskyEmbedVideo.Main | None:
try: try:
logging.info("🎬 Requesting Service Auth for Video Upload...")
# 1. Get a Service Auth token from your home PDS
VIDEO_SERVICE_DID = "did:web:video.bsky.app" VIDEO_SERVICE_DID = "did:web:video.bsky.app"
auth_response = client.com.atproto.server.get_service_auth({
# --- Token #1: bound to uploadVideo ---
logging.info("🎬 Requesting Service Auth for Video Upload...")
upload_auth = client.com.atproto.server.get_service_auth({
'aud': VIDEO_SERVICE_DID, 'aud': VIDEO_SERVICE_DID,
'exp': int(time.time()) + 60 * 30, # Token valid for 30 mins 'lxm': 'app.bsky.video.uploadVideo',
'exp': int(time.time()) + 60 * 30,
}) })
service_auth_token = auth_response.token upload_headers = {
"Authorization": f"Bearer {upload_auth.token}",
"Content-Type": "video/mp4",
}
logging.info("🎬 Uploading video to Bluesky Video Service...") logging.info("🎬 Uploading video to Bluesky Video Service...")
# 2. Upload the video using standard HTTP requests
upload_url = "https://video.bsky.app/xrpc/app.bsky.video.uploadVideo" upload_url = "https://video.bsky.app/xrpc/app.bsky.video.uploadVideo"
headers = { upload_resp = requests.post(upload_url, headers=upload_headers, data=video_data)
"Authorization": f"Bearer {service_auth_token}",
"Content-Type": "video/mp4" # Assuming mp4, adjust if needed
}
# The endpoint expects the raw bytes in the body
upload_resp = requests.post(upload_url, headers=headers, data=video_data)
if upload_resp.status_code != 200: if upload_resp.status_code != 200:
logging.error(f"❌ Video upload failed: {upload_resp.status_code} - {upload_resp.text}") logging.error(f"❌ Video upload failed: {upload_resp.status_code} - {upload_resp.text}")
return None return None
job_id = upload_resp.json().get("jobId") job_id = upload_resp.json().get("jobId")
if not job_id: if not job_id:
logging.error("❌ No Job ID returned from video service.") logging.error("❌ No Job ID returned from video service.")
return None return None
logging.info(f"⏳ Video uploaded! Job ID: {job_id}. Waiting for processing...") logging.info(f"⏳ Video uploaded! Job ID: {job_id}. Waiting for processing...")
# 3. Poll the job status # --- Token #2: bound to getJobStatus ---
status_auth = client.com.atproto.server.get_service_auth({
'aud': VIDEO_SERVICE_DID,
'lxm': 'app.bsky.video.getJobStatus',
'exp': int(time.time()) + 60 * 30,
})
status_headers = {"Authorization": f"Bearer {status_auth.token}"}
status_url = "https://video.bsky.app/xrpc/app.bsky.video.getJobStatus" status_url = "https://video.bsky.app/xrpc/app.bsky.video.getJobStatus"
params = {"jobId": job_id} params = {"jobId": job_id}
while True: while True:
status_resp = requests.get(status_url, headers=headers, params=params) status_resp = requests.get(status_url, headers=status_headers, params=params)
if status_resp.status_code != 200: if status_resp.status_code != 200:
logging.error(f"❌ Failed to get job status: {status_resp.status_code} - {status_resp.text}") logging.error(f"❌ Failed to get job status: {status_resp.status_code} - {status_resp.text}")
return None return None
job_status = status_resp.json().get("jobStatus", {}) job_status = status_resp.json().get("jobStatus", {})
state = job_status.get("state") state = job_status.get("state")
if state == 'JOB_STATE_COMPLETED': if state == 'JOB_STATE_COMPLETED':
logging.info("✅ Processing complete! Waiting 10s for CDN propagation...") logging.info("✅ Processing complete! Waiting 10s for CDN propagation...")
time.sleep(10) time.sleep(10)
# Construct the blob object exactly as the SDK expects it
blob_dict = job_status.get("blob") blob_dict = job_status.get("blob")
# Convert the raw dictionary into the SDK's BlobRef model
blob_ref = models.BlobRef.from_dict(blob_dict) blob_ref = models.BlobRef.from_dict(blob_dict)
return models.AppBskyEmbedVideo.Main( return models.AppBskyEmbedVideo.Main(
video=blob_ref, video=blob_ref,
alt=alt_text alt=alt_text
@@ -391,9 +391,9 @@ def upload_video_and_wait(
elif state == 'JOB_STATE_FAILED': elif state == 'JOB_STATE_FAILED':
logging.error("❌ Video processing failed on Bluesky's servers.") logging.error("❌ Video processing failed on Bluesky's servers.")
return None return None
logging.info(" ...still processing...") logging.info(" ...still processing...")
time.sleep(3) time.sleep(3)
except Exception as e: except Exception as e:
logging.error(f"❌ Failed to upload/process video: {repr(e)}") logging.error(f"❌ Failed to upload/process video: {repr(e)}")