spb/company-atlas

1"""Base class for structured job-board connectors (spec §9.3): the sensor URL is the public JSON/XML endpoint the vendor's own2career page loads. Each subclass implements `parse_jobs(data, sensor)`; the base turns the list into an `Extraction` whose blocks3mirror the jobs one-to-one (stable keys from external ids) so the block diff equals the job delta."""4from __future__ import annotations56import re7from collections.abc import Mapping8from typing import Any, ClassVar910from companyatlas.connectors._util import finish_job, job_blocks, jobs_text, load_json11from companyatlas.fetch import FetchResult12from companyatlas.sdk.connector import Connector13from companyatlas.sdk.models import ExtractedJob, Extraction14from companyatlas.taxonomy import Surface1516MAX_JOBS = 5000171819class AtsConnector(Connector):20    vendor: ClassVar[str] = ""21    token_re: ClassVar[re.Pattern[str] | None] = None2223    def token(self, sensor: Mapping[str, Any]) -> str | None:24        cfg = sensor.get("config") or {}25        if isinstance(cfg, Mapping) and cfg.get("token"):26            return str(cfg["token"])27        if self.token_re is not None:28            m = self.token_re.search(str(sensor.get("url") or ""))29            if m:30                return m.group(1)31        return None3233    def load(self, result: FetchResult) -> Any:34        return load_json(result)3536    def parse_jobs(self, data: Any, sensor: Mapping[str, Any]) -> list[ExtractedJob]:  # pragma: no cover - abstract37        raise NotImplementedError3839    def extract(self, sensor: Mapping[str, Any], result: FetchResult) -> Extraction:40        data = self.load(result)41        jobs = [finish_job(j) for j in self.parse_jobs(data, sensor)[:MAX_JOBS] if j.title]42        # dedupe by external id / url43        seen: set[str] = set()44        unique: list[ExtractedJob] = []45        for j in jobs:46            k = (j.external_id or j.url or f"{j.title}|{j.location_text}").lower()47            if k in seen:48                continue49            seen.add(k)50            unique.append(j)51        header = f"{self.vendor} board — {len(unique)} open positions"52        meta = {"vendor": self.vendor, "token": self.token(sensor), "job_count": len(unique), "source_url": result.final_url,53                "structured": True, "pages": result.headers.get("x-companyatlas-pages", "1")}54        return Extraction(text=jobs_text(unique, header), blocks=job_blocks(unique, path=f"{self.vendor} jobs"), title=header,55                          language=None, meta=meta, jobs=unique)565758def surface_jobs_board() -> str:59    return str(Surface.JOBS_BOARD)606162__all__ = ["MAX_JOBS", "AtsConnector", "surface_jobs_board"]63