pip install playwright beautifulsoup4 lxml pandas
playwright install

import asyncio
from playwright.async_api import async_playwright
from bs4 import BeautifulSoup
import pandas as pd
import re
from urllib.robotparser import RobotFileParser
from urllib.parse import urlparse
import time

class JobScraper:
    def __init__(self):
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15',
        ]

    async def scrape_url(self, url):
        """Asynchronously scrape a single URL with Playwright."""
        async with async_playwright() as p:
            # Randomize browser type and user agent for stealth
            browser = await p.chromium.launch(headless=True)
            context = await browser.new_context(user_agent=self.user_agents[0])
            page = await context.new_page()

            try:
                # Navigate and wait for network to be idle
                await page.goto(url, wait_until="networkidle", timeout=30000)
                # Wait for a content-specific selector if needed
                # await page.wait_for_selector('.job-description', timeout=10000)
                content = await page.content()
                await browser.close()
                return self._clean_html(content, url)
            except Exception as e:
                print(f"Error scraping {url}: {e}")
                await browser.close()
                return None

    def _clean_html(self, html, url):
        """Extract main text content, focusing on job description."""
        soup = BeautifulSoup(html, 'lxml')

        # Remove unnecessary tags
        for tag in soup(['script', 'style', 'nav', 'header', 'footer', 'iframe', 'aside']):
            tag.decompose()

        # Site-specific extraction logic (example for a generic site)
        # In production, you'd have specific selectors per target site
        job_content = soup.find('div', class_=re.compile(r'(job-description|description|content)'))
        if job_content:
            text = job_content.get_text(separator='\n', strip=True)
        else:
            # Fallback: get all text from body
            text = soup.body.get_text(separator='\n', strip=True) if soup.body else ""

        # Basic cleaning
        lines = (line.strip() for line in text.splitlines())
        chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
        text = '\n'.join(chunk for chunk in chunks if chunk)

        return {
            'url': url,
            'raw_html': html[:5000],  # Store snippet for debugging
            'clean_text': text[:10000],  # Limit text length for cost control
            'scraped_at': pd.Timestamp.now()
        }

# Usage example
async def main():
    scraper = JobScraper()
    urls = [
        'https://example-job-board.com/job/123',
        'https://another-board.org/careers/456'
    ]
    tasks = [scraper.scrape_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    valid_results = [r for r in results if r]
    df = pd.DataFrame(valid_results)
    df.to_csv('scraped_jobs_raw.csv', index=False)
    print(f"Scraped {len(valid_results)} jobs.")

# Run the async scraper
# asyncio.run(main())

import openai
import os
import json
import pandas as pd
from tenacity import retry, stop_after_attempt, wait_exponential

# Configure your API key
openai.api_key = os.getenv("OPENAI_API_KEY")

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def extract_job_data_with_ai(job_text, model="gpt-3.5-turbo-0125"):
    """Use LLM to extract structured data from job text."""
    prompt = f"""
    Analyze the following job posting and extract the information as a JSON object.
    Be precise and conservative. If information is not explicitly stated, use null.

    TEXT:
    {job_text[:6000]}  # Truncate to control token count

    Extract into this JSON structure:
    {{
      "job_title": "standardized title (e.g., Senior Software Engineer)",
      "company": "company name if mentioned",
      "salary_range": {{"min": number, "max": number, "currency": "USD/CAD/etc"}},
      "is_remote": boolean,
      "required_experience_years": number,
      "required_skills": ["list", "of", "specific", "technologies"],
      "preferred_skills": ["list", "of", "skills"],
      "seniority_level": ["Entry", "Mid", "Senior", "Lead", "Manager", "Executive"],
      "job_type": ["Full-time", "Part-time", "Contract"]
    }}

    Return ONLY the JSON object, no other text.
    """

    try:
        response = openai.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1,  # Low temperature for consistent output
            response_format={"type": "json_object"}  # Crucial for clean JSON
        )
        return json.loads(response.choices[0].message.content)
    except Exception as e:
        print(f"OpenAI API error: {e}")
        return None

def process_batch_with_ai(df, batch_size=20):
    """Process a batch of job texts for efficiency."""
    results = []
    for i in range(0, len(df), batch_size):
        batch = df.iloc[i:i+batch_size]
        print(f"Processing batch {i//batch_size + 1}/{(len(df)-1)//batch_size + 1}")

        for _, row in batch.iterrows():
            extracted = extract_job_data_with_ai(row['clean_text'])
            if extracted:
                extracted['source_url'] = row['url']  # Keep reference
                results.append(extracted)
        # Optional: brief pause between batches
        # time.sleep(1)
    return pd.DataFrame(results)

# Load your scraped data and process it
# raw_df = pd.read_csv('scraped_jobs_raw.csv')
# processed_df = process_batch_with_ai(raw_df)
# processed_df.to_csv('processed_jobs_structured.csv', index=False)

import numpy as np

def normalize_skills(df):
    """Clean and standardize skill names."""
    skill_mapping = {
        'reactjs': 'React', 'react.js': 'React', 'react': 'React',
        'nodejs': 'Node.js', 'node.js': 'Node.js', 'node': 'Node.js',
        'js': 'JavaScript', 'javascript': 'JavaScript',
        'ts': 'TypeScript', 'typescript': 'TypeScript',
        'py': 'Python', 'python': 'Python',
        'aws': 'Amazon Web Services',
        'gcp': 'Google Cloud Platform',
        'postgres': 'PostgreSQL', 'postgresql': 'PostgreSQL',
    }

    def map_skill_list(skill_list):
        if isinstance(skill_list, list):
            normalized = [skill_mapping.get(skill.lower().strip(), skill.strip()) for skill in skill_list]
            # Deduplicate
            return list(dict.fromkeys(normalized))
        return []

    df['required_skills_normalized'] = df['required_skills'].apply(map_skill_list)
    df['preferred_skills_normalized'] = df['preferred_skills'].apply(map_skill_list)
    return df

# Enrich with aggregated insights
def create_summary_statistics(processed_df):
    """Generate market insights from the processed data."""
    summary = {
        "total_jobs_analyzed": len(processed_df),
        "remote_ratio": f"{processed_df['is_remote'].mean():.1%}",
        "avg_experience_years": processed_df['required_experience_years'].mean(),
        "top_10_skills": pd.Series(
            [skill for sublist in processed_df['required_skills_normalized'].dropna() for skill in sublist]
        ).value_counts().head(10).to_dict(),
        "salary_ranges": processed_df['salary_range'].dropna().apply(
            lambda x: f"{x.get('currency', '')} {x.get('min', '')}-{x.get('max', '')}"
        ).tolist()
    }
    return summary

# normalized_df = normalize_skills(processed_df)
# insights = create_summary_statistics(normalized_df)
# print(json.dumps(insights, indent=2))

# Example using DeepSeek API (hypothetical, check current API)
def extract_with_deepseek(text):
    # Similar logic but with different API endpoint and cost structure
    # Potentially 1/10th the cost of GPT-4
    pass