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

Added

Browse Source
This commit is contained in:
Guillem Hernandez Sola
2026-05-19 09:15:59 +02:00
parent c7a623c386
commit a1f9a0dcb9
5 changed files with 9 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
.gitignore vendored
View File

@@ -9,6 +9,15 @@ config.json
*.png
*.tw_session
*.tw_session-journal
.vscode/
.venv/
.env/
*.env
*.env.*
*.pyc
__pycache__/
*.pyo
*.pyd
# =========================
# Created by https://www.toptal.com/developers/gitignore/api/python,visualstudiocode,macos,windows

5
.vscode/extensions.json vendored
View File

@@ -1,5 +0,0 @@
{
"recommendations": [
"monicaim.monica-code"
]
}

12
cookie_login.py
View File

@@ -1,12 +0,0 @@
from tweety import Twitter
# This creates a local session file so you only log in once
app = Twitter("my_account_session")
# Log in with your credentials
app.sign_in("grijanderm59258", "Bo3usi!!")
# Fetch the tweets
tweets = app.get_tweets("meteocat", pages=1)
for tweet in tweets:
print(tweet.text)

82
sync_runner.sh
View File

@@ -1,82 +0,0 @@
#!/bin/bash
# Exit immediately if a command fails (except when we explicitly allow it later)
set -e
echo "🚀 Starting Twitter to Bluesky Sync Job..."
# 1. Navigate to the directory containing the script
# (Uncomment and change the path below if running via cron so it knows where to look)
# cd /path/to/your/meteocatrss2bsky || exit
# 2. Load environment variables (Mimicking GitHub Secrets)
if [ -f ".env" ]; then
echo "📂 Loading secrets from .env file..."
export $(grep -v '^#' .env | xargs)
else
echo "⚠️ No .env file found. Make sure environment variables are exported manually!"
fi
# 3. Set up Python Virtual Environment (Mimicking actions/setup-python)
if [ ! -d "venv" ]; then
echo "🐍 Creating Python virtual environment..."
python3 -m venv venv
fi
echo "🔄 Activating virtual environment..."
source venv/bin/activate
# 4. Install dependencies
echo "📦 Installing dependencies..."
pip install --upgrade pip -q
pip install atproto tweety-ns playwright httpx arrow python-dotenv -q
# Install the browser
playwright install chromium
# Install the required Linux system libraries
playwright install-deps chromium
# 7. Clean up (Optional)
# Deactivate the virtual environment
deactivate
# Remove the virtual environment (if you want to start fresh next time)
# rm -rf venv
# 5. Run the script
echo "▶️ Running the sync script..."
# Temporarily disable "exit on error" so we can catch a script crash and save the screenshot
set +e
python3 twitter2bsky_daemon.py \
--twitter-username "$TWITTER_USERNAME" \
--twitter-password "$TWITTER_PASSWORD" \
--twitter-email "$TWITTER_EMAIL" \
--twitter-handle "$TWITTER_HANDLE" \
--bsky-handle "$BSKY_HANDLE" \
--bsky-password "$BSKY_APP_PASSWORD"
EXIT_CODE=$?
# Re-enable "exit on error"
set -e
# 6. Handle Errors and Screenshots (Mimicking actions/upload-artifact)
if [ $EXIT_CODE -ne 0 ]; then
echo "❌ Script failed with exit code $EXIT_CODE."
# Check if any screenshots were generated by Playwright
if ls screenshot_*.png 1> /dev/null 2>&1; then
echo "📸 Error screenshots found. Moving them to 'error_artifacts' folder..."
mkdir -p error_artifacts
mv screenshot_*.png error_artifacts/
echo "✅ Screenshots saved in ./error_artifacts/"
else
echo "⚠️ No screenshots found."
fi
exit $EXIT_CODE
fi
echo "✅ Sync job completed successfully!"

200
testlogin.py
View File

@@ -1,200 +0,0 @@
import argparse
import logging
import os
import random
import sys
import time
import httpx
from atproto import Client
# --- Logging ---
LOG_PATH = "bsky_login_test.log"
logging.basicConfig(
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler(LOG_PATH, encoding="utf-8"),
logging.StreamHandler(),
],
level=logging.INFO,
)
EXIT_OK = 0
EXIT_BAD_CREDS = 2
EXIT_RATE_LIMIT = 3
EXIT_NETWORK = 4
EXIT_OTHER = 5
def parse_wait_seconds_from_exception(exc, default_delay=15, max_delay=900):
"""
Parse common rate-limit headers from atproto exceptions:
- retry-after (seconds)
- x-ratelimit-after (seconds)
- ratelimit-reset (unix timestamp)
"""
try:
headers = getattr(exc, "headers", None) or {}
retry_after = headers.get("retry-after") or headers.get("Retry-After")
if retry_after:
return min(max(int(retry_after), 1), max_delay)
x_after = headers.get("x-ratelimit-after") or headers.get("X-RateLimit-After")
if x_after:
return min(max(int(x_after), 1), max_delay)
reset = headers.get("ratelimit-reset") or headers.get("RateLimit-Reset")
if reset:
wait_s = max(int(reset) - int(time.time()) + 1, 1)
return min(wait_s, max_delay)
except Exception:
pass
return default_delay
def classify_error(exc):
"""
Classify exception into:
- rate_limit
- bad_creds
- network
- other
"""
text = repr(exc).lower()
status_code = getattr(exc, "status_code", None)
if status_code == 429 or "429" in text or "too many requests" in text or "ratelimit" in text:
return "rate_limit"
if status_code in (401, 403) or "invalid identifier or password" in text or "authentication" in text:
return "bad_creds"
transient_signals = [
"timeout",
"connecterror",
"remoteprotocolerror",
"readtimeout",
"writetimeout",
"503",
"502",
"504",
"connection",
]
if any(sig in text for sig in transient_signals):
return "network"
return "other"
def preflight_health(service_url, timeout=8):
url = f"{service_url.rstrip('/')}/xrpc/_health"
try:
r = httpx.get(url, timeout=timeout)
logging.info(f"🩺 Health check {url} -> HTTP {r.status_code}")
return True
except Exception as e:
logging.warning(f"🩺 Health check failed: {e}")
return False
def build_client(service_url):
normalized = service_url.strip().rstrip("/")
try:
return Client(base_url=normalized)
except TypeError:
logging.warning("⚠️ Client(base_url=...) unsupported in this atproto version. Falling back.")
c = Client()
try:
if hasattr(c, "base_url"):
c.base_url = normalized
elif hasattr(c, "_base_url"):
c._base_url = normalized
except Exception as e:
logging.warning(f"⚠️ Could not apply custom base URL: {e}")
return c
def main():
parser = argparse.ArgumentParser(description="Bluesky login test only.")
parser.add_argument("--bsky-handle", required=True, help="Bluesky handle (e.g. user.example.social)")
parser.add_argument(
"--bsky-app-password",
default=None,
help="Bluesky app password (prefer env BSKY_APP_PASSWORD)",
)
parser.add_argument(
"--service",
default="https://bsky.social",
help="PDS base URL (default: https://bsky.social)",
)
parser.add_argument("--max-attempts", type=int, default=3, help="Retry attempts (default: 3)")
parser.add_argument("--base-delay", type=int, default=10, help="Base retry delay in seconds (default: 10)")
parser.add_argument("--jitter-max", type=float, default=2.0, help="Random jitter max seconds (default: 2.0)")
args = parser.parse_args()
handle = args.bsky_handle.strip()
service_url = args.service.strip().rstrip("/")
app_password = (args.bsky_app_password or os.getenv("BSKY_APP_PASSWORD", "")).strip()
if not app_password:
logging.error("❌ Missing app password. Use --bsky-app-password or env BSKY_APP_PASSWORD.")
print("LOGIN_FAILED_BAD_CREDS")
sys.exit(EXIT_BAD_CREDS)
logging.info(f"🔐 Testing login against: {service_url}")
logging.info(f"👤 Handle: {handle}")
# Optional but useful diagnostics
preflight_health(service_url)
client = build_client(service_url)
last_kind = "other"
for attempt in range(1, args.max_attempts + 1):
try:
logging.info(f"➡️ Login attempt {attempt}/{args.max_attempts}")
client.login(handle, app_password)
logging.info("✅ Login successful.")
print("LOGIN_OK")
sys.exit(EXIT_OK)
except Exception as e:
last_kind = classify_error(e)
logging.exception(f"❌ Login failed [{last_kind}]")
if last_kind == "bad_creds":
print("LOGIN_FAILED_BAD_CREDS")
sys.exit(EXIT_BAD_CREDS)
if attempt >= args.max_attempts:
break
if last_kind == "rate_limit":
wait_s = parse_wait_seconds_from_exception(e, default_delay=args.base_delay)
elif last_kind == "network":
wait_s = min(args.base_delay * attempt, 60)
else:
wait_s = min(args.base_delay * attempt, 45)
wait_s = wait_s + random.uniform(0, max(args.jitter_max, 0.0))
logging.warning(f"⏳ Waiting {wait_s:.1f}s before retry...")
time.sleep(wait_s)
if last_kind == "rate_limit":
print("LOGIN_FAILED_RATE_LIMIT")
sys.exit(EXIT_RATE_LIMIT)
if last_kind == "network":
print("LOGIN_FAILED_NETWORK")
sys.exit(EXIT_NETWORK)
print("LOGIN_FAILED")
sys.exit(EXIT_OTHER)
if __name__ == "__main__":
main()