scrape.py

import json
import pandas as pd
import requests


def get_test_type(keys_list):
    """Maps the JSON 'keys' array to the single-character 'test_type'

    expected by the SHL automated grader schema (K=Knowledge,
    P=Personality/Behavior, S=Simulation).
    """
    if not isinstance(keys_list, list):
        return "K"  # Default fallback

    # Standardize string checking to prevent casing issues
    keys_lower = [str(k).lower() for k in keys_list]

    if any("simulation" in k for k in keys_lower):
        return "S"
    elif any("personality" in k or "behavior" in k for k in keys_lower):
        return "P"
    else:
        return "K"  # Default maps to Knowledge & Skills


def process_shl_dataset():
    url = "https://tcp-us-prod-rnd.shl.com/voiceRater/shl-ai-hiring/shl_product_catalog.json"
    print("🚀 Fetching original SHL product catalog stream...")

    try:
        response = requests.get(url)

        # CRITICAL FIX: strict=False bypasses invalid control characters (e.g., raw tabs/newlines)
        # that cause standard json.loads to crash.
        raw_data = json.loads(response.text, strict=False)

    except Exception as e:
        print(f"❌ Network fetch failed: {e}. Attempting to read local raw JSON instead.")
        # Backup if network environment blocks the domain
        with open("shl_product_catalog.json", "r", encoding="utf-8") as f:
            raw_data = json.loads(f.read(), strict=False)

    # Convert the list of dictionaries to a DataFrame
    raw_df = pd.DataFrame(raw_data)
    print(f"📦 Total raw elements loaded: {len(raw_df)}")

    # ==========================================
    # REQUIREMENT 1: FILTER OUT JOB SOLUTIONS
    # Keep only Individual Test Solutions.
    # We strip out rows containing 'Job Solutions' or packaged templates
    # ==========================================
    if "link" in raw_df.columns:
        # Pre-packaged job solutions are explicitly out of scope.
        # Looking at the URLs, Individual tests follow: /view/test-name
        # While job solution packages often have 'solution' in names/links.
        filtered_df = raw_df[
            ~raw_df["name"].str.lower().str.contains("solution", na=False)
        ].copy()
    else:
        filtered_df = raw_df.copy()

    # ==========================================
    # REQUIREMENT 2: SCHEMA ALIGNMENT & MAPPING
    # ==========================================
    # 1. Map 'link' from JSON directly to 'url'
    filtered_df["url"] = filtered_df["link"]

    # 2. Extract and assign the strict 'test_type' required by the schema matrix
    filtered_df["test_type"] = filtered_df["keys"].apply(get_test_type)

    # Select only the clean columns mandatory for your final RAG indexing and API responses
    final_columns = ["name", "url", "test_type", "description"]
    clean_catalog_df = filtered_df[final_columns]

    # Drop any row completely missing a name or url to secure Hard Evals integrity
    clean_catalog_df = clean_catalog_df.dropna(subset=["name", "url"])

    # Save to final CSV
    output_path = "shl_individual_tests_catalog.csv"
    clean_catalog_df.to_csv(output_path, index=False, encoding="utf-8")

    print(f"\n🎉 Success! Output dataset perfectly structured.")
    print(f"Saved to: '{output_path}'")
    print(f"Total valid assessment records matching parameters: {len(clean_catalog_df)}")
    print("\nPreview of schema parameters matches:")
    print(clean_catalog_df[["name", "url", "test_type"]].head())


if __name__ == "__main__":
    process_shl_dataset()