1"""Normalisation helpers for each supported ATS.
2
3Every provider exposes a public, unauthenticated job-board API that the ATS
4publishes so that job aggregators can consume it. No login, no anti-bot
5circumvention, no proxies.
6
7robots.txt status checked 2026-09-10:
8 boards-api.greenhouse.io -> "Disallow: /embed/" only, /v1/boards/* allowed
9 api.lever.co -> "Allow: /" with "Crawl-delay: 1"
10 api.ashbyhq.com -> serves no robots.txt (401); documented public API
11 apply.workable.com -> "Disallow:" (empty, i.e. everything allowed)
12
13Deliberately NOT supported because their robots.txt forbids it for generic
14user agents:
15 api.smartrecruiters.com -> "User-agent: * / Disallow: /"
16 api.recruitee.com -> "User-agent: * / Disallow: /"
17"""
18
19from __future__ import annotations
20
21import html
22import re
23from datetime import datetime, timezone
24from typing import Any
25from urllib.parse import urlparse
26
27PROVIDERS = ('greenhouse', 'lever', 'ashby', 'workable')
28
29
30TOKEN_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$')
31
32
33MIN_DELAY = {'greenhouse': 0.0, 'lever': 1.0, 'ashby': 0.0, 'workable': 0.0}
34
35_TAG_RE = re.compile(r'<[^>]+>')
36_WS_RE = re.compile(r'[ \t\r\f\v]+')
37
38
39def strip_html(raw: str | None) -> str | None:
40 """Turn an HTML fragment into readable plain text."""
41 if not raw:
42 return None
43 text = html.unescape(raw)
44
45 if '<' in text or '>' in text:
46 text = html.unescape(text)
47 text = re.sub(r'<br\s*/?>', '\n', text, flags=re.I)
48 text = re.sub(r'</(p|div|li|h[1-6]|tr)>', '\n', text, flags=re.I)
49 text = _TAG_RE.sub('', text)
50 text = html.unescape(text)
51 text = _WS_RE.sub(' ', text)
52 text = re.sub(r'\n{3,}', '\n\n', text)
53 return text.strip() or None
54
55
56def _iso(value: Any) -> str | None:
57 """Best-effort conversion of an ATS timestamp to an ISO-8601 UTC string."""
58 if value in (None, '', 0):
59 return None
60 try:
61 if isinstance(value, (int, float)) or (isinstance(value, str) and value.isdigit()):
62 num = float(value)
63
64 if num > 1e11:
65 num /= 1000.0
66 return datetime.fromtimestamp(num, tz=timezone.utc).isoformat()
67 text = str(value).strip().replace('Z', '+00:00')
68 parsed = datetime.fromisoformat(text)
69 if parsed.tzinfo is None:
70 parsed = parsed.replace(tzinfo=timezone.utc)
71 return parsed.astimezone(timezone.utc).isoformat()
72 except (ValueError, OverflowError, OSError):
73 return None
74
75
76def parse_company(entry: str, default_provider: str) -> tuple[str, str]:
77 """Resolve one user-supplied company entry to a (provider, token) pair.
78
79 Raises ValueError when the entry cannot be understood.
80 """
81 entry = (entry or '').strip()
82 if not entry:
83 raise ValueError('empty entry')
84
85 if entry.startswith(('http://', 'https://')):
86 url = urlparse(entry)
87 host = (url.hostname or '').lower()
88 parts = [p for p in url.path.split('/') if p]
89 if 'greenhouse.io' in host:
90
91
92 token = url.query.split('for=')[-1].split('&')[0] if 'for=' in url.query else None
93 if not token and parts:
94 token = parts[-1] if parts[0] == 'embed' else parts[0]
95 if token:
96 return 'greenhouse', token
97 elif 'lever.co' in host and parts:
98 return 'lever', parts[0]
99 elif 'ashbyhq.com' in host and parts:
100 return 'ashby', parts[0]
101 elif 'workable.com' in host and parts:
102
103
104
105 if parts[0].lower() == 'j':
106 raise ValueError(
107 f'{entry} is a Workable job short-link, not a company board '
108 '(use https://apply.workable.com/<company>/)'
109 )
110 return 'workable', parts[0]
111 raise ValueError(f'unrecognised job board URL: {entry}')
112
113 if ':' in entry:
114 provider, _, token = entry.partition(':')
115 provider = provider.strip().lower()
116 token = token.strip().strip('/')
117 if provider not in PROVIDERS:
118 raise ValueError(
119 f'unknown provider in "{entry}" (expected one of {", ".join(PROVIDERS)})'
120 )
121 if not TOKEN_RE.match(token):
122 raise ValueError(f'"{token}" is not a valid board token')
123 return provider, token
124
125 token = entry.strip('/')
126 if not TOKEN_RE.match(token):
127 raise ValueError(
128 f'"{entry}" is neither a job board URL, a provider:token pair, nor a valid board token'
129 )
130 return default_provider, token
131
132
133def board_url(provider: str, token: str, include_description: bool) -> str:
134 if provider == 'greenhouse':
135 flag = 'true' if include_description else 'false'
136 return f'https://boards-api.greenhouse.io/v1/boards/{token}/jobs?content={flag}'
137 if provider == 'lever':
138 return f'https://api.lever.co/v0/postings/{token}?mode=json'
139 if provider == 'ashby':
140 return f'https://api.ashbyhq.com/posting-api/job-board/{token}?includeCompensation=true'
141 if provider == 'workable':
142 return f'https://apply.workable.com/api/v1/widget/accounts/{token}?details=true'
143 raise ValueError(f'unsupported provider {provider}')
144
145
146def _base(provider: str, token: str) -> dict[str, Any]:
147 return {
148 'source': provider,
149 'companySlug': token,
150 'companyName': None,
151 'jobId': None,
152 'title': None,
153 'url': None,
154 'applyUrl': None,
155 'location': None,
156 'locations': [],
157 'country': None,
158 'department': None,
159 'team': None,
160 'employmentType': None,
161 'isRemote': None,
162 'workplaceType': None,
163 'postedAt': None,
164 'updatedAt': None,
165 'salaryText': None,
166 'descriptionText': None,
167 'descriptionHtml': None,
168 'boardUrl': None,
169 }
170
171
172def _remote_from_text(*values: Any) -> bool:
173 return any('remote' in str(v).lower() for v in values if v)
174
175
176def normalise(provider: str, token: str, payload: Any, include_description: bool) -> list[dict[str, Any]]:
177 """Convert a raw board payload into a list of normalised job records."""
178 if provider == 'greenhouse':
179 return _norm_greenhouse(token, payload, include_description)
180 if provider == 'lever':
181 return _norm_lever(token, payload, include_description)
182 if provider == 'ashby':
183 return _norm_ashby(token, payload, include_description)
184 if provider == 'workable':
185 return _norm_workable(token, payload, include_description)
186 return []
187
188
189def _norm_greenhouse(token: str, payload: Any, include_description: bool) -> list[dict[str, Any]]:
190 out = []
191 for job in (payload or {}).get('jobs', []) or []:
192 item = _base('greenhouse', token)
193 loc = (job.get('location') or {}).get('name')
194 offices = [o.get('name') for o in (job.get('offices') or []) if o.get('name')]
195 departments = [d.get('name') for d in (job.get('departments') or []) if d.get('name')]
196 item.update({
197 'companyName': job.get('company_name'),
198 'jobId': str(job.get('id')) if job.get('id') is not None else None,
199 'title': (job.get('title') or '').strip() or None,
200 'url': job.get('absolute_url'),
201 'applyUrl': job.get('absolute_url'),
202 'location': loc,
203 'locations': offices or ([loc] if loc else []),
204 'department': departments[0] if departments else None,
205 'postedAt': _iso(job.get('first_published')),
206 'updatedAt': _iso(job.get('updated_at')),
207 'isRemote': _remote_from_text(loc, *offices),
208 'boardUrl': f'https://job-boards.greenhouse.io/{token}',
209 })
210 if include_description and job.get('content'):
211 item['descriptionHtml'] = html.unescape(job['content'])
212 item['descriptionText'] = strip_html(job['content'])
213 out.append(item)
214 return out
215
216
217def _norm_lever(token: str, payload: Any, include_description: bool) -> list[dict[str, Any]]:
218 out = []
219 for job in payload or []:
220 cats = job.get('categories') or {}
221 item = _base('lever', token)
222 all_locations = [x for x in (cats.get('allLocations') or []) if x]
223 item.update({
224 'jobId': job.get('id'),
225 'title': (job.get('text') or '').strip() or None,
226 'url': job.get('hostedUrl'),
227 'applyUrl': job.get('applyUrl'),
228 'location': cats.get('location'),
229 'locations': all_locations or ([cats['location']] if cats.get('location') else []),
230 'country': job.get('country'),
231 'department': cats.get('department'),
232 'team': cats.get('team'),
233 'employmentType': cats.get('commitment'),
234 'workplaceType': job.get('workplaceType'),
235 'isRemote': (job.get('workplaceType') == 'remote')
236 or _remote_from_text(cats.get('location'), *all_locations),
237 'postedAt': _iso(job.get('createdAt')),
238 'salaryText': (job.get('salaryRange') or {}).get('text')
239 if isinstance(job.get('salaryRange'), dict) else None,
240 'boardUrl': f'https://jobs.lever.co/{token}',
241 })
242 if include_description:
243 item['descriptionHtml'] = job.get('description')
244 item['descriptionText'] = job.get('descriptionPlain') or strip_html(job.get('description'))
245 out.append(item)
246 return out
247
248
249def _norm_ashby(token: str, payload: Any, include_description: bool) -> list[dict[str, Any]]:
250 out = []
251 for job in (payload or {}).get('jobs', []) or []:
252 if job.get('isListed') is False:
253 continue
254 item = _base('ashby', token)
255 secondary = [
256 s.get('location') for s in (job.get('secondaryLocations') or []) if s.get('location')
257 ]
258 addr = ((job.get('address') or {}).get('postalAddress') or {})
259 comp = job.get('compensation') or {}
260 item.update({
261 'jobId': job.get('id'),
262 'title': (job.get('title') or '').strip() or None,
263 'url': job.get('jobUrl'),
264 'applyUrl': job.get('applyUrl'),
265 'location': job.get('location'),
266 'locations': ([job['location']] if job.get('location') else []) + secondary,
267 'country': addr.get('addressCountry'),
268 'department': job.get('department'),
269 'team': job.get('team'),
270 'employmentType': job.get('employmentType'),
271 'workplaceType': job.get('workplaceType'),
272 'isRemote': bool(job.get('isRemote')),
273 'postedAt': _iso(job.get('publishedAt')),
274 'salaryText': comp.get('scrapeableCompensationSalarySummary')
275 or comp.get('compensationTierSummary'),
276 'boardUrl': f'https://jobs.ashbyhq.com/{token}',
277 })
278 if include_description:
279 item['descriptionHtml'] = job.get('descriptionHtml')
280 item['descriptionText'] = job.get('descriptionPlain') or strip_html(job.get('descriptionHtml'))
281 out.append(item)
282 return out
283
284
285def _norm_workable(token: str, payload: Any, include_description: bool) -> list[dict[str, Any]]:
286 payload = payload or {}
287 company = payload.get('name') or token
288 out = []
289 for job in payload.get('jobs', []) or []:
290
291
292
293 texts: list[str] = []
294
295 def _add(value: Any) -> None:
296 if isinstance(value, dict):
297 text = ', '.join(
298 str(x)
299 for x in (
300 value.get('city'),
301 value.get('region') or value.get('state'),
302 value.get('country'),
303 )
304 if x
305 )
306 else:
307 text = str(value).strip() if value else ''
308 if text and text not in texts:
309 texts.append(text)
310
311 raw_locations = job.get('locations')
312 if isinstance(raw_locations, list):
313 for entry in raw_locations:
314 _add(entry)
315 _add(job.get('location'))
316 _add(
317 ', '.join(
318 str(x)
319 for x in (job.get('city'), job.get('state'), job.get('country'))
320 if x
321 )
322 )
323
324 loc_text = texts[0] if texts else None
325 country = job.get('country')
326 if not country and isinstance(job.get('location'), dict):
327 country = job['location'].get('country')
328 if not country and isinstance(raw_locations, list) and raw_locations:
329 first = raw_locations[0]
330 if isinstance(first, dict):
331 country = first.get('country')
332
333 item = _base('workable', token)
334 item.update({
335 'companyName': company,
336 'jobId': job.get('shortcode') or job.get('id'),
337 'title': (job.get('title') or '').strip() or None,
338 'url': job.get('url') or job.get('shortlink'),
339 'applyUrl': job.get('application_url') or job.get('url'),
340 'location': loc_text,
341 'locations': texts,
342 'country': country,
343 'department': job.get('department') or None,
344 'team': job.get('function') or None,
345 'employmentType': job.get('employment_type') or None,
346 'isRemote': bool(job.get('telecommuting')) or _remote_from_text(*texts),
347 'workplaceType': 'Remote' if job.get('telecommuting') else None,
348 'postedAt': _iso(job.get('published_on') or job.get('created_at')),
349 'boardUrl': f'https://apply.workable.com/{token}/',
350 })
351 if include_description:
352 item['descriptionHtml'] = job.get('description')
353 item['descriptionText'] = strip_html(job.get('description'))
354 out.append(item)
355 return out
