Hot Take: Stop Using LinkedIn – Switch to Blind and Twitter for Tech Jobs

import tweepy
import os
import json
from datetime import datetime, timedelta
import time

class TwitterJobScraper:
    """Scrapes X (Twitter) for tech job postings using Tweepy and X API v2."""

    def __init__(self, bearer_token: str):
        # Initialize Tweepy client with bearer token for API v2
        try:
            self.client = tweepy.Client(bearer_token=bearer_token)
            # Test authentication to catch invalid tokens early
            self.client.get_me()
            print("✅ Twitter API authentication successful")
        except tweepy.TweepyException as e:
            raise ValueError(f"Failed to authenticate with Twitter API: {str(e)}")

        # Define search query targeting tech job posts
        self.search_query = (
            "hiring software engineer OR hiring senior engineer OR tech job opening "
            "OR backend engineer role OR frontend engineer position "
            "-filter:retweets -filter:links lang:en"
        )
        self.results = []

    def fetch_recent_tweets(self, max_results: int = 100) -> list:
        """Fetch recent tweets matching the search query, handling rate limits."""
        try:
            # Search recent tweets with max results per request (max 100 for free tier)
            response = self.client.search_recent_tweets(
                query=self.search_query,
                max_results=max_results,
                tweet_fields=["created_at", "author_id", "public_metrics"],
                expansions=["author_id"],
                user_fields=["username", "verified"]
            )

            # Handle rate limiting: wait 15 minutes if rate limit exceeded
            if response.errors and any("rate limit" in str(err).lower() for err in response.errors):
                print("⚠️ Rate limit exceeded. Waiting 15 minutes...")
                time.sleep(15 * 60)
                return self.fetch_recent_tweets(max_results)

            # Process response data
            if not response.data:
                print("ℹ️ No matching tweets found in this batch")
                return []

            # Map user IDs to usernames for enrichment
            users = {u.id: u for u in response.includes.get("users", [])}

            for tweet in response.data:
                author = users.get(tweet.author_id, None)
                tweet_data = {
                    "tweet_id": tweet.id,
                    "text": tweet.text,
                    "created_at": tweet.created_at.isoformat(),
                    "author_username": author.username if author else "unknown",
                    "author_verified": author.verified if author else False,
                    "retweet_count": tweet.public_metrics["retweet_count"],
                    "like_count": tweet.public_metrics["like_count"],
                    "scraped_at": datetime.utcnow().isoformat()
                }
                self.results.append(tweet_data)

            print(f"✅ Fetched {len(response.data)} tweets in this batch")
            return response.data

        except tweepy.TweepyException as e:
            print(f"❌ Tweepy error fetching tweets: {str(e)}")
            return []
        except Exception as e:
            print(f"❌ Unexpected error fetching tweets: {str(e)}")
            return []

    def save_results(self, output_path: str = "twitter_job_posts.json") -> None:
        """Save scraped results to a JSON file."""
        try:
            with open(output_path, "w") as f:
                json.dump(self.results, f, indent=2)
            print(f"✅ Saved {len(self.results)} results to {output_path}")
        except IOError as e:
            print(f"❌ Failed to save results: {str(e)}")

if __name__ == "__main__":
    # Load bearer token from environment variable (never hardcode!)
    bearer_token = os.getenv("TWITTER_BEARER_TOKEN")
    if not bearer_token:
        raise ValueError("TWITTER_BEARER_TOKEN environment variable not set")

    scraper = TwitterJobScraper(bearer_token)

    # Fetch 3 batches of 100 tweets each (300 total, respecting rate limits)
    for batch in range(3):
        print(f"🔍 Fetching batch {batch + 1}/3...")
        scraper.fetch_recent_tweets(max_results=100)
        # Wait 2 seconds between batches to avoid rate limits
        time.sleep(2)

    scraper.save_results()
    print(f"🎉 Total scraped job posts: {len(scraper.results)}")

import requests
import json
from typing import List, Dict, Optional
import statistics
from datetime import datetime

class BlindSalaryBenchmarker:
    """Fetches and analyzes salary data from Blind's public API endpoints."""

    def __init__(self):
        self.base_url = "https://blindapi.herokuapp.com"  # Unofficial Blind API endpoint
        self.salary_data = []
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        }

    def fetch_salary_data(self, company: str, role: str, years_experience: int = 0) -> List[Dict]:
        """Fetch salary entries for a specific company, role, and experience level."""
        endpoint = f"{self.base_url}/salary"
        params = {
            "company": company,
            "title": role,
            "min_years": years_experience
        }

        try:
            response = requests.get(endpoint, params=params, headers=self.headers, timeout=10)
            response.raise_for_status()  # Raise HTTPError for bad responses (4xx, 5xx)

            data = response.json()
            if not isinstance(data, list):
                print(f"⚠️ Unexpected response format for {company}/{role}: {data}")
                return []

            # Enrich data with timestamp
            for entry in data:
                entry["fetched_at"] = datetime.utcnow().isoformat()
                entry["company"] = company
                entry["role"] = role

            self.salary_data.extend(data)
            print(f"✅ Fetched {len(data)} salary entries for {company} {role}")
            return data

        except requests.exceptions.HTTPError as e:
            print(f"❌ HTTP error fetching salary data: {str(e)}")
            return []
        except requests.exceptions.Timeout as e:
            print(f"❌ Request timed out: {str(e)}")
            return []
        except json.JSONDecodeError as e:
            print(f"❌ Failed to decode JSON response: {str(e)}")
            return []
        except Exception as e:
            print(f"❌ Unexpected error fetching salary data: {str(e)}")
            return []

    def calculate_benchmarks(self) -> Optional[Dict]:
        """Calculate median, p90, p10 for collected salary data."""
        if not self.salary_data:
            print("⚠️ No salary data available to calculate benchmarks")
            return None

        try:
            # Extract base salaries, filter invalid entries
            base_salaries = [
                entry["base_salary"] for entry in self.salary_data
                if isinstance(entry.get("base_salary"), (int, float)) and entry["base_salary"] > 0
            ]

            if not base_salaries:
                print("⚠️ No valid base salary entries found")
                return None

            benchmarks = {
                "total_entries": len(self.salary_data),
                "valid_salary_entries": len(base_salaries),
                "median_salary": statistics.median(base_salaries),
                "p10_salary": statistics.quantiles(base_salaries, n=10)[0] if len(base_salaries) >= 10 else None,
                "p90_salary": statistics.quantiles(base_salaries, n=10)[8] if len(base_salaries) >= 10 else None,
                "average_salary": statistics.mean(base_salaries),
                "calculated_at": datetime.utcnow().isoformat()
            }

            print(f"✅ Calculated benchmarks for {len(base_salaries)} entries")
            return benchmarks

        except statistics.StatisticsError as e:
            print(f"❌ Statistics error calculating benchmarks: {str(e)}")
            return None
        except Exception as e:
            print(f"❌ Unexpected error calculating benchmarks: {str(e)}")
            return None

    def save_data(self, output_path: str = "blind_salary_data.json") -> None:
        """Save collected salary data and benchmarks to JSON."""
        try:
            benchmarks = self.calculate_benchmarks()
            output = {
                "salary_entries": self.salary_data,
                "benchmarks": benchmarks,
                "generated_at": datetime.utcnow().isoformat()
            }
            with open(output_path, "w") as f:
                json.dump(output, f, indent=2)
            print(f"✅ Saved salary data and benchmarks to {output_path}")
        except IOError as e:
            print(f"❌ Failed to save data: {str(e)}")

if __name__ == "__main__":
    benchmarker = BlindSalaryBenchmarker()

    # Fetch salary data for common tech roles at top companies
    target_companies = ["Google", "Meta", "Amazon", "Microsoft", "Apple"]
    target_roles = ["Senior Software Engineer", "Staff Software Engineer", "Backend Engineer"]

    for company in target_companies:
        for role in target_roles:
            print(f"🔍 Fetching {role} salaries at {company}...")
            benchmarker.fetch_salary_data(company=company, role=role, years_experience=5)

    # Calculate and save results
    benchmarks = benchmarker.calculate_benchmarks()
    if benchmarks:
        print(f"📊 Median salary across all entries: ${benchmarks['median_salary']:,.2f}")
        print(f"📊 P90 salary: ${benchmarks['p90_salary']:,.2f}")

    benchmarker.save_data()

import requests
from bs4 import BeautifulSoup
import re
from typing import List, Dict
import json
from datetime import datetime

class LinkedInFakeJobDetector:
    """Detects fake or phishing job posts on LinkedIn using heuristic rules."""

    # Known phishing patterns in job descriptions
    FAKE_PATTERNS = [
        r"send your resume to [\w\.-]+@(gmail|yahoo|hotmail)\.com",  # Personal email domains
        r"pay a fee to apply",  # Upfront fee requests
        r"immediate start no experience required \$[\d,]+/month",  # Unrealistic pay for no experience
        r"whatsapp: \+[\d\s]+",  # WhatsApp contact instead of company email
        r"click here to apply: https?://(?!linkedin\.com)",  # External links not on LinkedIn
    ]

    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        })
        self.results = []

    def fetch_job_post(self, job_url: str) -> Optional[Dict]:
        """Fetch and parse a LinkedIn job post page."""
        try:
            # Validate LinkedIn job URL format
            if not re.match(r"https://www\.linkedin\.com/jobs/view/[\d]+", job_url):
                print(f"⚠️ Invalid LinkedIn job URL: {job_url}")
                return None

            response = self.session.get(job_url, timeout=10)
            response.raise_for_status()

            soup = BeautifulSoup(response.text, "html.parser")

            # Extract job details
            job_title = soup.find("h1", class_="top-card-layout__title")
            company = soup.find("a", class_="topcard__org-name-link")
            description = soup.find("div", class_="show-more-less-html__markup")

            if not all([job_title, company, description]):
                print(f"⚠️ Missing job details for {job_url}")
                return None

            job_data = {
                "job_url": job_url,
                "title": job_title.text.strip(),
                "company": company.text.strip(),
                "description": description.text.strip(),
                "fetched_at": datetime.utcnow().isoformat()
            }

            return job_data

        except requests.exceptions.HTTPError as e:
            print(f"❌ HTTP error fetching job post: {str(e)}")
            return None
        except requests.exceptions.Timeout as e:
            print(f"❌ Request timed out for {job_url}: {str(e)}")
            return None
        except Exception as e:
            print(f"❌ Unexpected error fetching job post: {str(e)}")
            return None

    def detect_fake_patterns(self, job_data: Dict) -> List[str]:
        """Check job description against known fake patterns."""
        fake_matches = []
        description = job_data.get("description", "")

        for pattern in self.FAKE_PATTERNS:
            matches = re.findall(pattern, description, re.IGNORECASE)
            if matches:
                fake_matches.append(f"Pattern matched: {pattern} (matches: {matches})")

        # Additional heuristic: check company email domain
        company = job_data.get("company", "")
        # This would require a company domain lookup, simplified for example
        if "gmail" in description.lower() or "yahoo" in description.lower():
            fake_matches.append("Personal email domain found in description")

        return fake_matches

    def analyze_job(self, job_url: str) -> Dict:
        """Full analysis of a LinkedIn job post for fake indicators."""
        job_data = self.fetch_job_post(job_url)
        if not job_data:
            return {"job_url": job_url, "is_fake": None, "error": "Failed to fetch job post"}

        fake_matches = self.detect_fake_patterns(job_data)
        job_data["fake_indicators"] = fake_matches
        job_data["is_fake"] = len(fake_matches) > 0

        self.results.append(job_data)
        return job_data

    def save_results(self, output_path: str = "linkedin_fake_job_analysis.json") -> None:
        """Save analysis results to JSON."""
        try:
            with open(output_path, "w") as f:
                json.dump(self.results, f, indent=2)
            print(f"✅ Saved analysis of {len(self.results)} jobs to {output_path}")
        except IOError as e:
            print(f"❌ Failed to save results: {str(e)}")

if __name__ == "__main__":
    detector = LinkedInFakeJobDetector()

    # Example LinkedIn job URLs (mix of real and fake for testing)
    test_job_urls = [
        "https://www.linkedin.com/jobs/view/senior-software-engineer-google-1234567890",
        "https://www.linkedin.com/jobs/view/immediate-hiring-data-entry-9876543210",  # Likely fake
    ]

    for url in test_job_urls:
        print(f"🔍 Analyzing {url}...")
        result = detector.analyze_job(url)
        if result.get("is_fake"):
            print(f"❌ Fake job detected! Indicators: {result['fake_indicators']}")
        elif result.get("is_fake") is False:
            print(f"✅ Job appears legitimate")
        else:
            print(f"⚠️ Could not analyze job: {result.get('error')}")

    detector.save_results()

// Example Blind Pro profile JSON snippet for senior engineers
{
  "role": "Senior Backend Engineer",
  "years_experience": 8,
  "tech_stack": ["Python", "Django", "PostgreSQL", "Kubernetes"],
  "salary_expectation_usd": 220000,
  "open_to": ["Full-time", "Remote"],
  "hide_company": true
}

import tweepy

client = tweepy.Client(bearer_token="YOUR_BEARER_TOKEN")
# Add user to tech recruiters list (replace list ID and user ID with your own)
client.add_list_member(list_id="1234567890123456789", user_id="9876543210")

benchmarker = BlindSalaryBenchmarker()
benchmarker.fetch_salary_data(company="Series B Startup", role="Senior Backend Engineer", years_experience=8)
benchmarks = benchmarker.calculate_benchmarks()
if benchmarks:
    print(f"Target ask (110% of median): ${benchmarks['median_salary'] * 1.1:,.2f}")