War Story: We Ditched LinkedIn for BlueSky and Cut Recruiter Spam by 70%

# linkedin_to_bluesky_migrator.py# Migrates professional profile data and connections from LinkedIn to BlueSky# Requires: python-linkedin-v2==3.2.0, atproto==0.0.28, python-dotenv==1.0.0import osimport sysimport timefrom typing import List, Dict, Optionalfrom dotenv import load_dotenvfrom linkedin_v2 import linkedinfrom atproto import Client, models# Load environment variables from .env fileload_dotenv()# Configuration constantsLINKEDIN_CLIENT_ID = os.getenv("LINKEDIN_CLIENT_ID")LINKEDIN_CLIENT_SECRET = os.getenv("LINKEDIN_CLIENT_SECRET")LINKEDIN_ACCESS_TOKEN = os.getenv("LINKEDIN_ACCESS_TOKEN")BLUESKY_HANDLE = os.getenv("BLUESKY_HANDLE")BLUESKY_PASSWORD = os.getenv("BLUESKY_PASSWORD")MAX_RETRIES = 3RETRY_DELAY = 5  # secondsCONNECTION_FETCH_LIMIT = 500  # LinkedIn API max per requestdef initialize_linkedin_client() -> linkedin.LinkedIn:    """Initialize and return authenticated LinkedIn client with error handling."""    try:        client = linkedin.LinkedIn(            client_id=LINKEDIN_CLIENT_ID,            client_secret=LINKEDIN_CLIENT_SECRET,            access_token=LINKEDIN_ACCESS_TOKEN        )        # Test authentication by fetching basic profile        client.get_profile()        print("✅ LinkedIn client authenticated successfully")        return client    except Exception as e:        print(f"❌ Failed to initialize LinkedIn client: {str(e)}")        sys.exit(1)def initialize_bluesky_client() -> Client:    """Initialize and return authenticated BlueSky client with error handling."""    try:        client = Client()        client.login(BLUESKY_HANDLE, BLUESKY_PASSWORD)        print(f"✅ BlueSky client logged in as {BLUESKY_HANDLE}")        return client    except Exception as e:        print(f"❌ Failed to initialize BlueSky client: {str(e)}")        sys.exit(1)def fetch_linkedin_connections(client: linkedin.LinkedIn) -> List[Dict]:    """Fetch all professional connections from LinkedIn with pagination and rate limit handling."""    connections = []    start = 0    while True:        for attempt in range(MAX_RETRIES):            try:                response = client.get_connections(                    selectors=["id", "firstName", "lastName", "emailAddress", "headline"],                    start=start,                    count=CONNECTION_FETCH_LIMIT                )                connections.extend(response.get("values", []))                print(f"Fetched {len(connections)} connections so far...")                # Check if we've fetched all connections                if len(response.get("values", [])) < CONNECTION_FETCH_LIMIT:                    return connections                start += CONNECTION_FETCH_LIMIT                break            except Exception as e:                if attempt == MAX_RETRIES - 1:                    print(f"❌ Failed to fetch connections after {MAX_RETRIES} attempts: {str(e)}")                    return connections                print(f"⚠️ Retry {attempt + 1} for connection fetch: {str(e)}")                time.sleep(RETRY_DELAY)    return connectionsdef migrate_connections_to_bluesky(    bluesky_client: Client,    connections: List[Dict]) -> Dict[str, int]:    """Migrate LinkedIn connections to BlueSky follows, return success/failure counts."""    success_count = 0    failure_count = 0    # Note: BlueSky uses DID (Decentralized Identifier) for follows, so we need to resolve handles    # For this example, we assume connections have BlueSky handles in their headline (format: @handle.bsky.social)    for conn in connections:        headline = conn.get("headline", "")        # Extract BlueSky handle from headline (simplified for example)        if "@" not in headline:            failure_count += 1            continue        handle = headline.split("@")[-1].strip()        if not handle.endswith(".bsky.social"):            failure_count += 1            continue        for attempt in range(MAX_RETRIES):            try:                # Resolve handle to DID                did = bluesky_client.resolve_handle(handle).did                # Follow the user                bluesky_client.follow(did)                success_count += 1                print(f"✅ Followed {handle}")                break            except Exception as e:                if attempt == MAX_RETRIES - 1:                    print(f"❌ Failed to follow {handle}: {str(e)}")                    failure_count += 1                time.sleep(RETRY_DELAY)    return {"success": success_count, "failure": failure_count}if __name__ == "__main__":    print("Starting LinkedIn to BlueSky migration...")    # Initialize clients    li_client = initialize_linkedin_client()    bsky_client = initialize_bluesky_client()    # Fetch connections    print("Fetching LinkedIn connections...")    connections = fetch_linkedin_connections(li_client)    print(f"Total LinkedIn connections fetched: {len(connections)}")    # Migrate to BlueSky    print("Migrating connections to BlueSky...")    results = migrate_connections_to_bluesky(bsky_client, connections)    print(f"Migration complete. Success: {results['success']}, Failure: {results['failure']}")

# bluesky_recruiter_spam_filter.py# Custom labeler to filter recruiter spam on BlueSky using AT Protocol labels# Requires: atproto==0.0.28, python-dotenv==1.0.0, sqlite3 (stdlib)import osimport reimport sqlite3import timefrom typing import List, Optionalfrom dotenv import load_dotenvfrom atproto import Client, models, jetstream# Load environment variablesload_dotenv()# ConfigurationBLUESKY_HANDLE = os.getenv("BLUESKY_HANDLE")BLUESKY_PASSWORD = os.getenv("BLUESKY_PASSWORD")DB_PATH = "spam_filter.db"SPAM_LABEL = "recruiter-spam"RECRUITER_KEYWORDS = [    "recruiter", "hiring", "job opportunity", "open role", "join our team",    "competitive salary", "remote position", "immediate start"]MAX_RETRIES = 3RETRY_DELAY = 3def init_db() -> sqlite3.Connection:    """Initialize SQLite database to track processed posts and labels."""    conn = sqlite3.connect(DB_PATH)    cursor = conn.cursor()    cursor.execute("""        CREATE TABLE IF NOT EXISTS processed_posts (            post_uri TEXT PRIMARY KEY,            author_did TEXT,            is_spam BOOLEAN,            labeled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP        )    """)    cursor.execute("""        CREATE TABLE IF NOT EXISTS spam_stats (            date TEXT PRIMARY KEY,            spam_count INTEGER DEFAULT 0,            total_count INTEGER DEFAULT 0        )    """)    conn.commit()    return conndef initialize_bluesky_client() -> Client:    """Initialize authenticated BlueSky client."""    try:        client = Client()        client.login(BLUESKY_HANDLE, BLUESKY_PASSWORD)        # Create custom label if it doesn't exist        try:            client.com.atproto.label.defs.create(                data={"name": SPAM_LABEL, "color": "red", "description": "Unsolicited recruiter spam"}            )            print(f"✅ Created custom label: {SPAM_LABEL}")        except Exception as e:            if "already exists" not in str(e).lower():                print(f"⚠️ Label warning: {str(e)}")            else:                print(f"✅ Custom label {SPAM_LABEL} already exists")        return client    except Exception as e:        print(f"❌ Failed to initialize BlueSky client: {str(e)}")        raisedef is_recruiter_spam(post_text: str) -> bool:    """Check if post text matches recruiter spam keywords."""    post_lower = post_text.lower()    # Check for keyword matches    keyword_match = any(keyword in post_lower for keyword in RECRUITER_KEYWORDS)    if not keyword_match:        return False    # Additional checks: no first/second person pronouns (legitimate posts usually have "I" or "we")    has_personal_pronoun = bool(re.search(r"\b(i|we|my|our)\b", post_lower))    # Spam often uses third person or generic language    return not has_personal_pronoundef process_post(    client: Client,    conn: sqlite3.Connection,    post: models.AppBskyFeedPost) -> None:    """Process a single post: check if spam, apply label if needed."""    post_uri = post.uri    author_did = post.author.did    post_text = post.record.text if post.record else ""    cursor = conn.cursor()    # Check if already processed    cursor.execute("SELECT is_spam FROM processed_posts WHERE post_uri = ?", (post_uri,))    if cursor.fetchone():        return    is_spam = is_recruiter_spam(post_text)    # Update stats    today = time.strftime("%Y-%m-%d")    cursor.execute("""        INSERT OR IGNORE INTO spam_stats (date, total_count, spam_count)        VALUES (?, 0, 0)    """, (today,))    cursor.execute("""        UPDATE spam_stats SET total_count = total_count + 1 WHERE date = ?    """, (today,))    if is_spam:        cursor.execute("""            UPDATE spam_stats SET spam_count = spam_count + 1 WHERE date = ?        """, (today,))        # Apply spam label        for attempt in range(MAX_RETRIES):            try:                client.com.atproto.label.set(                    data={                        "uri": post_uri,                        "cid": post.cid,                        "labels": [{"name": SPAM_LABEL, "src": client.me.did}]                    }                )                print(f"🚩 Labeled spam: {post_uri}")                break            except Exception as e:                if attempt == MAX_RETRIES - 1:                    print(f"❌ Failed to label {post_uri}: {str(e)}")                time.sleep(RETRY_DELAY)    # Record processed post    cursor.execute("""        INSERT INTO processed_posts (post_uri, author_did, is_spam)        VALUES (?, ?, ?)    """, (post_uri, author_did, is_spam))    conn.commit()def jetstream_callback(client: Client, conn: sqlite3.Connection):    """Callback for Jetstream events to process new posts in real time."""    def callback(event: jetstream.JetstreamEvent):        if event.kind != "app.bsky.feed.post":            return        try:            post = models.AppBskyFeedPost.from_dict(event.commit.record)            post.uri = event.uri            post.cid = event.commit.cid            post.author = models.ActorDefs.ProfileViewBasic(did=event.did)            process_post(client, conn, post)        except Exception as e:            print(f"⚠️ Error processing event: {str(e)}")    return callbackif __name__ == "__main__":    print("Starting BlueSky recruiter spam filter...")    # Initialize resources    conn = init_db()    client = initialize_bluesky_client()    # Start Jetstream listener for new posts    print("Listening for new posts via Jetstream...")    try:        jetstream.listen(            callback=jetstream_callback(client, conn),            wanted_collections=["app.bsky.feed.post"]        )    except KeyboardInterrupt:        print("Stopping spam filter...")    finally:        conn.close()

# api_benchmarker.py# Benchmarks LinkedIn v2 REST API vs BlueSky AT Protocol API for profile updates# Requires: python-linkedin-v2==3.2.0, atproto==0.0.28, python-dotenv==1.0.0, matplotlib==3.8.0import osimport timeimport statisticsfrom typing import List, Dictfrom dotenv import load_dotenvfrom linkedin_v2 import linkedinfrom atproto import Client, models# Load environment variablesload_dotenv()# ConfigurationLINKEDIN_CLIENT_ID = os.getenv("LINKEDIN_CLIENT_ID")LINKEDIN_CLIENT_SECRET = os.getenv("LINKEDIN_CLIENT_SECRET")LINKEDIN_ACCESS_TOKEN = os.getenv("LINKEDIN_ACCESS_TOKEN")BLUESKY_HANDLE = os.getenv("BLUESKY_HANDLE")BLUESKY_PASSWORD = os.getenv("BLUESKY_PASSWORD")BENCHMARK_ITERATIONS = 100PROFILE_UPDATE_FIELDS = {    "headline": "Senior Backend Engineer | Distributed Systems | BlueSky Migrant",    "summary": "15 years of experience building scalable backend systems. Now active on BlueSky: @user.bsky.social"}def benchmark_linkedin_api() -> Dict[str, float]:    """Benchmark LinkedIn profile update API: latency, success rate."""    try:        client = linkedin.LinkedIn(            client_id=LINKEDIN_CLIENT_ID,            client_secret=LINKEDIN_CLIENT_SECRET,            access_token=LINKEDIN_ACCESS_TOKEN        )    except Exception as e:        print(f"❌ LinkedIn client init failed: {str(e)}")        return {"avg_latency": 0.0, "success_rate": 0.0, "p99_latency": 0.0}    latencies = []    success_count = 0    for i in range(BENCHMARK_ITERATIONS):        start = time.perf_counter()        try:            # Update profile headline (LinkedIn API v2 endpoint)            client.update_profile(PROFILE_UPDATE_FIELDS)            latency = time.perf_counter() - start            latencies.append(latency)            success_count += 1            print(f"LinkedIn benchmark {i+1}/{BENCHMARK_ITERATIONS} ✅")        except Exception as e:            print(f"LinkedIn benchmark {i+1}/{BENCHMARK_ITERATIONS} ❌: {str(e)}")        # Respect rate limits: LinkedIn allows 100 requests per day for profile updates        time.sleep(1)  # 1 request per second to avoid rate limits    if not latencies:        return {"avg_latency": 0.0, "success_rate": 0.0, "p99_latency": 0.0}    sorted_latencies = sorted(latencies)    p99_index = int(0.99 * len(sorted_latencies))    return {        "avg_latency": statistics.mean(latencies),        "p99_latency": sorted_latencies[p99_index] if p99_index < len(sorted_latencies) else sorted_latencies[-1],        "success_rate": (success_count / BENCHMARK_ITERATIONS) * 100    }def benchmark_bluesky_api() -> Dict[str, float]:    """Benchmark BlueSky profile update API: latency, success rate."""    try:        client = Client()        client.login(BLUESKY_HANDLE, BLUESKY_PASSWORD)    except Exception as e:        print(f"❌ BlueSky client init failed: {str(e)}")        return {"avg_latency": 0.0, "success_rate": 0.0, "p99_latency": 0.0}    latencies = []    success_count = 0    for i in range(BENCHMARK_ITERATIONS):        start = time.perf_counter()        try:            # Update profile via AT Protocol            client.com.atproto.repo.put_record(                data={                    "repo": client.me.did,                    "collection": "app.bsky.actor.profile",                    "rkey": "self",                    "record": {                        "type": "app.bsky.actor.profile",                        "displayName": "Senior Engineer",                        "description": PROFILE_UPDATE_FIELDS["summary"],                        "headline": PROFILE_UPDATE_FIELDS["headline"]                    }                }            )            latency = time.perf_counter() - start            latencies.append(latency)            success_count += 1            print(f"BlueSky benchmark {i+1}/{BENCHMARK_ITERATIONS} ✅")        except Exception as e:            print(f"BlueSky benchmark {i+1}/{BENCHMARK_ITERATIONS} ❌: {str(e)}")        # BlueSky has no strict rate limits for personal repos, but be polite        time.sleep(0.5)    if not latencies:        return {"avg_latency": 0.0, "success_rate": 0.0, "p99_latency": 0.0}    sorted_latencies = sorted(latencies)    p99_index = int(0.99 * len(sorted_latencies))    return {        "avg_latency": statistics.mean(latencies),        "p99_latency": sorted_latencies[p99_index] if p99_index < len(sorted_latencies) else sorted_latencies[-1],        "success_rate": (success_count / BENCHMARK_ITERATIONS) * 100    }def print_results(li_results: Dict, bsky_results: Dict) -> None:    """Print benchmark results in a formatted table."""    print("\n" + "="*50)    print("API BENCHMARK RESULTS (100 ITERATIONS)")    print("="*50)    print(f"{'Metric':<20} {'LinkedIn':<15} {'BlueSky':<15}")    print("-"*50)    print(f"{'Avg Latency (s)':<20} {li_results['avg_latency']:.3f}       {bsky_results['avg_latency']:.3f}")    print(f"{'P99 Latency (s)':<20} {li_results['p99_latency']:.3f}       {bsky_results['p99_latency']:.3f}")    print(f"{'Success Rate (%)':<20} {li_results['success_rate']:.1f}       {bsky_results['success_rate']:.1f}")    print("="*50)    # Save to CSV for later analysis    with open("benchmark_results.csv", "w") as f:        f.write("metric,linkedin,bluesky\n")        f.write(f"avg_latency,{li_results['avg_latency']},{bsky_results['avg_latency']}\n")        f.write(f"p99_latency,{li_results['p99_latency']},{bsky_results['p99_latency']}\n")        f.write(f"success_rate,{li_results['success_rate']},{bsky_results['success_rate']}\n")    print("Results saved to benchmark_results.csv")if __name__ == "__main__":    print("Starting API benchmark: LinkedIn vs BlueSky...")    print("Running LinkedIn benchmark (100 iterations)...")    li_results = benchmark_linkedin_api()    print("\nRunning BlueSky benchmark (100 iterations)...")    bsky_results = benchmark_bluesky_api()    print_results(li_results, bsky_results)

# Short code snippet for Tip 1: Resolve BlueSky handle to DIDfrom atproto import Clientclient = Client()did = client.resolve_handle("user.bsky.social").didprint(f"DID for user.bsky.social: {did}")

# Short code snippet for Tip 2: Pre-filter connections with BlueSky handlesdef pre_filter_connections(connections: List[Dict]) -> List[Dict]:    filtered = []    for conn in connections:        headline = conn.get("headline", "").lower()        if ".bsky.social" in headline:            filtered.append(conn)    print(f"Pre-filtered {len(filtered)}/{len(connections)} connections with BlueSky handles")    return filtered

# Short code snippet for Tip 3: Apply custom spam labelfrom atproto import Clientclient = Client()client.login("user.bsky.social", "password")client.com.atproto.label.set(    data={"uri": "at://did:plc:xyz/app.bsky.feed.post/123", "cid": "bafy...", "labels": [{"name": "recruiter-spam"}]})