Google Jobs API：如何从搜索结果采集职位数据

Google Jobs 搜索结果很适合用于了解 hiring demand、追踪 job-market trends、监测 competitors，或建立 recruiting research workflow。

例如，你可能想采集：

• New York 的 Python developer jobs

• remote data analyst jobs

• Texas 的 nursing jobs

• London 的 marketing manager jobs

• 本周发布的 AI engineer jobs

• 按 role 或 location 监测 competitor hiring activity

SERP API 会比手动 browser scraping 更干净。你传入 job search query、location、language 和 result type，API 返回 structured search data，后续可以 normalize、analyze、store，或传给 AI workflow。Talordata SERP API 提供来自 Google 和主要搜索引擎的 structured search-result data，支持 JSON / HTML response formats 和 geo-targeted SERP data。

我们要做什么？

我们会建立一个 Python script：

Job search query
→ SERP API request
→ Google Jobs results
→ Extract job listing fields
→ Normalize data
→ Export listings to CSV

最终 CSV 可以包含：

• job title

• company name

• location

• source platform

• posting date

• job type

• salary information

• description

• apply links

• job ID

• collected timestamp

这个 workflow 适合 recruiting analytics、labor-market research、competitor hiring monitoring、job aggregation 和 AI-assisted market reports。

为什么用 SERP API 采集 Job Listings？

你可以手动采集 job results，但 job search pages 通常是 dynamic 且 location-sensitive 的。

常见问题包括：

• dynamic rendering

• changing layouts

• inconsistent job-card fields

• duplicate listings across platforms

• missing salary data

• different job sources for the same role

• location-specific result changes

• pagination and result expansion

• blocked requests or CAPTCHA challenges

使用 SERP API 后，你可以把精力放在 data workflow，而不是 collection layer。

基本流程变成：

Send query + location
→ Receive structured job results
→ Normalize fields
→ Save to CSV, database, or dashboard

Step 1：准备 API Key

创建 Talordata 账号，并从 dashboard 获取 SERP API key。Talordata 产品页说明，新用户可以获得 1,000 free API responses，并支持 Python、JavaScript、cURL、Go 等开发语言集成。

把 API key 和 endpoint 设为 environment variables：

export TALORDATA_API_KEY="your_api_key_here"
export TALORDATA_SERP_ENDPOINT="your_talordata_serp_endpoint_here"

Windows PowerShell：

$env:TALORDATA_API_KEY="your_api_key_here"
$env:TALORDATA_SERP_ENDPOINT="your_talordata_serp_endpoint_here"

安装 Python package：

pip install requests

请使用 Talordata dashboard 或 API docs 中显示的 endpoint 和 parameter names。下面的 code 将 request layer 独立出来，方便你调整实际 endpoint、engine name 或 authentication method。

Step 2：发送 Google Jobs Request

Job search request 通常需要 query、location、language、country 和 result type。

import os
import requests


def require_env(name):
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def search_google_jobs(query, location="United States", language="en", country="us"):
    api_key = require_env("TALORDATA_API_KEY")
    endpoint = require_env("TALORDATA_SERP_ENDPOINT")

    payload = {
        "engine": "google_jobs",
        "q": query,
        "location": location,
        "hl": language,
        "gl": country,
        "output": "json"
    }

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }

    response = requests.post(
        endpoint,
        json=payload,
        headers=headers,
        timeout=30
    )

    response.raise_for_status()
    return response.json()

有些 API 可能使用 query 而不是 q，或使用 type=jobs 而不是 engine=google_jobs。把这个 function 独立出来，可以避免其他 parsing 和 export code 被 provider format 绑死。

Step 3：提取 Job Listings

不同 API 可能会把 job results 放在不同 keys 下：

jobs_results
job_results
jobs
organic_results

使用 flexible parser 会更容易调整。

from datetime import datetime, timezone


def clean_text(value):
    if not value:
        return ""
    return " ".join(str(value).split())


def join_list(value):
    if not value:
        return ""

    if isinstance(value, list):
        return " | ".join(clean_text(item) for item in value if item)

    return clean_text(value)


def get_job_results(serp_json):
    return (
        serp_json.get("jobs_results")
        or serp_json.get("job_results")
        or serp_json.get("jobs")
        or serp_json.get("organic_results")
        or []
    )


def extract_apply_links(item):
    apply_options = (
        item.get("apply_options")
        or item.get("apply_links")
        or item.get("related_links")
        or []
    )

    links = []

    for option in apply_options:
        if isinstance(option, dict):
            title = clean_text(option.get("title") or option.get("source"))
            link = option.get("link") or option.get("url")
            if link:
                links.append(f"{title}: {link}" if title else link)

    return " | ".join(links)


def normalize_job_results(serp_json, query, location):
    results = get_job_results(serp_json)
    collected_at = datetime.now(timezone.utc).isoformat()

    rows = []

    for index, item in enumerate(results, start=1):
        rows.append({
            "query": query,
            "location": location,
            "position": item.get("position") or item.get("rank") or index,
            "job_title": clean_text(item.get("title")),
            "company_name": clean_text(item.get("company_name") or item.get("company")),
            "job_location": clean_text(item.get("location")),
            "source": clean_text(item.get("via") or item.get("source")),
            "posted_at": clean_text(item.get("detected_extensions", {}).get("posted_at") or item.get("posted_at") or item.get("date")),
            "job_type": clean_text(item.get("detected_extensions", {}).get("schedule_type") or item.get("job_type") or item.get("schedule_type")),
            "salary": clean_text(item.get("detected_extensions", {}).get("salary") or item.get("salary")),
            "description": clean_text(item.get("description")),
            "extensions": join_list(item.get("extensions")),
            "job_id": item.get("job_id") or item.get("job_highlight_token") or item.get("id") or "",
            "job_link": item.get("link") or item.get("url") or "",
            "apply_links": extract_apply_links(item),
            "collected_at": collected_at
        })

    return rows

这样整理后的格式，更适合放入 spreadsheets、dashboards、databases 和 AI workflows。

Step 4：导出 Job Listings 到 CSV

接着把 normalized job rows 写入 CSV file。

import csv


CSV_COLUMNS = [
    "query",
    "location",
    "position",
    "job_title",
    "company_name",
    "job_location",
    "source",
    "posted_at",
    "job_type",
    "salary",
    "description",
    "extensions",
    "job_id",
    "job_link",
    "apply_links",
    "collected_at"
]


def export_to_csv(rows, filename="google_jobs_results.csv"):
    if not rows:
        print("No job listings found.")
        return

    with open(filename, mode="w", newline="", encoding="utf-8") as file:
        writer = csv.DictWriter(file, fieldnames=CSV_COLUMNS)
        writer.writeheader()
        writer.writerows(rows)

    print(f"Exported {len(rows)} job listings to {filename}")

完整 Python 示例

import os
import csv
import requests

from datetime import datetime, timezone


CSV_COLUMNS = [
    "query",
    "location",
    "position",
    "job_title",
    "company_name",
    "job_location",
    "source",
    "posted_at",
    "job_type",
    "salary",
    "description",
    "extensions",
    "job_id",
    "job_link",
    "apply_links",
    "collected_at"
]


def require_env(name):
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


def clean_text(value):
    if not value:
        return ""
    return " ".join(str(value).split())


def join_list(value):
    if not value:
        return ""

    if isinstance(value, list):
        return " | ".join(clean_text(item) for item in value if item)

    return clean_text(value)


def search_google_jobs(query, location="United States", language="en", country="us"):
    api_key = require_env("TALORDATA_API_KEY")
    endpoint = require_env("TALORDATA_SERP_ENDPOINT")

    payload = {
        "engine": "google_jobs",
        "q": query,
        "location": location,
        "hl": language,
        "gl": country,
        "output": "json"
    }

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }

    response = requests.post(
        endpoint,
        json=payload,
        headers=headers,
        timeout=30
    )

    response.raise_for_status()
    return response.json()


def get_job_results(serp_json):
    return (
        serp_json.get("jobs_results")
        or serp_json.get("job_results")
        or serp_json.get("jobs")
        or serp_json.get("organic_results")
        or []
    )


def extract_apply_links(item):
    apply_options = (
        item.get("apply_options")
        or item.get("apply_links")
        or item.get("related_links")
        or []
    )

    links = []

    for option in apply_options:
        if isinstance(option, dict):
            title = clean_text(option.get("title") or option.get("source"))
            link = option.get("link") or option.get("url")
            if link:
                links.append(f"{title}: {link}" if title else link)

    return " | ".join(links)


def normalize_job_results(serp_json, query, location):
    results = get_job_results(serp_json)
    collected_at = datetime.now(timezone.utc).isoformat()

    rows = []

    for index, item in enumerate(results, start=1):
        detected = item.get("detected_extensions") or {}

        rows.append({
            "query": query,
            "location": location,
            "position": item.get("position") or item.get("rank") or index,
            "job_title": clean_text(item.get("title")),
            "company_name": clean_text(item.get("company_name") or item.get("company")),
            "job_location": clean_text(item.get("location")),
            "source": clean_text(item.get("via") or item.get("source")),
            "posted_at": clean_text(detected.get("posted_at") or item.get("posted_at") or item.get("date")),
            "job_type": clean_text(detected.get("schedule_type") or item.get("job_type") or item.get("schedule_type")),
            "salary": clean_text(detected.get("salary") or item.get("salary")),
            "description": clean_text(item.get("description")),
            "extensions": join_list(item.get("extensions")),
            "job_id": item.get("job_id") or item.get("job_highlight_token") or item.get("id") or "",
            "job_link": item.get("link") or item.get("url") or "",
            "apply_links": extract_apply_links(item),
            "collected_at": collected_at
        })

    return rows


def export_to_csv(rows, filename="google_jobs_results.csv"):
    if not rows:
        print("No job listings found.")
        return

    with open(filename, mode="w", newline="", encoding="utf-8") as file:
        writer = csv.DictWriter(file, fieldnames=CSV_COLUMNS)
        writer.writeheader()
        writer.writerows(rows)

    print(f"Exported {len(rows)} job listings to {filename}")


if __name__ == "__main__":
    query = "python developer jobs"
    location = "New York, United States"

    serp_json = search_google_jobs(
        query=query,
        location=location,
        language="en",
        country="us"
    )

    rows = normalize_job_results(serp_json, query, location)
    export_to_csv(rows)

执行后会得到：

google_jobs_results.csv

输出示例

CSV 可能像这样：

query,location,position,job_title,company_name,job_location,source,posted_at,job_type,salary,description,extensions,job_id,job_link,apply_links,collected_at
python developer jobs,"New York, United States",1,Python Developer,Example Company,"New York, NY",LinkedIn,2 days ago,Full-time,"$120K-$150K","Build backend services in Python...",Full-time | Remote,abc123,https://example.com/job,"LinkedIn: https://example.com/apply",2026-06-22T00:00:00Z

接下来可以把数据发送到 Google Sheets、Airtable、database、recruiting dashboard 或 AI workflow。

常用参数

Google Jobs收集工作流程通常会用到这些参数：

Location 很重要。同一个 job title 的结果，会因 city、state、country 和 remote-work wording 发生明显差异。

常见使用场景

招聘市场调研

追踪不同城市或产业中哪些 roles 需求更高。

竞争对手招聘监控

用 company names 加 job titles 搜索，观察 competitors 正在招聘哪些职位。

薪资信号分析

在 salary ranges 可用时，按 role、location 或 company 比较薪资信号。

职位聚合

把 Google Jobs results 当成 discovery layer，再将 normalized rows 写入 database 或 matching system。

人工智能招聘工作流程

AI agent 可以根据 job search data 总结 hiring trends、按 function 分组 roles、识别 repeated skills，或生成 market snapshots。

Talordata 的 SERP API 产品页将 AI search 和 agent integration 列为使用场景之一，说明 real-time SERP data 可以作为 RAG pipelines 和 AI agents 的输入。

最佳实践

开发阶段保存 raw responses。Job result fields 可能会因 query、country 和 source platform 而变化。

仔细 normalize company names。同一家公司可能会以略有不同的名称出现。

尽可能保存 job IDs。它们有助于在重复采集时做 deduplication。

保存 timestamps。Job listings 会过期、移动或重新发布，所以每一行都应包含 collected_at。

分离 discovery 和 verification。SERP data 帮助你找到 listings，但 workflow 可能还需要抓取 original job page，以确认完整 job description、requirements 和 apply options。

不要假设每条 listing 都有 salary data。很多 job cards 不会展示 salary，因此分析时要处理 missing values。

FAQ

可以用 Python 采集 Google Jobs results 吗？

可以。你可以用 Python 搭配 SERP API 发送 job search query，并获取 structured job listings。上面的 script 示范了如何 normalize 常见 job fields，并导出为 CSV。

Google Jobs API 是 Google 官方 API 吗？

不是。这篇文章中的 Google Jobs API 指的是通过 SERP API 采集 Google Jobs-style search results，不是 Google 官方的一方 hiring 或 recruiting API。

可以从 job search results 采集哪些数据？

常见字段包括 job title、company name、location、source、posted date、job type、salary information、description、job ID、job link 和 apply links。

可以用于 recruiting analytics 吗？

可以。Job listing data 可以用来分析 hiring demand、competitor hiring activity、salary signals、role distribution 和 location trends。

应该把 job listings 保存为 CSV 还是 database？

小型测试和简单报告可以使用 CSV。若要做 recurring collection、deduplication、trend tracking 和 dashboards，database 更合适。