jobbsok/scripts/vurdering.py

"""Hard filters and the weighted score, kept out of the model (plan Step 10).

Scoring is split in two. Judgement -- how well a listing matches a criterion --
is the model's. Arithmetic is this module's, and nothing here calls a model.
Two reasons, both measured risks. A number produced end to end by a model is
not reproducible (H5), and M6's learning loop reads `score_da` as a value on a
weight-dependent scale (M8r), so a later change of weights has to be visible
rather than silently rescaling the history. Hence :func:`vurder` returns the
sub-scores it was handed and a hash of the weight vector alongside the score.

It persists nothing. `kandidatvurdering` writes nothing (build-brief 7), which
is what makes it safe to run on unguarded pasted text in M1, before the
ingestion guard exists. The sub-scores and the weight hash are written later,
at decision time, by `beslutninger.py` in Step 22.

Hard filters run first and are collected, not short-circuited into the first
one found: an operator who sees only "rejected on salary" cannot tell whether
fixing the salary would help. Every rejection names the frontmatter key it
rejected on, and the listing is scored anyway, because a rejection without a
number is a rejection that cannot be argued with.

Two rules here are precedents rather than mechanics, and both come from the
operator's real profile.

**Travel time is soft when the profile names its work locations.** Build-brief
5.1 says frontmatter is absolute, and for a profile with no `arbeidssteder`
key it still is. But a profile that lists the places it will actually work has
already answered the question the travel-time ceiling was asking, and answered
it more precisely. So a listing whose location is on that list turns
`maks_reisetid_min` into a warning -- for that location only. Anywhere else,
the ceiling rejects exactly as before.

**Seniority is free text ordered by a rank table.** 5.1 gives no enum, and the
operator's own values are `seniorrådgiver` and `fagdirektør/sjefsarkitekt`.
Without an ordering the seniority filter cannot decide anything and is
decoration. :data:`SENIORITETSRANG` supplies one. A title the table does not
know produces a warning and never a rejection: refusing to place a word is not
the same as placing it below the floor.

Absolute-no matching is normalised and word-bounded. Both sides fold through
`kandidat_schema.normaliser`, which is `paths.slug`'s rules, so `Ålesund` and `alesund` are one word, and a term matches only
as a whole token sequence -- `turnus` is a hit, `turnusplanleggeren` is not.
The alternatives contract from `kandidat_schema` carries over unchanged: a
slash separates spellings of one no, whitespace inside one spelling is a
phrase.
"""

import hashlib
import json

import kandidat_schema
from jobbsok_lib import frontmatter

#: Seniority titles the ordering knows, folded through `paths.slug`. The
#: numbers are ordinals with gaps, so a level can be inserted later without
#: renumbering a scale that is already recorded in decisions.
SENIORITETSRANG = {
    "junior": 10, "nyutdannet": 10, "trainee": 10, "graduate": 10, "laerling": 10,
    "radgiver": 20, "konsulent": 20, "medarbeider": 20, "ingenior": 20,
    "saksbehandler": 20,
    "senior": 30, "seniorradgiver": 30, "seniorkonsulent": 30, "senioringenior": 30,
    "spesialist": 30, "spesialradgiver": 30, "arkitekt": 30, "seniorarkitekt": 30,
    "fagleder": 40, "teamleder": 40, "teamlead": 40, "leder": 40, "fagansvarlig": 40,
    "sjefkonsulent": 40, "sjefingenior": 40, "losningsarkitekt": 40,
    "principal": 40, "prinsipal": 40,
    "fagdirektor": 50, "sjefsarkitekt": 50, "avdelingsdirektor": 50, "direktor": 50,
    "teknisk-direktor": 50, "cto": 50,
    "administrerende-direktor": 60, "adm-direktor": 60, "konsernsjef": 60, "ceo": 60,
}

#: The affirmative word that turns a listing flag into an attribute of the
#: listing. `turnus: ja` contributes the token `turnus` to what the
#: absolute-no list is matched against; `turnus: nei` contributes nothing.
JA = "ja"

MIN_DELSCORE = 0
MAX_DELSCORE = 100


class DelscoreError(Exception):
    """A sub-score payload that is missing a criterion or out of range."""


def les_profil(text):
    """Parse a `kandidat.md` document into what scoring needs from it.

    Goes through :func:`kandidat_schema.validate` rather than re-reading the
    frontmatter, so the comma/slash contract and the weight resolution are
    defined once. The raw metadata comes along because the numeric hard
    filters read keys the report does not restate.
    """
    rapport = kandidat_schema.validate(text)
    meta, _body = frontmatter.parse(text)
    return {
        "meta": meta,
        "rapport": rapport,
        "vekter": rapport["vekter"],
        "vekter_kilde": rapport["vekter_kilde"],
    }


def les_profil_fil(root, *parts):
    """Read and parse the profile at ``parts`` under ``root``."""
    meta, body = frontmatter.read(root, *parts)
    return les_profil(frontmatter.render(meta, body))


def vurder(profil, annonse, brodtekst, delscore_payload, vekter=None):
    """Score ``annonse`` against ``profil`` and run the hard filters.

    ``delscore_payload`` is the model's contribution: a mapping of criterion to
    an integer 0-100 under ``delscore``, and a free-text ``bekymringer`` list
    that is returned untouched. Nothing is written anywhere.
    """
    delscore = _valider_delscore(delscore_payload)
    vekter = dict(vekter if vekter is not None else profil["vekter"])

    avvisninger = []
    advarsler = []
    _filter_lonn(profil, annonse, avvisninger, advarsler)
    _filter_reisetid(profil, annonse, avvisninger, advarsler)
    _filter_hjemmekontor(profil, annonse, avvisninger, advarsler)
    _filter_ansettelsesform(profil, annonse, avvisninger, advarsler)
    _filter_absolutte_nei(profil, annonse, brodtekst, avvisninger, advarsler)
    _filter_senioritet(profil, annonse, avvisninger, advarsler)

    return {
        "score": _score(delscore, vekter),
        "delscore": dict(delscore),
        "vekter": vekter,
        "vekter_kilde": profil["vekter_kilde"],
        "vekt_hash": vekt_hash(vekter),
        "verdikt": "avvist" if avvisninger else "vurderes",
        "avvisninger": avvisninger,
        "advarsler": advarsler,
        "bekymringer": list(delscore_payload.get("bekymringer", [])),
    }


def vekt_hash(vekter):
    """A stable fingerprint of a weight vector, for recording beside a score."""
    canonical = json.dumps(vekter, sort_keys=True, separators=(",", ":"))
    return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def rang(tekst):
    """Rank a free-text seniority title, or ``None`` when it is not placeable.

    Slash alternatives are resolved to the **highest** rank any spelling
    reaches. For a ceiling that is the generous read and for a floor the
    conservative one, which is the right way round: a profile that writes two
    names for the same level should not be filtered by whichever name the
    table happened to rank lower.
    """
    rangeringer = []
    for alternativ in kandidat_schema.del_alternativer(tekst):
        folded = kandidat_schema.normaliser(alternativ)
        if folded in SENIORITETSRANG:
            rangeringer.append(SENIORITETSRANG[folded])
            continue
        tokens = [SENIORITETSRANG[t] for t in folded.split("-") if t in SENIORITETSRANG]
        if tokens:
            rangeringer.append(max(tokens))
    return max(rangeringer) if rangeringer else None


def treffer(termer, tekst):
    """True when any alternative in ``termer`` appears as whole tokens in ``tekst``."""
    hoystakk = _tokens(tekst)
    for alternativ in termer:
        naal = _tokens(alternativ)
        if naal and _delsekvens(naal, hoystakk):
            return True
    return False


def _valider_delscore(payload):
    raw = (payload or {}).get("delscore")
    if not isinstance(raw, dict):
        raise DelscoreError("delscore mangler; forventet et kart over kriterium til 0-100")
    delscore = {}
    for kriterium in kandidat_schema.STANDARDVEKTER:
        if kriterium not in raw:
            raise DelscoreError("delscore mangler kriteriet %r" % kriterium)
        verdi = raw[kriterium]
        if isinstance(verdi, bool) or not isinstance(verdi, int):
            raise DelscoreError(
                "delscore for %r er %r; forventet et heltall" % (kriterium, verdi)
            )
        if not MIN_DELSCORE <= verdi <= MAX_DELSCORE:
            raise DelscoreError(
                "delscore for %r er %d; utenfor %d-%d"
                % (kriterium, verdi, MIN_DELSCORE, MAX_DELSCORE)
            )
        delscore[kriterium] = verdi
    for kriterium in raw:
        if kriterium not in kandidat_schema.STANDARDVEKTER:
            raise DelscoreError("delscore har ukjent kriterium %r" % kriterium)
    return delscore


def _score(delscore, vekter):
    sum_vekt = sum(vekter.get(k, 0) for k in delscore)
    if sum_vekt <= 0:
        return 0
    total = sum(delscore[k] * vekter.get(k, 0) for k in delscore)
    # Half up, and explicitly: round() rounds half to even, which would make
    # the same inputs land differently either side of .5.
    return max(0, min(100, int(total / sum_vekt + 0.5)))


def _filter_lonn(profil, annonse, avvisninger, advarsler):
    gulv = _tall(profil["meta"].get("lonn"), "gulv_nok")
    tilbudt = annonse.get("lonn_nok")
    if gulv is None:
        return
    if not isinstance(tilbudt, int):
        # Not stated is not below the floor. Treating the two alike would
        # reject every listing that keeps its salary out of the advert.
        _si(advarsler, "lonn.gulv_nok",
            "annonsen oppgir ingen lonn; gulvet paa %d kunne ikke sjekkes" % gulv)
        return
    if tilbudt < gulv:
        _si(avvisninger, "lonn.gulv_nok",
            "annonsen oppgir %d, gulvet er %d" % (tilbudt, gulv))


def _filter_reisetid(profil, annonse, avvisninger, advarsler):
    tak = _tall(profil["meta"].get("geografi"), "maks_reisetid_min")
    reisetid = annonse.get("reisetid_min")
    if tak is None:
        return
    if not isinstance(reisetid, int):
        _si(advarsler, "geografi.maks_reisetid_min",
            "annonsen oppgir ingen reisetid; taket paa %d min kunne ikke sjekkes" % tak)
        return
    if reisetid <= tak:
        return
    if _paa_arbeidsstedslista(profil, annonse.get("sted")):
        _si(advarsler, "geografi.maks_reisetid_min",
            "%d min mot et tak paa %d, men %r staar i arbeidssteder; nedvektes, "
            "avvises ikke" % (reisetid, tak, annonse.get("sted")))
        return
    _si(avvisninger, "geografi.maks_reisetid_min",
        "%d min mot et tak paa %d" % (reisetid, tak))


def _filter_hjemmekontor(profil, annonse, avvisninger, advarsler):
    krav = _tall(profil["meta"].get("geografi"), "hjemmekontor_min_dager")
    dager = annonse.get("hjemmekontor_dager")
    if krav is None:
        return
    if not isinstance(dager, int):
        _si(advarsler, "geografi.hjemmekontor_min_dager",
            "annonsen oppgir ingen hjemmekontordager; kravet paa %d kunne ikke sjekkes"
            % krav)
        return
    if dager < krav:
        _si(avvisninger, "geografi.hjemmekontor_min_dager",
            "annonsen gir %d dag(er), kravet er %d" % (dager, krav))


def _filter_ansettelsesform(profil, annonse, avvisninger, advarsler):
    former = profil["rapport"]["ansettelsesform"]
    oppgitt = annonse.get("ansettelsesform")
    if not (former["aksepterer"] or former["avviser"]):
        return
    if oppgitt is None:
        _si(advarsler, "ansettelsesform.aksepterer",
            "annonsen oppgir ingen ansettelsesform")
        return
    for alternativer in former["avviser"]:
        if treffer(alternativer, str(oppgitt)):
            _si(avvisninger, "ansettelsesform.avviser",
                "annonsen er %r, som staar paa avviser-lista" % oppgitt)
            return
    if not former["aksepterer"]:
        return
    for alternativer in former["aksepterer"]:
        if treffer(alternativer, str(oppgitt)):
            return
    _si(avvisninger, "ansettelsesform.aksepterer",
        "annonsen er %r, som ikke staar paa aksepterer-lista" % oppgitt)


def _filter_absolutte_nei(profil, annonse, brodtekst, avvisninger, _advarsler):
    hoystakk = _hoystakk(annonse, brodtekst)
    for alternativer in profil["rapport"]["absolutte_nei"]:
        if treffer(alternativer, hoystakk):
            _si(avvisninger, "absolutte_nei",
                "annonsen treffer %r" % " / ".join(alternativer))


def _filter_senioritet(profil, annonse, avvisninger, advarsler):
    niva = profil["meta"].get("senioritet")
    if not isinstance(niva, dict):
        return
    oppgitt = annonse.get("senioritet")
    if oppgitt is None:
        _si(advarsler, "senioritet.min", "annonsen oppgir ingen senioritet")
        return
    annonserang = rang(oppgitt)
    if annonserang is None:
        # Not placeable is not below the floor. A stillingskode the table has
        # never seen must not be filtered as if it were junior.
        _si(advarsler, "senioritet.min",
            "%r finnes ikke i rangtabellen; senioritet ble ikke filtrert" % oppgitt)
        return
    minimum = rang(niva.get("min")) if niva.get("min") is not None else None
    maksimum = rang(niva.get("maks")) if niva.get("maks") is not None else None
    if minimum is not None and annonserang < minimum:
        _si(avvisninger, "senioritet.min",
            "%r ligger under %r" % (oppgitt, niva.get("min")))
    if maksimum is not None and annonserang > maksimum:
        _si(avvisninger, "senioritet.maks",
            "%r ligger over %r" % (oppgitt, niva.get("maks")))


def _paa_arbeidsstedslista(profil, sted):
    if sted is None:
        return False
    maal = kandidat_schema.normaliser(sted)
    if not maal:
        return False
    return any(
        kandidat_schema.normaliser(kandidat) == maal
        for kandidat in profil["rapport"]["arbeidssteder"]
    )


def _hoystakk(annonse, brodtekst):
    """What the absolute-no list is matched against.

    The role, the location and the body, plus the name of every flag the
    listing set to `ja` -- `turnus: ja` says the listing has turnus as plainly
    as a sentence would, and a match that only read prose would miss it.
    """
    deler = [str(annonse.get(key, "")) for key in ("rolle", "sted", "arbeidsgiver")]
    deler.append(brodtekst or "")
    for key, verdi in annonse.items():
        if not isinstance(verdi, dict) and str(verdi).strip().lower() == JA:
            deler.append(key)
    return " ".join(deler)


def _tokens(tekst):
    folded = kandidat_schema.normaliser(tekst)
    return [token for token in folded.split("-") if token]


def _delsekvens(naal, hoystakk):
    for start in range(len(hoystakk) - len(naal) + 1):
        if hoystakk[start:start + len(naal)] == naal:
            return True
    return False


def _tall(blokk, key):
    if not isinstance(blokk, dict):
        return None
    verdi = blokk.get(key)
    return verdi if isinstance(verdi, int) and not isinstance(verdi, bool) else None


def _si(samling, nokkel, begrunnelse):
    samling.append({"nokkel": nokkel, "begrunnelse": begrunnelse})