import asyncio import random import sqlite3 import os from datetime import datetime from typing import Optional, Dict from playwright.async_api import async_playwright from browserforge.injectors.playwright import AsyncNewContext class LinkedInJobScraper: def __init__( self, engine, db_path: str = "linkedin_jobs.db", human_speed: float = 1.0 ): self.engine = engine self.db_path = db_path self.human_speed = human_speed self._init_db() def _init_db(self): os.makedirs(os.path.dirname(self.db_path) if os.path.dirname(self.db_path) else ".", exist_ok=True) with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS jobs ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT, title TEXT, company TEXT, location TEXT, salary TEXT, description TEXT, url TEXT UNIQUE, workplace_type TEXT, scraped_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() async def _human_click(self, page, element, wait_after: bool = True): if not element: return False await element.scroll_into_view_if_needed() await asyncio.sleep(random.uniform(0.3, 0.8) * self.human_speed) try: await element.click() if wait_after: await asyncio.sleep(random.uniform(2, 4) * self.human_speed) return True except: return False async def _login(self, page, credentials: Dict) -> bool: """Human-realistic LinkedIn login""" print("🔐 Navigating to LinkedIn login page...") await page.goto("https://www.linkedin.com/login", timeout=60000) await asyncio.sleep(random.uniform(2.0, 3.5) * self.human_speed) email_field = await page.query_selector('input[name="session_key"]') if not email_field: print("❌ Email field not found.") return False print("✍ī¸ Typing username...") await email_field.click() await asyncio.sleep(random.uniform(0.4, 0.9) * self.human_speed) for char in credentials["email"]: await page.keyboard.type(char) await asyncio.sleep(random.uniform(0.06, 0.14) * self.human_speed) await asyncio.sleep(random.uniform(1.0, 1.8) * self.human_speed) password_field = await page.query_selector('input[name="session_password"]') if not password_field: print("❌ Password field not found.") return False print("🔒 Typing password...") await password_field.click() await asyncio.sleep(random.uniform(0.3, 0.7) * self.human_speed) for char in credentials["password"]: await page.keyboard.type(char) await asyncio.sleep(random.uniform(0.08, 0.16) * self.human_speed) await asyncio.sleep(random.uniform(0.8, 1.5) * self.human_speed) print("✅ Submitting login form...") await page.keyboard.press("Enter") for _ in range(15): current_url = page.url if "/feed" in current_url or "/jobs" in current_url: if "login" not in current_url: print("✅ Login successful!") await asyncio.sleep(random.uniform(2.0, 3.0) * self.human_speed) return True await asyncio.sleep(1) print("❌ Login may have failed.") return False async def _extract_job_details(self, page) -> Dict: """Extract from ANY job page: LinkedIn Easy Apply OR external site""" await asyncio.sleep(2 * self.human_speed) async def get_text(selector: str) -> str: try: el = await page.query_selector(selector) if el: text = await el.inner_text() return text.strip() if text else "N/A" except: pass return "N/A" # Try multiple strategies for each field title = await get_text("h1.t-24") # LinkedIn if title == "N/A": title = await get_text("h1, h2") # External company = await get_text("a.app-aware-link[href*='/company/']") # LinkedIn if company == "N/A": company = await get_text("div.org, .company, [class*='company']") # External location = await get_text("span[class*='location']") # LinkedIn if location == "N/A": location = await get_text(".location, [class*='location']") description = await get_text("div[class*='description__text']") # LinkedIn if description == "N/A": description = await get_text(".job-desc, .description, main, body") # Workplace & salary — LinkedIn only (external may not have) workplace = await get_text("span.job-workplace-type") or "N/A" salary = await get_text("span.salary") or "N/A" return { "title": title, "company": company, "location": location, "workplace_type": workplace, "salary": salary, "description": description, "url": page.url } async def _save_to_markdown(self, job_data: Dict, keyword: str): os.makedirs("linkedin_jobs", exist_ok=True) clean_keyword = keyword.replace(" ", "_") filename = f"linkedin_{clean_keyword}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md" filepath = os.path.join("linkedin_jobs", filename) with open(filepath, "w", encoding="utf-8") as f: f.write(f"# {job_data['title']}\n\n") f.write(f"- **Company**: {job_data['company']}\n") f.write(f"- **Location**: {job_data['location']}\n") f.write(f"- **Workplace**: {job_data['workplace_type']}\n") f.write(f"- **Salary**: {job_data['salary']}\n") f.write(f"- **URL**: <{job_data['url']}>\n\n") f.write(f"## Description\n\n{job_data['description']}\n") async def _save_to_db(self, job_data: Dict, keyword: str): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' INSERT OR IGNORE INTO jobs (keyword, title, company, location, salary, description, url, workplace_type) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ''', ( keyword, job_data["title"], job_data["company"], job_data["location"], job_data["salary"], job_data["description"], job_data["url"], job_data["workplace_type"] )) conn.commit() async def scrape_jobs( self, search_keywords: str, max_pages: int = 1, credentials: Optional[Dict] = None ): encoded_keywords = search_keywords.replace(" ", "%20") search_url = f"https://www.linkedin.com/jobs/search/?keywords={encoded_keywords}" profile = self.engine._select_profile() renderer = random.choice(self.engine.common_renderers[self.engine.os]) vendor = random.choice(self.engine.common_vendors) spoof_script = self.engine._get_spoof_script(renderer, vendor) async with async_playwright() as pw: browser = await pw.chromium.launch( headless=False, args=['--disable-blink-features=AutomationControlled'] ) context = await AsyncNewContext(browser, fingerprint=profile) await context.add_init_script(f""" Object.defineProperty(navigator, 'hardwareConcurrency', {{ get: () => {profile.navigator.hardwareConcurrency} }}); Object.defineProperty(navigator, 'deviceMemory', {{ get: () => {profile.navigator.deviceMemory} }}); Object.defineProperty(navigator, 'platform', {{ get: () => '{profile.navigator.platform}' }}); """) await context.add_init_script(spoof_script) page = await context.new_page() if credentials: print("🔐 Attempting LinkedIn login...") if not await self._login(page, credentials): print("❌ Login failed. Exiting.") await browser.close() return else: print("ℹī¸ No credentials — proceeding as guest.") print(f"🔍 Searching for: {search_keywords}") await page.goto(search_url, wait_until='load', timeout=60000) await asyncio.sleep(random.uniform(4.0, 6.0) * self.human_speed) scraped_count = 0 all_job_links = [] # Collect job links for page_num in range(1, max_pages + 1): print(f"📄 Collecting job links from page {page_num}/{max_pages}") for _ in range(50): links = await page.query_selector_all("a[href*='/jobs/view/']") if links: for link in links: href = await link.get_attribute("href") if href and href not in all_job_links: all_job_links.append(href) break await asyncio.sleep(1) print(f" ➕ Found {len(links) if 'links' in locals() else 0} new job links.") if page_num < max_pages: next_btn = await page.query_selector("button[aria-label='Next']") if next_btn and await next_btn.is_enabled(): await self._human_click(page, next_btn) await asyncio.sleep(4 * self.human_speed) else: print("🔚 No next page.") break # Process each job for idx, href in enumerate(all_job_links): try: full_url = href if href.startswith("http") else f"https://www.linkedin.com{href}" print(f" → Opening job {idx+1}: {full_url}") await page.goto(full_url, wait_until='load', timeout=60000) await asyncio.sleep(3 * self.human_speed) if not await page.query_selector("h1.t-24"): print(f" ⚠ī¸ Invalid job page, skipping.") continue # Find and click the main "Apply" button apply_btn = None apply_selectors = [ "button[aria-label*='Apply']", "button:has-text('Apply')", "a:has-text('Apply')", "button:has-text('Easy Apply')" ] for selector in apply_selectors: apply_btn = await page.query_selector(selector) if apply_btn: break if not apply_btn: print(f" ⚠ī¸ No 'Apply' button found, skipping.") continue # Click "Apply" print(f" → Clicking 'Apply' / 'Easy Apply' button...") await self._human_click(page, apply_btn, wait_after=False) await asyncio.sleep(4 * self.human_speed) # Wait for next page/form to load # Now scrape WHATEVER page is displayed (Easy Apply form OR external site) job_data = await self._extract_job_details(page) if job_data["title"] == "N/A" and "linkedin.com" in page.url: # On LinkedIn but no title → likely Easy Apply form; use job ID as title job_id = full_url.split("/")[-2] if "/jobs/view/" in full_url else "unknown" job_data["title"] = f"Easy Apply Job - ID {job_id}" await self._save_to_db(job_data, search_keywords) await self._save_to_markdown(job_data, search_keywords) scraped_count += 1 domain = "LinkedIn (Easy Apply)" if "linkedin.com" in page.url else "External Site" print(f" ✅ Scraped ({domain}): {job_data['title'][:50]}...") except Exception as e: print(f" ⚠ī¸ Failed on job {idx+1}: {str(e)[:100]}") finally: # Return to search results print(" ↩ī¸ Returning to LinkedIn search results...") await page.goto(search_url, timeout=60000) await asyncio.sleep(4 * self.human_speed) await browser.close() print(f"✅ Completed! Scraped {scraped_count} job pages (internal + external) for '{search_keywords}'.")