"""SmartRecruiters postings — public endpoint behind `careers.smartrecruiters.com/`: `GET https://api.smartrecruiters.com/v1/companies/{token}/postings?limit=100&offset=N` → {offset, limit, totalFound, content:[{id, name, uuid, refNumber, releasedDate, location:{city, region, country(iso2 lower), remote, hybrid, fullLocation}, department:{label}, function:{label}, typeOfEmployment:{label}, experienceLevel:{label}}]}. Paginated by offset (bounded). Verified 2026-09-12 against `smartrecruiters`.""" from __future__ import annotations import re from collections.abc import Mapping from typing import Any from companyatlas.connectors._ats_base import AtsConnector from companyatlas.connectors._util import country_code, load_json, merged_result, parse_date, text_of from companyatlas.fetch import Fetcher, FetchResult from companyatlas.sdk.connector import ConnectorContext, ConnectorMeta, register from companyatlas.sdk.models import ExtractedJob from companyatlas.taxonomy import FetchMode, Surface PAGE_SIZE = 100 MAX_PAGES = 10 @register class SmartRecruitersConnector(AtsConnector): vendor = "smartrecruiters" token_re = re.compile(r"api\.smartrecruiters\.com/v1/companies/([a-z0-9_-]+)/postings", re.IGNORECASE) meta = ConnectorMeta(connector_id="smartrecruiters-v1", name="SmartRecruiters postings", version="1", category=Surface.JOBS_BOARD, fetch_mode=FetchMode.JSON, default_interval_s=6 * 3600, url_pattern=r"api\.smartrecruiters\.com/v1/companies/", priority=50, pattern_required=True, accept="application/json", description="Public SmartRecruiters posting API with offset pagination") async def fetch(self, ctx: ConnectorContext, sensor: Mapping[str, Any], fetcher: Fetcher) -> FetchResult: base = str(sensor["url"]).split("&offset=")[0] first = await fetcher.get(base, etag=sensor.get("etag"), last_modified=sensor.get("last_modified"), accept="application/json") data = load_json(first) content = list(data.get("content") or []) if isinstance(data, dict) else [] total = int(data.get("totalFound") or len(content)) if isinstance(data, dict) else len(content) pages = 1 offset = len(content) while offset < total and pages < MAX_PAGES and content: page = await fetcher.get(f"{base}&offset={offset}", accept="application/json") more = load_json(page).get("content") or [] if not more: break content.extend(more) offset += len(more) pages += 1 if pages == 1: return first return merged_result(first, {"totalFound": total, "content": content}, pages=pages) def parse_jobs(self, data: Any, sensor: Mapping[str, Any]) -> list[ExtractedJob]: token = self.token(sensor) or "" out: list[ExtractedJob] = [] for j in (data.get("content") if isinstance(data, dict) else []) or []: if not isinstance(j, dict) or not j.get("name"): continue loc = j.get("location") or {} jid = str(j.get("id") or j.get("uuid") or "") url = f"https://jobs.smartrecruiters.com/{token}/{jid}" if token and jid else None remote = True if loc.get("remote") else (False if loc.get("hybrid") is not None or loc.get("city") else None) out.append(ExtractedJob( title=str(j["name"]).strip(), url=url, external_id=jid, department=text_of(j.get("department")), location_text=text_of(loc.get("fullLocation")) or ", ".join(x for x in (loc.get("city"), loc.get("region"), loc.get("country")) if x) or None, city=text_of(loc.get("city")), region=text_of(loc.get("region")), country=country_code(loc.get("country")), remote=remote, employment_type=text_of(j.get("typeOfEmployment")), seniority=None, posted_at=parse_date(j.get("releasedDate")), raw={"refNumber": j.get("refNumber"), "function": text_of(j.get("function")), "experienceLevel": text_of(j.get("experienceLevel"))}, )) return out