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

fixes

Browse Source
This commit is contained in:
Guillem Hernandez Sola
2026-05-08 14:48:50 +02:00
parent f9aaa8b3ed
commit 022af846a4
1 changed files with 67 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
bsky_post.py
View File

@@ -2,12 +2,10 @@
"""
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 --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 --allow-pds-video-fallback
Examples:
python3 bsky_post.py "DIVENDRES!!!!" --video media/divendres.mp4 --username you --password app-pass --service https://eurosky.social
python3 bsky_post.py "Dijous!!!!" --image media/dijous.jpg --username you --password app-pass --service https://eurosky.social
python3 bsky_post.py "Bon dia!" --username you --password app-pass --service https://eurosky.social
"""
import argparse
@@ -22,7 +20,7 @@ from dataclasses import dataclass
from urllib.parse import urlparse
import requests
from atproto import Client, client_utils, models
from atproto import Client, models
# ============================================================
@@ -47,47 +45,6 @@ def setup_logging() -> None:
)
# ============================================================
# Text builder
# ============================================================
def make_rich(content: str):
text_builder = client_utils.TextBuilder()
content = content.strip()
lines = content.splitlines()
for line_idx, line in enumerate(lines):
if not line.strip():
if line_idx < len(lines) - 1:
text_builder.text("\n")
continue
words = line.split(" ")
for i, word in enumerate(words):
if not word:
if i < len(words) - 1:
text_builder.text(" ")
continue
if word.startswith("http://") or word.startswith("https://"):
text_builder.link(word, word)
elif word.startswith("#") and len(word) > 1:
tag_name = word[1:].rstrip(".,;:!?)'\"")
if tag_name:
text_builder.tag(word, tag_name)
else:
text_builder.text(word)
else:
text_builder.text(word)
if i < len(words) - 1:
text_builder.text(" ")
if line_idx < len(lines) - 1:
text_builder.text("\n")
return text_builder
# ============================================================
# Error helpers
# ============================================================
@@ -147,27 +104,7 @@ def is_transient_error(error_obj) -> bool:
def get_rate_limit_wait_seconds(error_obj, default_delay: float, max_delay: float) -> float:
try:
now_ts = int(time.time())
headers = getattr(error_obj, "headers", None) or {}
retry_after = headers.get("retry-after") or headers.get("Retry-After")
if retry_after:
return min(max(float(retry_after), 1.0), max_delay)
x_after = headers.get("x-ratelimit-after") or headers.get("X-RateLimit-After")
if x_after:
return min(max(float(x_after), 1.0), max_delay)
reset_value = headers.get("ratelimit-reset") or headers.get("RateLimit-Reset")
if reset_value:
wait_seconds = max(float(reset_value) - now_ts + 1.0, default_delay)
return min(wait_seconds, max_delay)
except Exception:
pass
try:
response = getattr(error_obj, "response", None)
headers = getattr(response, "headers", None) or {}
now_ts = int(time.time())
retry_after = headers.get("retry-after") or headers.get("Retry-After")
if retry_after:
@@ -232,36 +169,25 @@ def login_with_backoff(
if is_rate_limited_error(e):
if attempt < max_attempts:
wait = get_rate_limit_wait_seconds(e, default_delay=base_delay, max_delay=max_delay)
wait = wait + random.uniform(0, jitter)
logging.warning(
f"⏳ Rate-limited on login (attempt {attempt}/{max_attempts}). "
f"Retrying in {wait:.1f}s..."
)
wait += random.uniform(0, jitter)
logging.warning(f"⏳ Rate-limited. Retrying in {wait:.1f}s...")
time.sleep(wait)
continue
logging.error("❌ Exhausted login retries due to rate limiting.")
return False
if is_network_error(e) or is_transient_error(e):
if attempt < max_attempts:
wait = min(base_delay * attempt, max_delay) + random.uniform(0, jitter)
logging.warning(
f"⏳ Transient login error (attempt {attempt}/{max_attempts}). "
f"Retrying in {wait:.1f}s..."
)
logging.warning(f"⏳ Transient error. Retrying in {wait:.1f}s...")
time.sleep(wait)
continue
logging.error("❌ Exhausted login retries after transient/network errors.")
return False
if attempt < max_attempts:
wait = min(base_delay * attempt, max_delay) + random.uniform(0, jitter)
logging.warning(
f"⏳ Unknown login error (attempt {attempt}/{max_attempts}). "
f"Retrying in {wait:.1f}s..."
)
logging.warning(f"⏳ Unknown login error. Retrying in {wait:.1f}s...")
time.sleep(wait)
continue
@@ -289,10 +215,9 @@ def detect_mime_type(path: str) -> str:
return fallbacks.get(ext, "application/octet-stream")
def wait_with_heartbeat(total_seconds: float, label: str = "indexing") -> None:
def wait_with_heartbeat(total_seconds: float, label: str = "processing") -> None:
if total_seconds <= 0:
return
logging.info(f"⏳ Waiting {total_seconds:.0f}s for {label}...")
remaining = total_seconds
while remaining > 0:
@@ -311,6 +236,16 @@ def pds_did_from_service_url(service_url: str) -> str:
return f"did:web:{host}"
def model_to_dict(obj):
if obj is None:
return None
if hasattr(obj, "model_dump"):
return obj.model_dump(by_alias=True, exclude_none=True)
if hasattr(obj, "dict"):
return obj.dict(by_alias=True, exclude_none=True)
return obj
# ============================================================
# Media upload — Image
# ============================================================
@@ -320,6 +255,10 @@ def upload_image(
alt_text: str = "",
) -> models.AppBskyEmbedImages.Image | None:
try:
if not os.path.exists(image_path):
logging.error(f"❌ Image file not found: {image_path}")
return None
mime = detect_mime_type(image_path)
with open(image_path, "rb") as f:
data = f.read()
@@ -336,7 +275,7 @@ def upload_image(
# ============================================================
# Media upload — Video (PDS direct fallback only)
# Media upload — Video via PDS direct fallback
# ============================================================
def upload_video_via_pds(
client: Client,
@@ -346,7 +285,7 @@ def upload_video_via_pds(
) -> models.AppBskyEmbedVideo.Main | None:
"""
Direct upload to home PDS via upload_blob.
Kept only as optional fallback; playback can be unreliable in clients.
Fallback only. Can be less reliable for playback in clients.
"""
try:
if not os.path.exists(video_path):
@@ -373,7 +312,7 @@ def upload_video_via_pds(
# ============================================================
# Media upload — Video (video.bsky.app primary)
# Media upload — Video via video.bsky.app (primary)
# ============================================================
def _extract_service_auth_token(upload_auth) -> str | None:
token = getattr(upload_auth, "token", None)
@@ -393,9 +332,9 @@ def upload_video_via_bsky_service(
"""
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.
Critical compatibility fixes:
- aud must be user's PDS DID (e.g. did:web:eurosky.social)
- lxm must be com.atproto.repo.uploadBlob
"""
try:
if not os.path.exists(video_path):
@@ -411,19 +350,19 @@ def upload_video_via_bsky_service(
VIDEO_HOST = "https://video.bsky.app"
pds_did = pds_did_from_service_url(service_url)
# Robust for different atproto versions
# Some atproto versions prefer typed params, others accept dict
try:
params = models.ComAtprotoServerGetServiceAuth.Params(
aud=pds_did, # <-- critical fix
lxm="app.bsky.video.uploadVideo",
aud=pds_did,
lxm="com.atproto.repo.uploadBlob", # <-- critical fix
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": pds_did, # <-- critical fix
"lxm": "app.bsky.video.uploadVideo",
"aud": pds_did,
"lxm": "com.atproto.repo.uploadBlob", # <-- critical fix
"exp": int(time.time()) + 60 * 30,
}
)
@@ -505,10 +444,7 @@ def upload_video_smart(
settle_delay_seconds: float = 30.0,
allow_pds_video_fallback: bool = False,
) -> models.AppBskyEmbedVideo.Main | None:
logging.info(
f"🌍 PDS ({service_url}). Trying video.bsky.app first for playback reliability."
)
logging.info(f"🌍 PDS ({service_url}). Trying video.bsky.app first.")
embed = upload_video_via_bsky_service(
client=client,
video_path=video_path,
@@ -519,10 +455,7 @@ def upload_video_smart(
return embed
if allow_pds_video_fallback:
logging.warning(
"⚠️ video.bsky.app failed; trying direct PDS fallback "
"(may be unplayable in some clients)."
)
logging.warning("⚠️ video.bsky.app failed; trying direct PDS fallback.")
return upload_video_via_pds(
client=client,
video_path=video_path,
@@ -530,15 +463,12 @@ def upload_video_smart(
settle_delay_seconds=settle_delay_seconds,
)
logging.error(
"❌ video.bsky.app failed. Not posting unreliable direct-PDS video. "
"Use --allow-pds-video-fallback to override."
)
logging.error("❌ video.bsky.app failed. Not using direct fallback unless enabled.")
return None
# ============================================================
# Post
# Post creation (explicit record to guarantee text string)
# ============================================================
def post_to_bsky(
client: Client,
@@ -553,7 +483,6 @@ def post_to_bsky(
) -> bool:
post_text = text.strip()
# Allow empty text only if media exists
if not post_text and not image_path and not video_path:
logging.error("❌ Empty post text with no media is not allowed.")
return False
@@ -563,7 +492,7 @@ def post_to_bsky(
if video_path:
logging.info(f"🎬 Preparing video upload: {video_path}")
video_embed = upload_video_smart(
embed_obj = upload_video_smart(
client=client,
video_path=video_path,
service_url=service_url,
@@ -571,10 +500,9 @@ def post_to_bsky(
settle_delay_seconds=video_settle_delay,
allow_pds_video_fallback=allow_pds_video_fallback,
)
if not video_embed:
if not embed_obj:
logging.error("❌ Aborting post: video upload/processing failed.")
return False
embed_obj = video_embed
elif image_path:
image = upload_image(client, image_path, alt_text=alt_text)
@@ -583,10 +511,9 @@ def post_to_bsky(
return False
embed_obj = models.AppBskyEmbedImages.Main(images=[image])
# Build record explicitly (most reliable)
record = {
"$type": "app.bsky.feed.post",
"text": post_text, # <-- guaranteed string
"text": post_text, # guaranteed plain string
"createdAt": time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime()),
}
@@ -594,23 +521,27 @@ def post_to_bsky(
record["langs"] = langs
if embed_obj is not None:
# atproto models -> plain dict
if hasattr(embed_obj, "model_dump"):
record["embed"] = embed_obj.model_dump(by_alias=True, exclude_none=True)
elif hasattr(embed_obj, "dict"):
record["embed"] = embed_obj.dict(by_alias=True, exclude_none=True)
else:
record["embed"] = embed_obj
record["embed"] = model_to_dict(embed_obj)
logging.info(f"🧾 Final record text={record.get('text')!r}, has_embed={'embed' in record}")
resp = client.com.atproto.repo.create_record(
models.ComAtprotoRepoCreateRecord.Data(
repo=client.me.did,
collection="app.bsky.feed.post",
record=record,
# typed first, dict fallback for compatibility
try:
resp = client.com.atproto.repo.create_record(
models.ComAtprotoRepoCreateRecord.Data(
repo=client.me.did,
collection="app.bsky.feed.post",
record=record,
)
)
except Exception:
resp = client.com.atproto.repo.create_record(
{
"repo": client.me.did,
"collection": "app.bsky.feed.post",
"record": record,
}
)
)
uri = getattr(resp, "uri", None) or (resp.get("uri") if isinstance(resp, dict) else None)
logging.info(f"✅ Post published! URI: {uri}")
@@ -620,33 +551,32 @@ def post_to_bsky(
logging.error(f"❌ Failed to send post: {repr(e)}")
return False
# ============================================================
# CLI
# ============================================================
def main():
setup_logging()
parser = argparse.ArgumentParser(
description="Post text + optional image or video to Bluesky/federated PDS."
)
parser = argparse.ArgumentParser(description="Post text + optional image/video to Bluesky/federated PDS.")
parser.add_argument("text", help="Post text content")
parser.add_argument("--username", required=True, help="Bluesky handle or email")
parser.add_argument("--password", required=True, help="Bluesky app password")
parser.add_argument("--service", default="https://bsky.social", help="Bluesky PDS URL")
parser.add_argument("--service", default="https://bsky.social", help="PDS URL")
parser.add_argument("--lang", default="ca", help="Comma-separated language codes (e.g. ca,es)")
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("--image", default=None, help="Path to image file")
parser.add_argument("--video", default=None, help="Path to video file")
parser.add_argument("--alt", default="", help="Alt text for media")
parser.add_argument(
"--video-settle-delay",
type=float,
default=30.0,
help="Seconds to wait after direct-PDS fallback upload before posting (default: 30).",
help="Seconds to wait after direct-PDS fallback upload before posting.",
)
parser.add_argument(
"--allow-pds-video-fallback",
action="store_true",
help="Allow direct-PDS fallback if video.bsky.app fails (less reliable playback).",
help="Allow direct PDS video fallback if video.bsky.app fails.",
)
args = parser.parse_args()
@@ -656,6 +586,7 @@ def main():
sys.exit(1)
client = Client(base_url=args.service)
success = login_with_backoff(
client=client,
username=args.username,