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

retries

Browse Source
This commit is contained in:
Guillem Hernandez Sola
2026-05-07 09:34:16 +02:00
parent 0511f0a087
commit de54458fd7
1 changed files with 63 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
bsky_post.py
View File

@@ -16,11 +16,22 @@ import os
import sys import sys
import time import time
import random import random
import re
from dataclasses import dataclass
from atproto import Client, client_utils, models from atproto import Client, client_utils, models
# ============================================================
# Config
# ============================================================
@dataclass(frozen=True)
class RetryConfig:
login_max_attempts: int = 5
login_base_delay_seconds: float = 2.0
login_max_delay_seconds: float = 600.0
login_jitter_seconds: float = 1.5
# ============================================================ # ============================================================
# Logging # Logging
# ============================================================ # ============================================================
@@ -83,25 +94,31 @@ def is_rate_limited_error(e) -> bool:
def is_auth_error(e) -> bool: def is_auth_error(e) -> bool:
text = repr(e).lower() text = repr(e).lower()
return any(s in text for s in ["401", "403", "invalid identifier or password", return any(s in text for s in [
"authenticationrequired", "invalidtoken"]) "401", "403", "invalid identifier or password",
"authenticationrequired", "invalidtoken",
])
def is_network_error(e) -> bool: def is_network_error(e) -> bool:
text = repr(e) text = repr(e)
return any(s in text for s in ["ConnectError", "RemoteProtocolError", "ReadTimeout", return any(s in text for s in [
"WriteTimeout", "TimeoutException", "503", "502", "504", "ConnectError", "RemoteProtocolError", "ReadTimeout",
"ConnectionResetError"]) "WriteTimeout", "TimeoutException", "503", "502", "504",
"ConnectionResetError",
])
def is_timeout_error(e) -> bool: def is_timeout_error(e) -> bool:
text = repr(e) text = repr(e)
return any(s in text for s in ["InvokeTimeoutError", "ReadTimeout", return any(s in text for s in [
"WriteTimeout", "TimeoutException"]) "InvokeTimeoutError", "ReadTimeout",
"WriteTimeout", "TimeoutException",
])
# ============================================================ # ============================================================
# Login with backoff (same pattern as rss2bsky / twitter bot) # Login with backoff
# ============================================================ # ============================================================
def login_with_backoff( def login_with_backoff(
client: Client, client: Client,
@@ -110,7 +127,7 @@ def login_with_backoff(
service_url: str, service_url: str,
max_attempts: int = 5, max_attempts: int = 5,
base_delay: float = 2.0, base_delay: float = 2.0,
max_delay: float = 120.0, max_delay: float = 600.0,
jitter: float = 1.5, jitter: float = 1.5,
) -> bool: ) -> bool:
for attempt in range(1, max_attempts + 1): for attempt in range(1, max_attempts + 1):
@@ -125,32 +142,25 @@ def login_with_backoff(
except Exception as e: except Exception as e:
logging.exception("❌ Login exception") logging.exception("❌ Login exception")
if is_rate_limited_error(e):
if attempt < max_attempts:
delay = min(base_delay * (2 ** (attempt - 1)), max_delay) + random.uniform(0, jitter)
logging.warning(f"⏳ Rate-limited on login. Retrying in {delay:.1f}s...")
time.sleep(delay)
continue
logging.error("❌ Login rate-limited and retries exhausted.")
return False
if is_auth_error(e): if is_auth_error(e):
logging.error("❌ Bad credentials. Check handle/password.") logging.error("❌ Bad credentials. Check handle/password.")
return False return False
if attempt < max_attempts and (is_network_error(e) or is_timeout_error(e)): if attempt >= max_attempts:
logging.error("❌ Login retries exhausted.")
return False
if is_rate_limited_error(e):
delay = min(base_delay * (2 ** (attempt - 1)), max_delay) + random.uniform(0, jitter)
logging.warning(f"⏳ Rate-limited on login. Retrying in {delay:.1f}s...")
elif is_network_error(e) or is_timeout_error(e):
delay = min(base_delay * attempt, max_delay) + random.uniform(0, jitter) delay = min(base_delay * attempt, max_delay) + random.uniform(0, jitter)
logging.warning(f"⏳ Transient login error. Retrying in {delay:.1f}s...") logging.warning(f"⏳ Transient login error. Retrying in {delay:.1f}s...")
time.sleep(delay) else:
continue
if attempt < max_attempts:
delay = min(base_delay * attempt, max_delay) + random.uniform(0, jitter) delay = min(base_delay * attempt, max_delay) + random.uniform(0, jitter)
logging.warning(f"⏳ Unknown login error. Retrying in {delay:.1f}s...") logging.warning(f"⏳ Unknown login error. Retrying in {delay:.1f}s...")
time.sleep(delay)
continue
return False time.sleep(delay)
return False return False
@@ -173,7 +183,11 @@ def detect_mime_type(path: str) -> str:
return fallbacks.get(ext, "application/octet-stream") return fallbacks.get(ext, "application/octet-stream")
def upload_image(client: Client, image_path: str, alt_text: str = "") -> models.AppBskyEmbedImages.Image | None: def upload_image(
client: Client,
image_path: str,
alt_text: str = "",
) -> models.AppBskyEmbedImages.Image | None:
try: try:
mime = detect_mime_type(image_path) mime = detect_mime_type(image_path)
with open(image_path, "rb") as f: with open(image_path, "rb") as f:
@@ -193,7 +207,11 @@ def upload_image(client: Client, image_path: str, alt_text: str = "") -> models.
return None return None
def upload_video(client: Client, video_path: str, alt_text: str = "") -> models.AppBskyEmbedVideo.Main | None: def upload_video(
client: Client,
video_path: str,
alt_text: str = "",
) -> models.AppBskyEmbedVideo.Main | None:
try: try:
mime = detect_mime_type(video_path) mime = detect_mime_type(video_path)
with open(video_path, "rb") as f: with open(video_path, "rb") as f:
@@ -262,14 +280,14 @@ def main():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="Post text + optional image or video to a Bluesky instance." description="Post text + optional image or video to a Bluesky instance."
) )
parser.add_argument("text", help="Post text content") parser.add_argument("text", help="Post text content")
parser.add_argument("--username", required=True, help="Bluesky handle or email") parser.add_argument("--username", required=True, help="Bluesky handle or email")
parser.add_argument("--password", required=True, help="Bluesky app password") parser.add_argument("--password", required=True, help="Bluesky app password")
parser.add_argument("--service", default="https://eurosky.social", help="Bluesky PDS URL") parser.add_argument("--service", default="https://eurosky.social", help="Bluesky PDS URL")
parser.add_argument("--lang", default="ca", help="Comma-separated language codes (e.g. ca,es)") 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("--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("--video", default=None, help="Path to video file to attach")
parser.add_argument("--alt", default="", help="Alt text for the attached media") parser.add_argument("--alt", default="", help="Alt text for the attached media")
args = parser.parse_args() args = parser.parse_args()
if args.image and args.video: if args.image and args.video:
@@ -284,16 +302,23 @@ def main():
logging.error(f"❌ Video file not found: {args.video}") logging.error(f"❌ Video file not found: {args.video}")
sys.exit(1) sys.exit(1)
langs = [l.strip() for l in args.lang.split(",") if l.strip()] or ["ca"] langs = [lang.strip() for lang in args.lang.split(",") if lang.strip()] or ["ca"]
logging.info(f"🌍 Language(s): {langs}") logging.info(f"🌍 Language(s): {langs}")
cfg = RetryConfig()
client = Client(base_url=args.service) client = Client(base_url=args.service)
logged_in = login_with_backoff( logged_in = login_with_backoff(
client=client, client=client,
username=args.username, username=args.username,
password=args.password, password=args.password,
service_url=args.service, service_url=args.service,
max_attempts=cfg.login_max_attempts,
base_delay=cfg.login_base_delay_seconds,
max_delay=cfg.login_max_delay_seconds,
jitter=cfg.login_jitter_seconds,
) )
if not logged_in: if not logged_in:
logging.error("❌ Could not log in. Exiting.") logging.error("❌ Could not log in. Exiting.")
sys.exit(1) sys.exit(1)