LinkedIn Job Scraper

import os
import time
from datetime import datetime
from dotenv import load_dotenv
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from bs4 import BeautifulSoup
import telegram

# Load environment variables
load_dotenv()

# Configuration
LINKEDIN_EMAIL = os.getenv('LINKEDIN_EMAIL')
LINKEDIN_PASSWORD = os.getenv('LINKEDIN_PASSWORD')
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')
TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')
WEBDRIVER_PATH = os.getenv('WEBDRIVER_PATH', 'chromedriver')

# Job search parameters
JOB_KEYWORDS = "python developer"
JOB_LOCATION = "United States"
EXPERIENCE_LEVEL = "Entry level"  # Can be "Internship", "Entry level", "Associate", etc.

# Initialize Telegram bot
bot = telegram.Bot(token=TELEGRAM_TOKEN)

# Track already seen jobs to avoid duplicates
seen_jobs = set()

def init_driver():
    """Initialize and return a Chrome WebDriver with options."""
    chrome_options = Options()
    chrome_options.add_argument("--headless")  # Run in headless mode
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--disable-dev-shm-usage")
    return webdriver.Chrome(executable_path=WEBDRIVER_PATH, options=chrome_options)

def login_to_linkedin(driver):
    """Log in to LinkedIn."""
    driver.get("https://www.linkedin.com/login")
    time.sleep(2)
    
    # Fill in login form
    email_field = driver.find_element(By.ID, "username")
    email_field.send_keys(LINKEDIN_EMAIL)
    
    password_field = driver.find_element(By.ID, "password")
    password_field.send_keys(LINKEDIN_PASSWORD)
    
    # Click login button
    login_button = driver.find_element(By.XPATH, "//button[@type='submit']")
    login_button.click()
    time.sleep(3)

def construct_job_search_url(keywords, location, experience):
    """Construct LinkedIn job search URL based on parameters."""
    base_url = "https://www.linkedin.com/jobs/search/"
    params = {
        "keywords": keywords.replace(" ", "%20"),
        "location": location.replace(" ", "%20"),
        "f_E": "2" if experience == "Entry level" else "1",  # Experience level filter
        "sortBy": "DD",  # Sort by most recent
        "f_TPR": "r86400",  # Past 24 hours
    }
    query_string = "&".join(f"{k}={v}" for k, v in params.items())
    return f"{base_url}?{query_string}"

def scrape_jobs(driver):
    """Scrape job listings from LinkedIn."""
    url = construct_job_search_url(JOB_KEYWORDS, JOB_LOCATION, EXPERIENCE_LEVEL)
    driver.get(url)
    time.sleep(3)  # Wait for page to load
    
    # Scroll to load more jobs (adjust based on needs)
    for _ in range(2):
        driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
        time.sleep(1)
    
    # Parse page with BeautifulSoup
    soup = BeautifulSoup(driver.page_source, "html.parser")
    jobs = soup.find_all("li", class_="jobs-search-results__list-item")
    
    new_jobs = []
    for job in jobs:
        try:
            title = job.find("h3", class_="base-search-card__title").get_text(strip=True)
            company = job.find("h4", class_="base-search-card__subtitle").get_text(strip=True)
            location = job.find("span", class_="job-search-card__location").get_text(strip=True)
            posted = job.find("time", class_="job-search-card__listdate") or \
                     job.find("time", class_="job-search-card__listdate--new")
            posted = posted.get_text(strip=True) if posted else "Recently"
            
            job_url = job.find("a", class_="base-card__full-link")["href"].split("?")[0]
            
            job_id = hash(f"{title}{company}{location}")
            if job_id not in seen_jobs:
                seen_jobs.add(job_id)
                new_jobs.append({
                    "title": title,
                    "company": company,
                    "location": location,
                    "posted": posted,
                    "url": job_url
                })
        except Exception as e:
            print(f"Error parsing job: {e}")
            continue
    
    return new_jobs

def send_telegram_notification(job):
    """Send job details to Telegram."""
    message = (
        f"🚀 New Job Alert! 🚀\n\n"
        f"Title: {job['title']}\n"
        f"Company: {job['company']}\n"
        f"Location: {job['location']}\n"
        f"Posted: {job['posted']}\n\n"
        f"View Job on LinkedIn"
    )
    
    try:
        bot.send_message(
            chat_id=TELEGRAM_CHAT_ID,
            text=message,
            parse_mode="HTML",
            disable_web_page_preview=False
        )
    except Exception as e:
        print(f"Failed to send Telegram notification: {e}")

def main():
    """Main function to run the job scraper."""
    driver = init_driver()
    
    try:
        # Login to LinkedIn
        login_to_linkedin(driver)
        
        # Scrape jobs and send notifications
        while True:
            print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Checking for new jobs...")
            new_jobs = scrape_jobs(driver)
            
            if new_jobs:
                print(f"Found {len(new_jobs)} new jobs. Sending notifications...")
                for job in new_jobs:
                    send_telegram_notification(job)
                    time.sleep(1)  # Avoid rate limiting
            else:
                print("No new jobs found.")
            
            # Wait before next check (e.g., every 6 hours)
            time.sleep(6 * 60 * 60)
            
    except KeyboardInterrupt:
        print("\nScript stopped by user.")
    finally:
        driver.quit()

if __name__ == "__main__":
    main()