1"""LinkedIn Company Employees Scraper — Multi-mode actor with enrichment and confidence scoring."""
2import asyncio
3import datetime
4import hashlib
5import json
6import re
7from typing import Any
8
9import httpx
10from apify import Actor
11from playwright.async_api import async_playwright
12
13EMAIL_REGEX = re.compile(r'[a-zA-Z0-9._-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}')
14SOCIAL_REGEX = re.compile(r'(?:https?:\/\/)?(?:www\.)?(facebook|instagram|twitter|linkedin|youtube|tiktok)\.com\/[a-zA-Z0-9._-]+')
15
16
17def now_iso() -> str:
18 return datetime.datetime.utcnow().isoformat() + 'Z'
19
20
21def make_cache_key(actor_id: str, mode: str, query: str) -> str:
22 raw = f'{actor_id}:{mode}:{query.lower().strip()}'
23 return hashlib.md5(raw.encode()).hexdigest()
24
25
26async def check_cache(key: str) -> list | None:
27 try:
28 store = await Actor.open_key_value_store()
29 entry = await store.get_value(key)
30 if entry and isinstance(entry, dict):
31 expires = entry.get('expiresAt')
32 if expires and datetime.datetime.fromisoformat(expires.replace('Z', '')) > datetime.datetime.utcnow():
33 return entry.get('value')
34 except Exception:
35 pass
36 return None
37
38
39async def set_cache(key: str, value: list, ttl_seconds: int = 86400) -> None:
40 try:
41 store = await Actor.open_key_value_store()
42 await store.set_value(key, {
43 'value': value,
44 'createdAt': now_iso(),
45 'expiresAt': (datetime.datetime.utcnow() + datetime.timedelta(seconds=ttl_seconds)).isoformat() + 'Z',
46 })
47 except Exception:
48 pass
49
50
51async def extract_contacts_from_website(url: str, timeout: float = 8.0) -> dict:
52 result: dict[str, Any] = {'emails': [], 'phones': [], 'socialMedia': [], 'techStack': []}
53 try:
54 async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
55 response = await client.get(url, headers={
56 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
57 })
58 text = response.text or ''
59 result['emails'] = list(set(EMAIL_REGEX.findall(text)))[:20]
60 result['socialMedia'] = list(set(SOCIAL_REGEX.findall(text)))[:10]
61 except Exception:
62 pass
63 return result
64
65
66def score_confidence(result: dict) -> float:
67 score = 0.3
68 if result.get('emails'): score += 0.3
69 if result.get('phones'): score += 0.15
70 if result.get('url') or result.get('website'): score += 0.15
71 if result.get('techStack'): score += 0.1
72 return min(1.0, round(score, 2))
73
74
75async def _google_search_linkedin_profiles(query: str, max_results: int) -> list[str]:
76 """Find LinkedIn profile URLs for people at a company via Google search (no API key).
77
78 Uses Google's site: operator with a company query. No-login, httpx-only.
79 Returns a list of linkedin.com/in/ URLs.
80 """
81
82
83
84 company = query.strip()
85 m = re.search(r'linkedin\.com/company/([^/?#\s]+)', company, re.IGNORECASE)
86 if m:
87 slug = m.group(1).strip()
88
89 company = ' '.join(part.capitalize() for part in re.split(r'[-_]+', slug) if part)
90 else:
91
92 company = re.sub(r'\b(employees|people|who work|working)\b.*', '', company, flags=re.IGNORECASE).strip()
93 company = re.sub(r'[^a-zA-Z0-9 ]+', ' ', company).strip() or query.strip()
94 q = f'site:linkedin.com/in "{company}"'
95 urls: list[str] = []
96 seen: set[str] = set()
97 try:
98 async with httpx.AsyncClient(timeout=12.0, follow_redirects=True) as client:
99 resp = await client.get('https://www.google.com/search', params={'q': q, 'num': min(max(max_results * 3, 20), 50)}, headers={
100 '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',
101 'Accept-Language': 'en-US,en;q=0.9',
102 })
103 text = resp.text or ''
104
105 for m in re.finditer(r'linkedin\.com/in/([a-zA-Z0-9_\-%\-]+)', text):
106 slug = m.group(1).rstrip('-').rstrip('/')
107 if slug and slug not in seen and not slug.lower().startswith('company'):
108 seen.add(slug)
109 urls.append(f'https://www.linkedin.com/in/{slug}')
110 except Exception as e:
111 Actor.log.warning(f'Google search for company profiles failed: {e}')
112 return urls[:max_results]
113
114
115async def _fetch_linkedin_profile_mobile(profile_url: str) -> dict:
116 """Fetch a public LinkedIn profile via mobile HTML (no login). Returns structured fields."""
117 profile: dict[str, Any] = {'url': profile_url, 'name': '', 'headline': '', 'location': '', 'company': '', 'profilePic': ''}
118 try:
119 async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
120 resp = await client.get(profile_url, headers={
121 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1',
122 'Accept-Language': 'en-US,en;q=0.9',
123 })
124 text = resp.text or ''
125 m = re.search(r'<meta property="og:title" content="([^"]+)"', text)
126 if m: profile['name'] = m.group(1).strip()
127 m = re.search(r'<meta property="og:description" content="([^"]+)"', text)
128 if m: profile['headline'] = m.group(1).strip()
129 m = re.search(r'<meta property="og:image" content="([^"]+)"', text)
130 if m: profile['profilePic'] = m.group(1).strip()
131
132 if profile['headline']:
133 cm = re.search(r'(?:at|@)\s+([A-Z][^,|•\-]{2,40})', profile['headline'])
134 if cm: profile['company'] = cm.group(1).strip()
135
136 m = re.search(r'<script type="application/ld\+json"[^>]*>(.*?)</script>', text, re.DOTALL)
137 if m:
138 try:
139 j = json.loads(m.group(1))
140 if isinstance(j, dict):
141 if j.get('name') and not profile['name']: profile['name'] = j['name']
142 if j.get('jobTitle'): profile['headline'] = j['jobTitle'] if isinstance(j['jobTitle'], str) else profile['headline']
143 if j.get('worksFor'): profile['company'] = (j['worksFor'].get('name') if isinstance(j['worksFor'], dict) else j['worksFor']) or profile['company']
144 if j.get('address'): profile['location'] = (j['address'].get('addressLocality') if isinstance(j['address'], dict) else j['address']) or profile['location']
145 except Exception:
146 pass
147 emails = EMAIL_REGEX.findall(text)
148 if emails: profile['emails'] = list(set(emails))[:5]
149 except Exception as e:
150 profile['error'] = str(e)[:200]
151 return profile
152
153
154def _parse_cookies(cookie_str: str, domain: str = '.linkedin.com') -> list[dict]:
155 """Parse comma-separated name=value pairs into Playwright cookie dicts."""
156 cookies: list[dict] = []
157 for pair in cookie_str.split(','):
158 pair = pair.strip()
159 if not pair or '=' not in pair:
160 continue
161 name, _, value = pair.partition('=')
162 name = name.strip()
163 value = value.strip()
164 if name and value:
165 cookies.append({'name': name, 'value': value, 'domain': domain, 'path': '/'})
166 return cookies
167
168
169def _extract_company_slug(query: str) -> str | None:
170 """Extract the LinkedIn company slug from a URL or query string."""
171 m = re.search(r'linkedin\.com/company/([^/?#\s]+)', query, re.IGNORECASE)
172 return m.group(1).strip() if m else None
173
174
175async def _scrape_linkedin_people_page(query: str, session_cookies: str, max_results: int) -> list[dict]:
176 """Navigate to a company's /people/ page with session cookies and extract employee profiles."""
177 slug = _extract_company_slug(query)
178 if not slug:
179 Actor.log.warning(f'Could not extract company slug from query: {query}')
180 return []
181 cookies = _parse_cookies(session_cookies, domain='.linkedin.com')
182 if not cookies:
183 Actor.log.warning('No valid cookies parsed from sessionCookies input')
184 return []
185 people_url = f'https://www.linkedin.com/company/{slug}/people/'
186 results: list[dict] = []
187 async with async_playwright() as p:
188 browser = await p.chromium.launch(headless=True)
189 try:
190 context = await browser.new_context(
191 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',
192 viewport={'width': 1366, 'height': 768},
193 )
194 await context.add_cookies(cookies)
195 page = await context.new_page()
196 Actor.log.info(f'Navigating to company people page: {people_url}')
197 await page.goto(people_url, wait_until='domcontentloaded', timeout=30000)
198 await page.wait_for_timeout(3000)
199
200 for _ in range(3):
201 await page.evaluate('window.scrollBy(0, document.body.scrollHeight)')
202 await page.wait_for_timeout(1500)
203
204 profile_hrefs = await page.eval_on_selector_all(
205 'a[href*="/in/"]',
206 '''els => els.map(e => e.href).filter(h => h.includes('linkedin.com/in/'))''',
207 )
208 seen: set[str] = set()
209 profile_urls: list[str] = []
210 for href in profile_hrefs:
211
212 clean = href.split('?')[0].rstrip('/')
213 if clean not in seen:
214 seen.add(clean)
215 profile_urls.append(clean)
216 profile_urls = profile_urls[:max_results]
217 Actor.log.info(f'Found {len(profile_urls)} employee profile URLs on /people/ page')
218
219 sem = asyncio.Semaphore(4)
220
221 async def _fetch_one(url: str) -> dict:
222 async with sem:
223 profile_page = await context.new_page()
224 try:
225 await profile_page.goto(url, wait_until='domcontentloaded', timeout=20000)
226 await profile_page.wait_for_timeout(2000)
227 content = await profile_page.content()
228 except Exception as e:
229 Actor.log.warning(f'Failed to fetch profile {url}: {e}')
230 return {'url': url, 'name': '', 'headline': '', 'location': '', 'company': '', 'profilePic': ''}
231 finally:
232 await profile_page.close()
233 profile: dict[str, Any] = {'url': url, 'name': '', 'headline': '', 'location': '', 'company': '', 'profilePic': ''}
234 m = re.search(r'<meta property="og:title" content="([^"]+)"', content)
235 if m:
236 profile['name'] = m.group(1).strip()
237 m = re.search(r'<meta property="og:description" content="([^"]+)"', content)
238 if m:
239 profile['headline'] = m.group(1).strip()
240 m = re.search(r'<meta property="og:image" content="([^"]+)"', content)
241 if m:
242 profile['profilePic'] = m.group(1).strip()
243 if profile['headline']:
244 cm = re.search(r'(?:at|@)\s+([A-Z][^,|•\-]{2,40})', profile['headline'])
245 if cm:
246 profile['company'] = cm.group(1).strip()
247 m = re.search(r'<script type="application/ld\+json"[^>]*>(.*?)</script>', content, re.DOTALL)
248 if m:
249 try:
250 j = json.loads(m.group(1))
251 if isinstance(j, dict):
252 if j.get('name') and not profile['name']:
253 profile['name'] = j['name']
254 if j.get('jobTitle') and isinstance(j['jobTitle'], str):
255 profile['headline'] = j['jobTitle']
256 if j.get('worksFor'):
257 profile['company'] = (j['worksFor'].get('name') if isinstance(j['worksFor'], dict) else j['worksFor']) or profile['company']
258 if j.get('address'):
259 profile['location'] = (j['address'].get('addressLocality') if isinstance(j['address'], dict) else j['address']) or profile['location']
260 except Exception:
261 pass
262 emails = EMAIL_REGEX.findall(content)
263 if emails:
264 profile['emails'] = list(set(emails))[:5]
265 return profile
266
267 profiles = await asyncio.gather(*[_fetch_one(u) for u in profile_urls], return_exceptions=True)
268 for p in profiles:
269 if isinstance(p, BaseException) or not isinstance(p, dict) or not p.get('name'):
270 continue
271 results.append({
272 'query': query,
273 'name': p.get('name', ''),
274 'url': p.get('url', ''),
275 'headline': p.get('headline', ''),
276 'location': p.get('location', ''),
277 'company': p.get('company', ''),
278 'emails': p.get('emails', []),
279 'phones': [],
280 'socialMedia': ['linkedin'],
281 'techStack': [],
282 'metrics': {'profilePic': p.get('profilePic', '')},
283 'confidenceScore': 0.7 if p.get('name') else 0.3,
284 'sources': ['linkedin_people_page', 'linkedin_profile'],
285 'extractedAt': now_iso(),
286 })
287 finally:
288 await browser.close()
289 return results[:max_results]
290
291
292async def fetch_primary(query: str, max_results: int, session_cookies: str = '') -> list[dict]:
293 """Primary data fetch — discovers LinkedIn employees for a company query.
294
295 When session_cookies is provided, uses Playwright with an authenticated LinkedIn
296 session to navigate the company's /people/ page and extract employee profiles.
297 When no cookies are provided, falls back to Google site:linkedin.com/in discovery
298 (which often returns 0 results due to LinkedIn's auth wall).
299 """
300 try:
301 max_results = max(1, min(int(max_results or 10), 50))
302 except (TypeError, ValueError):
303 max_results = 10
304 if session_cookies and session_cookies.strip():
305 Actor.log.info('Using session-cookie authed Playwright path for LinkedIn /people/ page')
306 cookie_results = await _scrape_linkedin_people_page(query, session_cookies, max_results)
307 if cookie_results:
308 return cookie_results
309 Actor.log.warning('Cookie-authed /people/ page yielded no results; falling back to Google search')
310 else:
311 Actor.log.warning('No li_at session cookie provided — falling back to Google search which often returns 0 results. Provide a li_at cookie for reliable employee discovery.')
312 profile_urls = await _google_search_linkedin_profiles(query, max_results)
313 if not profile_urls:
314 Actor.log.warning(f'No LinkedIn profiles found for query: {query}')
315 return []
316
317 sem = asyncio.Semaphore(4)
318 async def _one(u):
319 async with sem:
320 return await _fetch_linkedin_profile_mobile(u)
321 profiles = await asyncio.gather(*[_one(u) for u in profile_urls], return_exceptions=True)
322 results: list[dict] = []
323 for p in profiles:
324 if isinstance(p, BaseException) or not isinstance(p, dict) or not p.get('name'):
325 continue
326 results.append({
327 'query': query,
328 'name': p.get('name', ''),
329 'url': p.get('url', ''),
330 'headline': p.get('headline', ''),
331 'location': p.get('location', ''),
332 'company': p.get('company', ''),
333 'emails': p.get('emails', []),
334 'phones': [],
335 'socialMedia': ['linkedin'],
336 'techStack': [],
337 'metrics': {'profilePic': p.get('profilePic', '')},
338 'confidenceScore': 0.7 if p.get('name') else 0.3,
339 'sources': ['google_search', 'linkedin_mobile'],
340 'extractedAt': now_iso(),
341 })
342 return results[:max_results]
343
344
345async def run_fast_lookup(payload: dict) -> list[dict]:
346 queries = payload.get('queries', [])
347 session_cookies = payload.get('sessionCookies', '')
348 try:
349 max_results = int(payload.get('maxResults', 25))
350 except (TypeError, ValueError):
351 max_results = 25
352 results: list[dict] = []
353 for query in queries:
354 cache_key = make_cache_key('linkedin-company-employees', 'fast_lookup', query)
355 cached = await check_cache(cache_key)
356 if cached:
357 results.extend(cached)
358 continue
359 businesses = await fetch_primary(query, max_results, session_cookies)
360 base_results = [{
361 'query': query, 'name': b.get('name', ''), 'url': b.get('url', ''),
362 'emails': [], 'phones': [], 'socialMedia': [], 'techStack': [],
363 'metrics': b.get('metrics', {}), 'confidenceScore': 0.3,
364 'sources': ['primary'], 'cacheStatus': 'miss', 'mode': 'fast_lookup',
365 'extractedAt': now_iso(),
366 } for b in businesses]
367 await set_cache(cache_key, base_results)
368 results.extend(base_results)
369 return results
370
371
372async def run_enrich(payload: dict) -> list[dict]:
373 queries = payload.get('queries', [])
374 session_cookies = payload.get('sessionCookies', '')
375 try:
376 max_results = int(payload.get('maxResults', 25))
377 except (TypeError, ValueError):
378 max_results = 25
379 results: list[dict] = []
380 for query in queries:
381 businesses = await fetch_primary(query, max_results, session_cookies)
382 for biz in businesses[:max_results]:
383 base: dict[str, Any] = {
384 'query': query, 'name': biz.get('name', ''), 'url': biz.get('url', ''),
385 'emails': [], 'phones': [], 'socialMedia': [], 'techStack': [],
386 'metrics': biz.get('metrics', {}), 'confidenceScore': 0.3,
387 'sources': ['primary'], 'cacheStatus': 'miss', 'mode': 'enrich',
388 'extractedAt': now_iso(),
389 }
390 if biz.get('url'):
391 try:
392 contact = await asyncio.wait_for(extract_contacts_from_website(biz['url']), timeout=8.0)
393 base['emails'] = contact['emails']
394 base['socialMedia'] = contact['socialMedia']
395 base['sources'].append('website_scraper')
396 base['confidenceScore'] = score_confidence(base)
397 except (asyncio.TimeoutError, httpx.RequestError, httpx.HTTPError):
398 pass
399 await Actor.push_data(base)
400 results.append(base)
401 return results
402
403
404async def run_batch(payload: dict) -> list[dict]:
405 queries = payload.get('queries', [])
406 session_cookies = payload.get('sessionCookies', '')
407 try:
408 max_results = int(payload.get('maxResults', 50))
409 except (TypeError, ValueError):
410 max_results = 50
411 results: list[dict] = []
412 for query in queries:
413 try:
414 batch_results = await run_enrich({'queries': [query], 'maxResults': max_results, 'sessionCookies': session_cookies})
415 results.extend(batch_results)
416 except Exception as e:
417 Actor.log.error(f'Batch item failed: {query} — {e}')
418 return results
419
420
421async def main() -> None:
422 async with Actor:
423 actor_input = await Actor.get_input() or {}
424 payload = actor_input.get('input', actor_input)
425 mode = payload.get('mode', 'enrich')
426 Actor.log.info(f'Starting in mode: {mode}')
427 if mode == 'fast_lookup':
428 results = await run_fast_lookup(payload)
429 elif mode == 'enrich':
430 results = await run_enrich(payload)
431 elif mode == 'batch':
432 results = await run_batch(payload)
433 else:
434 Actor.log.error(f'Unknown mode: {mode}')
435 return
436 if mode == 'fast_lookup' and results:
437 await Actor.push_data(results)
438 Actor.log.info(f'Extraction complete. Total results: {len(results)}')
439 webhook_url = payload.get('webhookUrl')
440 if webhook_url and results:
441 try:
442 async with httpx.AsyncClient(timeout=10.0) as client:
443 await client.post(webhook_url, json={'status': 'completed', 'mode': mode, 'resultCount': len(results), 'timestamp': now_iso()})
444 except Exception:
445 pass
446
447
448if __name__ == "__main__":
449 asyncio.run(main())