job_apply_ai/scraper/linkedin.py

"""
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()