Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
51 kB · 1416 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417# LinkedIn profile scraper over LinkedIn's flagship RSC (React Server Components) wire format.## HOW IT SCRAPES# --------------# 1. GET https://www.linkedin.com/flagship-web/in/<vanity>/ returns an RSC stream carrying only# the profile TOP CARD (name, headline, location, images, pronouns, followers/connections).# 2. The viewee profileId (ACo...) is extracted from that stream.# 3. POST https://www.linkedin.com/flagship-web/rsc-action/actions/component is fired for each of# 5 profileCards* components, concurrently:# profileCardsAboveActivity -> About# profileCardsExperienceOnly -> Experience# profileCardsBelowActivityPart1... -> Education + Certifications# profileCardsBelowActivityPart4 -> Languages# profileCardsBelowActivityPart7 -> Skills# Each POST returns an RSC stream with that section's content.# 4. Every RSC stream is parsed: lines of `chunkId:JSON`, with `$Lxx`/`$xx` references that# resolve to other chunks. The resolved React tree is walked to pull out text and images.# 5. The flat text plus the component tree are structured into a validated Profile# (see src.schemas).## CHALLENGES AND HOW THEY WERE SOLVED# ----------------------------------# - RSC wire format, not HTML: the page is a React Server Components stream, not rendered HTML.# Solved with a small parser: parse_chunks splits `chunkId:JSON` lines, resolve() follows# `$Lxx`/`$xx` references, and tree walkers collect text/image leaves. A stream can hold# several independent root chunks, so each walker visits chunk 0 plus every unreferenced chunk.# - The full sections are NOT in the initial GET: the browser loads them as separate async POSTs# to /flagship-web/rsc-action/actions/component. Those POSTs were reverse-engineered: the# componentId/sduiid/parentSpanId query params, per-component JSON payloads, csrf-token =# JSESSIONID (quotes stripped), and the page-instance headers copied from the top-card# response.# - TLS fingerprinting: httpx/httpcore's TLS fingerprint is rejected by LinkedIn's CDN on the# server-action POSTs (GETs return 200, POSTs return 500 "Server internal error"). Solved by# using curl_cffi with impersonate='firefox', which presents a real browser TLS fingerprint.# - Duplicate chunks: the RSC stream re-emits the same section content in several root chunks,# triplicating the flat text. Solved with _dedupe_repeated (collapse exact block repetition).# Sections that repeat field values across items (certifications share issuer/issued dates,# languages can share proficiency) are NOT value-deduplicated, only de-triplicated.# - Flat text -> items: structured via the component tree using a componentKey stack (the# list-wrapper is depth 1, items are depth 2, deeper sub-items merge into their depth-2# ancestor). Certifications are parsed from the flat text by their terminal field line# (Credential ID, or Issued when no credential id follows) because their names live in# referenced child chunks the tree grouping does not resolve."""Async LinkedIn profile scraper over the flagship RSC wire format."""
import asyncioimport base64import jsonimport osimport reimport urllib.parsefrom collections import Counterfrom types import TracebackTypefrom typing import Any
from curl_cffi.requests import ( AsyncSession, Response,)from loguru import logger
from src.schemas import Profile
HEX: set[str] = set("0123456789abcdef")
def parse_chunks(data: bytes) -> dict[str, Any]: """Split an RSC stream into its chunks.
An RSC stream is a sequence of `chunkId:JSON` lines. Each line is parsed and stored by its chunk id; lines that are not valid JSON are skipped.
Args: data: Raw bytes of an RSC stream.
Returns: A dict mapping each chunk id to its parsed JSON value. """ text = data.decode("utf-8", "replace") chunks: dict[str, Any] = {} for line in text.split("\n"): if ":" not in line: continue cid, body = line.split(":", 1) try: chunks[cid] = json.loads(body) except json.JSONDecodeError: continue return chunks
def _is_module_import(o: Any) -> bool: """Return True if `o` is an RSC module-import marker (a 3-item list whose first item is a 32-hex hash).""" return ( isinstance(o, list) and len(o) == 3 and isinstance(o[0], str) and len(o[0]) == 32 and all(c in HEX for c in o[0]) and isinstance(o[2], str) )
def _is_ref(s: Any) -> bool: """Return True if `s` is an RSC reference token like `$Lxx` or `$xx` (hex id).""" if not isinstance(s, str) or not s.startswith("$") or len(s) < 2: return False rest = s[2:] if s[1] == "L" else s[1:] return len(rest) >= 1 and all(c in HEX for c in rest)
def resolve(node: Any, chunks: dict[str, Any], seen: set[str]) -> Any: """Resolve RSC `$Lxx`/`$xx` references into the actual chunk content, recursively.
`$undefined` becomes None. References are looked up in `chunks`; `seen` guards against reference cycles. Lists and dicts are resolved element-wise.
Args: node: Any node from an RSC stream (str, list, dict, or scalar). chunks: Chunk id to parsed JSON (from :func:`parse_chunks`). seen: Set of reference ids already visited, for cycle protection.
Returns: The node with every reference replaced by its resolved content. """ if isinstance(node, str): if node == "$undefined": return None if _is_ref(node): ref = node[2:] if node[1] == "L" else node[1:] if ref in seen: return None return resolve(chunks.get(ref), chunks, seen | {ref}) return node if isinstance(node, list): return [resolve(x, chunks, seen) for x in node] if isinstance(node, dict): return {k: resolve(v, chunks, seen) for k, v in node.items()} return node
SECTION_KEYS: dict[str, set[str]] = { "about": {"about"}, "experience": {"experience"}, "education": {"education"}, "skills": {"skills"}, "certifications": { "licences & certifications", "licenses & certifications", "licences and certifications", "certifications", }, "languages": {"languages"},}DELIMITER_HEADINGS: set[str] = { "services", "featured", "recommendation", "recommendations", "projects", "activity", "honors & awards", "honors", "organizations", "volunteer experience", "volunteering", "publications", "patents", "courses", "test scores", "interests", "people you may know", "you might like", "similar profiles",}HEADING_SET: set[str] = set()for _v in SECTION_KEYS.values(): HEADING_SET |= _vHEADING_SET |= DELIMITER_HEADINGS
def heading_key(name: str) -> str | None: """Map a section/card title to its canonical section key.
Args: name: The heading text as it appears in the stream.
Returns: One of the SECTION_KEYS (e.g. 'about', 'experience'), or None if the title is not a known profile section. """ low = name.strip().lower() for k, variants in SECTION_KEYS.items(): if low in variants: return k return None
TAGS: set[str] = { "div", "section", "p", "h1", "h2", "h3", "h4", "span", "hr", "a", "ul", "li", "img", "meta", "title", "button", "figure", "svg", "path", "input", "label", "form", "nav", "header", "footer", "main", "article", "aside", "table", "tr", "td", "th", "tbody", "thead",}BUTTON_TEXTS: set[str] = { "Send profile in a message", "Save to PDF", "Report / Block", "About this member", "Connect", "Pending", "Follow", "Following", "Cancel", "Unfollow", "Contact info", "Message", "View settings", "Saved items", "Activity", "Explore Premium profiles",}PRONOUNS: set[str] = {"She/Her", "He/Him", "They/Them", "She/They", "He/They"}FOOTER: set[str] = { "About", "Accessibility", "Talent Solutions", "Community Guidelines", "Careers", "Marketing Solutions", "Privacy Policy", "User Agreement", "Pages Terms", "Cookie Policy", "Copyright Policy", "Privacy & Terms", "Ad Choices", "Advertising", "Sales Solutions", "Mobile", "Small Business", "Safety Center", "Questions?", "Recommendation transparency", "LinkedIn Corporation © 2026", "Visit our Help Center.", "Go to your Settings.", "Manage your account and privacy", "Learn more about Recommended Content.",}DIST_RE = re.compile(r"^·\s*\d")FOLL_RE = re.compile(r"^[\d,]+\s+followers$")NUM_RE = re.compile(r"^[\d,]+\+?$")LOC_RE = re.compile(r"^[^|]+?,\s*(?:[^|]+?,\s*)?[^|]+$")SECTION_NOISE: set[str] = set(BUTTON_TEXTS) | { "·", "Show all", "Show more", "See all", "Edit", "Add", "Add profile", "Add role", "Add school", "Add experience", "Add skill", "Add language", "Add certification", "Add profile section", "Show more activity", "See all activity",}UPSELL_HINTS: tuple[str, ...] = ("trial ends.", "customization features.", "Premium", "stand out", "stand out and")SELFVIEW_SKIP: tuple[str, ...] = ( "Finding a new job", "Show recruiters", "Share that you", "Providing services", "Finding volunteer", "Show that you are open", "Showcase services", "Make a strong first impression", "Unlock your", "Use AI profile writing", "Add a custom button", "Hiring", "Enhance your profile", "Feature your career", "Rohan and millions", "dataX.ai", "Stand out and build",)
def _is_skip(t: str) -> bool: """Return True if a text leaf is page chrome (footer, upsell, self-view placeholders).""" return ( t in FOOTER or any(h in t for h in UPSELL_HINTS) or t in SELFVIEW_SKIP or any(t.startswith(s) for s in SELFVIEW_SKIP) )
def _is_text(s: Any) -> bool: """Return True if `s` is a real rendered text leaf (non-empty, not a ref, not a tag name).""" return isinstance(s, str) and bool(s.strip()) and not s.startswith("$") and s not in TAGS
def _emit_image(props: dict[str, Any], images: list[list[str]]) -> None: """Append any image URL(s) found in `props` to the `images` collection.""" ia = props.get("imageAssetValue") if isinstance(ia, dict): src = ia.get("source", {}) if isinstance(src, dict) and src.get("$case") == "renderPayload": rp = src.get("renderPayload", {}) root = rp.get("rootUrl", "") urls = [root + r.get("suffixUrl", "") for r in rp.get("imageRenditions", []) if root and r.get("suffixUrl")] if urls: images.append(urls) if isinstance(props.get("src"), str) and props["src"].startswith("http"): images.append([props["src"]])
def _walk_content(node: Any, texts: list[str], images: list[list[str]]) -> None: """Walk a resolved React tree, collecting rendered text leaves and image URLs.""" if node is None: return if isinstance(node, list): if len(node) >= 2 and node[0] == "$": props = node[3] if len(node) > 3 and isinstance(node[3], dict) else {} tp = props.get("textProps") if isinstance(tp, dict) and isinstance(tp.get("children"), list): for c in tp["children"]: if _is_text(c): texts.append(c.strip()) ch = props.get("children") if isinstance(ch, str) and _is_text(ch): texts.append(ch.strip()) elif isinstance(ch, list): for c in ch: if _is_text(c): texts.append(c.strip()) _emit_image(props, images) for v in props.values(): _walk_content(v, texts, images) for e in node[4:]: _walk_content(e, texts, images) else: for x in node: _walk_content(x, texts, images) elif isinstance(node, dict): _emit_image(node, images) for v in node.values(): _walk_content(v, texts, images)
def _walk_stream(chunks: dict[str, Any], texts: list[str], images: list[list[str]]) -> None: """Walk chunk 0 plus every other unreferenced content chunk, collecting text and images.""" seen: set[str] = set() _walk_content(resolve(chunks.get("0"), chunks, seen), texts, images) for cid, o in chunks.items(): if cid == "0" or cid in seen or _is_module_import(o): continue _walk_content(resolve(o, chunks, set()), texts, images)
def _collect_card_events(chunks: dict[str, Any], events: list[tuple[str, str]]) -> None: """Collect ('heading', name) and ('text', s) events from a card stream.
Headings are either literal text leaves or bound ICU-message `prefix` values that match a known heading. All root chunks are visited in file order. The collected `events` list is appended to in place. """
def emit_str(s: str) -> None: s = s.strip() if not s or s.startswith("$"): return events.append(("heading", s) if s.lower() in HEADING_SET else ("text", s))
def visit_children(node: Any) -> None: if node is None: return if isinstance(node, str): emit_str(node) elif isinstance(node, list): if len(node) >= 2 and node[0] == "$": visit(node) else: for x in node: visit_children(x) elif isinstance(node, dict): visit(node)
def visit(node: Any) -> None: if node is None: return if isinstance(node, list): if len(node) >= 2 and node[0] == "$": props = node[3] if len(node) > 3 and isinstance(node[3], dict) else {} pf = props.get("prefix") if isinstance(pf, str) and pf.strip().lower() in HEADING_SET: events.append(("heading", pf.strip())) tp = props.get("textProps") if isinstance(tp, dict): visit_children(tp.get("children")) visit_children(props.get("children")) for k, v in props.items(): if k in ("textProps", "children", "prefix"): continue visit(v) for e in node[4:]: visit(e) else: for x in node: visit(x) elif isinstance(node, dict): pf = node.get("prefix") if isinstance(pf, str) and pf.strip().lower() in HEADING_SET: events.append(("heading", pf.strip())) for v in node.values(): visit(v)
seen: set[str] = set() visit(resolve(chunks.get("0"), chunks, seen)) for cid, o in chunks.items(): if cid == "0" or cid in seen or _is_module_import(o): continue visit(resolve(o, chunks, set()))
def _find_topcard_chunk(chunks: dict[str, Any]) -> str | None: """Return the chunk id whose JSON contains the profile-top-card view, or None.""" for cid, o in chunks.items(): if '"viewName":"profile-top-card"' in json.dumps(o, separators=(",", ":")): return cid return None
UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")_ITEM_SECTION_TOP: dict[str, str] = { "experience": "experiencetoplevel", "education": "educationtoplevel", "certifications": "certificationtoplevel", "languages": "languagetoplevel", "skills": "skills",}_BOUNDARY_RE = re.compile( r"(com\.linkedin\.sdui\.profile\.card\.ref|Profile_Top_Level_|ProfileNullStateCardAnchor|TopLevel)", re.I,)
EMP_TYPES: set[str] = { "Full-time", "Part-time", "Internship", "Contract", "Self-employed", "Freelance", "Temporary", "Seasonal",}WORK_TYPES: set[str] = {"On-site", "Remote", "Hybrid"}MONTHS: tuple[str, ...] = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec")SKIP_TAGS: set[str] = { "div", "section", "p", "span", "hr", "figure", "svg", "path", "ul", "li", "h1", "h2", "h3", "h4",}
def _section_of_item(ck: Any) -> str | None: """Return the canonical section key for a componentKey, or None if it is not a section marker.""" if not isinstance(ck, str): return None low = ck.lower() for k, marker in _ITEM_SECTION_TOP.items(): if marker in low: return k return None
def _is_item_key(ck: Any) -> bool: """Return True if a componentKey identifies a list item (UUID, entity-collection-item, or skill chip).""" if not isinstance(ck, str): return False return bool( UUID_RE.match(ck) or ck.startswith("entity-collection-item") or ck.startswith("com.linkedin.sdui.profile.skill"), )
def group_items(data: bytes) -> dict[str, dict[str, list[str]]]: """Group a card stream's text into per-section items by component structure.
A stack of open item-key components is tracked. The first item-key in a section is the list-wrapper (stack depth 1); item-keys nested directly under it (depth 2) are the items; deeper sub-items merge into their depth-2 ancestor. A repeated componentKey on a wrapper+inner pair is deduped, and section-boundary componentKeys reset the stack.
Args: data: Raw bytes of a card RSC stream.
Returns: A dict of `{section: {itemKey: [text, ...]}}`. """ chunks = parse_chunks(data) items: dict[str, dict[str, list[str]]] = {}
def emit(props: dict[str, Any], sec: str | None, item: str | None) -> None: if not (sec and item): return bucket = items.setdefault(sec, {}).setdefault(item, []) tp = props.get("textProps") if isinstance(tp, dict) and isinstance(tp.get("children"), list): for c in tp["children"]: if isinstance(c, str) and c.strip() and not c.strip().startswith("$"): bucket.append(c.strip()) ch = props.get("children") if isinstance(ch, list): for c in ch: if isinstance(c, str) and c.strip() and not c.strip().startswith("$"): bucket.append(c.strip())
def visit(node: Any, section: str | None, stack: list[str]) -> None: if node is None: return if isinstance(node, list): if len(node) >= 2 and node[0] == "$": props = node[3] if len(node) > 3 and isinstance(node[3], dict) else {} ck = props.get("componentKey") or props.get("componentkey") sec, stk = section, stack if ck: if _BOUNDARY_RE.search(ck): sec = _section_of_item(ck) stk = [] if _is_item_key(ck) and (not stk or stk[-1] != ck): stk = stk + [ck] item = stk[1] if len(stk) >= 2 else None emit(props, sec, item) for v in props.values(): visit(v, sec, stk) for e in node[4:]: visit(e, sec, stk) else: for x in node: visit(x, section, stack) elif isinstance(node, dict): for v in node.values(): visit(v, section, stack)
seen: set[str] = set() visit(resolve(chunks.get("0"), chunks, seen), None, []) for cid, o in chunks.items(): if cid == "0" or cid in seen or _is_module_import(o): continue visit(resolve(o, chunks, set()), None, []) return items
def _clean_texts(texts: list[str]) -> list[str]: """Drop tag names and skill-chip component identifiers from a list of text leaves.""" return [t for t in texts if t not in SKIP_TAGS and not t.startswith("com.linkedin.sdui.profile.skill")]
def _is_dates_line(t: str) -> bool: """Return True if a line looks like an experience/education date range or duration.""" if " · " in t: left = t.split(" · ", 1)[0] if left in EMP_TYPES: return True if re.search(r"\b(19|20)\d{2}\b", left): return True if any(m in left for m in MONTHS) and re.search(r"\d{4}", left): return True if re.search(r"\b(19|20)\d{2}\b\s*[-–]\s*(\b(19|20)\d{2}\b|Present)", t): return True return False
def _parse_dates(t: str) -> dict[str, str | None]: """Parse a date line into start/end/duration (and optionally employment_type).
Args: t: A date line such as 'Jul 2026 - Present · 2 mos' or 'Full-time · 6 mos'.
Returns: A dict with any of `start_date`, `end_date`, `duration`, `employment_type` that could be parsed. """ duration: str | None = None if " · " in t: rng, duration = t.split(" · ", 1) else: rng = t rng = rng.strip() if rng in EMP_TYPES: return {"employment_type": rng, "duration": duration} for sep in (" - ", " – ", " — "): if sep in rng: s, e = rng.split(sep, 1) return {"start_date": s.strip(), "end_date": e.strip(), "duration": duration} return {"start_date": rng, "end_date": None, "duration": duration}
def _split_suffix(t: str, suffixes: set[str]) -> tuple[str, str] | None: """If `t` ends with ' · <suffix>' for a known suffix, return (prefix, suffix), else None.""" for s in suffixes: if t.endswith(" · " + s): return t[: -(len(s) + 3)].strip(), s return None
def structure_experience(item_texts: dict[str, list[str]]) -> list[dict[str, Any]]: """Structure grouped experience items into a list of experience dicts.
Each item's text leaves are classified into title, company + employment type, dates, location + work type, and a description list.
Args: item_texts: `{itemKey: [text, ...]}` from :func:`group_items`.
Returns: A list of dicts with keys `title`, `company`, `employment_type`, `start_date`, `end_date`, `duration`, `location`, `work_type` and `description`. """ out: list[dict[str, Any]] = [] for key in item_texts: texts = _clean_texts(item_texts[key]) if not texts: continue entry: dict[str, Any] = { "title": None, "company": None, "employment_type": None, "start_date": None, "end_date": None, "duration": None, "location": None, "work_type": None, "description": [], } for t in texts: if t.startswith("Activities and societies"): continue if _is_dates_line(t): entry.update({k: v for k, v in _parse_dates(t).items() if v is not None}) continue suf = _split_suffix(t, EMP_TYPES) if suf is not None: entry["company"], entry["employment_type"] = suf continue suf = _split_suffix(t, WORK_TYPES) if suf is not None: entry["location"], entry["work_type"] = suf continue if entry["title"] is None: entry["title"] = t else: entry["description"].append(t) out.append(entry) return out
def structure_education(item_texts: dict[str, list[str]]) -> list[dict[str, Any]]: """Structure grouped education items into a list of education dicts.
The first text leaf is the school; a line containing a year range becomes the start/end; the degree line may split into degree and field on ', '. Anything after an 'Activities and societies:' line is collected as activities.
Args: item_texts: `{itemKey: [text, ...]}` from :func:`group_items`.
Returns: A list of dicts with keys `school`, `degree`, `field`, `start`, `end` and `activities`. """ out: list[dict[str, Any]] = [] for key in item_texts: texts = _clean_texts(item_texts[key]) if not texts: continue entry: dict[str, Any] = { "school": None, "degree": None, "field": None, "start": None, "end": None, "activities": [], } for i, t in enumerate(texts): if t.startswith("Activities and societies:"): entry["activities"].append(t[len("Activities and societies:") :].strip()) for rest in texts[i + 1 :]: if rest.startswith("Activities and societies:"): entry["activities"].append(rest[len("Activities and societies:") :].strip()) else: entry["activities"].append(rest) break if entry["school"] is None: entry["school"] = t elif re.search(r"\b(19|20)\d{2}\b", t) and ("–" in t or "-" in t or "·" in t): for sep in (" – ", " - ", "–", "-"): if sep in t: s, e = t.split(sep, 1) entry["start"], entry["end"] = s.strip(), e.strip() break else: entry["start"] = t elif entry["degree"] is None: if ", " in t: deg, field = t.split(", ", 1) entry["degree"], entry["field"] = deg.strip(), field.strip() else: entry["degree"] = t out.append(entry) return out
def _is_endorser_line(t: str) -> bool: """Return True if a line is skill-endorser context (a 'Job at Company' line or a skill chip id).""" return " at " in t or "·" in t or t.startswith("com.linkedin.sdui.profile.skill")
def _dedupe_repeated[T](lst: list[T]) -> list[T]: """Collapse exact block repetitions.
RSC card streams re-emit the same section content in several independent root chunks, so the flat text is one block repeated k times. This returns the shortest prefix block whose repetition reconstructs the list, or the list unchanged if it is not a clean repetition.
Args: lst: A list that may be a block repeated some number of times.
Returns: The de-triplicated block, or `lst` unchanged if no exact repetition is found. """ n = len(lst) for p in range(1, n // 2 + 1): if n % p: continue if all(lst[i] == lst[i % p] for i in range(p, n)): return lst[:p] return lst
def structure_skills(item_texts: dict[str, list[str]]) -> list[dict[str, str]]: """Structure grouped skill items into a list of `{name}` dicts.
Endorser-context lines (a 'Job at Company' line or a skill chip id) and duplicates are dropped.
Args: item_texts: `{itemKey: [text, ...]}` from :func:`group_items`.
Returns: A list of `{'name': str}` dicts, de-duplicated, in first-seen order. """ out: list[dict[str, str]] = [] seen: set[str] = set() for key in item_texts: for t in _clean_texts(item_texts[key]): if t in seen or _is_endorser_line(t): continue seen.add(t) out.append({"name": t}) return out
def _build_cert(block: list[str]) -> dict[str, str | None] | None: """Build one certification dict from its flat text block.
Args: block: The text lines belonging to a single certification (name, optional issuer, optional 'Issued <date>', optional 'Credential ID <id>').
Returns: A cert dict, or None if the block had no recognizable content. """ c: dict[str, str | None] = { "name": None, "issuer": None, "issued": None, "credential_id": None, "description": None, } for t in block: if t.startswith("Credential ID "): c["credential_id"] = t[len("Credential ID ") :] elif t.startswith("Issued "): c["issued"] = t[len("Issued ") :] rest = [t for t in block if not t.startswith("Credential ID ") and not t.startswith("Issued ")] if rest: c["name"] = rest[0] if len(rest) > 1 and not _is_endorser_line(rest[1]): c["issuer"] = rest[1] desc = [t for t in rest[2:] if not _is_endorser_line(t)] else: desc = [t for t in rest[1:] if not _is_endorser_line(t)] if desc: c["description"] = " ".join(desc) if c["name"] is not None or c["issuer"] or c["issued"] or c["credential_id"]: return c return None
def structure_certifications( flat: list[str], item_texts: dict[str, list[str]] | None = None,) -> list[dict[str, str | None]]: """Split certifications into a list of certification dicts.
Each certification block ends at its terminal field line: a `Credential ID <id>` line when the cert exposes one, otherwise an `Issued <date>` line that is not immediately followed by a `Credential ID` line. Within a block the first non-field line is the cert name and the next is the issuer (when present).
The flat text is de-triplicated first (RSC streams emit the same content in several root chunks). The tree (`item_texts`) is not relied on because cert names live in referenced child chunks that the item grouping does not resolve.
Args: flat: The certifications section's flat text leaves. item_texts: Unused; kept for signature compatibility.
Returns: A list of dicts with keys `name`, `issuer`, `issued`, `credential_id` and `description`. """ flat = _dedupe_repeated(flat) out: list[dict[str, str | None]] = [] block: list[str] = [] for i, t in enumerate(flat): block.append(t) nxt = flat[i + 1] if i + 1 < len(flat) else None if t.startswith("Credential ID "): pass elif t.startswith("Issued ") and (nxt is None or not nxt.startswith("Credential ID ")): pass else: continue cert = _build_cert(block) if cert is not None: out.append(cert) block = [] if block: cert = _build_cert(block) if cert is not None: out.append(cert) return out
def structure_languages(flat: list[str]) -> list[dict[str, str | None]]: """Structure the languages section into a list of `{language, proficiency}` dicts.
Languages are flat name/proficiency pairs in the wrapper.
Args: flat: The languages section's flat text leaves.
Returns: A list of `{'language': str, 'proficiency': str | None}` dicts. """ PROF = { "Native or bilingual proficiency", "Full professional proficiency", "Professional working proficiency", "Limited working proficiency", "Elementary proficiency", "Conversational", } out: list[dict[str, str | None]] = [] i = 0 while i < len(flat): lang = flat[i] prof = flat[i + 1] if i + 1 < len(flat) and flat[i + 1] in PROF else None out.append({"language": lang, "proficiency": prof}) i += 2 if prof else 1 return out
def parse_topcard(vanity: str, data: bytes) -> dict[str, Any]: """Parse the top-card RSC stream into the top-card profile fields.
Extracts name, headline, location, pronouns, followers, connections, and the profile/background image URLs. The card-sourced sections (about/experience/...) are left empty here and filled by :func:`merge_card_sections`.
Args: vanity: The LinkedIn vanity name. data: Raw bytes of the top-card RSC stream.
Returns: A profile dict with all Profile keys present (card sections set to None until :func:`merge_card_sections` runs). """ chunks = parse_chunks(data) profile: dict[str, Any] = dict.fromkeys( [ "name", "headline", "location", "about", "experience", "education", "skills", "certifications", "languages", "profile_image", "background_image", "pronouns", "followers", "connections", "vanity", ], None, ) profile["vanity"] = vanity
texts: list[str] = [] images: list[list[str]] = [] _walk_content(resolve(chunks.get("0"), chunks, set()), texts, images) for v in texts: if isinstance(v, str) and v.endswith(" | LinkedIn"): profile["name"] = v[: -len(" | LinkedIn")] break
disp: list[str] = [] bg: list[str] = [] for urls in images: if not urls: continue u = urls[0] if "displaybackgroundimage" in u: bg = urls elif "displayphoto" in u and not disp: disp = urls profile["profile_image"] = disp[0] if disp else None profile["background_image"] = bg[0] if bg else None
cnt: Counter[str] = Counter(texts)
def ok_repeat(t: str) -> bool: return ( cnt[t] >= 2 and t != profile["name"] and t not in BUTTON_TEXTS and t not in PRONOUNS and not DIST_RE.match(t) and not FOLL_RE.match(t) and not NUM_RE.match(t) and not t.startswith("View ") and not t.startswith("auto-component") and "Drawer" not in t and not _is_skip(t) )
repeat_long: list[str] = [t for t in cnt if ok_repeat(t)] headline: str | None = max(repeat_long, key=len) if repeat_long else None if not headline: tcid = _find_topcard_chunk(chunks) if tcid: tt: list[str] = [] _walk_content(resolve(chunks.get(tcid), chunks, set()), tt, []) cand = [ t for t in tt if t != profile["name"] and t not in BUTTON_TEXTS and t not in PRONOUNS and not DIST_RE.match(t) and not FOLL_RE.match(t) and not NUM_RE.match(t) and not _is_skip(t) and len(t) > 10 ] if cand: headline = max(cand, key=len)
profile["headline"] = headline profile["location"] = next( ( t for t in texts if LOC_RE.match(t) and t != headline and t not in BUTTON_TEXTS and t not in PRONOUNS and not _is_skip(t) ), None, ) profile["pronouns"] = next((t for t in cnt if t in PRONOUNS), None) profile["followers"] = next((t for t in cnt if FOLL_RE.match(t)), None) profile["connections"] = next((t for t in texts if NUM_RE.match(t) and t != "·"), None) return profile
def merge_card_sections(profile: dict[str, Any], cards: dict[str, bytes]) -> None: """Parse card streams and fill the section fields of `profile` in place.
Builds flat text and a per-section item tree from every card stream, then structures about/experience/education/skills/certifications/languages. Certifications and languages are de-triplicated (not value-deduplicated) because their items can repeat field values.
Args: profile: The profile dict from :func:`parse_topcard`, mutated in place. cards: `{component_short: bytes}` from :meth:`LinkedInScraper.fetch_cards`. """ flat: dict[str, list[str]] = {k: [] for k in SECTION_KEYS} tree: dict[str, dict[str, list[str]]] = {} for data in cards.values(): ch = parse_chunks(data) events: list[tuple[str, str]] = [] _collect_card_events(ch, events) current: str | None = None for kind, val in events: if kind == "heading": current = heading_key(val) continue if current is None: continue v = val.strip() if ( not v or v in SECTION_NOISE or v == profile.get("name") or v.startswith("·") or v.startswith("Add ") or v == profile.get("headline") or v in (profile.get("headline") or "") ): continue if current in ("certifications", "languages"): flat[current].append(v) elif v not in flat[current]: flat[current].append(v) for sec, items in group_items(data).items(): tree.setdefault(sec, {}).update(items)
flat["certifications"] = _dedupe_repeated(flat["certifications"]) flat["languages"] = _dedupe_repeated(flat["languages"])
profile["about"] = " ".join(flat["about"]) if flat["about"] else None profile["experience"] = structure_experience(tree.get("experience", {})) or None profile["education"] = structure_education(tree.get("education", {})) or None profile["skills"] = structure_skills(tree.get("skills", {})) or None profile["certifications"] = structure_certifications(flat["certifications"], tree.get("certifications", {})) or None profile["languages"] = structure_languages(flat["languages"]) or None
class LinkedInScraper: """Async scraper for LinkedIn profiles over the flagship RSC wire format.
Uses curl_cffi (which impersonates a real browser's TLS fingerprint) because LinkedIn's CDN rejects the TLS fingerprint of plain httpx/httpcore on the `rsc-action` POST endpoints (GETs pass, the sensitive server-action POSTs do not). curl_cffi's `AsyncSession` has an httpx-like async API.
Use it as an async context manager so the underlying session is closed::
async with LinkedInScraper(cookies) as scraper: profile = await scraper.scrape("ankita-gupta-5a311322")
Args: cookies: LinkedIn session cookies (must include `li_at` and `JSESSIONID`; the `csrf-token` header is auto-synced to `JSESSIONID`). headers: Optional extra/override request headers. impersonate: curl_cffi browser profile to impersonate (TLS fingerprint). """
CARD_COMPONENTS: list[str] = [ "profileCardsAboveActivity", "profileCardsExperienceOnly", "profileCardsBelowActivityPart1WithoutExp", "profileCardsBelowActivityPart4", "profileCardsBelowActivityPart7", ]
DEFAULT_HEADERS: dict[str, str] = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:154.0) Gecko/20100101 Firefox/154.0", "Accept": "*/*", "Accept-Language": "en-US,en;q=0.9", "Referer": "https://www.linkedin.com/", "x-li-rsc-stream": "true", "x-li-initial-url": "/in/", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Priority": "u=4", }
BASE: str = "https://www.linkedin.com/flagship-web" TIMEOUT: float = 30.0 IMPERSONATE: str = "firefox"
def __init__( self, cookies: dict[str, str], headers: dict[str, str] | None = None, impersonate: str = IMPERSONATE, ) -> None: self.cookies: dict[str, str] = dict(cookies) h: dict[str, str] = dict(self.DEFAULT_HEADERS) if headers: h.update(headers) h["csrf-token"] = str(self.cookies.get("JSESSIONID", "")).strip('"') self.headers: dict[str, str] = h self.impersonate: str = impersonate self._session: AsyncSession | None = None
@property def session(self) -> AsyncSession: """The lazy curl_cffi `AsyncSession` used for all requests.""" if self._session is None: self._session = AsyncSession( impersonate=self.impersonate, cookies=self.cookies, timeout=self.TIMEOUT, ) return self._session
async def close(self) -> None: """Close the underlying HTTP session, if one was created.""" if self._session is not None: await self._session.close() self._session = None
async def __aenter__(self) -> LinkedInScraper: return self
async def __aexit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None, ) -> None: await self.close()
@staticmethod def _vurl(vanity: str) -> str: """URL-encode a vanity name, keeping '/' and '%' untouched.""" return urllib.parse.quote(vanity, safe="/%")
async def fetch_topcard(self, vanity: str) -> Response: """GET the initial RSC stream (top card) for a profile.
Args: vanity: The LinkedIn vanity name.
Returns: The top-card HTTP response (status 200).
Raises: RuntimeError: If the GET does not return 200. """ logger.debug("fetching top card for vanity={!r}", vanity) resp = await self.session.get( f"{self.BASE}/in/{self._vurl(vanity)}/", params={"skipRedirect": "true"}, headers=self.headers, ) if resp.status_code != 200: logger.error("top-card GET failed for {!r}: HTTP {}", vanity, resp.status_code) raise RuntimeError(f"top-card GET failed: {resp.status_code}") logger.debug("top card fetched for {!r} ({} bytes)", vanity, len(resp.content)) return resp
@staticmethod def extract_profile_id(topcard_bytes: bytes) -> tuple[str | None, bool]: """Extract the viewee profileId and self-view flag from a top-card stream.
Args: topcard_bytes: Raw bytes of the top-card RSC stream.
Returns: A `(profile_id, is_self_view)` tuple. `profile_id` is the `ACo...` id (or None if not found); `is_self_view` is True when the stream indicates the viewer is viewing their own profile. """ data = topcard_bytes.decode("utf-8", "replace") m = ( re.search(r'"vieweeProfileId":"(ACo[A-Za-z0-9_-]+)"', data) or re.search(r'"nonIterableProfileId":"(ACo[A-Za-z0-9_-]+)"', data) or re.search(r'"profileUrn":"urn:li:fsd_profile:(ACo[A-Za-z0-9_-]+)"', data) or re.search(r'"profileId":"(ACo[A-Za-z0-9_-]+)"', data) ) pid = m.group(1) if m else None return pid, '"isSelfView":true' in data
def _binding(self, key: str, profile_id: str) -> dict[str, Any]: """Build a single SDUI BindingImpl value for a profile-component-state key.""" return { "type": "com.linkedin.sdui.components.core.BindingImpl", "value": {"key": f"{key}{profile_id}ProfileComponentState", "namespace": "MemoryNamespace"}, }
def _full_payload(self, vanity: str, pid: str) -> dict[str, Any]: """Build the full card POST payload for the non-About card components.""" keys = [ ("ProfileComponentStateShouldRefreshScreen", "shouldRefreshScreenOnReappear"), ("ProfileComponentStateFetchFromCache", "shouldFetchFromCache"), ("ProfileComponentStateLoadedProfileSections", "loadedSections"), ("ProfileComponentStateShouldDisplayTabAnchors", "shouldDisplayTabAnchors"), ("ProfileComponentStateShouldReloadTopCardOnReappear", "shouldReloadTopCardOnReappear"), ("ProfileComponentStateShouldDisplayStickyHeader", "shouldDisplayStickyHeader"), ("ProfileComponentStateShouldRefreshLanguageDetails", "shouldRefreshLanguageDetailScreen"), ("ProfileComponentStateLastPerformedActionRef", "lastPerformedActionRef"), ("ProfileComponentStateShouldFocusOnReappear", "shouldFocusOnReappear"), ("ProfileComponentStateShouldFocusFeaturedOnReappear", "shouldFocusFeaturedOnReappear"), ("ProfileComponentStateLastFeaturedActionRef", "lastFeaturedActionRef"), ("ProfileComponentStateProfileHideCards", "shouldHideProfileCards"), ] pcs: dict[str, Any] = {"profileId": vanity} for src, dst in keys: pcs[dst] = self._binding(src, vanity) return { "isSelfView": False, "vanityName": vanity, "replaceableSectionArgs": { "vanityName": vanity, "hideCardsForGoldenGate": False, "shouldSetupReplaceableComponent": True, "vieweeProfileId": pid, "isSelfView": False, "isSelfViewResolved": False, }, "profileComponentState": pcs, }
def _card_headers(self, vanity: str, trk: str, app_inst: str, app_ver: str) -> dict[str, str]: """Build the headers for the card POSTs, copying page-instance values from the top card.""" return { "User-Agent": self.headers["User-Agent"], "Accept": "*/*", "Accept-Language": "en-US,en;q=0.9", "Referer": f"https://www.linkedin.com/in/{self._vurl(vanity)}/", "x-li-rsc-stream": "true", "Content-Type": "application/json", "csrf-token": self.headers["csrf-token"], "x-li-page-instance-tracking-id": trk, "x-li-application-instance": app_inst, "x-li-anchor-page-key": "d_flagship3_profile_view_base", "x-li-application-version": app_ver, "x-li-page-instance": f"urn:li:page:d_flagship3_profile_view_base;{trk}", "x-li-track": '{"clientVersion":"0.2.6951","mpVersion":"0.2.6951","osName":"web",' '"timezoneOffset":5.5,"timezone":"Asia/Kolkata","deviceFormFactor":"DESKTOP",' '"mpName":"web","displayDensity":1,"displayWidth":2560,"displayHeight":1440}', "Origin": "https://www.linkedin.com", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Priority": "u=4", }
async def fetch_cards(self, vanity: str, topcard_resp: Response | None = None) -> dict[str, bytes]: """POST all profile-card RSC streams concurrently and return their bytes.
Args: vanity: The LinkedIn vanity name. topcard_resp: An optional pre-fetched top-card response, reused for the viewee profileId and page-instance headers. If None, a top-card GET is performed first.
Returns: A `{component_short: bytes}` mapping of fetched card streams. Components that did not return 200 are omitted.
Raises: ValueError: If no profileId could be found in the top-card stream. """ if topcard_resp is None: topcard_resp = await self.fetch_topcard(vanity) pid, is_self = self.extract_profile_id(topcard_resp.content) if not pid: raise ValueError(f"could not find profileId for {vanity!r}") gh = topcard_resp.headers trk = gh.get("x-li-page-instance-tracking-id", "") H = self._card_headers( vanity, trk, gh.get("x-li-application-instance", ""), gh.get("x-li-application-version", "0.2.6951"), ) req_meta: dict[str, str] = {"$type": "proto.sdui.common.RequestMetadata"}
async def fetch_one(short: str) -> tuple[str, bytes | None]: nid = f"com.linkedin.sdui.generated.profile.dsl.impl.{short}" if short == "profileCardsAboveActivity": payload: dict[str, Any] = {"isSelfView": is_self, "vanityName": vanity} screen = "com.linkedin.sdui.flagshipnav.home.Home" else: payload = self._full_payload(vanity, pid) screen = "com.linkedin.sdui.flagshipnav.profile.Profile" body = { "clientArguments": { "payload": payload, "states": [], "requestMetadata": req_meta, "screenId": screen, "knownTemplateIds": [], }, } parent_span = base64.b64encode(os.urandom(12)).decode().rstrip("=") url = ( f"{self.BASE}/rsc-action/actions/component" f"?componentId={urllib.parse.quote(nid)}&sduiid={urllib.parse.quote(nid)}" f"&parentSpanId={urllib.parse.quote(parent_span)}" ) r = await self.session.post( url, headers=H, json=body, params={"skipRedirect": "true"}, timeout=40.0, ) return short, r.content if r.status_code == 200 else None
logger.debug("requesting {} card streams for {!r}", len(self.CARD_COMPONENTS), vanity) results = await asyncio.gather(*(fetch_one(s) for s in self.CARD_COMPONENTS)) cards = {s: c for s, c in results if c is not None} logger.debug("fetched {}/{} card streams for {!r}", len(cards), len(self.CARD_COMPONENTS), vanity) return cards
@staticmethod def parse_profile(vanity: str, topcard_bytes: bytes, cards: dict[str, bytes]) -> Profile: """Parse a top-card stream plus card streams into a validated Profile.
Args: vanity: The LinkedIn vanity name. topcard_bytes: Raw bytes of the top-card RSC stream. cards: `{component_short: bytes}` from :meth:`fetch_cards`.
Returns: A validated :class:`src.schemas.Profile`. """ profile = parse_topcard(vanity, topcard_bytes) if cards: merge_card_sections(profile, cards) return Profile(**profile)
async def scrape(self, vanity: str) -> Profile: """Fetch and parse a profile in one call.
Args: vanity: The LinkedIn vanity name.
Returns: A validated :class:`src.schemas.Profile`. """ logger.info("scraping profile {!r}", vanity) resp = await self.fetch_topcard(vanity) cards = await self.fetch_cards(vanity, resp) profile = self.parse_profile(vanity, resp.content, cards) logger.debug( "scraped {!r}: name={!r} experience={} education={} skills={} certs={} languages={}", vanity, profile.name, len(profile.experience), len(profile.education), len(profile.skills), len(profile.certifications), len(profile.languages), ) return profile