1
2"""ATS 职位板抓取 —— 工厂的第一个产品类型。
3
4为什么是这个类型(每一条都是实测,不是判断):
5
6 · **缺口层实测**:workday/wellfound/ashby/lever/remoteok 整簇落在「可进的空位」里,
7 头名运行 2,160-11,984,而总需求 6,926-26,096。有人在搜,没人认真做。
8 · **判断层实测**:招聘赛道自营占比 **0%**(平台没有自营 actor 挡路),
9 干净度 0.90(商业数据、不碰个人数据、不需要登录)。
10 · **技术前提实测**(2026-08-01,四个平台各打一次真实请求):
11 Greenhouse 548 条 (stripe) Lever 302 条 (palantir)
12 Ashby 126 条 (ramp) Workable 1,404 条 (toloka-ai)
13 全部 HTTP 200、无需鉴权、返回结构化 JSON。
14
15 这三条合起来才构成「造一次就不用维护」:**不是因为爬虫好写,
16 是因为这些平台把职位板当公开产品在维护,接口不会随便改。**
17 而普通网页爬虫的易碎正来自「目标站没义务对你稳定」。
18
19⚠️ 边界(必须写在代码里,不能只写在文档里):
20 · 只取**职位**数据 —— 公司名、岗位、地点、部门、链接。**不取任何候选人信息。**
21 · 不登录、不绕过任何鉴权。四个端点都是平台公开发布的。
22 · 抓到的是**企业发布的招聘信息**,不是个人数据。
23"""
24from __future__ import annotations
25
26import gzip
27import json
28import time
29import urllib.error
30import urllib.parse
31import urllib.request
32from typing import Any, Iterable
33
34UA = "Mozilla/5.0 (compatible; foundry-ats/1.0)"
35TIMEOUT = 30
36
37
38def _fetch(url: str, tries: int = 3) -> Any:
39 """带重试和 gzip 的 JSON 取数。
40
41 ⛔ gzip 这一段是实测逼出来的:第一次测四个平台,三个返回 HTTP 200 但
42 json.loads 全部失败,看着像「接口返回的不是 JSON」。
43 真相是 urllib 不会自动解压,而这些平台默认发 gzip。
44 **一个 200 加一个解析失败,长得很像「接口坏了」,其实是量具坏了。**
45 """
46 last = None
47 for i in range(tries):
48 try:
49 req = urllib.request.Request(
50 url, headers={"User-Agent": UA, "Accept-Encoding": "gzip",
51 "Accept": "application/json"})
52 with urllib.request.urlopen(req, timeout=TIMEOUT) as f:
53 raw = f.read()
54 if f.headers.get("Content-Encoding") == "gzip":
55 raw = gzip.decompress(raw)
56 return json.loads(raw.decode("utf-8", "replace"))
57 except urllib.error.HTTPError as e:
58 if e.code in (404, 403, 401):
59 raise LookupError(f"HTTP {e.code}") from e
60 last = e
61 except Exception as e:
62 last = e
63 time.sleep(1.2 * (i + 1))
64 raise RuntimeError(f"{type(last).__name__}: {str(last)[:80]}")
65
66
67def _s(v: Any) -> str:
68 return "" if v is None else str(v).strip()
69
70
71def _slug(v: str) -> str:
72 """把 board 标识安全地放进 URL。
73
74 ⛔ 自检抓到的真缺陷:标识里有非 ASCII 字符时,urllib 在构造请求那一刻
75 抛 UnicodeEncodeError,而不是干净地拿到 404。
76 两者的区别很要命 —— **一个是「这家公司不存在」,一个是「我们的代码有 bug」**,
77 混在一起之后,批量跑一千个目标时你分不出哪些是真没招人。
78 """
79 return urllib.parse.quote(str(v).strip(), safe="")
80
81
82
83
84FIELDS = ("board", "company", "job_id", "title", "location", "department",
85 "employment_type", "remote", "posted_at", "url", "source")
86
87
88def _row(**kw) -> dict:
89 r = {k: "" for k in FIELDS}
90 r.update({k: v for k, v in kw.items() if k in FIELDS})
91 return r
92
93
94
95
96def greenhouse(board: str) -> list[dict]:
97 """Greenhouse。board = 公司在 boards.greenhouse.io 上的标识,如 stripe。"""
98 d = _fetch(f"https://boards-api.greenhouse.io/v1/boards/{_slug(board)}/jobs?content=false")
99 out = []
100 for j in d.get("jobs") or []:
101 offices = j.get("offices") or []
102 loc = _s((j.get("location") or {}).get("name")) or \
103 ", ".join(_s(o.get("name")) for o in offices if o.get("name"))
104 depts = j.get("departments") or []
105 out.append(_row(
106 board="greenhouse", company=board, job_id=_s(j.get("id")),
107 title=_s(j.get("title")), location=loc,
108 department=", ".join(_s(x.get("name")) for x in depts if x.get("name")),
109 posted_at=_s(j.get("updated_at")), url=_s(j.get("absolute_url")),
110 remote="yes" if "remote" in loc.lower() else "",
111 source="boards-api.greenhouse.io"))
112 return out
113
114
115def lever(board: str) -> list[dict]:
116 """Lever。board = 公司标识,如 palantir。"""
117 d = _fetch(f"https://api.lever.co/v0/postings/{_slug(board)}?mode=json")
118 out = []
119 for j in d if isinstance(d, list) else []:
120 cat = j.get("categories") or {}
121 loc = _s(cat.get("location"))
122 out.append(_row(
123 board="lever", company=board, job_id=_s(j.get("id")),
124 title=_s(j.get("text")), location=loc,
125 department=_s(cat.get("team")) or _s(cat.get("department")),
126 employment_type=_s(cat.get("commitment")),
127 remote="yes" if "remote" in (loc + _s(cat.get("allLocations"))).lower() else "",
128 posted_at=_s(j.get("createdAt")), url=_s(j.get("hostedUrl")),
129 source="api.lever.co"))
130 return out
131
132
133def ashby(board: str) -> list[dict]:
134 """Ashby。board = 公司标识,如 ramp。"""
135 d = _fetch(f"https://api.ashbyhq.com/posting-api/job-board/{_slug(board)}")
136 out = []
137 for j in d.get("jobs") or []:
138 out.append(_row(
139 board="ashby", company=board, job_id=_s(j.get("id")),
140 title=_s(j.get("title")), location=_s(j.get("location")),
141 department=_s(j.get("department")) or _s(j.get("team")),
142 employment_type=_s(j.get("employmentType")),
143 remote="yes" if j.get("isRemote") else "",
144 posted_at=_s(j.get("publishedAt")), url=_s(j.get("jobUrl")),
145 source="api.ashbyhq.com"))
146 return out
147
148
149def workable(board: str) -> list[dict]:
150 """Workable。board = 公司标识,如 toloka-ai。"""
151 d = _fetch(f"https://apply.workable.com/api/v1/widget/accounts/{_slug(board)}")
152 out = []
153 for j in d.get("jobs") or []:
154 loc = ", ".join(x for x in (_s(j.get("city")), _s(j.get("state")),
155 _s(j.get("country"))) if x)
156 out.append(_row(
157 board="workable", company=_s(d.get("name")) or board,
158 job_id=_s(j.get("shortcode")), title=_s(j.get("title")),
159 location=loc, department=_s(j.get("department")),
160 employment_type=_s(j.get("employment_type")),
161 remote="yes" if j.get("telecommuting") else "",
162 posted_at=_s(j.get("published_on")), url=_s(j.get("url")),
163 source="apply.workable.com"))
164 return out
165
166
167def smartrecruiters(board: str) -> list[dict]:
168 """SmartRecruiters。board = 公司标识,如 Bosch。分页取全量。"""
169 out, offset = [], 0
170 while True:
171 d = _fetch("https://api.smartrecruiters.com/v1/companies/"
172 f"{_slug(board)}/postings?limit=100&offset={offset}")
173 items = d.get("content") or []
174 for j in items:
175 loc = j.get("location") or {}
176 out.append(_row(
177 board="smartrecruiters", company=board, job_id=_s(j.get("id")),
178 title=_s(j.get("name")),
179 location=", ".join(x for x in (_s(loc.get("city")),
180 _s(loc.get("region")),
181 _s(loc.get("country"))) if x),
182 department=_s((j.get("department") or {}).get("label")),
183 employment_type=_s((j.get("typeOfEmployment") or {}).get("label")),
184 remote="yes" if loc.get("remote") else "",
185 posted_at=_s(j.get("releasedDate")),
186 url=_s((j.get("ref") or {}).get("jobAd")) or _s(j.get("applyUrl")),
187 source="api.smartrecruiters.com"))
188 total = d.get("totalFound") or 0
189 offset += len(items)
190 if not items or offset >= total:
191 break
192 time.sleep(0.3)
193 return out
194
195
196BOARDS = {
197 "greenhouse": greenhouse,
198 "lever": lever,
199 "ashby": ashby,
200 "workable": workable,
201 "smartrecruiters": smartrecruiters,
202}
203
204
205def detect(url_or_slug: str) -> tuple[str, str]:
206 """从一个网址或标识里认出平台和 board。
207
208 买家给的往往是一个招聘页网址,而不是"平台名 + slug"。
209 认不出来就抛错,**不猜** —— 猜错会静默返回空列表,
210 而空列表和"这家公司没在招人"长得一模一样。
211 """
212 u = url_or_slug.strip().rstrip("/")
213 low = u.lower()
214 pats = [
215 ("greenhouse", ("boards.greenhouse.io/", "job-boards.greenhouse.io/")),
216 ("lever", ("jobs.lever.co/",)),
217 ("ashby", ("jobs.ashbyhq.com/",)),
218 ("workable", ("apply.workable.com/",)),
219 ("smartrecruiters", ("careers.smartrecruiters.com/", "jobs.smartrecruiters.com/")),
220 ]
221 for name, marks in pats:
222 for m in marks:
223 if m in low:
224 tail = low.split(m, 1)[1].split("/")[0].split("?")[0]
225 if tail:
226 return name, tail
227 if "/" not in u and ":" in u:
228 b, _, s = u.partition(":")
229 if b in BOARDS and s:
230 return b, s
231 raise ValueError(
232 "认不出这个招聘板:%s —— 支持 boards.greenhouse.io / jobs.lever.co / "
233 "jobs.ashbyhq.com / apply.workable.com / careers.smartrecruiters.com,"
234 "或者写成 平台:标识(如 greenhouse:stripe)" % url_or_slug[:60])
235
236
237def fetch(target: str) -> list[dict]:
238 """给一个网址或 平台:标识,返回统一结构的职位列表。"""
239 board, slug = detect(target)
240 return BOARDS[board](slug)
241
242
243def dedupe(rows: list[dict]) -> list[dict]:
244 """按 (平台, 公司, job_id) 去重,把多地点合并成一行。
245
246 ⛔ 实测抓到的真问题:Workable **同一个岗位按每个可接受地点各返回一行**。
247 toloka-ai 返回 1,404 行,唯一 job_id 只有 **293** ——
248 买家直接数行数会以为这家在招 1,404 个人,实际是 293 个。
249 (Greenhouse / Lever / Ashby 零重复,这是 Workable 特有的行为。)
250
251 两种读法都成立:要"有多少个岗位"就去重,要"岗位 × 地点"就别去重。
252 **错的是不告诉人家有这回事** —— 那是典型的「看起来没问题但是错的」。
253 所以默认保真不去重,但两个数永远一起报出来。
254 """
255 seen, out = set(), []
256 for r in rows:
257 k = (r["board"], r["company"], r["job_id"])
258 if k in seen:
259
260 for o in out:
261 if (o["board"], o["company"], o["job_id"]) == k:
262 if r["location"] and r["location"] not in o["location"]:
263 o["location"] = (o["location"] + " | " + r["location"])[:300]
264 break
265 continue
266 seen.add(k)
267 out.append(dict(r))
268 return out
269
270
271def unique_count(rows: list[dict]) -> int:
272 return len({(r["board"], r["company"], r["job_id"]) for r in rows})
273
274
275def fetch_many(targets: Iterable[str], pause: float = 0.4) -> tuple[list[dict], list[dict]]:
276 """批量。返回 (职位, 失败记录)。**失败必须被返回,不能吞掉** ——
277 吞掉之后「这家没在招」和「我们抓挂了」就分不开了。"""
278 rows, errs = [], []
279 for t in targets:
280 try:
281 got = fetch(t)
282 rows.extend(got)
283 if not got:
284 errs.append({"target": t, "error": "取到 0 条(可能确实没在招)",
285 "fatal": False})
286 except Exception as e:
287 errs.append({"target": t, "error": f"{type(e).__name__}: {str(e)[:120]}",
288 "fatal": True})
289 time.sleep(pause)
290 return rows, errs
