scripts/validate_dataset.py

"""Validate the extracted salary knowledge base.

Layers:
  1. JSON Schema (Draft 2020-12) — structural validity
  2. Percentile ordering — p25 <= median <= p75 <= p90 (within tolerance)
  3. Value sanity — monthly salaries in a plausible DKK range
  4. Digit cross-check — every extracted measure value must appear verbatim in
     the source page's raw text (proves no digits were hallucinated/altered)
  5. Coverage — records per page vs the table inventory
"""
from __future__ import annotations

import json
import re
from pathlib import Path

import pdfplumber
from jsonschema import Draft202012Validator

BASE = Path("/home/simonl/Desktop/competiton/pay-equity-for-eu")
OUT = BASE / "data/processed/lonstatistik"
RAW = BASE / "data/raw"
PDFS = {"IDA": RAW / "ida-loenstatistik-2025.pdf",
        "Djøf": RAW / "Privatansatte lnstatistik 2025.pdf"}

MONEY_KEYS = ["mean", "p25", "median", "p75", "p90", "base", "supplement", "pension", "bonus_avg"]


def dk_num(v):
    """Format a number the Danish way for raw-text matching: 81629 -> '81.629'."""
    if isinstance(v, float) and not v.is_integer():
        # percent / decimal value -> '41,6'
        return f"{v:.1f}".replace(".", ",")
    iv = int(v)
    s = f"{iv:,}".replace(",", ".")
    return s


def main():
    records = json.loads((OUT / "salary_records.json").read_text())
    schema = json.loads((OUT / "schema.json").read_text())
    validator = Draft202012Validator(schema)

    report = {"total_records": len(records)}

    # 1) schema
    errs = []
    for r in records:
        for e in validator.iter_errors(r):
            errs.append({"id": r.get("id"), "path": list(e.path), "msg": e.message})
            if len(errs) > 50:
                break
    report["schema_valid"] = len(errs) == 0
    report["schema_errors"] = errs[:50]

    # 2) percentile ordering
    ord_viol = []
    for r in records:
        m = r["measure"]
        seq = [(k, m[k]) for k in ["p25", "median", "p75", "p90"] if k in m]
        for (k1, v1), (k2, v2) in zip(seq, seq[1:]):
            if v1 > v2 + 1:  # allow rounding noise
                ord_viol.append({"id": r["id"], "where": f"{k1}={v1} > {k2}={v2}"})
                break
    report["percentile_order_violations"] = len(ord_viol)
    report["percentile_order_examples"] = ord_viol[:15]

    # 3) value sanity (monthly gross in 10k..400k DKK; counts < 100000)
    sanity = []
    for r in records:
        m = r["measure"]
        pc = r["pay_concept"]
        for k in ["mean", "p25", "median", "p75", "p90"]:
            if k in m and pc in ("gross_monthly", "gross_monthly_with_components", "net_monthly"):
                if not (5000 <= m[k] <= 500000):
                    sanity.append({"id": r["id"], "where": f"{k}={m[k]} ({pc})"})
    report["value_sanity_violations"] = len(sanity)
    report["value_sanity_examples"] = sanity[:15]

    # 4) digit cross-check against raw page text
    page_text = {}
    for union, path in PDFS.items():
        with pdfplumber.open(path) as pdf:
            for i, pg in enumerate(pdf.pages):
                t = (pg.extract_text() or "")
                # normalize: remove spaces inside numbers and soft hyphens
                page_text[(union, i + 1)] = t.replace("\xad", "")
    mismatches = []
    checked = 0
    for r in records:
        key = (r["union"], r["source_page"])
        txt = page_text.get(key, "")
        txt_nospace = txt.replace(" ", "")
        for k in MONEY_KEYS:
            if k in r["measure"]:
                checked += 1
                needle = dk_num(r["measure"][k])
                # accept Danish-grouped ('78.065'), plain ('78065'), or
                # space-separated forms (line-wrap can drop the separator)
                plain = needle.replace(".", "").replace(",", "")
                if (needle in txt or plain in txt_nospace
                        or needle.replace(".", "") in txt_nospace):
                    continue
                mismatches.append({"id": r["id"], "key": k,
                                   "value": r["measure"][k], "needle": needle})
    report["digit_checks"] = checked
    report["digit_mismatches"] = len(mismatches)
    report["digit_mismatch_rate"] = round(len(mismatches) / max(checked, 1), 4)
    report["digit_mismatch_examples"] = mismatches[:20]

    # 5) coverage
    from collections import Counter
    by_union = Counter(r["union"] for r in records)
    pages_with = {u: sorted({r["source_page"] for r in records if r["union"] == u}) for u in by_union}
    report["records_by_union"] = dict(by_union)
    report["pages_with_records"] = {u: len(p) for u, p in pages_with.items()}
    report["measure_field_coverage"] = dict(Counter(
        k for r in records for k in r["measure"]))
    report["records_with_experience"] = sum(
        1 for r in records if r.get("experience_years_min") is not None)
    report["records_with_age"] = sum(
        1 for r in records if r.get("age_min") is not None)
    report["sector_distribution"] = dict(Counter(r["sector"] for r in records))
    report["pay_concept_distribution"] = dict(Counter(r["pay_concept"] for r in records))

    (OUT / "validation_report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2))

    # console summary
    print("=== VALIDATION SUMMARY ===")
    print("records:", report["total_records"])
    print("schema_valid:", report["schema_valid"], "| errors:", len(errs))
    print("percentile order violations:", report["percentile_order_violations"])
    print("value sanity violations:", report["value_sanity_violations"])
    print(f"digit cross-check: {report['digit_mismatches']}/{checked} mismatches "
          f"({report['digit_mismatch_rate']*100:.2f}%)")
    print("records by union:", report["records_by_union"])
    print("with experience:", report["records_with_experience"],
          "| with age:", report["records_with_age"])
    if mismatches:
        print("--- sample digit mismatches ---")
        for mm in mismatches[:10]:
            print("   ", mm)


if __name__ == "__main__":
    main()