| """ |
| SHL Catalog Scraper |
| ==================== |
| Scrapes the SHL product catalog and saves to data/shl_catalog.json |
| matching the schema expected by the retrieval module: |
| |
| { |
| "name": str, |
| "url": str, |
| "description": str, |
| "test_type": str, # single letter code e.g. "A", "K", "P", "B", "S" |
| "keys": [str], |
| "duration": str, |
| "languages": [str], |
| "job_levels": [str] |
| } |
| |
| SHL exposes a paginated JSON API used by its catalog page. |
| We query that directly instead of parsing HTML (which is JS-rendered). |
| """ |
|
|
| import json |
| import os |
| import re |
| import time |
| import requests |
| from pathlib import Path |
|
|
| |
| HERE = Path(__file__).resolve().parent |
| OUTPUT_PATH = HERE.parent / "data" / "shl_catalog.json" |
|
|
| |
| |
| CATALOG_API = "https://www.shl.com/api/product-catalog/" |
|
|
| HEADERS = { |
| "User-Agent": ( |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " |
| "AppleWebKit/537.36 (KHTML, like Gecko) " |
| "Chrome/124.0.0.0 Safari/537.36" |
| ), |
| "Accept": "application/json, text/plain, */*", |
| "Referer": "https://www.shl.com/products/product-catalog/", |
| } |
|
|
| |
| TEST_TYPE_MAP = { |
| "ability & aptitude": "A", |
| "biodata & situational judgment": "B", |
| "competencies": "C", |
| "development & 360": "D", |
| "knowledge & skills": "K", |
| "personality & behavior": "P", |
| "simulations": "S", |
| "assessment exercises": "E", |
| } |
|
|
|
|
| def _letter_from_keys(keys): |
| """Derive the primary test_type letter from a list of category keys.""" |
| for k in keys: |
| code = TEST_TYPE_MAP.get(k.lower().strip()) |
| if code: |
| return code |
| return "K" |
|
|
|
|
| def fetch_catalog_via_api(start=0, limit=100): |
| """ |
| Try to fetch a page of products from SHL JSON catalog API. |
| Returns parsed JSON dict, or None on failure. |
| """ |
| params = { |
| "start": start, |
| "limit": limit, |
| "type": "", |
| "industry": "", |
| "jobLevel": "", |
| } |
| try: |
| resp = requests.get(CATALOG_API, headers=HEADERS, params=params, timeout=20) |
| resp.raise_for_status() |
| return resp.json() |
| except Exception as e: |
| print(f" [API] Error at start={start}: {e}") |
| return None |
|
|
|
|
| def scrape_via_api(): |
| """Collect all products using the JSON API endpoint.""" |
| catalog = [] |
| start = 0 |
| limit = 100 |
|
|
| while True: |
| print(f" Fetching API page: start={start} ...") |
| data = fetch_catalog_via_api(start, limit) |
|
|
| if data is None: |
| print(" API request failed -- stopping.") |
| break |
|
|
| products = ( |
| data.get("products") |
| or data.get("results") |
| or data.get("items") |
| or data.get("data") |
| or [] |
| ) |
|
|
| if not products: |
| print(f" No products returned at start={start}. Done.") |
| break |
|
|
| for p in products: |
| item = normalize_api_product(p) |
| if item: |
| catalog.append(item) |
|
|
| print(f" Got {len(products)} products (total so far: {len(catalog)})") |
|
|
| if len(products) < limit: |
| break |
|
|
| start += limit |
| time.sleep(1) |
|
|
| return catalog |
|
|
|
|
| def normalize_api_product(p): |
| """Normalize a raw API product dict into our catalog schema.""" |
| name = p.get("name") or p.get("title") or "" |
| if not name: |
| return None |
|
|
| url = p.get("url") or p.get("link") or p.get("href") or "" |
| if url and not url.startswith("http"): |
| url = "https://www.shl.com" + url |
|
|
| description = p.get("description") or p.get("summary") or p.get("body") or "" |
|
|
| raw_keys = ( |
| p.get("keys") |
| or p.get("categories") |
| or p.get("tags") |
| or p.get("types") |
| or [] |
| ) |
| if isinstance(raw_keys, str): |
| raw_keys = [k.strip() for k in raw_keys.split(",") if k.strip()] |
| keys = [str(k) for k in raw_keys] |
|
|
| test_type = ( |
| p.get("test_type") |
| or p.get("testType") |
| or p.get("type") |
| or _letter_from_keys(keys) |
| ) |
| if len(str(test_type)) > 2: |
| test_type = TEST_TYPE_MAP.get(str(test_type).lower().strip(), _letter_from_keys(keys)) |
|
|
| duration = p.get("duration") or p.get("time") or p.get("timing") or "" |
| if isinstance(duration, (int, float)): |
| duration = f"{duration} minutes" |
|
|
| raw_langs = p.get("languages") or p.get("language") or [] |
| if isinstance(raw_langs, str): |
| raw_langs = [l.strip() for l in raw_langs.split(",") if l.strip()] |
| languages = [str(l) for l in raw_langs] |
|
|
| raw_levels = p.get("job_levels") or p.get("jobLevels") or p.get("levels") or [] |
| if isinstance(raw_levels, str): |
| raw_levels = [lv.strip() for lv in raw_levels.split(",") if lv.strip()] |
| job_levels = [str(lv) for lv in raw_levels] |
|
|
| return { |
| "name": str(name).strip(), |
| "url": str(url).strip(), |
| "description": str(description).strip(), |
| "test_type": str(test_type).strip(), |
| "keys": keys, |
| "duration": str(duration).strip(), |
| "languages": languages, |
| "job_levels": job_levels, |
| } |
|
|
|
|
| def scrape_via_html_sitemap(): |
| """ |
| Fallback: collect product URLs from sitemap.xml. |
| Returns a list of product page URLs. |
| """ |
| SITEMAP_URL = "https://www.shl.com/sitemap.xml" |
| urls = [] |
| try: |
| resp = requests.get(SITEMAP_URL, headers=HEADERS, timeout=20) |
| resp.raise_for_status() |
| pattern = re.compile( |
| r"<loc>(https://www\.shl\.com/products/product-catalog/view/[^<]+)</loc>" |
| ) |
| urls = pattern.findall(resp.text) |
| print(f" Found {len(urls)} product URLs in sitemap.") |
| except Exception as e: |
| print(f" Sitemap fetch failed: {e}") |
| return urls |
|
|
|
|
| def scrape_product_page(url): |
| """ |
| Scrape an individual SHL product page using BeautifulSoup. |
| Uses multiple selector strategies to handle different page layouts. |
| """ |
| from bs4 import BeautifulSoup |
|
|
| try: |
| resp = requests.get(url, headers=HEADERS, timeout=20) |
| resp.raise_for_status() |
| except Exception as e: |
| print(f" Error fetching {url}: {e}") |
| return None |
|
|
| soup = BeautifulSoup(resp.text, "html.parser") |
|
|
| |
| name_tag = soup.find("h1") or soup.find( |
| "h2", class_=lambda c: c and "title" in c.lower() if c else False |
| ) |
| name = name_tag.get_text(strip=True) if name_tag else "" |
|
|
| |
| desc_tag = ( |
| soup.find("div", class_=lambda c: c and "description" in c.lower() if c else False) |
| or soup.find("div", class_=lambda c: c and "body" in c.lower() if c else False) |
| or soup.find("div", class_=lambda c: c and "content" in c.lower() if c else False) |
| ) |
| if desc_tag: |
| description = desc_tag.get_text(separator=" ", strip=True) |
| else: |
| first_p = soup.find("p") |
| description = first_p.get_text(strip=True) if first_p else "" |
|
|
| |
| meta = {} |
|
|
| for dl in soup.find_all("dl"): |
| terms = dl.find_all("dt") |
| defs = dl.find_all("dd") |
| for dt, dd in zip(terms, defs): |
| key = dt.get_text(strip=True).lower().rstrip(":") |
| vals = [v.strip() for v in dd.get_text(separator=",").split(",") if v.strip()] |
| meta[key] = vals |
|
|
| for table in soup.find_all("table"): |
| for row in table.find_all("tr"): |
| cells = row.find_all(["td", "th"]) |
| if len(cells) >= 2: |
| key = cells[0].get_text(strip=True).lower().rstrip(":") |
| vals = [v.strip() for v in cells[1].get_text(separator=",").split(",") if v.strip()] |
| meta[key] = vals |
|
|
| for container in soup.find_all( |
| ["div", "section", "li"], |
| class_=lambda c: c and any( |
| x in c.lower() for x in ["detail", "spec", "meta", "attr", "info"] |
| ) if c else False, |
| ): |
| label_el = container.find(["label", "span", "strong", "b", "dt"]) |
| value_el = container.find(["p", "span", "dd"]) |
| if label_el and value_el and label_el != value_el: |
| key = label_el.get_text(strip=True).lower().rstrip(":") |
| vals = [v.strip() for v in value_el.get_text(separator=",").split(",") if v.strip()] |
| if key and vals: |
| meta.setdefault(key, vals) |
|
|
| def _get(keys_list): |
| for k in keys_list: |
| if k in meta: |
| return meta[k] |
| return [] |
|
|
| keys_val = _get(["assessment type", "keys", "categories", "type", "solution type"]) |
| test_type = _letter_from_keys(keys_val) |
| duration_vals = _get(["duration", "timing", "time limit", "test duration"]) |
| duration = duration_vals[0] if duration_vals else "" |
| languages = _get(["languages", "language", "available languages"]) |
| job_levels = _get(["job levels", "job level", "levels", "level"]) |
|
|
| if not name: |
| slug = url.rstrip("/").split("/")[-1] |
| name = slug.replace("-", " ").title() |
|
|
| return { |
| "name": name, |
| "url": url, |
| "description": description, |
| "test_type": test_type, |
| "keys": keys_val, |
| "duration": duration, |
| "languages": languages, |
| "job_levels": job_levels, |
| } |
|
|
|
|
| def main(): |
| print("=" * 60) |
| print("SHL Catalog Scraper") |
| print("=" * 60) |
|
|
| catalog = [] |
|
|
| |
| print("\n[Step 1] Trying SHL JSON API ...") |
| catalog = scrape_via_api() |
|
|
| if catalog: |
| print(f"\nAPI scrape succeeded: {len(catalog)} products collected.") |
| else: |
| |
| print("\n[Step 2] API returned nothing. Falling back to sitemap + HTML scraping ...") |
| product_urls = scrape_via_html_sitemap() |
|
|
| if not product_urls: |
| print(" No URLs found in sitemap either. Aborting.") |
| return |
|
|
| total = len(product_urls) |
| for i, url in enumerate(product_urls, 1): |
| print(f" [{i}/{total}] Scraping: {url}") |
| item = scrape_product_page(url) |
| if item: |
| catalog.append(item) |
| time.sleep(1.5) |
|
|
| print(f"\nHTML scrape complete: {len(catalog)} products collected.") |
|
|
| |
| OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) |
| with open(OUTPUT_PATH, "w", encoding="utf-8") as f: |
| json.dump(catalog, f, indent=2, ensure_ascii=False) |
|
|
| print(f"\nSaved {len(catalog)} products to: {OUTPUT_PATH}") |
| print("=" * 60) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
