553 lines
18 KiB
Python
553 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from abc import ABC, abstractmethod
|
|
from urllib.parse import urljoin, urlsplit
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
from .base import ExtractedJob, ExtractionResult, FieldEvidence
|
|
|
|
CLOSED_STATUSES = {
|
|
"closed",
|
|
"inactive",
|
|
"withdrawn",
|
|
"removed",
|
|
"filled",
|
|
"expired",
|
|
"archived",
|
|
}
|
|
|
|
|
|
def _to_text(value):
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, str):
|
|
return value.strip()
|
|
if isinstance(value, bool):
|
|
return str(value).lower()
|
|
if isinstance(value, int | float):
|
|
return str(value)
|
|
return ""
|
|
|
|
|
|
def _find_nested(data, path: str):
|
|
if data is None:
|
|
return None
|
|
node = data
|
|
for part in path.split("."):
|
|
if not isinstance(node, dict) or part not in node:
|
|
return None
|
|
node = node[part]
|
|
return node
|
|
|
|
|
|
def _first_text(data, *candidates):
|
|
for candidate in candidates:
|
|
if "." in candidate:
|
|
value = _find_nested(data, candidate)
|
|
else:
|
|
value = data.get(candidate) if isinstance(data, dict) else None
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, list | tuple):
|
|
for item in value:
|
|
text = _to_text(item)
|
|
if text:
|
|
return text
|
|
continue
|
|
text = _to_text(value)
|
|
if text:
|
|
return text
|
|
if isinstance(value, dict):
|
|
nested = value.values()
|
|
for nested_value in nested:
|
|
nested_text = _to_text(nested_value)
|
|
if nested_text:
|
|
return nested_text
|
|
return ""
|
|
|
|
|
|
def _collect_texts(value) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, str):
|
|
return [value.strip()] if value.strip() else []
|
|
if isinstance(value, list):
|
|
out: list[str] = []
|
|
for item in value:
|
|
for inner in _collect_texts(item):
|
|
if inner:
|
|
out.append(inner)
|
|
return out
|
|
if isinstance(value, dict):
|
|
out: list[str] = []
|
|
for key in ("name", "city", "location", "address", "raw", "region", "country"):
|
|
if key in value:
|
|
for inner in _collect_texts(value[key]):
|
|
if inner:
|
|
out.append(inner)
|
|
return out
|
|
if isinstance(value, bool | int | float):
|
|
return [_to_text(value)]
|
|
return []
|
|
|
|
|
|
def _to_list(value) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, list):
|
|
return [item.strip() for item in (_to_text(v) for v in value) if item.strip()]
|
|
text = _to_text(value)
|
|
if not text:
|
|
return []
|
|
return [part.strip() for part in text.replace(";", ",").split(",") if part.strip()]
|
|
|
|
|
|
def _as_text(value) -> str:
|
|
text = _to_text(value)
|
|
if not text:
|
|
return ""
|
|
return BeautifulSoup(text, "lxml").get_text(" ", strip=True)
|
|
|
|
|
|
def _extract_payload(content: str):
|
|
try:
|
|
return json.loads(content.lstrip("\ufeff"))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
soup = BeautifulSoup(content, "lxml")
|
|
for script in soup.find_all("script", attrs={"type": "application/json"}):
|
|
script_text = script.string or script.get_text("", strip=True)
|
|
if not script_text:
|
|
continue
|
|
try:
|
|
return json.loads(script_text.lstrip("\ufeff"))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return None
|
|
|
|
|
|
class _AtsAdapter(ABC):
|
|
parser_key = "ats-provider"
|
|
parser_version = "1.0.0"
|
|
source_hosts: tuple[str, ...] = ()
|
|
support_markers: tuple[str, ...] = ()
|
|
listing_paths: tuple[tuple[str, ...], ...] = ()
|
|
detail_paths: tuple[tuple[str, ...], ...] = ()
|
|
closed_statuses = CLOSED_STATUSES
|
|
|
|
@abstractmethod
|
|
def _extract_records(self, payload) -> list[dict[str, object]]: ...
|
|
|
|
def _supports_url(self, url: str) -> bool:
|
|
host = (urlsplit(url).hostname or "").lower()
|
|
return any(host.endswith(suffix) for suffix in self.source_hosts)
|
|
|
|
def _supports_payload(self, payload, content: str) -> bool:
|
|
if not payload:
|
|
return False
|
|
payload_text = str(payload).lower()
|
|
return any(
|
|
marker in payload_text or marker in content.lower() for marker in self.support_markers
|
|
)
|
|
|
|
def _job_is_closed(self, record: dict[str, object]) -> bool:
|
|
status = _first_text(record, "status", "state", "job_status", "data.status").lower()
|
|
if status in self.closed_statuses:
|
|
return True
|
|
active = record.get("active")
|
|
if isinstance(active, bool):
|
|
return not active
|
|
return False
|
|
|
|
def _coerce_url(self, raw_url: str, base_url: str) -> str:
|
|
if raw_url:
|
|
return urljoin(base_url, raw_url)
|
|
return base_url
|
|
|
|
def _extract_location(self, record: dict[str, object]) -> tuple[str, str, str, str]:
|
|
location_raw = _first_text(
|
|
record,
|
|
"location",
|
|
"city",
|
|
"cityName",
|
|
"place",
|
|
"office",
|
|
"location.address",
|
|
"address",
|
|
"data.location",
|
|
"officeLocation",
|
|
"locationName",
|
|
)
|
|
if not location_raw:
|
|
location_raw = _first_text(record, "data.city", "data.location")
|
|
if not location_raw:
|
|
location_values = []
|
|
for key in ("city", "address", "location", "region", "office"):
|
|
location_values.extend(_collect_texts(record.get(key, "")))
|
|
location_raw = ", ".join(location_values)
|
|
|
|
location = location_raw
|
|
[part.strip() for part in _to_text(location).split(",") if part.strip()]
|
|
region = _first_text(
|
|
record,
|
|
"region",
|
|
"data.region",
|
|
"location.region",
|
|
"address.region",
|
|
)
|
|
postal_code = _first_text(
|
|
record,
|
|
"postal_code",
|
|
"postalCode",
|
|
"location.postalCode",
|
|
"zipCode",
|
|
)
|
|
country = _first_text(
|
|
record,
|
|
"country",
|
|
"countryName",
|
|
"location.country",
|
|
"address.country",
|
|
"data.country",
|
|
)
|
|
return location, region, postal_code, country
|
|
|
|
def _extract_workplace(self, record: dict[str, object]) -> str:
|
|
workplace_raw = _first_text(
|
|
record,
|
|
"workplaceType",
|
|
"workplace",
|
|
"remoteType",
|
|
"remote_type",
|
|
"job_type",
|
|
).lower()
|
|
if "remote" in workplace_raw:
|
|
return "remote" if "hybrid" not in workplace_raw else "hybrid"
|
|
if "hybrid" in workplace_raw:
|
|
return "hybrid"
|
|
if "onsite" in workplace_raw or "on site" in workplace_raw or "on-site" in workplace_raw:
|
|
return "on_site"
|
|
return ""
|
|
|
|
def _to_job(self, record: dict[str, object], base_url: str) -> ExtractedJob:
|
|
external_id = _first_text(
|
|
record,
|
|
"id",
|
|
"jobId",
|
|
"requisitionId",
|
|
"postingId",
|
|
"referenceId",
|
|
"positionId",
|
|
)
|
|
title = _first_text(
|
|
record,
|
|
"title",
|
|
"name",
|
|
"position",
|
|
"positionName",
|
|
"jobTitle",
|
|
"text",
|
|
"title.value",
|
|
)
|
|
raw_url = _first_text(
|
|
record,
|
|
"url",
|
|
"jobUrl",
|
|
"job_url",
|
|
"applyUrl",
|
|
"link",
|
|
"absoluteUrl",
|
|
"data.url",
|
|
)
|
|
employer_name = _first_text(
|
|
record,
|
|
"company",
|
|
"employer",
|
|
"organization",
|
|
"organizationName",
|
|
"company.name",
|
|
"department",
|
|
"hiringOrganization",
|
|
)
|
|
location_text, region, postal_code, country = self._extract_location(record)
|
|
description_raw = _first_text(
|
|
record,
|
|
"description",
|
|
"jobDescription",
|
|
"content",
|
|
"descriptionHtml",
|
|
"descriptionText",
|
|
)
|
|
description_text = _as_text(description_raw)
|
|
date_posted = _first_text(
|
|
record,
|
|
"createdAt",
|
|
"created",
|
|
"datePosted",
|
|
"publishedAt",
|
|
"published_at",
|
|
"created_at",
|
|
"jobCreated",
|
|
)
|
|
valid_through = _first_text(
|
|
record,
|
|
"closeDate",
|
|
"expiresAt",
|
|
"validThrough",
|
|
"valid_until",
|
|
"expirationDate",
|
|
"expiryDate",
|
|
)
|
|
employment_types = _to_list(
|
|
_first_text(
|
|
record,
|
|
"employmentType",
|
|
"employment_types",
|
|
"jobType",
|
|
"type",
|
|
"data.employmentType",
|
|
)
|
|
)
|
|
workplace_type = self._extract_workplace(record)
|
|
|
|
evidence = [
|
|
FieldEvidence("external_id", "ats-id", 0.95, external_id[:240]),
|
|
FieldEvidence("url", "ats-url", 0.9, raw_url[:240]),
|
|
FieldEvidence("location_text", "ats-location", 0.88, location_text[:240]),
|
|
FieldEvidence("date_posted", "ats-date", 0.8, date_posted[:240]),
|
|
FieldEvidence("valid_through", "ats-date", 0.8, valid_through[:240]),
|
|
FieldEvidence(
|
|
"employment_types",
|
|
"ats-employment",
|
|
0.9,
|
|
", ".join(employment_types)[:240],
|
|
),
|
|
]
|
|
|
|
return ExtractedJob(
|
|
url=self._coerce_url(raw_url, base_url),
|
|
title=title,
|
|
employer_name=employer_name,
|
|
external_id=external_id,
|
|
location_text=location_text,
|
|
region=region,
|
|
postal_code=postal_code,
|
|
country=country,
|
|
description_html=description_raw,
|
|
description_text=description_text,
|
|
date_posted=date_posted,
|
|
valid_through=valid_through,
|
|
employment_types=employment_types,
|
|
workplace_type=workplace_type,
|
|
raw=record.get("__raw__", record),
|
|
evidence=evidence,
|
|
)
|
|
|
|
def extract(self, content: str, *, url: str) -> ExtractionResult:
|
|
if not self._supports_url(url):
|
|
return ExtractionResult(
|
|
[],
|
|
self.parser_key,
|
|
self.parser_version,
|
|
0.0,
|
|
["Onherkenbare ATS-host"],
|
|
)
|
|
|
|
payload = _extract_payload(content)
|
|
if not payload:
|
|
return ExtractionResult(
|
|
[],
|
|
self.parser_key,
|
|
self.parser_version,
|
|
0.0,
|
|
["Geen parseerbare ATS-response"],
|
|
)
|
|
|
|
if not self._supports_payload(payload, content):
|
|
return ExtractionResult(
|
|
[],
|
|
self.parser_key,
|
|
self.parser_version,
|
|
0.0,
|
|
["Geen herkenbare ATS-markup voor deze adapter"],
|
|
)
|
|
|
|
jobs: list[ExtractedJob] = []
|
|
for record in self._extract_records(payload):
|
|
if not isinstance(record, dict):
|
|
continue
|
|
if self._job_is_closed(record):
|
|
continue
|
|
job = self._to_job(record, base_url=url)
|
|
if job.title:
|
|
jobs.append(job)
|
|
|
|
warnings: list[str] = []
|
|
if not jobs:
|
|
warnings.append("Geen actieve ATS-vacatures gevonden")
|
|
return ExtractionResult(
|
|
jobs, self.parser_key, self.parser_version, 0.9 if jobs else 0.0, warnings
|
|
)
|
|
|
|
|
|
class GreenhouseAdapter(_AtsAdapter):
|
|
parser_key = "ats-greenhouse"
|
|
source_hosts = ("greenhouse.io", "boards.greenhouse.io")
|
|
support_markers = ("greenhouse", "job board", "jobboard")
|
|
listing_paths = (("jobs",), ("data", "jobs"), ("data", "results"))
|
|
detail_paths = (("job",), ("data", "job"), ("result", "job"), ("result", "position"))
|
|
closed_statuses = CLOSED_STATUSES | {"published", "draft", "deleted"}
|
|
|
|
def _extract_records(self, payload):
|
|
for path in self.listing_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
if isinstance(value, dict) and "results" in value:
|
|
possible = value.get("results")
|
|
if isinstance(possible, list):
|
|
return [item for item in possible if isinstance(item, dict)]
|
|
for path in self.detail_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, dict):
|
|
return [value]
|
|
return []
|
|
|
|
|
|
class LeverAdapter(_AtsAdapter):
|
|
parser_key = "ats-lever"
|
|
source_hosts = ("jobs.lever.co",)
|
|
support_markers = ("lever", "requisition", "posting")
|
|
listing_paths = (("data",), ("jobs",), ("results",))
|
|
detail_paths = (("data",), ("job",), ("position",), ("result",))
|
|
closed_statuses = CLOSED_STATUSES | {"archived", "deleted"}
|
|
|
|
def _extract_records(self, payload):
|
|
for path in self.listing_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
for path in self.detail_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, dict):
|
|
if path == ("position",):
|
|
raw_payload = {
|
|
"position": value,
|
|
"work_type": _to_text(value.get("workplaceType")),
|
|
}
|
|
return [{**value, "__raw__": raw_payload}]
|
|
return [value]
|
|
return []
|
|
|
|
|
|
class RecruiteeAdapter(_AtsAdapter):
|
|
parser_key = "ats-recruitee"
|
|
source_hosts = ("recruitee.com",)
|
|
support_markers = ("recruitee", "career", "vacancy")
|
|
listing_paths = (("jobs",), ("data", "jobs"), ("vacancies",))
|
|
detail_paths = (("job",), ("data", "job"), ("vacancy",), ("result",))
|
|
closed_statuses = CLOSED_STATUSES | {"hidden", "paused"}
|
|
|
|
def _extract_records(self, payload):
|
|
for path in self.listing_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
for path in self.detail_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, dict):
|
|
return [value]
|
|
return []
|
|
|
|
|
|
class SmartRecruitersAdapter(_AtsAdapter):
|
|
parser_key = "ats-smartrecruiters"
|
|
source_hosts = ("smartrecruiters.com",)
|
|
support_markers = ("smartrecruiters", "smart recruiter")
|
|
listing_paths = (("jobs",), ("data", "jobs"), ("results",))
|
|
detail_paths = (("job",), ("data", "job"), ("posting",), ("result",))
|
|
closed_statuses = CLOSED_STATUSES | {"unpublished", "expired"}
|
|
|
|
def _extract_records(self, payload):
|
|
for path in self.listing_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
for path in self.detail_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, dict):
|
|
return [value]
|
|
return []
|
|
|
|
|
|
class WorkableAdapter(_AtsAdapter):
|
|
parser_key = "ats-workable"
|
|
source_hosts = ("apply.workable.com",)
|
|
support_markers = ("workable", "workable job")
|
|
listing_paths = (("jobs",), ("data", "jobs"), ("results",))
|
|
detail_paths = (("job",), ("data", "job"), ("position",), ("result",))
|
|
closed_statuses = CLOSED_STATUSES | {"draft", "inactive"}
|
|
|
|
def _extract_records(self, payload):
|
|
for path in self.listing_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
for path in self.detail_paths:
|
|
value = payload
|
|
for key in path:
|
|
if not isinstance(value, dict) or key not in value:
|
|
value = None
|
|
break
|
|
value = value[key]
|
|
if isinstance(value, dict):
|
|
return [value]
|
|
return []
|