spb/company-atlas

1"""Personio job XML feed — public feed behind `<token>.jobs.personio.de`:2`GET https://{token}.jobs.personio.de/xml` → <workzag-jobs><position><id/><office/><additionalOffices><office/></additionalOffices><department/>3<name/><employmentType/><seniority/><schedule/><createdAt/>…</position></workzag-jobs>. Job URL `https://{token}.jobs.personio.de/job/{id}`.4Verified 2026-09-12 against `personio` (fixture)."""5from __future__ import annotations67import re8import xml.etree.ElementTree as ET9from collections.abc import Mapping10from typing import Any1112from companyatlas.connectors._ats_base import AtsConnector13from companyatlas.connectors._util import parse_date14from companyatlas.fetch import FetchResult15from companyatlas.sdk.connector import ConnectorMeta, register16from companyatlas.sdk.models import ExtractedJob17from companyatlas.taxonomy import FetchMode, Surface181920@register21class PersonioConnector(AtsConnector):22    vendor = "personio"23    token_re = re.compile(r"https?://([a-z0-9-]+)\.jobs\.personio\.(?:de|com)/xml", re.IGNORECASE)24    meta = ConnectorMeta(connector_id="personio-v1", name="Personio XML feed", version="1", category=Surface.JOBS_BOARD, fetch_mode=FetchMode.FEED,25                         default_interval_s=12 * 3600, url_pattern=r"\.jobs\.personio\.(de|com)/xml", pattern_required=True, priority=50, accept="application/xml,text/xml",26                         description="Public Personio job XML feed")2728    def load(self, result: FetchResult) -> Any:29        if len(result.content) > 8 * 1024 * 1024 or b"<!ENTITY" in result.content[:4096]:30            raise ValueError("refusing suspicious XML (size or entity declarations)")31        return ET.fromstring(result.content)3233    def parse_jobs(self, data: Any, sensor: Mapping[str, Any]) -> list[ExtractedJob]:34        token = self.token(sensor) or ""35        host = re.sub(r"/xml.*$", "", str(sensor.get("url") or "")) or f"https://{token}.jobs.personio.de"36        out: list[ExtractedJob] = []37        for pos in data.iter("position"):38            g = {c.tag: (c.text or "").strip() for c in pos if c.tag != "additionalOffices" and c.tag != "jobDescriptions"}39            name = g.get("name")40            if not name:41                continue42            offices = [g.get("office")] + [o.text.strip() for o in pos.iter("office") if o.text and o.text.strip() != g.get("office")]43            offices = [o for o in offices if o]44            jid = g.get("id")45            out.append(ExtractedJob(title=name, url=f"{host}/job/{jid}" if jid else None, external_id=jid or None, department=g.get("department") or None,46                                    location_text=offices[0] if offices else None, employment_type=g.get("schedule") or g.get("employmentType") or None,47                                    seniority=g.get("seniority") or None, posted_at=parse_date(g.get("createdAt")),48                                    raw={"recruitingCategory": g.get("recruitingCategory"), "occupation": g.get("occupation"), "subcompany": g.get("subcompany"),49                                         "yearsOfExperience": g.get("yearsOfExperience"), "offices": offices[:5]}))50        return out51