167 lines
6.7 KiB
Python
167 lines
6.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Iterable
|
|
from typing import Any
|
|
from urllib.parse import urljoin
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
from .base import ExtractedJob, ExtractionResult, FieldEvidence
|
|
|
|
|
|
class JsonLdJobPostingAdapter:
|
|
parser_key = "jsonld-jobposting"
|
|
parser_version = "1.0.0"
|
|
|
|
@staticmethod
|
|
def _is_jobposting(value: Any) -> bool:
|
|
types = value if isinstance(value, list) else [value]
|
|
return any(str(item).lower() == "jobposting" for item in types)
|
|
|
|
def _walk(self, value: Any) -> Iterable[dict[str, Any]]:
|
|
if isinstance(value, dict):
|
|
if self._is_jobposting(value.get("@type")):
|
|
yield value
|
|
graph = value.get("@graph")
|
|
if graph is not None:
|
|
yield from self._walk(graph)
|
|
for child in value.values():
|
|
if isinstance(child, dict | list):
|
|
yield from self._walk(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
yield from self._walk(child)
|
|
|
|
@staticmethod
|
|
def _name(value: Any) -> str:
|
|
if isinstance(value, str):
|
|
return value.strip()
|
|
if isinstance(value, dict):
|
|
return str(value.get("name") or value.get("legalName") or "").strip()
|
|
return ""
|
|
|
|
@staticmethod
|
|
def _location(value: Any) -> tuple[str, str, str, str]:
|
|
locations = value if isinstance(value, list) else [value]
|
|
parts: list[str] = []
|
|
region = postal = country = ""
|
|
for location in locations:
|
|
if not isinstance(location, dict):
|
|
continue
|
|
address = location.get("address", location)
|
|
if isinstance(address, str):
|
|
parts.append(address)
|
|
continue
|
|
if not isinstance(address, dict):
|
|
continue
|
|
locality = str(address.get("addressLocality") or "").strip()
|
|
region = region or str(address.get("addressRegion") or "").strip()
|
|
postal = postal or str(address.get("postalCode") or "").strip()
|
|
country_value = address.get("addressCountry")
|
|
country = (
|
|
country
|
|
or JsonLdJobPostingAdapter._name(country_value)
|
|
or str(country_value or "").strip()
|
|
)
|
|
label = ", ".join(part for part in [locality, region, postal, country] if part)
|
|
if label:
|
|
parts.append(label)
|
|
return " | ".join(dict.fromkeys(parts)), region, postal, country
|
|
|
|
@staticmethod
|
|
def _employment_types(value: Any) -> list[str]:
|
|
if isinstance(value, list):
|
|
return [str(item).strip() for item in value if str(item).strip()]
|
|
if value:
|
|
return [str(value).strip()]
|
|
return []
|
|
|
|
@staticmethod
|
|
def _identifier(value: Any) -> str:
|
|
if isinstance(value, dict):
|
|
return str(value.get("value") or value.get("name") or "").strip()
|
|
return str(value or "").strip()
|
|
|
|
@staticmethod
|
|
def _salary(value: Any) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
result: dict[str, Any] = {}
|
|
currency = value.get("currency")
|
|
if currency:
|
|
result["currency"] = currency
|
|
raw_value = value.get("value")
|
|
if isinstance(raw_value, dict):
|
|
for key in ("minValue", "maxValue", "value", "unitText"):
|
|
if key in raw_value:
|
|
result[key] = raw_value[key]
|
|
elif raw_value is not None:
|
|
result["value"] = raw_value
|
|
return result
|
|
|
|
def extract(self, content: str, *, url: str) -> ExtractionResult:
|
|
soup = BeautifulSoup(content, "lxml")
|
|
records: list[dict[str, Any]] = []
|
|
warnings: list[str] = []
|
|
for script in soup.find_all(
|
|
"script", attrs={"type": lambda value: value and "ld+json" in value}
|
|
):
|
|
raw = script.string or script.get_text("", strip=True)
|
|
if not raw:
|
|
continue
|
|
try:
|
|
payload = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
warnings.append("Ongeldige JSON-LD overgeslagen")
|
|
continue
|
|
records.extend(self._walk(payload))
|
|
|
|
jobs: list[ExtractedJob] = []
|
|
for record in records:
|
|
title = str(record.get("title") or record.get("name") or "").strip()
|
|
if not title:
|
|
warnings.append("JobPosting zonder titel overgeslagen")
|
|
continue
|
|
description_html = str(record.get("description") or "").strip()
|
|
description_text = BeautifulSoup(description_html, "lxml").get_text("\n", strip=True)
|
|
employer = self._name(record.get("hiringOrganization"))
|
|
location, region, postal, country = self._location(record.get("jobLocation"))
|
|
workplace_type = str(record.get("jobLocationType") or "").strip()
|
|
if not location and record.get("applicantLocationRequirements"):
|
|
location, region, postal, country = self._location(
|
|
record.get("applicantLocationRequirements")
|
|
)
|
|
job_url = str(record.get("url") or url).strip()
|
|
job_url = urljoin(url, job_url)
|
|
evidence = [
|
|
FieldEvidence("title", "jsonld", 0.98, title[:240]),
|
|
FieldEvidence("employer_name", "jsonld", 0.95, employer[:240]),
|
|
FieldEvidence("location_text", "jsonld", 0.92, location[:240]),
|
|
FieldEvidence("description", "jsonld", 0.95, description_text[:300]),
|
|
]
|
|
jobs.append(
|
|
ExtractedJob(
|
|
url=job_url,
|
|
title=title,
|
|
employer_name=employer,
|
|
external_id=self._identifier(record.get("identifier")),
|
|
location_text=location,
|
|
region=region,
|
|
postal_code=postal,
|
|
country=country,
|
|
description_html=description_html,
|
|
description_text=description_text,
|
|
language=str(record.get("inLanguage") or "").strip(),
|
|
date_posted=str(record.get("datePosted") or "").strip(),
|
|
valid_through=str(record.get("validThrough") or "").strip(),
|
|
employment_types=self._employment_types(record.get("employmentType")),
|
|
workplace_type=workplace_type,
|
|
compensation=self._salary(record.get("baseSalary")),
|
|
raw=record,
|
|
evidence=evidence,
|
|
)
|
|
)
|
|
confidence = 0.95 if jobs else 0.0
|
|
return ExtractionResult(jobs, self.parser_key, self.parser_version, confidence, warnings)
|