1"""
2LinkedIn Profile Scraper for HRX Verifier.
3Scrapes candidate profile experience, tenure dates, headline, and hiring activity using li_at session cookie.
4Handles proxy rotation, randomized delays, and CAPTCHA detection.
5"""
6import asyncio
7import json
8import logging
9import random
10import re
11from typing import Any, Dict, List, Optional
12import httpx
13from bs4 import BeautifulSoup
14
15logger = logging.getLogger("hrx-linkedin-verifier")
16
17LINKEDIN_HEADERS = {
18 "User-Agent": (
19 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
20 "AppleWebKit/537.36 (KHTML, like Gecko) "
21 "Chrome/125.0.0.0 Safari/537.36"
22 ),
23 "Accept-Language": "en-US,en;q=0.9",
24 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
25}
26
27HIRING_KEYWORDS = [
28 "hiring", "we're hiring", "we are hiring", "job opening",
29 "open position", "join our team", "join my team", "recruiting",
30 "hiring for", "looking for software engineers", "looking for developers",
31 "looking for product manager", "apply here", "job link"
32]
33
34
35class CaptchaError(Exception):
36 """Raised when LinkedIn presents a CAPTCHA or security checkpoint."""
37 pass
38
39
40class ProfileNotFoundError(Exception):
41 """Raised when the requested profile URL cannot be found."""
42 pass
43
44
45def normalize_profile_url(url: str) -> str:
46 """Ensure profile URL is in proper format."""
47 url = url.strip().rstrip("/")
48 if not url.startswith("http"):
49 url = f"https://www.linkedin.com/in/{url.lstrip('/')}"
50 return url
51
52
53def check_captcha_or_checkpoint(resp: httpx.Response) -> None:
54 """Fail loudly if LinkedIn returns a CAPTCHA, 429 rate limit, or security checkpoint."""
55 if resp.status_code in (429, 999):
56 raise CaptchaError(f"LinkedIn HTTP {resp.status_code}: Rate limited or security checkpoint triggered.")
57
58 url_str = str(resp.url).lower()
59 if "/checkpoint/" in url_str or "/challenge/" in url_str or "captcha" in url_str:
60 raise CaptchaError(f"LinkedIn redirected to Security Checkpoint/CAPTCHA page: {resp.url}")
61
62 soup = BeautifulSoup(resp.text[:50000], "lxml")
63 if soup.find("input", {"id": "captcha"}) or soup.find(id=re.compile(r"captcha", re.I)):
64 raise CaptchaError("LinkedIn CAPTCHA element detected on page response.")
65
66
67async def scrape_profile(
68 profile_url: str,
69 linkedin_cookie: Optional[str] = None,
70 proxy_url: Optional[str] = None
71) -> Dict[str, Any]:
72 """
73 Scrape candidate profile from LinkedIn using session cookie and proxy.
74 Applies randomized 2–5s delays and safety checks.
75 """
76 clean_url = normalize_profile_url(profile_url)
77
78
79 delay = random.uniform(2.0, 5.0)
80 logger.info(f"Applying randomized safety delay of {delay:.2f}s before requesting {clean_url}")
81 await asyncio.sleep(delay)
82
83 headers = dict(LINKEDIN_HEADERS)
84 cookies = {}
85 if linkedin_cookie:
86 cookies["li_at"] = linkedin_cookie
87
88 result: Dict[str, Any] = {
89 "url": clean_url,
90 "full_name": None,
91 "headline": None,
92 "current_employer": None,
93 "current_role_title": None,
94 "experiences": [],
95 "actual_tenure_years": 0.0,
96 "is_actively_hiring": False,
97 "recent_activity_posts": [],
98 "raw_text_snippet": None,
99 }
100
101 try:
102 async with httpx.AsyncClient(
103 headers=headers,
104 cookies=cookies,
105 follow_redirects=True,
106 timeout=20.0,
107 proxy=proxy_url,
108 ) as client:
109 resp = await client.get(clean_url)
110
111
112 check_captcha_or_checkpoint(resp)
113
114 if resp.status_code == 404:
115 raise ProfileNotFoundError(f"LinkedIn profile not found at {clean_url}")
116
117 if resp.status_code != 200:
118 raise RuntimeError(f"Failed to fetch profile: HTTP {resp.status_code}")
119
120 soup = BeautifulSoup(resp.text, "lxml")
121
122
123 json_ld_tags = soup.find_all("script", {"type": "application/ld+json"})
124 for tag in json_ld_tags:
125 if not tag.string:
126 continue
127 try:
128 data = json.loads(tag.string)
129 if isinstance(data, list):
130 data = data[0] if data else {}
131
132 if data.get("@type") == "Person":
133 result["full_name"] = data.get("name")
134 result["headline"] = data.get("description") or data.get("jobTitle")
135 works_for = data.get("worksFor")
136 if isinstance(works_for, list) and works_for:
137 result["current_employer"] = works_for[0].get("name")
138 elif isinstance(works_for, dict):
139 result["current_employer"] = works_for.get("name")
140 except Exception:
141 pass
142
143
144 if not result["full_name"]:
145 og_title = soup.find("meta", property="og:title")
146 if og_title and og_title.get("content"):
147 result["full_name"] = og_title["content"].split(" | ")[0].split(" - ")[0].strip()
148
149 og_desc = soup.find("meta", property="og:description")
150 if og_desc and og_desc.get("content") and not result["headline"]:
151 result["headline"] = og_desc["content"].strip()
152
153
154 experiences: List[Dict[str, Any]] = []
155
156
157 code_tags = soup.find_all("code")
158 for code in code_tags:
159 if not code.string:
160 continue
161 if "com.linkedin.voyager.dash.identity.profile.Position" in code.string or "experience" in code.string.lower():
162 try:
163 raw_data = json.loads(code.string)
164 included = raw_data.get("included", []) if isinstance(raw_data, dict) else []
165 for item in included:
166 if item.get("$type", "").endswith("Position"):
167 company_name = item.get("companyName")
168 title = item.get("title")
169 time_period = item.get("timePeriod", {})
170 start_dt = time_period.get("startDate", {})
171 end_dt = time_period.get("endDate", {})
172
173 start_str = f"{start_dt.get('year', '')}-{start_dt.get('month', '01')}" if start_dt else None
174 end_str = f"{end_dt.get('year', '')}-{end_dt.get('month', '01')}" if end_dt else "Present"
175
176 if company_name or title:
177 experiences.append({
178 "company": company_name,
179 "title": title,
180 "start_date": start_str,
181 "end_date": end_str,
182 "is_current": end_str == "Present"
183 })
184 except Exception:
185 pass
186
187
188 if not experiences:
189 exp_section = soup.find("section", id=re.compile(r"experience", re.I)) or soup.find("div", class_=re.compile(r"experience", re.I))
190 if exp_section:
191 items = exp_section.find_all(["li", "div"], class_=re.compile(r"position|item|experience", re.I))
192 for item in items:
193 txt = item.get_text(separator=" ").strip()
194 dates = re.findall(r'\b(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)?\s?\d{4}\s?-\s?(?:Present|(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)?\s?\d{4})', txt, re.I)
195 if dates:
196 experiences.append({
197 "raw": txt[:150],
198 "start_date": dates[0] if isinstance(dates[0], str) else "2020",
199 "end_date": "Present" if "Present" in txt else "2023",
200 "is_current": "Present" in txt
201 })
202
203 result["experiences"] = experiences
204
205
206 if experiences and not result["current_employer"]:
207 current_exp = next((e for e in experiences if e.get("is_current")), experiences[0])
208 result["current_employer"] = current_exp.get("company")
209 result["current_role_title"] = current_exp.get("title")
210
211
212 activity_url = f"{clean_url}/recent-activity/all/"
213 try:
214
215 await asyncio.sleep(random.uniform(1.5, 3.0))
216 act_resp = await client.get(activity_url)
217 if act_resp.status_code == 200:
218 check_captcha_or_checkpoint(act_resp)
219 act_soup = BeautifulSoup(act_resp.text, "lxml")
220 page_text = act_soup.get_text(separator=" ").lower()
221
222 for kw in HIRING_KEYWORDS:
223 if kw in page_text:
224 result["is_actively_hiring"] = True
225 result["recent_activity_posts"].append(f"Detected keyword '{kw}' in recent activity feed.")
226 break
227 except CaptchaError:
228 raise
229 except Exception as e:
230 logger.warning(f"Could not fetch activity feed: {e}")
231
232
233 if not result["is_actively_hiring"]:
234 main_text = soup.get_text(separator=" ").lower()
235 if "hiring" in main_text and ("team" in main_text or "role" in main_text):
236 result["is_actively_hiring"] = True
237 result["recent_activity_posts"].append("Detected hiring keywords in profile headline/about section.")
238
239 except (CaptchaError, ProfileNotFoundError):
240 raise
241 except Exception as exc:
242 logger.error(f"Error scraping profile {clean_url}: {exc}")
243 raise RuntimeError(f"LinkedIn scraping error for {clean_url}: {exc}") from exc
244
245 return result
