How to Scrape LinkedIn Profiles and Jobs Without Getting Banned (2026)

from curl_cffi import requests

BROWSER = "chrome124"


def make_session(proxy=None):
    """A session whose TLS handshake and header order match a real Chrome build."""
    session = requests.Session(impersonate=BROWSER)
    session.headers.update({
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9",
    })
    if proxy:
        session.proxies = {"http": proxy, "https": proxy}
    return session


def fetch_public_profile(session, public_id):
    """Fetch a logged-out profile page. Returns None if the profile is not public."""
    url = f"https://www.linkedin.com/in/{public_id}/"
    response = session.get(url, timeout=20, allow_redirects=True)

    if response.status_code == 999:
        raise RuntimeError("Refused at the edge: check client fingerprint and egress IP")
    if response.status_code == 429:
        raise RuntimeError("Rate limited: back off")
    if "/authwall" in response.url:
        return None  # Not public. Do not retry this from another IP.

    response.raise_for_status()
    return response.text

import json
from dataclasses import dataclass, field
from bs4 import BeautifulSoup


@dataclass
class LinkedInProfile:
    public_id: str
    name: str = ""
    headline: str = ""
    locality: str = ""
    country: str = ""
    positions: list[dict] = field(default_factory=list)
    education: list[dict] = field(default_factory=list)


def _person_node(html):
    """Pull the Person node out of the page's JSON-LD @graph."""
    soup = BeautifulSoup(html, "html.parser")
    for tag in soup.find_all("script", type="application/ld+json"):
        try:
            payload = json.loads(tag.string or "")
        except json.JSONDecodeError:
            continue
        for node in payload.get("@graph", []):
            if node.get("@type") == "Person":
                return node
    return None


def _org_entries(nodes):
    entries = []
    for node in nodes or []:
        role = node.get("member") or {}
        entries.append({
            "name": node.get("name", ""),
            "url": node.get("url", ""),
            "start": role.get("startDate"),
            "end": role.get("endDate"),
        })
    return entries


def parse_profile(html, public_id):
    person = _person_node(html)
    if person is None:
        return None

    address = person.get("address") or {}

    return LinkedInProfile(
        public_id=public_id,
        name=person.get("name", ""),
        headline=person.get("description", ""),
        locality=address.get("addressLocality", ""),
        country=address.get("addressCountry", ""),
        positions=_org_entries(person.get("worksFor")),
        education=_org_entries(person.get("alumniOf")),
    )

from dataclasses import dataclass
from bs4 import BeautifulSoup

GUEST_SEARCH = "https://www.linkedin.com/jobs-guest/jobs/api/seeMoreJobPostings/search"
GUEST_DETAIL = "https://www.linkedin.com/jobs-guest/jobs/api/jobPosting/{job_id}"


@dataclass
class JobListing:
    job_id: str
    title: str = ""
    company: str = ""
    location: str = ""
    posted_at: str = ""
    url: str = ""
    description: str = ""
    employment_type: str = ""
    seniority_level: str = ""


def _text(node):
    return node.get_text(strip=True) if node else ""


def search_jobs(session, keywords, location="", start=0):
    """One page of the logged-out job search."""
    response = session.get(
        GUEST_SEARCH,
        params={"keywords": keywords, "location": location, "start": start},
        timeout=20,
    )
    response.raise_for_status()

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

    for card in soup.select("div.job-search-card"):
        time_tag = card.select_one("time")
        link = card.select_one("a.base-card__full-link")
        listings.append(JobListing(
            job_id=card.get("data-entity-urn", "").rsplit(":", 1)[-1],
            title=_text(card.select_one("h3.base-search-card__title")),
            company=_text(card.select_one("h4.base-search-card__subtitle")),
            location=_text(card.select_one("span.job-search-card__location")),
            posted_at=time_tag.get("datetime", "") if time_tag else "",
            url=link["href"].split("?")[0] if link else "",
        ))
    return listings


def get_job_details(session, job_id):
    """Full public posting: description text plus the criteria block."""
    response = session.get(GUEST_DETAIL.format(job_id=job_id), timeout=20)
    response.raise_for_status()

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

    criteria = {}
    for item in soup.select("li.description__job-criteria-item"):
        label = _text(item.select_one("h3.description__job-criteria-subheader"))
        if label:
            key = label.lower().replace(" ", "_")
            criteria[key] = _text(item.select_one("span.description__job-criteria-text"))

    return {
        "description": _text(soup.select_one("div.show-more-less-html__markup")),
        "seniority_level": criteria.get("seniority_level", ""),
        "employment_type": criteria.get("employment_type", ""),
        "job_function": criteria.get("job_function", ""),
        "industries": criteria.get("industries", ""),
    }

import random
import time
from dataclasses import dataclass


@dataclass
class Pacer:
    """Request pacing driven by response codes, not by a guessed limit."""

    delay: float = 2.0
    min_delay: float = 1.5
    max_delay: float = 120.0
    jitter: float = 0.4

    def wait(self):
        time.sleep(self.delay * random.uniform(1 - self.jitter, 1 + self.jitter))

    def on_success(self):
        # Recover slowly. It took a backoff to get here.
        self.delay = max(self.min_delay, self.delay * 0.9)

    def on_throttle(self):
        # 429: halve the rate immediately.
        self.delay = min(self.max_delay, self.delay * 2)

    def on_refusal(self):
        # 999 is not a rate problem, so do not treat it as one.
        raise RuntimeError(
            "Refused at the edge. Check TLS fingerprint and egress before retrying."
        )


def fetch(session, url, pacer, attempts=5):
    for _ in range(attempts):
        pacer.wait()
        response = session.get(url, timeout=20, allow_redirects=True)

        if response.status_code == 999:
            pacer.on_refusal()
        if response.status_code == 429:
            pacer.on_throttle()
            continue
        if "/authwall" in response.url:
            return None  # Not public. Never retry this from another IP.

        response.raise_for_status()
        pacer.on_success()
        return response

    raise RuntimeError(f"Gave up after {attempts} attempts: {url}")

import json
import sqlite3
from dataclasses import asdict
from datetime import date, timedelta

SCHEMA = """
CREATE TABLE IF NOT EXISTS profiles (
    public_id   TEXT NOT NULL,
    scraped_at  DATE NOT NULL,
    name        TEXT,
    headline    TEXT,
    locality    TEXT,
    country     TEXT,
    raw_json    TEXT,
    PRIMARY KEY (public_id, scraped_at)
);
CREATE TABLE IF NOT EXISTS jobs (
    job_id           TEXT NOT NULL,
    scraped_at       DATE NOT NULL,
    title            TEXT,
    company          TEXT,
    location         TEXT,
    posted_at        TEXT,
    url              TEXT,
    description      TEXT,
    employment_type  TEXT,
    seniority_level  TEXT,
    raw_json         TEXT,
    PRIMARY KEY (job_id, scraped_at)
);
CREATE INDEX IF NOT EXISTS idx_profiles_scraped ON profiles(scraped_at);
CREATE INDEX IF NOT EXISTS idx_jobs_company ON jobs(company);
"""


def init_database(db_path):
    conn = sqlite3.connect(db_path)
    conn.executescript(SCHEMA)
    return conn


def save_profile(conn, profile):
    conn.execute(
        "INSERT OR REPLACE INTO profiles VALUES (?, ?, ?, ?, ?, ?, ?)",
        (
            profile.public_id,
            date.today().isoformat(),
            profile.name,
            profile.headline,
            profile.locality,
            profile.country,
            json.dumps(asdict(profile)),
        ),
    )
    conn.commit()


def save_job(conn, job):
    conn.execute(
        "INSERT OR REPLACE INTO jobs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (
            job.job_id,
            date.today().isoformat(),
            job.title,
            job.company,
            job.location,
            job.posted_at,
            job.url,
            job.description,
            job.employment_type,
            job.seniority_level,
            json.dumps(asdict(job)),
        ),
    )
    conn.commit()


def purge_expired(conn, retain_days):
    """Retention is a compliance requirement for personal data, not a nice-to-have."""
    cutoff = (date.today() - timedelta(days=retain_days)).isoformat()
    cursor = conn.execute(
        "DELETE FROM profiles WHERE scraped_at NOT BETWEEN ? AND date('now')",
        (cutoff,),
    )
    conn.commit()
    return cursor.rowcount