spb/company-atlas

1"""JobPosting JSON-LD / microdata on any careers page (schema.org). Shared helper `jobs_from_jsonld` is also used by the generic HTML2connector; this connector is picked when discovery has seen JobPosting structured data on a careers URL that has no ATS board."""3from __future__ import annotations45from collections.abc import Mapping6from typing import Any78from companyatlas.connectors._util import country_code, description_hash, finish_job, job_blocks, jobs_text, parse_date, text_of9from companyatlas.fetch import FetchResult10from companyatlas.sdk import normalize11from companyatlas.sdk.connector import Connector, ConnectorMeta, register12from companyatlas.sdk.models import ExtractedJob, Extraction13from companyatlas.taxonomy import FetchMode, Surface141516def _place(loc: Any) -> tuple[str | None, str | None, str | None, str | None]:17    """(location_text, city, region, country) from a schema.org Place / PostalAddress / string."""18    if loc is None:19        return None, None, None, None20    if isinstance(loc, list):21        loc = loc[0] if loc else None22        if loc is None:23            return None, None, None, None24    if isinstance(loc, str):25        return loc, None, None, None26    addr = loc.get("address") if isinstance(loc, dict) else None27    if isinstance(addr, str):28        return addr, None, None, None29    addr = addr if isinstance(addr, dict) else (loc if isinstance(loc, dict) and "addressLocality" in loc else {})30    city = text_of(addr.get("addressLocality"))31    region = text_of(addr.get("addressRegion"))32    country_raw = addr.get("addressCountry")33    if isinstance(country_raw, dict):34        country_raw = country_raw.get("name")35    country = country_code(country_raw) if isinstance(country_raw, str) else None36    parts = [x for x in (city, region, (country or (country_raw if isinstance(country_raw, str) else None))) if x]37    return (", ".join(parts) or text_of(loc.get("name")) if isinstance(loc, dict) else None), city, region, country383940def jobs_from_jsonld(items: list[dict[str, Any]] | None, *, page_url: str) -> list[ExtractedJob]:41    out: list[ExtractedJob] = []42    for jp in items or []:43        title = text_of(jp.get("title")) or text_of(jp.get("name"))44        if not title:45            continue46        loc_text, city, region, country = _place(jp.get("jobLocation"))47        ident = jp.get("identifier")48        ext = None49        if isinstance(ident, dict):50            ext = text_of(ident.get("value")) or text_of(ident.get("name"))51        elif isinstance(ident, str | int):52            ext = str(ident)53        remote = True if str(jp.get("jobLocationType") or "").upper() == "TELECOMMUTE" else None54        salary = jp.get("baseSalary") if isinstance(jp.get("baseSalary"), dict) else {}55        val = salary.get("value") if isinstance(salary.get("value"), dict) else {}56        emp = jp.get("employmentType")57        job = ExtractedJob(title=title, url=text_of(jp.get("url")) or page_url, external_id=ext, location_text=loc_text, city=city, region=region,58                           country=country, remote=remote, employment_type=text_of(emp[0] if isinstance(emp, list) and emp else emp),59                           posted_at=parse_date(jp.get("datePosted")), description_hash=description_hash(jp.get("description")),60                           salary_min=val.get("minValue") or val.get("value"), salary_max=val.get("maxValue"), salary_currency=salary.get("currency"),61                           salary_period=val.get("unitText"), department=text_of(jp.get("occupationalCategory")) if isinstance(jp.get("occupationalCategory"), str) else None,62                           raw={"validThrough": jp.get("validThrough"), "hiringOrganization": text_of(jp.get("hiringOrganization"))})63        out.append(finish_job(job))64    return out656667@register68class JsonLdJobsConnector(Connector):69    meta = ConnectorMeta(connector_id="jsonld-jobs-v1", name="JobPosting structured data", version="1", category=Surface.JOBS_BOARD, fetch_mode=FetchMode.HTTP,70                         default_interval_s=12 * 3600, surfaces=(), priority=2,71                         description="schema.org JobPosting JSON-LD / microdata (auto-picked only for non-ATS jobs_board URLs; careers HTML uses generic-html)")7273    def extract(self, sensor: Mapping[str, Any], result: FetchResult) -> Extraction:74        page = normalize.parse(result.text, url=result.final_url, surface=str(sensor.get("surface") or ""))75        items = list(page.jsonld.get("job_postings") or []) + list(page.microdata.get("job_postings") or [])76        jobs = jobs_from_jsonld(items, page_url=result.final_url)77        header = f"{page.title or 'Careers'} — {len(jobs)} structured job postings"78        ex = page.to_extraction()79        ex.jobs = jobs80        if jobs:81            ex.blocks = ex.blocks + job_blocks(jobs, path="JobPosting")82            ex.text = ex.text + "\n" + jobs_text(jobs, header)83        ex.meta.update({"job_count": len(jobs), "structured": bool(jobs)})84        return ex858687__all__ = ["JsonLdJobsConnector", "jobs_from_jsonld"]88