如何使用 Python 和 API 抢夺 Google 职位

Google Jobs 动态呈现列表并本地化结果，因此简单的 HTTP 请求很少返回可用的数据。最可靠的方法是使用处理 JavaScript 渲染和地理定位的抓取 API，然后将您需要的字段解析为 CSV。

您将构建什么：

每个查询和位置的 CSV 文件，包含职位、公司、位置、日期、工资、来源和共享 URL。
可跨多个地理位置同时运行多个查询的可扩展管道。

方法 1 — Oxylabs Web Scraper API（异步、可扩展）

Oxylabs 的 Web Scraper API（Google Jobs 来源）运行具有地理定位和自定义解析器的无头浏览器，因此您可以收到结构化字段而不是原始 HTML。

你需要什么

Oxylabs 帐户和 API 用户凭据（用户名/密码）。在 Oxylabs 仪表板中创建一个帐户，开始免费试用，然后从仪表板凭据页面复制您的 API 用户和密码。
您机器上的 Python 3.11+。

步骤一：创建一个 Oxylabs 帐户并从仪表板 (Oxylabs) 复制您的 API 用户名和密码。

步骤2：如果尚未安装，请在系统上安装 Python 3.11 或更高版本。

步骤3：安装异步抓取所需的库。

pip install aiohttp asyncio pandas

第4步：创建一个名为payload.json带有 Google Jobs 的解析规则（Oxylabs 自定义解析器）。

{
  "source": "google",
  "url": null,
  "geo_location": null,
  "user_agent_type": "desktop",
  "render": "html",
  "parse": true,
  "parsing_instructions": {
    "jobs": {
      "_fns": [
        { "_fn": "xpath", "_args": ["//div[@class='nJXhWc']//ul/li"] }
      ],
      "_items": {
        "job_title":     { "_fns": [{ "_fn": "xpath_one", "_args": [".//div[@class='BjJfJf PUpOsf']/text()"] }] },
        "company_name":  { "_fns": [{ "_fn": "xpath_one", "_args": [".//div[@class='vNEEBe']/text()"] }] },
        "location":      { "_fns": [{ "_fn": "xpath_one", "_args": [".//div[@class='Qk80Jf'][1]/text()"] }] },
        "date":          { "_fns": [{ "_fn": "xpath_one", "_args": [".//div[@class='PuiEXc']//span[@class='LL4CDc' and contains(@aria-label, 'Posted')]/span/text()"] }] },
        "salary":        { "_fns": [{ "_fn": "xpath_one", "_args": [".//div[@class='PuiEXc']//div[contains(@class,'I2Cbhb') and contains(@class,'bSuYSc')]//span[@aria-hidden='true']/text()"] }] },
        "posted_via":    { "_fns": [{ "_fn": "xpath_one", "_args": [".//div[@class='Qk80Jf'][2]/text()"] }] },
        "URL":           { "_fns": [{ "_fn": "xpath_one", "_args": [".//div[@data-share-url]/@data-share-url"] }] }
      }
    }
  }
}

第5步：创建一个名为的 Python 文件jobs_oxylabs.py.

第6步：粘贴导入并加载凭据 (Oxylabs)。

import asyncio
import aiohttp
import json
import pandas as pd
from aiohttp import ClientSession, BasicAuth

# Replace with your Oxylabs API username/password from the dashboard
OXY_USER = "USERNAME"
OXY_PASS = "PASSWORD"

credentials = BasicAuth(OXY_USER, OXY_PASS)

# Load parsing payload
with open("payload.json", "r") as f:
    PAYLOAD = json.load(f)

第7步：添加辅助函数来提交作业、轮询状态和获取结果（Oxylabs 推拉端点）。

async def submit_job(session: ClientSession, payload: dict) -> str:
    async with session.post("https://data.oxylabs.io/v1/queries", auth=credentials, json=payload) as resp:
        data = await resp.json()
        return data["id"]

async def check_status(session: ClientSession, job_id: str) -> str:
    async with session.get(f"https://data.oxylabs.io/v1/queries/{job_id}", auth=credentials) as resp:
        data = await resp.json()
        return data["status"]

async def fetch_jobs(session: ClientSession, job_id: str) -> list:
    async with session.get(f"https://data.oxylabs.io/v1/queries/{job_id}/results", auth=credentials) as resp:
        data = await resp.json()
        return data["results"][0]["content"]["jobs"]

步骤8：添加一个函数，用于将每个查询和位置的解析作业写入 CSV。

async def save_csv(query: str, location: str, jobs: list) -> None:
    rows = []
    for j in jobs:
        rows.append({
            "Job title": j.get("job_title"),
            "Company name": j.get("company_name"),
            "Location": j.get("location"),
            "Date": j.get("date"),
            "Salary": j.get("salary"),
            "Posted via": j.get("posted_via"),
            "URL": j.get("URL"),
        })
    df = pd.DataFrame(rows)
    filename = f"{query}_jobs_{location.replace(',', '_').replace(' ', '_')}.csv"
    await asyncio.to_thread(df.to_csv, filename, index=False)

第9步：定义用于设置 Google Jobs URL 和地理位置、提交工作、等待完成并保存结果的协程 (Oxylabs)。

async def scrape_jobs(session: ClientSession, query: str, country_code: str, location: str) -> None:
    url = f"https://www.google.com/search?q={query}&ibp=htl;jobs&hl=en&gl={country_code}"
    PAYLOAD["url"] = url
    PAYLOAD["geo_location"] = location

    job_id = await submit_job(session, PAYLOAD)

    # Give the backend time to render before polling
    await asyncio.sleep(12)

    while True:
        status = await check_status(session, job_id)
        if status == "done":
            break
        if status == "failed":
            print(f"Job {job_id} failed for {query} @ {location}.")
            return
        await asyncio.sleep(5)

    jobs = await fetch_jobs(session, job_id)
    await save_csv(query, location, jobs)

第10步：添加main()同时运行多个查询和位置。

URL_QUERIES = ["developer", "chef", "manager"]
LOCATIONS = {
    "US": ["California,United States", "Virginia,United States", "New York,United States"],
    "GB": ["United Kingdom"],
    "DE": ["Germany"]
}

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = []
        for cc, locs in LOCATIONS.items():
            for loc in locs:
                for q in URL_QUERIES:
                    tasks.append(asyncio.ensure_future(scrape_jobs(session, q, cc, loc)))
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())
    print("Completed.")

第11步：运行脚本并验证是否根据查询和位置创建了 CSV 文件。

笔记：

Google 招聘信息共享 URL 只能从抓取过程中使用的同一国家/地区的 IP 打开；打开它们时使用匹配的代理/VPN。
使用Ctrl+Shift+I在 Windows 或Option+Command+I如果您想扩展解析，请在 macOS 上检查选择器。

方法 2 — SerpApi Google Jobs API（简单的 REST、过滤器和半径）

SerpApi 的google_jobs引擎返回带有可选过滤器（如语言、国家/地区和半径）的结构化列表。

你需要什么

SerpApi 帐户和 API 密钥。在 serpapi.com 上注册，打开仪表板，然后从“API 密钥”页面复制您的 API 密钥。
Python 3.9+ 和requests和pandas包。

步骤一：创建 SerpApi 帐户并从仪表板 (SerpApi) 复制您的 API 密钥。

步骤2：安装依赖项。

pip install requests pandas

步骤3：发送您的第一个 Google Jobs 请求并将结果写入 CSV (SerpApi)。

import requests
import pandas as pd

API_KEY = "YOUR_SERPAPI_KEY"

params = {
    "engine": "google_jobs",
    "q": "developer new york",
    "hl": "en",
    "gl": "us",
    "api_key": API_KEY,
    # Optional: radius in kilometers (Google may not strictly enforce)
    "lrad": 25
}

r = requests.get("https://serpapi.com/search.json", params=params)
r.raise_for_status()
data = r.json()

rows = []
for j in data.get("jobs_results", []):
    rows.append({
        "Job title": j.get("title"),
        "Company name": j.get("company_name"),
        "Location": j.get("location"),
        "Posted via": j.get("via"),
        "Share link": j.get("share_link"),
        "Description": j.get("description")
    })

pd.DataFrame(rows).to_csv("jobs_serpapi.csv", index=False)

第4步：使用分页next_page_token从响应中获取其他页面（SerpApi）。

def fetch_all(params):
    out = []
    while True:
        res = requests.get("https://serpapi.com/search.json", params=params).json()
        out.extend(res.get("jobs_results", []))
        token = res.get("serpapi_pagination", {}).get("next_page_token")
        if not token:
            break
        params["next_page_token"] = token
    return out

jobs = fetch_all(params)

尖端：

使用location或者uule模拟城市级搜索；避免将两者组合在同一个请求中。
过滤器可以作为udsJSON 中的字符串；重复使用过滤器uds改进结果的新请求中的价值。

方法 3 — 使用 Selenium 的免费本地刮刀（最适合小型测试）

无头浏览器可以滚动 Google 职位列表并使用强大的选择器提取字段。这对于原型设计来说很实用，但需要维护，并且可能会大量攻击反机器人系统。

你需要什么

Google Chrome 和匹配的 ChromeDriver 二进制文件。
Python 3.9+ 与selenium,beautifulsoup4（可选），以及pandas.

步骤一：下载与您的 Chrome 版本匹配的 ChromeDriver 并将其添加到您的 PATH 中。

步骤2：安装 Selenium 和 pandas。

pip install selenium pandas

步骤3：启动 Chrome，并带有减少自动化指纹的选项。

from selenium import webdriver

options = webdriver.ChromeOptions()
options.add_argument("--window-size=1920,1080")
options.add_argument("--disable-blink-features=AutomationControlled")

driver = webdriver.Chrome(options=options)

第4步：打开根据您的查询和国家/地区代码构建的 Google 职位网址。

query = "developer"
country = "us"
url = f"https://www.google.com/search?q={query}&ibp=htl;jobs&hl=en&gl={country}"
driver.get(url)

第5步：滚动作业列表容器以加载更多结果。

import time
from selenium.webdriver.common.by import By

loaded = 0
while True:
    cards = driver.find_elements(By.XPATH, "//div[@class='nJXhWc']//ul/li")
    if len(cards) == loaded:
        break
    loaded = len(cards)
    driver.execute_script("arguments[0].scrollIntoView({block: 'end'});", cards[-1])
    time.sleep(1.5)

第6步：使用 XPath 提取稳定字段；如果可用，请使用基于图标的选择器来获取发布日期和工资。

from selenium.common.exceptions import NoSuchElementException

rows = []
for li in driver.find_elements(By.XPATH, "//div[@class='nJXhWc']//ul/li//div[@role='treeitem']/div/div"):
    def txt(xp):
        try:
            return li.find_element(By.XPATH, xp).get_attribute("innerText")
        except NoSuchElementException:
            return None

    def attr(xp, name):
        try:
            return li.find_element(By.XPATH, xp).get_attribute(name)
        except NoSuchElementException:
            return None

    rows.append({
        "Job title": txt("./div[2]"),
        "Company name": txt("./div[4]/div/div[1]"),
        "Location": txt("./div[4]/div/div[2]"),
        "Source": txt("./div[4]/div/div[3]"),
        "Posted": txt(".//*[name()='path'][contains(@d,'M11.99')]/ancestor::div[1]"),
        "Full/Part": txt(".//*[name()='path'][contains(@d,'M20 6')]/ancestor::div[1]"),
        "Salary": txt(".//*[name()='path'][@fill-rule='evenodd']/ancestor::div[1]"),
        "Logo src": attr("./div[1]//img", "src")
    })

第7步：将提取的数据保存到 CSV。

import pandas as pd
pd.DataFrame(rows).to_csv("jobs_selenium.csv", index=False)

警告：

预计布局变化；定期重新验证选择器。
限制请求并遵守站点条款以避免阻塞。

选择一种方法

对于生产和规模，Oxylabs 的异步管道可以跨多个查询和位置可靠地返回已解析的作业，从而减少工程时间和故障。 SerpApi 可以快速集成较小的脚本，并允许您添加半径和过滤器令牌。 Selenium 最适合快速实验或无法使用第三方 API 时。

快速维护技巧：将查询列表和位置字典保存在单独的 JSON 或 YAML 文件中，以便非开发人员可以更新搜索范围而无需接触代码。

方法 1 — Oxylabs Web Scraper API（异步、可扩展）

方法 2 — SerpApi Google Jobs API（简单的 REST、过滤器和半径）

方法 3 — 使用 Selenium 的免费本地刮刀（最适合小型测试）

选择一种方法

相关文章

如何将 Canva 文档转换为 PDF

如何在 Arc 命令栏中使用 ChatGPT

ARC Raiders 无法邀请朋友

修复 Windows 11 上电脑睡眠期间显示器未关闭的问题

You vs Homer 代码（2026 年 2 月）

如何在 Windows 11 PC 上录制视频