from __future__ import annotations import re from urllib.parse import parse_qs, urljoin, urlsplit from bs4 import BeautifulSoup, Tag from .base import ExtractedJob, ExtractionResult, FieldEvidence class KempenEmployerAdapter: """Extract public employer lists around Mol from exact reviewed routes.""" parser_key = "regional-kempen-employers" parser_version = "1.0.0" supported_routes = { "ziekenhuisgeel.careersite.be": {"/nl/vacatures"}, "geel.hro.be": {"/"}, "jobs.turnhout.be": {"/"}, "jobs.renotec.be": {"/nl/alle-jobs"}, "ravago.softgarden.io": {"/en/vacancies"}, "jobs.sanofi.com": {"/en/belgium"}, "www.daf.com": {"/nl-nl/werken-bij-daf/vacatures"}, } def _route(self, url: str) -> tuple[str, str] | None: parts = urlsplit(url) host = (parts.hostname or "").lower() path = parts.path.rstrip("/") or "/" if parts.scheme != "https" or path not in self.supported_routes.get(host, set()): return None return host, path def _supports_url(self, url: str) -> bool: return self._route(url) is not None @staticmethod def _same_host_url(base_url: str, href: str) -> str: job_url = urljoin(base_url, href) parts = urlsplit(job_url) if parts.scheme != "https": return "" if (parts.hostname or "").lower() != (urlsplit(base_url).hostname or "").lower(): return "" return job_url @staticmethod def _job( *, url: str, title: str, employer: str, location: str, postal_code: str, description: str, external_id: str, evidence_source: str, ) -> ExtractedJob: return ExtractedJob( url=url, title=title, employer_name=employer, external_id=external_id, location_text=location, postal_code=postal_code, region="Antwerpen", country="BE", description_text=description, raw={"regional_listing": evidence_source, "region_center": "2400 Mol"}, evidence=[ FieldEvidence("title", evidence_source, 0.94, title[:240]), FieldEvidence("employer_name", "reviewed-source", 0.97, employer), FieldEvidence("location_text", "kempen-location-marker", 0.9, location), ], ) def _ziekenhuis_geel(self, soup: BeautifulSoup, url: str) -> list[ExtractedJob]: jobs = [] for card in soup.select(".vacature-tegel"): link = card.select_one("a.vacature-tegel__link[href]") title_node = card.select_one(".vacature-tegel__titel") if not isinstance(link, Tag) or not isinstance(title_node, Tag): continue job_url = self._same_host_url(url, str(link.get("href") or "")) match = re.fullmatch(r"/nl/vacature/(\d+)/[^/]+", urlsplit(job_url).path) title = title_node.get_text(" ", strip=True) if not title or not match: continue jobs.append( self._job( url=job_url, title=title, employer="Ziekenhuis Geel", location="Geel, Antwerpen", postal_code="2440", description=card.get_text(" ", strip=True), external_id=match.group(1), evidence_source="ziekenhuis-geel-card", ) ) return jobs def _stad_geel(self, soup: BeautifulSoup, url: str) -> list[ExtractedJob]: jobs = [] for card in soup.select(".vacatureKader[data-href]"): href = str(card.get("data-href") or "") match = re.fullmatch(r"vacature\.php\?id=(\d+)", href) title_node = card.find("h4") title = title_node.get_text(" ", strip=True) if title_node else "" if not title or not match: continue jobs.append( self._job( url=self._same_host_url(url, href), title=title, employer="Lokaal bestuur Geel", location="Geel, Antwerpen", postal_code="2440", description=card.get_text(" ", strip=True), external_id=match.group(1), evidence_source="stad-geel-hro-card", ) ) return jobs def _stad_turnhout(self, soup: BeautifulSoup, url: str) -> list[ExtractedJob]: jobs = [] for link in soup.find_all("a", href=True): href = str(link.get("href") or "") match = re.fullmatch(r"vacature\.php\?id=(\d+)", href) title_node = link.select_one(".block-update__body__title__inner") title = title_node.get_text(" ", strip=True) if title_node else "" if not title or not match: continue jobs.append( self._job( url=self._same_host_url(url, href), title=title, employer="Stad Turnhout", location="Turnhout, Antwerpen", postal_code="2300", description=link.get_text(" ", strip=True), external_id=match.group(1), evidence_source="stad-turnhout-hro-card", ) ) return jobs def _renotec(self, soup: BeautifulSoup, url: str) -> list[ExtractedJob]: jobs = [] for card in soup.select(".s-tile.s-card"): card_text = card.get_text(" ", strip=True) if "Geel" not in card_text: continue title_node = card.find("h3") link = card.find("a", href=True) title = title_node.get_text(" ", strip=True) if title_node else "" job_url = self._same_host_url(url, str(link.get("href") or "")) if link else "" parts = urlsplit(job_url) query = parse_qs(parts.query) external_id = (query.get("id") or [""])[0] if parts.path != "/nl/detail/" or not external_id.isdigit() or not title: continue jobs.append( self._job( url=job_url, title=title, employer="Group Renotec", location="Geel, Antwerpen (mogelijk meerdere werfregio's)", postal_code="2440", description=card_text, external_id=external_id, evidence_source="renotec-geel-card", ) ) return jobs def _ravago(self, soup: BeautifulSoup, url: str) -> list[ExtractedJob]: jobs = [] for card in soup.select(".matchElement"): locations = { node.get_text(" ", strip=True) for node in card.select(".location-view-item") } local_places = locations.intersection({"Arendonk", "Olen"}) if not local_places: continue link = card.find("a", href=True) title = link.get_text(" ", strip=True) if link else "" job_url = self._same_host_url(url, str(link.get("href") or "")) if link else "" match = re.fullmatch(r"/job/(\d+)/[^/]+/?", urlsplit(job_url).path) if not title or not match: continue place = "Arendonk" if "Arendonk" in local_places else "Olen" jobs.append( self._job( url=job_url, title=title, employer="Ravago", location=f"{place}, Antwerpen", postal_code="2370" if place == "Arendonk" else "2250", description=card.get_text(" ", strip=True), external_id=match.group(1), evidence_source="ravago-softgarden-card", ) ) return jobs def _sanofi(self, soup: BeautifulSoup, url: str) -> list[ExtractedJob]: jobs = [] for link in soup.select(".job-list a[data-job-id][href]"): location_node = link.select_one(".job-location") title_node = link.select_one(".job-title") location = location_node.get_text(" ", strip=True) if location_node else "" title = title_node.get_text(" ", strip=True) if title_node else "" external_id = str(link.get("data-job-id") or "") job_url = self._same_host_url(url, str(link.get("href") or "")) if location != "Geel, Belgium" or not title or not external_id.isdigit(): continue if not re.fullmatch(r"/en/job/geel/[^/]+/\d+/\d+", urlsplit(job_url).path): continue jobs.append( self._job( url=job_url, title=title, employer="Sanofi", location="Geel, Antwerpen", postal_code="2440", description=link.get_text(" ", strip=True), external_id=external_id, evidence_source="sanofi-belgium-geel-card", ) ) return jobs def _daf(self, soup: BeautifulSoup, url: str) -> list[ExtractedJob]: jobs = [] for card in soup.select("li.itemlist__item"): location_node = card.select_one(".js-vac-metalocation") title_link = card.select_one("a.js-vac-title[href]") location = location_node.get_text(" ", strip=True) if location_node else "" title = title_link.get_text(" ", strip=True) if title_link else "" job_url = ( self._same_host_url(url, str(title_link.get("href") or "")) if title_link else "" ) path = urlsplit(job_url).path prefix = "/nl-nl/werken-bij-daf/vacatures/" if location != "Westerlo" or not title or not path.startswith(prefix): continue slug = path.removeprefix(prefix).strip("/") if not slug or "/" in slug: continue jobs.append( self._job( url=job_url, title=title, employer="DAF Trucks", location="Westerlo, Antwerpen", postal_code="2260", description=card.get_text(" ", strip=True), external_id=slug, evidence_source="daf-westerlo-card", ) ) return jobs def extract(self, content: str, *, url: str) -> ExtractionResult: route = self._route(url) if route is None: return ExtractionResult( [], self.parser_key, self.parser_version, 0.0, ["Onherkende Kempen-bron"] ) host, _ = route extractors = { "ziekenhuisgeel.careersite.be": self._ziekenhuis_geel, "geel.hro.be": self._stad_geel, "jobs.turnhout.be": self._stad_turnhout, "jobs.renotec.be": self._renotec, "ravago.softgarden.io": self._ravago, "jobs.sanofi.com": self._sanofi, "www.daf.com": self._daf, } jobs = extractors[host](BeautifulSoup(content, "lxml"), url) unique_jobs = list({job.url: job for job in jobs if job.url}.values()) return ExtractionResult( unique_jobs, self.parser_key, self.parser_version, 0.92 if unique_jobs else 0.0, [] if unique_jobs else ["Geen actuele regionale vacatures gevonden"], )