用 Python 调用 LinkedIn 职位搜索 API:远程岗位、时间筛选与分页
裴宁对接的不是招聘看板,是求职产品每周一发出去的那封订阅。通道只认附件,文件名写死 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 去重。