用 Python 调用 LinkedIn 职位搜索 API:远程岗位、时间筛选与分页

发布于 2026年10月16日 · 更新于2026年10月16日

裴宁对接的不是招聘看板,是求职产品每周一发出去的那封订阅。通道只认附件,文件名写死 remote-jobs.csv:关键词 software engineer,只要远程,时间窗是最近 24 小时,地点只能是 geocode,同一职位 id 不能出现两行。详情、公司主页、在招列表、岗位计数都不进这张表。订阅服务拿它和上周的文件做 diff,没有人会再去点领英。把城市名塞进搜索、把现场岗混进远程筛选、第二页把第一页的职位再写一遍——标题还没人看,这封邮件就已经错了。

LinkedIn 数据 API 是第三方只读接口,不是官方产品,也不是合作伙伴授权。某次首页样例返回过 25 条,不能当成筛选、翻页或更新时效的保证。打开本页或复制脚本不会发请求,只有你真正跑起来的调用才会。

登录后在控制台取有权使用的 Key,放进服务端环境变量 OPENDATA_API_KEY。文中 YOUR_API_KEY 只是占位。Key 走查询参数,不要写进工单、截图和浏览器前端。不需要领英账号、Cookie 或会话。

招聘订阅真正会发出的文件

产物就叫 remote-jobs.csv。每行共用:实际发出的 geocode、date_posted、采集时间 observed_at_utc。岗位列:字符串 id、title、url、listed_at、location、公司名。只要两次调用,不是接口目录。

search-location 把可读地名变成 location / geocode 候选。search-jobs 必填 keyword,这张表再带 geocode、remote、date_posted、sort_by、page。get-job-detail 是拿到 id 之后才补描述、办公方式、work_remote_allowed 和薪资元数据的;include_skills=true 可能多耗额度。公司资料、公司岗位、岗位计数属于雇主维度的另一张表,不进这份订阅。

easy_apply 不会代投。列表上的 is_easy_apply、is_promote 只是属性,不是招聘效果。响应里的 cost 是元数据,不是价目表。

先解析 geocode 再筛远程

职位搜索不认「美国」三个字,认 geocode。先用 search-location 查 keyword=United States,对着返回的地点标签挑一条,再把编码交给 search-jobs。文档里的美国示例是 103644278。不要因为第一条像就用,也不要把展示名直接填进搜索。人才搜索里同类编码叫 geocode_location,招聘这边不是这个参数名。

远程不是布尔值:onsite、remote、hybrid。远程岗仍然要带地理编码——远程不等于全球可入职。时间窗是 anytime、past_month、past_week、past_24_hours。这封每日订阅用 past_24_hours。不要套用帖子接口的 past_24h,也不要按别家招聘 API 的 start / count 翻页;这里按 page。盯新岗位时加 sort_by=recent。

先看 HTTP:4xx / 5xx 是传输、鉴权或额度,请求没跑成。再看信封 {code, message, data}:code 为 200 只说明平台受理了,不说明有岗位。信封里的 data 还有自己的 success。资料不可访问时曾出现 HTTP 200、success: false、data: null、业务 status_code: 403;HTTP 202 也可能带着业务失败。不要把空当成「市场里确实没有」,也不要把访问失败当成页码写错再重试。

翻页、去重、停下来

从 page=1 起。只有 has_more 为 true、本页至少新增一条 id、且未碰到页数上限,才翻下一页。去重用 str(job["id"])。结果会漂:第一页出现过的岗位,在 sort_by=recent 重排后可能在第二页再来一次。返回的 total 不是「一定能下载完」的承诺。

has_more 为 false、空页、整页都是重复、缺少继续信号、或额度用尽,就停。不要自己编 pagination_token——本 API 里别的接口才用游标,招聘搜索不用。订阅跨天跑,已见 id 要落盘,内存集合进程一停就没了。失败跑可能留下半份文件,别标成完整结果。

id 当字符串。listed_at 保持原样,另记你自己的 UTC 采集时间。以后如果补详情,薪资缺失是未知,不是零;这份 CSV 不调详情。

每周一寄出 remote-jobs.csv

安装 requests,在服务端导出 OPENDATA_API_KEY,当任务跑。脚本先解析美国 geocode,再按 remote=remote、date_posted=past_24_hours 翻 search-jobs,按职位 id 去重,写出 remote-jobs.csv。缺字段就空着。筛选项是文档里的输入,不保证有结果。

import csv
import os
import time
from datetime import datetime, timezone

import requests

BASE = "https://api.opendata-api.com/api/project/linkedin-scraper"
API_KEY = os.environ["OPENDATA_API_KEY"]
KEYWORD = "software engineer"
LOCATION_KEYWORD = "United States"
DATE_POSTED = "past_24_hours"
REMOTE = "remote"
MAX_PAGES = 5
OUT = "remote-jobs.csv"


def read(operation, **params):
    for attempt in range(3):
        try:
            response = requests.get(
                f"{BASE}/{operation}",
                params={"key": API_KEY, **params},
                timeout=30,
            )
        except requests.RequestException:
            if attempt < 2:
                time.sleep(2 ** (attempt + 1))
                continue
            raise RuntimeError("Network failure; request URL withheld") from None
        if response.status_code in (429, 500, 502, 503, 504) and attempt < 2:
            time.sleep(2 ** (attempt + 1))
            continue
        if not 200 <= response.status_code < 300:
            raise RuntimeError(f"HTTP failure: {response.status_code}")
        try:
            body = response.json()
        except ValueError:
            raise RuntimeError("Expected JSON") from None
        if not isinstance(body, dict) or body.get("code") != 200:
            raise RuntimeError("Gateway business failure")
        payload = body.get("data")
        if not isinstance(payload, dict) or payload.get("success") is not True:
            raise RuntimeError("Operation failed; review redacted status")
        return payload


def resolve_geocode(keyword):
    payload = read("search-location", keyword=keyword)
    rows = payload.get("data")
    if not isinstance(rows, list) or not rows:
        raise RuntimeError("No location matches")
    chosen = None
    for row in rows:
        if not isinstance(row, dict):
            continue
        if str(row.get("location", "")).lower() == keyword.lower():
            chosen = row
            break
    if chosen is None:
        chosen = rows[0]  # ranked fallback; store the geocode you actually sent
    geocode = chosen.get("geocode")
    if not geocode:
        raise RuntimeError("Match missing geocode")
    return str(geocode)


def collect(geocode):
    seen = set()
    for page in range(1, MAX_PAGES + 1):
        payload = read(
            "search-jobs",
            keyword=KEYWORD,
            geocode=geocode,
            remote=REMOTE,
            date_posted=DATE_POSTED,
            sort_by="recent",
            page=page,
        )
        jobs = payload.get("data")
        if not isinstance(jobs, list):
            raise RuntimeError("Expected job array")
        added = 0
        observed = datetime.now(timezone.utc).isoformat()
        for job in jobs:
            if not isinstance(job, dict) or not job.get("id"):
                raise RuntimeError("Missing job identifier")
            job_id = str(job["id"])
            if job_id in seen:
                continue
            seen.add(job_id)
            added += 1
            company = job.get("company")
            yield {
                "id": job_id,
                "title": job.get("title"),
                "url": job.get("url"),
                "listed_at": job.get("listed_at"),
                "location": job.get("location"),
                "company": company.get("name") if isinstance(company, dict) else None,
                "geocode": geocode,
                "date_posted": DATE_POSTED,
                "observed_at_utc": observed,
            }
        if not added or payload.get("has_more") is not True:
            break
        time.sleep(1)  # client spacing, not a published limit


def cell(value):
    text = "" if value is None else str(value)
    # Prevent returned text from becoming a spreadsheet formula.
    return "'" + text if text.lstrip().startswith(("=", "+", "-", "@")) else text


if __name__ == "__main__":
    geocode = resolve_geocode(LOCATION_KEYWORD)
    fields = [
        "id", "title", "url", "listed_at", "location",
        "company", "geocode", "date_posted", "observed_at_utc",
    ]
    rows = list(collect(geocode))
    if not rows:
        raise SystemExit("No job rows to write")
    with open(OUT, "w", newline="", encoding="utf-8") as output:
        writer = csv.DictWriter(output, fieldnames=fields)
        writer.writeheader()
        for row in rows:
            writer.writerow({key: cell(value) for key, value in row.items()})

整份交付就是这一张远程岗位 CSV:geocode、时间窗、按职位 id 去重。