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

revert 1f642299e5

Browse Source 
revert Text!
This commit is contained in:
Guillem Hernandez Sola
2026-05-08 19:34:02 +00:00
parent 5d2cec8b31
commit ef46ac5fd6
1 changed files with 88 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

304
twitter2bsky_daemon.py
View File

@@ -2,18 +2,18 @@
""" """
Post text + optional image/video to Bluesky/federated PDS. Post text + optional image/video to Bluesky/federated PDS.
Reliability strategy: Key reliability choices:
- Login via atproto SDK (only SDK usage). - Video uploads go through https://video.bsky.app first (best client playback compatibility).
- ALL media + record operations via raw XRPC (requests) to avoid - getServiceAuth uses:
atproto SDK BlobRef serialization bugs across versions. aud = did:web:<your-pds-host>
- Video uploads go through https://video.bsky.app (official path). lxm = com.atproto.repo.uploadBlob
- getServiceAuth uses aud=did:web:<pds-host>, lxm=com.atproto.repo.uploadBlob. - Handles 409 already_exists from video service by reusing jobId.
- Handles 409 already_exists by reusing jobId. - Uses raw lexicon dict embeds (NO AppBskyEmbedVideo typed model), avoiding BlobRef SDK mismatch.
- ffmpeg compression enabled by default. - Optional direct-PDS fallback for video.
- ffmpeg compression enabled by default (disable with --no-compress-video).
""" """
import argparse import argparse
import json
import logging import logging
import mimetypes import mimetypes
import os import os
@@ -124,99 +124,22 @@ def random_video_name(ext: str = ".mp4") -> str:
return f"{int(time.time())}_{token}{ext}" return f"{int(time.time())}_{token}{ext}"
# --------------------------- def extract_token_from_service_auth(resp_obj) -> str | None:
# Auth/session extraction (raw XRPC) tok = getattr(resp_obj, "token", None)
# --------------------------- if tok:
def get_session_info(client: Client) -> tuple[str, str, str]: return tok
""" if isinstance(resp_obj, dict):
Returns (did, access_jwt, pds_url) from the SDK client's session, return resp_obj.get("token")
avoiding any lazy-loaded properties that may trigger BlobRef bugs. return None
"""
# atproto SDK exposes session via _session or me; pull safely.
did = None
access_jwt = None
# Try common attribute paths across SDK versions
sess = getattr(client, "_session", None) or getattr(client, "session", None)
if sess is not None:
did = getattr(sess, "did", None) or (sess.get("did") if isinstance(sess, dict) else None)
access_jwt = (
getattr(sess, "access_jwt", None)
or getattr(sess, "accessJwt", None)
or (sess.get("accessJwt") if isinstance(sess, dict) else None)
or (sess.get("access_jwt") if isinstance(sess, dict) else None)
)
if not did:
# Last resort: client.me — but wrap to suppress BlobRef issues
try:
me = client.me
did = getattr(me, "did", None)
except Exception as e:
logging.error(f"❌ Could not read client.me: {repr(e)}")
if not access_jwt:
# Some SDK versions: client._session_dispatcher or similar
for attr in ("_access_jwt", "access_jwt"):
v = getattr(client, attr, None)
if v:
access_jwt = v
break
if not did or not access_jwt:
raise RuntimeError("Unable to extract DID/accessJwt from atproto Client session.")
pds_url = getattr(client, "_base_url", None) or getattr(client, "base_url", None) or "https://bsky.social"
return did, access_jwt, pds_url
def xrpc_get(pds_url: str, access_jwt: str, method: str, params: dict, timeout: int = 30) -> dict: def extract_blob_from_upload_blob_result(resp_obj):
url = f"{pds_url.rstrip('/')}/xrpc/{method}" blob = getattr(resp_obj, "blob", None)
headers = {"Authorization": f"Bearer {access_jwt}"} if blob is not None:
r = requests.get(url, headers=headers, params=params, timeout=timeout) return blob
r.raise_for_status() if isinstance(resp_obj, dict):
return r.json() return resp_obj.get("blob")
return None
def xrpc_post_json(pds_url: str, access_jwt: str, method: str, body: dict, timeout: int = 60) -> dict:
url = f"{pds_url.rstrip('/')}/xrpc/{method}"
headers = {
"Authorization": f"Bearer {access_jwt}",
"Content-Type": "application/json",
}
r = requests.post(url, headers=headers, data=json.dumps(body), timeout=timeout)
if r.status_code >= 400:
raise RuntimeError(f"XRPC {method} failed: {r.status_code} {r.text}")
return r.json()
def xrpc_post_bytes(pds_url: str, access_jwt: str, method: str, data: bytes, content_type: str, timeout: int = 240) -> dict:
url = f"{pds_url.rstrip('/')}/xrpc/{method}"
headers = {
"Authorization": f"Bearer {access_jwt}",
"Content-Type": content_type,
}
r = requests.post(url, headers=headers, data=data, timeout=timeout)
if r.status_code >= 400:
raise RuntimeError(f"XRPC {method} failed: {r.status_code} {r.text}")
return r.json()
def get_service_auth_token(pds_url: str, access_jwt: str, aud: str, lxm: str, exp_seconds: int = 1800) -> str:
"""
Raw XRPC call to com.atproto.server.getServiceAuth — avoids SDK typed coercion.
"""
body = xrpc_get(
pds_url=pds_url,
access_jwt=access_jwt,
method="com.atproto.server.getServiceAuth",
params={"aud": aud, "lxm": lxm, "exp": int(time.time()) + exp_seconds},
timeout=30,
)
token = body.get("token")
if not token:
raise RuntimeError(f"getServiceAuth returned no token: {body}")
return token
# --------------------------- # ---------------------------
@@ -317,40 +240,34 @@ def compress_video_ffmpeg(
# --------------------------- # ---------------------------
# Image upload (raw XRPC) # Media upload: image
# --------------------------- # ---------------------------
def upload_image_embed_dict( def upload_image_embed_dict(client: Client, image_path: str, alt_text: str = "") -> dict | None:
pds_url: str,
access_jwt: str,
image_path: str,
alt_text: str = "",
) -> dict | None:
if not os.path.exists(image_path): if not os.path.exists(image_path):
logging.error(f"❌ Image file not found: {image_path}") logging.error(f"❌ Image file not found: {image_path}")
return None return None
mime, _ = mimetypes.guess_type(image_path) mime, _ = mimetypes.guess_type(image_path)
mime = mime or "image/jpeg"
with open(image_path, "rb") as f: with open(image_path, "rb") as f:
data = f.read() data = f.read()
logging.info(f"🖼️ Uploading image: {image_path} ({len(data)/1024:.1f} KB, {mime})") logging.info(f"🖼️ Uploading image: {image_path} ({len(data)/1024:.1f} KB, {mime or 'unknown'})")
try: try:
body = xrpc_post_bytes( up = client.upload_blob(data)
pds_url=pds_url, blob = extract_blob_from_upload_blob_result(up)
access_jwt=access_jwt, if blob is None:
method="com.atproto.repo.uploadBlob", logging.error("❌ uploadBlob returned no blob for image.")
data=data,
content_type=mime,
timeout=180,
)
blob = body.get("blob")
if not blob:
logging.error(f"❌ uploadBlob returned no blob: {body}")
return None return None
# Raw lexicon dict embed (cross-SDK safe)
return { return {
"$type": "app.bsky.embed.images", "$type": "app.bsky.embed.images",
"images": [{"alt": alt_text or "", "image": blob}], "images": [
{
"alt": alt_text or "",
"image": blob,
}
],
} }
except Exception as e: except Exception as e:
logging.error(f"❌ Image upload failed: {repr(e)}") logging.error(f"❌ Image upload failed: {repr(e)}")
@@ -358,12 +275,10 @@ def upload_image_embed_dict(
# --------------------------- # ---------------------------
# Video upload via video.bsky.app # Media upload: video via video.bsky.app (primary)
# --------------------------- # ---------------------------
def upload_video_via_video_service_embed_dict( def upload_video_via_video_service_embed_dict(
did: str, client: Client,
access_jwt: str,
pds_url: str,
video_path: str, video_path: str,
service_url: str, service_url: str,
alt_text: str = "", alt_text: str = "",
@@ -381,30 +296,31 @@ def upload_video_via_video_service_embed_dict(
video_host = "https://video.bsky.app" video_host = "https://video.bsky.app"
pds_did = pds_did_from_service_url(service_url) pds_did = pds_did_from_service_url(service_url)
# Service auth token from user's PDS # getServiceAuth from user's PDS with correct audience + method binding
try: try:
token = get_service_auth_token( auth_resp = client.com.atproto.server.get_service_auth(
pds_url=pds_url, {"aud": pds_did, "lxm": "com.atproto.repo.uploadBlob", "exp": int(time.time()) + 1800}
access_jwt=access_jwt,
aud=pds_did,
lxm="com.atproto.repo.uploadBlob",
exp_seconds=1800,
) )
except Exception as e: except Exception as e:
logging.error(f"❌ getServiceAuth failed: {repr(e)}") logging.error(f"❌ getServiceAuth failed: {repr(e)}")
return None return None
token = extract_token_from_service_auth(auth_resp)
if not token:
logging.error("❌ getServiceAuth returned no token.")
return None
upload_name = random_video_name(".mp4") upload_name = random_video_name(".mp4")
logging.info(f"🎞️ Upload name: {upload_name}") logging.info(f"🎞️ Upload name: {upload_name}")
upload_url = f"{video_host}/xrpc/app.bsky.video.uploadVideo?did={did}&name={upload_name}" upload_url = f"{video_host}/xrpc/app.bsky.video.uploadVideo?did={client.me.did}&name={upload_name}"
headers = { headers = {
"Authorization": f"Bearer {token}", "Authorization": f"Bearer {token}",
"Content-Type": "video/mp4", "Content-Type": "video/mp4",
} }
try: try:
r = requests.post(upload_url, headers=headers, data=video_bytes, timeout=300) r = requests.post(upload_url, headers=headers, data=video_bytes, timeout=240)
except Exception as e: except Exception as e:
logging.error(f"❌ video upload request failed: {repr(e)}") logging.error(f"❌ video upload request failed: {repr(e)}")
return None return None
@@ -413,12 +329,9 @@ def upload_video_via_video_service_embed_dict(
logging.error(f"❌ video.bsky.app upload failed: {r.status_code} - {r.text}") logging.error(f"❌ video.bsky.app upload failed: {r.status_code} - {r.text}")
return None return None
try: payload = r.json()
payload = r.json()
except Exception as e:
logging.error(f"❌ Could not parse upload response JSON: {repr(e)} body={r.text[:500]}")
return None
# Dedupe path: reuse existing job
if r.status_code == 409: if r.status_code == 409:
if payload.get("error") == "already_exists" and payload.get("jobId"): if payload.get("error") == "already_exists" and payload.get("jobId"):
logging.info("ℹ️ Video already processed on video.bsky.app. Reusing existing job.") logging.info("ℹ️ Video already processed on video.bsky.app. Reusing existing job.")
@@ -435,16 +348,10 @@ def upload_video_via_video_service_embed_dict(
status_url = f"{video_host}/xrpc/app.bsky.video.getJobStatus" status_url = f"{video_host}/xrpc/app.bsky.video.getJobStatus"
deadline = time.time() + 600 # 10 min max poll deadline = time.time() + 600 # 10 min max poll
last_state = None
while time.time() < deadline: while time.time() < deadline:
try: try:
s = requests.get( s = requests.get(status_url, params={"jobId": job_id}, timeout=30)
status_url,
params={"jobId": job_id},
headers={"Authorization": f"Bearer {token}"},
timeout=30,
)
except Exception as e: except Exception as e:
logging.warning(f"⚠️ Status poll request failed once: {repr(e)}") logging.warning(f"⚠️ Status poll request failed once: {repr(e)}")
time.sleep(3) time.sleep(3)
@@ -454,19 +361,10 @@ def upload_video_via_video_service_embed_dict(
logging.error(f"❌ Job status failed: {s.status_code} - {s.text}") logging.error(f"❌ Job status failed: {s.status_code} - {s.text}")
return None return None
try: body = s.json()
body = s.json() job_status = body.get("jobStatus", {})
except Exception as e:
logging.error(f"❌ Could not parse status JSON: {repr(e)}")
return None
job_status = body.get("jobStatus", {}) or {}
state = job_status.get("state") state = job_status.get("state")
if state != last_state:
logging.info(f" state → {state}")
last_state = state
if state == "JOB_STATE_COMPLETED": if state == "JOB_STATE_COMPLETED":
blob = job_status.get("blob") blob = job_status.get("blob")
if not blob: if not blob:
@@ -475,7 +373,7 @@ def upload_video_via_video_service_embed_dict(
wait_with_heartbeat(8, "CDN propagation") wait_with_heartbeat(8, "CDN propagation")
# RAW embed dict — no BlobRef typed conversion anywhere. # RAW embed dict; no BlobRef conversion at all.
return { return {
"$type": "app.bsky.embed.video", "$type": "app.bsky.embed.video",
"video": blob, "video": blob,
@@ -486,6 +384,7 @@ def upload_video_via_video_service_embed_dict(
logging.error(f"❌ Video processing failed: {job_status}") logging.error(f"❌ Video processing failed: {job_status}")
return None return None
logging.info(f" ...still processing (state={state})...")
time.sleep(3) time.sleep(3)
logging.error("❌ Video processing timed out.") logging.error("❌ Video processing timed out.")
@@ -493,11 +392,10 @@ def upload_video_via_video_service_embed_dict(
# --------------------------- # ---------------------------
# Direct PDS video fallback (rarely works on third-party PDS) # Media upload: direct PDS fallback (optional)
# --------------------------- # ---------------------------
def upload_video_via_pds_embed_dict( def upload_video_via_pds_embed_dict(
pds_url: str, client: Client,
access_jwt: str,
video_path: str, video_path: str,
alt_text: str = "", alt_text: str = "",
settle_delay_seconds: float = 30.0, settle_delay_seconds: float = 30.0,
@@ -509,20 +407,13 @@ def upload_video_via_pds_embed_dict(
try: try:
with open(video_path, "rb") as f: with open(video_path, "rb") as f:
b = f.read() b = f.read()
size_mb = len(b) / (1024 * 1024) size_mb = len(b) / (1024 * 1024)
logging.warning(f"🎬 [PDS-direct fallback] Uploading: {video_path} ({size_mb:.2f} MB)") logging.warning(f"🎬 [PDS-direct fallback] Uploading: {video_path} ({size_mb:.2f} MB)")
up = client.upload_blob(b)
body = xrpc_post_bytes( blob = extract_blob_from_upload_blob_result(up)
pds_url=pds_url, if blob is None:
access_jwt=access_jwt, logging.error("❌ PDS uploadBlob returned no blob.")
method="com.atproto.repo.uploadBlob",
data=b,
content_type="video/mp4",
timeout=600, # large timeout for direct PDS upload
)
blob = body.get("blob")
if not blob:
logging.error(f"❌ PDS uploadBlob returned no blob: {body}")
return None return None
wait_with_heartbeat(settle_delay_seconds, "PDS/AppView indexing") wait_with_heartbeat(settle_delay_seconds, "PDS/AppView indexing")
@@ -538,9 +429,7 @@ def upload_video_via_pds_embed_dict(
def upload_video_smart_embed_dict( def upload_video_smart_embed_dict(
did: str, client: Client,
access_jwt: str,
pds_url: str,
video_path: str, video_path: str,
service_url: str, service_url: str,
alt_text: str = "", alt_text: str = "",
@@ -549,9 +438,7 @@ def upload_video_smart_embed_dict(
) -> dict | None: ) -> dict | None:
logging.info(f"🌍 PDS ({service_url}). Trying video.bsky.app first.") logging.info(f"🌍 PDS ({service_url}). Trying video.bsky.app first.")
embed = upload_video_via_video_service_embed_dict( embed = upload_video_via_video_service_embed_dict(
did=did, client=client,
access_jwt=access_jwt,
pds_url=pds_url,
video_path=video_path, video_path=video_path,
service_url=service_url, service_url=service_url,
alt_text=alt_text, alt_text=alt_text,
@@ -562,8 +449,7 @@ def upload_video_smart_embed_dict(
if allow_pds_video_fallback: if allow_pds_video_fallback:
logging.warning("⚠️ video.bsky.app failed; trying direct PDS fallback.") logging.warning("⚠️ video.bsky.app failed; trying direct PDS fallback.")
return upload_video_via_pds_embed_dict( return upload_video_via_pds_embed_dict(
pds_url=pds_url, client=client,
access_jwt=access_jwt,
video_path=video_path, video_path=video_path,
alt_text=alt_text, alt_text=alt_text,
settle_delay_seconds=settle_delay_seconds, settle_delay_seconds=settle_delay_seconds,
@@ -574,12 +460,16 @@ def upload_video_smart_embed_dict(
# --------------------------- # ---------------------------
# Create post (raw XRPC) # Create post
# --------------------------- # ---------------------------
def create_post_record(text: str, langs: list[str], embed_dict: dict | None = None) -> dict: def create_post_record(
text: str,
langs: list[str],
embed_dict: dict | None = None,
) -> dict:
record = { record = {
"$type": "app.bsky.feed.post", "$type": "app.bsky.feed.post",
"text": text.strip(), "text": text.strip(), # must be plain string
"createdAt": time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime()), "createdAt": time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime()),
} }
if langs: if langs:
@@ -589,20 +479,17 @@ def create_post_record(text: str, langs: list[str], embed_dict: dict | None = No
return record return record
def publish_post(pds_url: str, access_jwt: str, did: str, record: dict) -> bool: def publish_post(client: Client, record: dict) -> bool:
try: try:
body = xrpc_post_json( # Use dict payload directly for max cross-version compatibility.
pds_url=pds_url, resp = client.com.atproto.repo.create_record(
access_jwt=access_jwt, {
method="com.atproto.repo.createRecord", "repo": client.me.did,
body={
"repo": did,
"collection": "app.bsky.feed.post", "collection": "app.bsky.feed.post",
"record": record, "record": record,
}, }
timeout=60,
) )
uri = body.get("uri") uri = getattr(resp, "uri", None) or (resp.get("uri") if isinstance(resp, dict) else None)
logging.info(f"✅ Post published! URI: {uri}") logging.info(f"✅ Post published! URI: {uri}")
return True return True
except Exception as e: except Exception as e:
@@ -625,9 +512,10 @@ def main():
parser.add_argument("--image", default=None, help="Image path") parser.add_argument("--image", default=None, help="Image path")
parser.add_argument("--video", default=None, help="Video path") parser.add_argument("--video", default=None, help="Video path")
parser.add_argument("--alt", default="", help="Alt text") parser.add_argument("--alt", default="", help="Alt text")
parser.add_argument("--video-settle-delay", type=float, default=30.0) parser.add_argument("--video-settle-delay", type=float, default=30.0, help="Fallback indexing wait")
parser.add_argument("--allow-pds-video-fallback", action="store_true") parser.add_argument("--allow-pds-video-fallback", action="store_true")
# Compression ON by default
parser.add_argument("--compress-video", dest="compress_video", action="store_true", default=True) parser.add_argument("--compress-video", dest="compress_video", action="store_true", default=True)
parser.add_argument("--no-compress-video", dest="compress_video", action="store_false") parser.add_argument("--no-compress-video", dest="compress_video", action="store_false")
parser.add_argument("--max-video-mb", type=float, default=45.0) parser.add_argument("--max-video-mb", type=float, default=45.0)
@@ -650,17 +538,6 @@ def main():
): ):
sys.exit(1) sys.exit(1)
# Capture session info ONCE — avoid lazy SDK property access later.
try:
did, access_jwt, pds_url = get_session_info(client)
except Exception as e:
logging.error(f"❌ Could not extract session: {repr(e)}")
sys.exit(1)
# Override pds_url with the user-specified service (in case SDK normalized it)
pds_url = args.service.rstrip("/")
logging.info(f"🆔 DID: {did} | PDS: {pds_url}")
langs = [x.strip() for x in args.lang.split(",") if x.strip()] langs = [x.strip() for x in args.lang.split(",") if x.strip()]
video_path_for_upload = args.video video_path_for_upload = args.video
@@ -686,9 +563,7 @@ def main():
if video_path_for_upload: if video_path_for_upload:
logging.info(f"🎬 Preparing video upload: {video_path_for_upload}") logging.info(f"🎬 Preparing video upload: {video_path_for_upload}")
embed_dict = upload_video_smart_embed_dict( embed_dict = upload_video_smart_embed_dict(
did=did, client=client,
access_jwt=access_jwt,
pds_url=pds_url,
video_path=video_path_for_upload, video_path=video_path_for_upload,
service_url=args.service, service_url=args.service,
alt_text=args.alt, alt_text=args.alt,
@@ -702,20 +577,17 @@ def main():
sys.exit(1) sys.exit(1)
elif args.image: elif args.image:
embed_dict = upload_image_embed_dict( embed_dict = upload_image_embed_dict(client=client, image_path=args.image, alt_text=args.alt)
pds_url=pds_url,
access_jwt=access_jwt,
image_path=args.image,
alt_text=args.alt,
)
if embed_dict is None: if embed_dict is None:
logging.error("❌ Aborting post: image upload failed.") logging.error("❌ Aborting post: image upload failed.")
if temp_compressed_path and os.path.exists(temp_compressed_path):
os.remove(temp_compressed_path)
sys.exit(1) sys.exit(1)
record = create_post_record(text=args.text, langs=langs, embed_dict=embed_dict) record = create_post_record(text=args.text, langs=langs, embed_dict=embed_dict)
logging.info(f"🧾 Final record text={record.get('text')!r}, has_embed={'embed' in record}") logging.info(f"🧾 Final record text={record.get('text')!r}, has_embed={'embed' in record}")
ok = publish_post(pds_url=pds_url, access_jwt=access_jwt, did=did, record=record) ok = publish_post(client=client, record=record)
if temp_compressed_path and os.path.exists(temp_compressed_path): if temp_compressed_path and os.path.exists(temp_compressed_path):
try: try: