""" LinkedIn Job Scraper Module This module provides functionality to scrape job listings from LinkedIn, including job titles, company names, links, and full job descriptions. """ import time import logging import os import shutil from datetime import datetime, timedelta from urllib.parse import quote_plus import pandas as pd import requests import undetected_chromedriver as uc from bs4 import BeautifulSoup from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, NoSuchElementException # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) HTTP_TIMEOUT_SECONDS = 20 class LinkedInScraper: """ A class to scrape job listings from LinkedIn. """ def __init__(self, headless=True): """ Initialize the LinkedIn scraper. Args: headless (bool): Whether to run the browser in headless mode. """ self.headless = headless runtime_root = os.environ.get("JOB_APPLY_AI_DATA_DIR", os.path.join(os.getcwd(), ".runtime")) self.runtime_root = os.path.join(runtime_root, "scraper") self.uc_data_dir = os.path.join(self.runtime_root, "uc_driver") self.chrome_profile_root = os.path.join(self.runtime_root, "chrome_profiles") os.makedirs(self.uc_data_dir, exist_ok=True) os.makedirs(self.chrome_profile_root, exist_ok=True) def _configure_driver(self): """ Configure and return a Chrome WebDriver. Returns: WebDriver: Configured Chrome WebDriver instance. """ options = webdriver.ChromeOptions() if self.headless: options.add_argument("--headless=new") options.add_argument("--disable-dev-shm-usage") options.add_argument("--disable-notifications") options.add_argument("--disable-extensions") options.add_argument("--disable-background-networking") options.add_argument("--no-first-run") options.add_argument("--no-default-browser-check") options.add_argument("--window-size=1920,1080") options.add_argument("--remote-debugging-port=0") # GPU settings from environment disable_gpu = os.environ.get("CHROME_DISABLE_GPU", "1") == "1" if disable_gpu: options.add_argument("--disable-gpu") options.add_argument("--disable-software-rasterizer") disable_sandbox = os.environ.get("CHROME_DISABLE_SANDBOX", "1") == "1" if disable_sandbox: options.add_argument("--no-sandbox") # Add user agent to avoid detection options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36") options.add_argument("--disable-features=RendererCodeIntegrity") # Chrome binary path from environment chrome_binary = os.environ.get("CHROME_BINARY_PATH") if chrome_binary and os.path.exists(chrome_binary): options.binary_location = chrome_binary logger.info(f"Using Chrome binary: {chrome_binary}") # Keep undetected-chromedriver artifacts local to the project. uc.patcher.Patcher.data_path = self.uc_data_dir chrome_version_main = os.environ.get("UC_CHROME_VERSION_MAIN") session_profile_dir = os.path.join( self.chrome_profile_root, f"profile_{int(time.time())}_{os.getpid()}" ) os.makedirs(session_profile_dir, exist_ok=True) chrome_args = { "options": options, "user_data_dir": session_profile_dir, "use_subprocess": False, } if chrome_version_main and chrome_version_main.isdigit(): chrome_args["version_main"] = int(chrome_version_main) if chrome_binary and os.path.exists(chrome_binary): chrome_args["browser_executable_path"] = chrome_binary try: driver = uc.Chrome(**chrome_args) # Save profile dir for cleanup after driver.quit(). driver._job_apply_profile_dir = session_profile_dir return driver except Exception as e: logger.warning(f"undetected-chromedriver failed, trying Selenium fallback: {str(e)}") fallback_options = webdriver.ChromeOptions() for argument in options.arguments: fallback_options.add_argument(argument) if options.binary_location: fallback_options.binary_location = options.binary_location fallback_options.add_argument(f"--user-data-dir={session_profile_dir}") try: fallback_driver_path = os.path.join(self.uc_data_dir, "undetected_chromedriver.exe") if os.path.exists(fallback_driver_path): service = Service(executable_path=fallback_driver_path) driver = webdriver.Chrome(service=service, options=fallback_options) else: driver = webdriver.Chrome(options=fallback_options) driver._job_apply_profile_dir = session_profile_dir logger.info("Selenium fallback driver initialized successfully") return driver except Exception as fallback_error: logger.error(f"Failed to create Chrome driver: {str(fallback_error)}") logger.error("Try setting CHROME_BINARY_PATH to your Chrome installation path") shutil.rmtree(session_profile_dir, ignore_errors=True) raise def _parse_days_ago(self, raw_text): """Parse LinkedIn relative time text into integer day count when possible.""" if not raw_text: return "Unknown" text = raw_text.strip().lower() if "today" in text or "just now" in text: return 0 if "hour" in text or "minute" in text: return 0 parts = text.split() try: value = int(parts[0]) except (ValueError, IndexError): return "Unknown" if "day" in text: return value if "week" in text: return value * 7 if "month" in text: return value * 30 return "Unknown" def _scrape_job_listings_http(self, keyword, location, max_jobs=10, max_days_old=14): """Fallback scraping using LinkedIn public guest endpoints (no browser).""" logger.info("Using HTTP fallback for LinkedIn job scraping") session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36" }) jobs = [] start = 0 while len(jobs) < max_jobs: url = ( "https://www.linkedin.com/jobs-guest/jobs/api/seeMoreJobPostings/search" f"?keywords={quote_plus(keyword)}&location={quote_plus(location)}&start={start}" ) response = session.get(url, timeout=HTTP_TIMEOUT_SECONDS) response.raise_for_status() soup = BeautifulSoup(response.text, "lxml") cards = soup.select("li") if not cards: break added_this_page = 0 for card in cards: if len(jobs) >= max_jobs: break title_elem = card.select_one("h3.base-search-card__title") company_elem = card.select_one("h4.base-search-card__subtitle") link_elem = card.select_one("a.base-card__full-link") time_elem = card.select_one("time") if not title_elem or not company_elem or not link_elem: continue title = title_elem.get_text(" ", strip=True) company = company_elem.get_text(" ", strip=True) link = (link_elem.get("href") or "").strip() days_ago = self._parse_days_ago(time_elem.get_text(" ", strip=True) if time_elem else "") if isinstance(days_ago, int) and days_ago > max_days_old: continue jobs.append({ "title": title, "company": company, "link": link, "source": "LinkedIn", "posted_days_ago": days_ago }) added_this_page += 1 if added_this_page == 0: break start += 25 logger.info(f"HTTP fallback scraped {len(jobs)} job listings") return jobs def _fetch_job_description_http(self, job_url): """Fallback description fetch using HTTP requests only.""" session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36" }) response = session.get(job_url, timeout=HTTP_TIMEOUT_SECONDS) response.raise_for_status() soup = BeautifulSoup(response.text, "lxml") title_elem = soup.select_one("h1.top-card-layout__title") or soup.select_one("h1.topcard__title") company_elem = soup.select_one("a.topcard__org-name-link") or soup.select_one("span.topcard__flavor") desc_elem = soup.select_one("div.show-more-less-html__markup") or soup.select_one("div.description__text") job_title = title_elem.get_text(" ", strip=True) if title_elem else "" company_name = company_elem.get_text(" ", strip=True) if company_elem else "" job_description = desc_elem.get_text("\n", strip=True) if desc_elem else "" return job_title, company_name, job_description def scrape_job_listings(self, keyword, location, max_jobs=10, max_days_old=14): """ Scrape job listings from LinkedIn based on keyword and location. Args: keyword (str): Job title or keyword to search for. location (str): Location to search in. max_jobs (int): Maximum number of jobs to scrape. max_days_old (int): Maximum age of job postings in days. Returns: list: List of dictionaries containing job details. """ logger.info(f"Scraping LinkedIn jobs for '{keyword}' in '{location}'") use_browser = os.environ.get("LINKEDIN_USE_BROWSER", "1") == "1" if not use_browser: logger.info("LINKEDIN_USE_BROWSER=0, using HTTP fallback directly") try: jobs = self._scrape_job_listings_http(keyword, location, max_jobs=max_jobs, max_days_old=max_days_old) if jobs: return jobs logger.info("HTTP fallback returned no jobs, retrying with browser mode") except Exception as http_error: logger.error(f"HTTP fallback failed: {http_error}") jobs = [] try: driver = self._configure_driver() except Exception as browser_error: logger.error(f"Browser retry unavailable: {browser_error}") return jobs try: search_url = f"https://www.linkedin.com/jobs/search?keywords={keyword.replace(' ', '%20')}&location={location.replace(' ', '%20')}" driver.get(search_url) for _ in range(3): driver.execute_script("window.scrollBy(0, 800);") time.sleep(2) wait = WebDriverWait(driver, 15) try: wait.until(EC.presence_of_element_located((By.CLASS_NAME, "base-card"))) except TimeoutException: logger.warning("Browser retry found no job listings") return jobs jobs = [] today = datetime.today() job_elements = driver.find_elements(By.CLASS_NAME, "base-card") for job in job_elements[:max_jobs]: try: title = job.find_element(By.CSS_SELECTOR, "h3").text.strip() company = job.find_element(By.CSS_SELECTOR, "h4").text.strip() link = job.find_element(By.TAG_NAME, "a").get_attribute("href") try: date_element = job.find_element(By.CSS_SELECTOR, "time") posted_time = date_element.get_attribute("datetime") if posted_time: posted_date = datetime.strptime(posted_time[:10], "%Y-%m-%d") days_ago = (today - posted_date).days if days_ago > max_days_old: logger.info(f"Skipping job: {title} (Posted {days_ago} days ago)") continue else: days_ago = "Unknown" except NoSuchElementException: logger.warning(f"Could not find post time for: {title}, assuming it's recent") days_ago = "Unknown" jobs.append({ "title": title, "company": company, "link": link, "source": "LinkedIn", "posted_days_ago": days_ago }) except Exception as e: logger.error(f"Error processing job listing: {str(e)}") continue logger.info(f"Browser retry scraped {len(jobs)} job listings") return jobs finally: if driver: profile_dir = getattr(driver, "_job_apply_profile_dir", None) driver.quit() if profile_dir: shutil.rmtree(profile_dir, ignore_errors=True) driver = None try: driver = self._configure_driver() except Exception as browser_error: logger.warning(f"Browser scraper unavailable, using HTTP fallback: {browser_error}") try: return self._scrape_job_listings_http(keyword, location, max_jobs=max_jobs, max_days_old=max_days_old) except Exception as http_error: logger.error(f"HTTP fallback failed: {http_error}") return [] search_url = f"https://www.linkedin.com/jobs/search?keywords={keyword.replace(' ', '%20')}&location={location.replace(' ', '%20')}" try: driver.get(search_url) # Scroll to load more jobs for _ in range(3): driver.execute_script("window.scrollBy(0, 800);") time.sleep(2) # Wait for job listings to appear wait = WebDriverWait(driver, 15) try: wait.until(EC.presence_of_element_located((By.CLASS_NAME, "base-card"))) except TimeoutException: logger.warning("No job listings found") driver.quit() return [] jobs = [] today = datetime.today() job_elements = driver.find_elements(By.CLASS_NAME, "base-card") for job in job_elements[:max_jobs]: try: title = job.find_element(By.CSS_SELECTOR, "h3").text.strip() company = job.find_element(By.CSS_SELECTOR, "h4").text.strip() link = job.find_element(By.TAG_NAME, "a").get_attribute("href") # Check job posting date try: date_element = job.find_element(By.CSS_SELECTOR, "time") posted_time = date_element.get_attribute("datetime") if posted_time: posted_date = datetime.strptime(posted_time[:10], "%Y-%m-%d") days_ago = (today - posted_date).days if days_ago > max_days_old: logger.info(f"Skipping job: {title} (Posted {days_ago} days ago)") continue else: days_ago = "Unknown" except NoSuchElementException: logger.warning(f"Could not find post time for: {title}, assuming it's recent") days_ago = "Unknown" jobs.append({ "title": title, "company": company, "link": link, "source": "LinkedIn", "posted_days_ago": days_ago }) except Exception as e: logger.error(f"Error processing job listing: {str(e)}") continue logger.info(f"Successfully scraped {len(jobs)} job listings") return jobs except Exception as e: logger.error(f"Error during job scraping: {str(e)}") return [] finally: if driver: profile_dir = getattr(driver, "_job_apply_profile_dir", None) driver.quit() if profile_dir: shutil.rmtree(profile_dir, ignore_errors=True) def fetch_job_description(self, job_url): """ Fetch the full job description from a LinkedIn job URL. Args: job_url (str): URL of the LinkedIn job posting. Returns: tuple: (job_title, company_name, job_description) """ logger.info(f"Fetching job description from {job_url}") use_browser = os.environ.get("LINKEDIN_USE_BROWSER", "1") == "1" if not use_browser: logger.info("LINKEDIN_USE_BROWSER=0, fetching description via HTTP") try: job_title, company_name, job_description = self._fetch_job_description_http(job_url) if job_description: return job_title, company_name, job_description logger.info("HTTP description fetch returned empty content, retrying with browser mode") except Exception as http_error: logger.error(f"HTTP description fallback failed: {http_error}") return "", "", "" driver = None try: driver = self._configure_driver() except Exception as browser_error: logger.error(f"Browser description retry unavailable: {browser_error}") return "", "", "" try: driver.get(job_url) wait = WebDriverWait(driver, 15) try: title_elem = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "h1.topcard__title"))) job_title = title_elem.text.strip() except TimeoutException: job_title = "" try: company_elem = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "a.topcard__org-name-link, span.topcard__flavor"))) company_name = company_elem.text.strip() except TimeoutException: company_name = "" try: desc_elem = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div.show-more-less-html__markup, div.description__text"))) job_description = desc_elem.text.strip() except TimeoutException: job_description = "" return job_title, company_name, job_description finally: if driver: profile_dir = getattr(driver, "_job_apply_profile_dir", None) driver.quit() if profile_dir: shutil.rmtree(profile_dir, ignore_errors=True) driver = None try: driver = self._configure_driver() except Exception as browser_error: logger.warning(f"Browser description fetch unavailable, using HTTP fallback: {browser_error}") try: return self._fetch_job_description_http(job_url) except Exception as http_error: logger.error(f"HTTP description fallback failed: {http_error}") return "", "", "" try: driver.get(job_url) wait = WebDriverWait(driver, 15) # Job Title try: title_elem = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "h1.topcard__title"))) job_title = title_elem.text.strip() except TimeoutException: logger.warning("Could not find job title") job_title = "" # Company Name try: company_elem = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "a.topcard__org-name-link"))) company_name = company_elem.text.strip() except TimeoutException: logger.warning("Could not find company name") company_name = "" # Job Description try: desc_elem = wait.until(EC.presence_of_element_located((By.CLASS_NAME, "description__text"))) job_description = desc_elem.text.strip() except TimeoutException: logger.warning("Could not find job description") job_description = "" return job_title, company_name, job_description except Exception as e: logger.error(f"Error fetching job description: {str(e)}") return "", "", "" finally: if driver: profile_dir = getattr(driver, "_job_apply_profile_dir", None) driver.quit() if profile_dir: shutil.rmtree(profile_dir, ignore_errors=True) def save_jobs_to_excel(self, jobs, filename=None): """ Save scraped jobs to an Excel file. Args: jobs (list): List of job dictionaries. filename (str, optional): Output filename. If None, generates a filename with today's date. Returns: str: Path to the saved Excel file. """ if not jobs: logger.warning("No jobs to save") return None df = pd.DataFrame(jobs) if filename is None: today_date = datetime.today().strftime("%Y-%m-%d") filename = f"linkedin_jobs_{today_date}.xlsx" df.to_excel(filename, index=False) logger.info(f"Saved {len(jobs)} jobs to {filename}") return filename def main(): """ Main function to demonstrate the LinkedIn scraper. """ keyword = input("Enter job title (e.g., Software Engineer): ") location = input("Enter location (e.g., Remote, New York, Berlin): ") scraper = LinkedInScraper(headless=True) jobs = scraper.scrape_job_listings(keyword, location) if jobs: # Fetch full job descriptions for i, job in enumerate(jobs): logger.info(f"Fetching description for job {i+1}/{len(jobs)}: {job['title']}") title, company, description = scraper.fetch_job_description(job['link']) jobs[i]['description'] = description # Save to Excel filename = scraper.save_jobs_to_excel(jobs) print(f"\n✅ Jobs saved to {filename}") else: print("\n❌ No LinkedIn jobs found.") if __name__ == "__main__": main()