使用 Bright Data Web Scraper API + Python 高效抓取 Glassdoor 数据：从配置到结构化输出全流程实战

使用 Bright Data Web Scraper API + Python 高效抓取 Glassdoor 数据：从配置到结构化输出全流程实战

摘要

本文详细介绍了如何使用 Bright Data 的 Web Scraper API 搭配 Python，实现对 Glassdoor 平台信息的高效抓取。通过 API 请求构建器、反爬机制集成与结构化数据输出，开发者可轻松获取高质量网页数据，适用于招聘分析、AI 训练与商业情报等场景，同时介绍了 Bright Data 的 Deep Lookup 功能，通过自然语言指令实现深度数据挖掘，进一步拓展数据采集的智能化能力。

前言

数字化商业时代，网页数据蕴含着市场洞察的宝藏，从 AI 模型训练的高质量素材，到商业分析、市场调研与竞争情报的核心依据，结构化网页数据成为开发者的 “必争之地”，然而传统爬虫开发成本高企、反爬封锁频发、长期维护艰难，让数据采集寸步难行。

亮数据 Bright Data 的 Web Scraper API 能很好解决这些问题，本文将聚焦 “Web Scraper API + Python” 组合，详解如何精准抓取 Glassdoor 平台信息：从 API 请求构建、反爬策略集成，到数据解析与自动化流程落地，带你避开开发陷阱，高效获取结构化数据，为 AI 训练、商业决策筑牢数据根基 。

亮数据 Bright Data 网页爬虫 API 简介

亮数据网页爬虫 API 高效、稳定采集网页数据设计的接口服务，集成核心的全球代理网络、反爬突破技术与智能解析能力，该 API 能自动处理 IP 封锁、验证码拦截、动态页面渲染等常见爬虫难题，开发者无需搭建复杂代理池或反爬逻辑，通过简单的 API 调用即可实现对各类网站的数据抓取，不仅支持定制化配，兼容主流编程语言直接返回结构化数据，大幅降低爬虫开发门槛，让开发者专注于数据应用而非采集技术细节，尤其适合大规模、高稳定性要求的商业级数据采集场景。

✅卓越反爬突破：依托 7200 万住宅代理 IP 池高频切换 IP，模拟真实用户行为避开封锁，精准处理 Cloudflare/Akamai 等验证码，稳定抓取动态 JavaScript 页面，采集成功率达 99%+

✅零代码便捷操作：提供可视化界面，无需编程基础，通过输入网址、框选字段、设置输出格式与调度频率，快速完成数据采集配置，降低技术门槛

✅高度定制化适配：支持自定义采集规则，可灵活配置字段、分页逻辑，适配电商、房产、舆情等多行业场景，满足多样化数据需求，助力业务分析

前提准备：亮数据 Bright Data 注册

1、亮数据 Bright Data 注册

2、Sign up 注册

3、注册方式选择

4、访问 Bright Data

Python 自动化爬虫 调用 API 抓取 Glassdoor 信息

1、点击 Web Scrapers（亮数据提供的网络数据抓取与解析工具集）

2、亮数据 Web Scrapers Library 网页数据抓取取库中搜索 Glassdoor

3、可以看到亮数据模板库中 glassdoor 信息的爬取方案有许多种类，可以根据个人需要进行选择

• 此处也有现成的爬取后的数据，如有需要也可以进行选择

4、此处我们不用现成的数据，选择通过URL来收集

5、选择爬取方式（选择Scraper API）

✅Scraper API：需通过代码调用，可深度定制抓取逻辑，支持与系统集成，适合技术用户实现大规模、高定制化的数据采集

✅No-Code Scraper：纯可视化操作，无需代码，依赖模板快速配置，对非技术用户友好，适合简单到中等复杂度标准化数据采集

6、点击 API request builder API 请求构建器（此处可以配置请求参数 URL、headers、参数等，自动生成对应语言、返回数据结构、错误处理与重试机制、自动化工具或数据库的集成方式）

7、API request builder API 参数信息配置

8、复制右侧自动生成的 API 调用代码

9、PyCharm粘贴代码

import requests

url = "https://api.brightdata.com/datasets/v3/trigger"
headers = {
	"Authorization": "Bearer 6c03e5947c0ac330f4fcd3f363c0038736bfcf27089c7deec066dc65a48d12b2",
	"Content-Type": "application/json",
}
params = {
	"dataset_id": "gd_l7j0bx501ockwldaqf",
	"include_errors": "true",
}
data = [
	{"url":"https://www.glassdoor.co.uk/Overview/Working-at-Department-for-International-Trade-EI_IE313977.11,45.htm"},
	{"url":"https://www.glassdoor.com/partner/jobListing.htm?pos=1621&ao=1136043&s=21&guid=000001984b2811aa9b3021c052e9e184&src=GD_JOB_AD&t=ESR&vt=w&ea=1&cs=1_7b1a4a81&cb=1753655687500&jobListingId=1009767232500&jrtk=5-yul1-0-1j15ig4h8j0ol800-ec3098cf4fbb1304"},
	{"url":"https://www.glassdoor.com/partner/jobListing.htm?pos=122&ao=1136043&s=21&guid=000001984b2811c6bc3aa87ef9131188&src=GD_JOB_AD&t=ESR&vt=w&cs=1_dd419c0f&cb=1753655688710&jobListingId=1006323743607&jrtk=5-yul1-0-1j15ig4hth73h800-72475e0676a4abef"},
	{"url":"https://www.glassdoor.com/partner/jobListing.htm?pos=1624&ao=1136043&s=21&guid=000001984b2811aa9b3021c052e9e184&src=GD_JOB_AD&t=ESR&vt=w&ea=1&cs=1_e5b2064c&cb=1753655688089&jobListingId=1009612278091&jrtk=5-yul1-0-1j15ig4h8j0ol800-e101a229fb38e5ba"},
]

response = requests.post(url, headers=headers, params=params, json=data)
print(response.json())

• 代码进行优化：结构化的 Glassdoor 数据抓取工具，通过封装成GlassdoorScraper类实现了完整的 API 交互流程，初始化认证、触发 URL 抓取，到分页获取结果、清洗数据（提取公司名、职位等关键信息并处理缺失值），最终将结果保存为 JSON 文件，同时配备全面的异常处理和日志记录，确保抓取过程稳定可靠且结果可用

import requests
import time
import json
import logging
from datetime import datetime
from typing import List, Dict, Optional

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("scraper.log"),
        logging.StreamHandler()
    ]
)

class GlassdoorScraper:
    def __init__(self, api_token: str, dataset_id: str):
        """初始化Glassdoor数据抓取器

        Args:
            api_token: BrightData API令牌
            dataset_id: 要使用的数据集ID
        """
        self.api_url = "https://api.brightdata.com/datasets/v3/trigger"
        self.headers = {
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json",
        }
        self.dataset_id = dataset_id
        self.session = requests.Session()  # 使用会话提高性能
        self.session.headers.update(self.headers)

    def _handle_response(self, response: requests.Response) -> Dict:
        """处理API响应，检查状态码并返回解析后的JSON

        Args:
            response: API响应对象

        Returns:
            解析后的JSON数据

        Raises:
            ValueError: 当API返回错误状态码时
        """
        try:
            response.raise_for_status()  # 抛出HTTP错误
            return response.json()
        except requests.exceptions.HTTPError as e:
            error_msg = f"HTTP错误: {str(e)}"
            if response.text:
                try:
                    error_details = response.json()
                    error_msg += f" 详情: {json.dumps(error_details)}"
                except json.JSONDecodeError:
                    error_msg += f" 响应内容: {response.text}"
            logging.error(error_msg)
            raise ValueError(error_msg) from e
        except json.JSONDecodeError as e:
            error_msg = f"解析JSON响应失败: {str(e)}"
            logging.error(error_msg)
            raise ValueError(error_msg) from e

    def trigger_scrape(self, urls: List[str], include_errors: bool = True) -> Dict:
        """触发对指定URL的抓取

        Args:
            urls: 要抓取的URL列表
            include_errors: 是否包含错误信息

        Returns:
            API响应结果
        """
        if not urls:
            logging.warning("没有提供要抓取的URL")
            return {}

        params = {
            "dataset_id": self.dataset_id,
            "include_errors": str(include_errors).lower(),
        }

        data = [{"url": url} for url in urls]

        try:
            logging.info(f"开始抓取 {len(urls)} 个URL")
            response = self.session.post(
                self.api_url,
                params=params,
                json=data,
                timeout=30  # 设置超时时间
            )
            result = self._handle_response(response)
            logging.info(f"成功触发抓取，返回ID: {result.get('id')}")
            return result

        except requests.exceptions.RequestException as e:
            error_msg = f"请求失败: {str(e)}"
            logging.error(error_msg)
            raise ConnectionError(error_msg) from e

    def get_results(self, job_id: str, page: int = 1, page_size: int = 100) -> Optional[Dict]:
        """获取抓取结果

        Args:
            job_id: 抓取任务ID
            page: 页码
            page_size: 每页结果数量

        Returns:
            抓取结果数据或None
        """
        results_url = f"https://api.brightdata.com/datasets/v3/results/{job_id}"

        params = {
            "page": page,
            "page_size": page_size,
            "dataset_id": self.dataset_id,
        }

        try:
            logging.info(f"获取任务 {job_id} 的结果，第 {page} 页")
            response = self.session.get(
                results_url,
                params=params,
                timeout=30
            )
            return self._handle_response(response)

        except requests.exceptions.RequestException as e:
            error_msg = f"获取结果失败: {str(e)}"
            logging.error(error_msg)
            raise ConnectionError(error_msg) from e

    def paginated_results(self, job_id: str, page_size: int = 100, max_pages: int = None) -> List[Dict]:
        """分页获取所有结果

        Args:
            job_id: 抓取任务ID
            page_size: 每页结果数量
            max_pages: 最大页数，None表示获取所有页

        Returns:
            所有页面的结果列表
        """
        all_results = []
        page = 1

        while True:
            try:
                results = self.get_results(job_id, page, page_size)

                # 检查是否有结果
                if not results.get('results'):
                    logging.info(f"任务 {job_id} 第 {page} 页没有结果，停止分页")
                    break

                all_results.extend(results['results'])
                logging.info(f"已获取 {len(all_results)} 条结果")

                # 检查是否有下一页
                total_pages = results.get('total_pages', 1)
                if page >= total_pages:
                    logging.info(f"已获取所有 {total_pages} 页结果")
                    break

                # 检查是否超过最大页数限制
                if max_pages and page >= max_pages:
                    logging.info(f"已达到最大页数限制 {max_pages}")
                    break

                page += 1
                time.sleep(1)  # 延迟避免请求过于频繁

            except Exception as e:
                logging.error(f"获取第 {page} 页结果时出错: {str(e)}")
                # 尝试再获取一次
                time.sleep(3)
                continue

        return all_results

    @staticmethod
    def clean_data(raw_data: List[Dict]) -> List[Dict]:
        """清洗原始数据，提取有用字段并处理缺失值

        Args:
            raw_data: 原始数据列表

        Returns:
            清洗后的数据列表
        """
        cleaned = []

        for item in raw_data:
            # 根据实际API返回结构调整字段提取
            cleaned_item = {
                'url': item.get('url'),
                'timestamp': item.get('timestamp'),
                'status': item.get('status'),
                'company_name': item.get('data', {}).get('company_name'),
                'job_title': item.get('data', {}).get('job_title'),
                'location': item.get('data', {}).get('location'),
                'salary': item.get('data', {}).get('salary'),
                'description': item.get('data', {}).get('description', '')[:500]  # 截断长描述
            }

            # 处理缺失值
            for key, value in cleaned_item.items():
                if value is None:
                    cleaned_item[key] = "N/A"

            cleaned.append(cleaned_item)

        logging.info(f"已清洗 {len(cleaned)} 条数据")
        return cleaned

    @staticmethod
    def save_results(data: List[Dict], filename: str = None) -> None:
        """保存结果到JSON文件

        Args:
            data: 要保存的数据
            filename: 文件名，None则使用时间戳生成
        """
        if not data:
            logging.warning("没有数据可保存")
            return

        if not filename:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"glassdoor_results_{timestamp}.json"

        try:
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            logging.info(f"结果已保存到 {filename}")
        except IOError as e:
            logging.error(f"保存文件失败: {str(e)}")


def main():
    # 配置参数
    API_TOKEN = "6c03e5947c0ac330f4fcd3f363c0038736bfcf27089c7deec066dc65a48d12b2"
    DATASET_ID = "gd_l7j0bx501ockwldaqf"

    # 要抓取的URL列表
    urls = [
        "https://www.glassdoor.co.uk/Overview/Working-at-Department-for-International-Trade-EI_IE313977.11,45.htm",
        "https://www.glassdoor.com/partner/jobListing.htm?pos=1621&ao=1136043&s=21&guid=000001984b2811aa9b3021c052e9e184&src=GD_JOB_AD&t=ESR&vt=w&ea=1&cs=1_7b1a4a81&cb=1753655687500&jobListingId=1009767232500&jrtk=5-yul1-0-1j15ig4h8j0ol800-ec3098cf4fbb1304",
        "https://www.glassdoor.com/partner/jobListing.htm?pos=122&ao=1136043&s=21&guid=000001984b2811c6bc3aa87ef9131188&src=GD_JOB_AD&t=ESR&vt=w&cs=1_dd419c0f&cb=1753655688710&jobListingId=1006323743607&jrtk=5-yul1-0-1j15ig4hth73h800-72475e0676a4abef",
        "https://www.glassdoor.com/partner/jobListing.htm?pos=1624&ao=1136043&s=21&guid=000001984b2811aa9b3021c052e9e184&src=GD_JOB_AD&t=ESR&vt=w&ea=1&cs=1_e5b2064c&cb=1753655688089&jobListingId=1009612278091&jrtk=5-yul1-0-1j15ig4h8j0ol800-e101a229fb38e5ba",
    ]

    try:
        # 初始化抓取器
        scraper = GlassdoorScraper(API_TOKEN, DATASET_ID)

        # 触发抓取
        scrape_result = scraper.trigger_scrape(urls)
        job_id = scrape_result.get('id')

        if not job_id:
            logging.error("未能获取到任务ID，无法继续")
            return

        # 等待抓取完成（实际应用中可能需要更长时间或轮询状态）
        logging.info("等待抓取完成...")
        time.sleep(10)  # 根据实际情况调整等待时间

        # 获取所有结果
        raw_data = scraper.paginated_results(job_id)

        if not raw_data:
            logging.warning("没有获取到任何数据")
            return

        # 清洗数据
        cleaned_data = GlassdoorScraper.clean_data(raw_data)

        # 保存结果
        GlassdoorScraper.save_results(cleaned_data)

        logging.info("抓取任务完成")

    except Exception as e:
        logging.error(f"程序执行出错: {str(e)}", exc_info=True)


if __name__ == "__main__":
    main()

10、运行代码开始数据爬取，运行成功即可在亮数据 Bright Data 看到一条正在爬取的记录

11、等待数据爬取完成选择JSON格式下载即可

12、打开下载后的 JSON和CSV 数据可以看到已经获取到爬取的 Glassdoor 信息

Bright Data Deep lookup：AI深度查找

Bright Data Deep Lookup 是AI 驱动深度数据搜索工具，核心在于通过直观的自然语言指令（如 “Find all + 实体类型 + 条件”），快速将复杂查询转化为结构化数据集，支持结构化查询进阶与自定义列功能，能精准挖掘非结构化数据中的关键信息（如特定公司、产品、事件等），无需复杂技术操作，可帮助用户从海量信息中高效提取精准、定制化的深度洞察。

1、输入需要查询的信息

查找所有查找所有使用 OpenAI 或 GPT 技术的 SaaS 公司，并在其网站上列出实时集成

2、通过亮数据 Bright Data Deep lookup 使用 OpenAI 或 GPT 技术的 SaaS 公司

• 如下是 V1 版本的信息

3、如果给我的结果不满意可以继续让 Bright Data Deep lookup 优化

补充 live_integrations_evidence / openai_gpt_usage_evidence字段的具体证明（文档、链路、数据）

• 如下是 V2 版本的信息

亮数据 Web Scraper API 技术亮点

亮数据 Web Scraper API 核心优势显著：高效采集上，可批量处理大量 URL，数分钟获取海量数据，支持定时采集保障数据时效；智能处理时，能精准解析 HTML 与 JavaScript 复杂页面，提取关键数据并初步清洗，输出结构化数据直接可用；便捷使用方面，零代码界面让非技术人员快速配置启动任务，多语言 API 调用便于系统集成

✅数据获取更高效便捷：无需维护代理或解析 HTML，可直接获取结构化数据，简化技术流程

✅灵活适配多样需求：支持按需调用与批量处理，兼顾零散查询与大规模数据需求

✅成本友好且支付灵活：提供按量、套餐等多种定价模式，起步价低至 $0.79 / 千条记录；支持免费试用及支付宝付款，更贴合中国开发者使用习惯

总结

亮数据 Bright Data Web Scraper API 凭借 7200 万住宅代理池与智能反爬技术轻松突破网站封锁，支持零代码快速配置与 Python 深度定制两种模式，直接输出结构化数据，广泛适配电商、招聘、舆情等多场景需求， AI 驱动的 Deep Lookup 功能更能通过自然语言指令挖掘深度洞察，搭配免费试用、按量付费灵活方案，让开发者无需投入复杂维护成本，即可高效获取高质量数据，为 AI 训练与商业决策筑牢数据根基。

亮数据 Bright Data 免费体验通道 已开启，即刻解锁高效数据采集新体验！