scripts/reextract_multipanel.py

"""Re-extract the multi-panel IDA tables whose second axis was flattened away.

The original ``build_records_ida`` keyed ``measure[stat]`` by stat name, so for
multi-panel tables (region t11, gender×leadership t17, management-span t14,
public position t27) each panel overwrote the previous — only the last panel's
numbers survived and the panel identity was lost. This script adds the lattice
analogue of the Djøf ``split_panels`` logic: it finds the repeated stat band,
splits columns into panels, reads each panel's segment title from the header
rows above, and emits one ``IdaSalaryRecord`` per (row × panel) with the typed
dimension recovered.

Deterministic pdfplumber positional parsing (no OCR) — the IDA PDF is
digital-born; this keeps the 0-digit-error guarantee. Output:
``data/processed/lonstatistik/ida_reextracted.jsonl`` (picked up by
``build_ida_records.py``).

Usage::  uv run python scripts/reextract_multipanel.py
"""

from __future__ import annotations

import json
import re
import sys
from pathlib import Path

import pdfplumber

REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO))
sys.path.insert(0, str(REPO / "scripts"))

from extract_engine import (  # noqa: E402
    cluster_rows, get_rulings, is_number, norm, parse_number, xcenter,
)

from src.backend.indexing.ida_transform import (  # noqa: E402
    POSITION_RANK, SPAN_BOUNDS, reliability_tier,
)
from src.backend.schemas import IdaSalaryRecord, Measure  # noqa: E402

IDA_PDF = REPO / "data" / "raw" / "ida-loenstatistik-2025.pdf"
OUT = REPO / "data" / "processed" / "lonstatistik" / "ida_reextracted.jsonl"
SOURCE_DOC = "IDA Lønstatistik 2025"


def _stat_of(text: str) -> str | None:
    """Map a stat-header token to a measure key (these tables carry count+mean)."""
    t = norm(text).replace(" ", "")
    if "antal" in t:
        return "count"
    if "bruttolon" in t or "gennemsnit" in t or "brutolon" in t:
        return "mean"
    return None


def extract_panels(pg) -> dict | None:
    """Geometry pass (ruling-based): stat band → panels → titles → per-(row,panel) values.

    Uses the vertical rulings to define exact column boundaries (robust to
    missing cells / dashes, which otherwise shift nearest-center assignment).
    Only columns whose header carries a stat token (Antal/Bruttoløn/Gennemsnit)
    are kept, which also drops spurious narrow rulings. A new panel begins at
    each 'count'. Returns ``{panel_titles, rows:[{label, panels:[{stat:val}]}]}``.
    """
    bounds = get_rulings(pg)
    if len(bounds) < 4:
        return None
    ncol = len(bounds) - 1

    def colof(cx: float) -> int | None:
        for i in range(ncol):
            if bounds[i] - 1 <= cx <= bounds[i + 1] + 1:
                return i
        return None

    words = pg.extract_words(keep_blank_chars=False)
    rows = cluster_rows(words)

    # 1) stat header row = the row with the most count/mean tokens.
    best_i, best_n = None, 0
    for i, r in enumerate(rows):
        n = sum(1 for w in r if _stat_of(w["text"]))
        if n > best_n:
            best_i, best_n = i, n
    if best_i is None or best_n < 4:
        return None

    # 2) classify each column's stat from the header row tokens.
    col_stat: dict[int, str] = {}
    for w in rows[best_i]:
        s = _stat_of(w["text"])
        ci = colof(xcenter(w))
        if s and ci is not None and ci > 0:  # col 0 is the row label
            col_stat[ci] = s
    stat_cols = sorted(col_stat)  # column indices carrying a statistic
    if len(stat_cols) < 2:
        return None

    # 3) split stat columns into panels — a new panel begins at each 'count'.
    panels: list[list[int]] = []
    for ci in stat_cols:
        if col_stat[ci] == "count" or not panels:
            panels.append([ci])
        else:
            panels[-1].append(ci)

    # panel x-range from the ruling bounds of its first/last columns.
    p_lo = [bounds[p[0]] for p in panels]
    p_hi = [bounds[p[-1] + 1] for p in panels]
    first_stat_x = bounds[stat_cols[0]]

    # 4) panel titles: words above the stat row, bucketed by panel x-range.
    title_frags: list[list[str]] = [[] for _ in panels]
    for ri in range(best_i):
        for w in rows[ri]:
            cx = xcenter(w)
            if cx < first_stat_x - 4 or is_number(w["text"]):
                continue
            for pi in range(len(panels)):
                if p_lo[pi] - 4 <= cx <= p_hi[pi] + 4:
                    title_frags[pi].append(w["text"])
                    break
    panel_titles = [re.sub(r"\s+", " ", " ".join(f)).strip() for f in title_frags]

    # 5) data rows: label in col 0; each numeric cell placed by its column.
    out_rows = []
    for ri in range(best_i + 1, len(rows)):
        r = rows[ri]
        nums = [w for w in r if is_number(w["text"])]
        if len(nums) < 2:
            continue
        # Column 0 is the label column by ruling, so keep all of it — including
        # numeric-only labels like '1996' (single-year cohort) or '0' (span band),
        # which a numeric filter would wrongly drop.
        label = re.sub(r"\s+", " ", " ".join(
            w["text"] for w in sorted(r, key=lambda w: w["x0"])
            if colof(xcenter(w)) == 0)).strip()
        if not label:
            continue
        cellvals: dict[int, float] = {}
        for w in nums:
            ci = colof(xcenter(w))
            if ci is None:
                continue
            v, _pct, missing = parse_number(w["text"])
            if not missing:
                cellvals[ci] = v
        panel_measures = [
            {col_stat[ci]: cellvals[ci] for ci in p if ci in cellvals}
            for p in panels
        ]
        out_rows.append({"label": label, "panels": panel_measures})

    return {"panel_titles": panel_titles, "rows": out_rows}


# --- cohort / span row-label parsing ---------------------------------------
def parse_cohort(label: str) -> tuple[int | None, int | None] | None:
    """Cohort label → (start, end) with the 'Før YYYY' half-open convention."""
    low = norm(label)
    if low in ("alle", "ialt", "ialt"):
        return (None, None)
    m = re.search(r"(?:for)\s*(\d{4})", low)  # 'før' folds to 'for'
    if m:
        return (None, int(m.group(1)) - 1)
    m = re.search(r"(\d{4})(?:[-–](\d{2,4}))?", low)
    if not m:
        return None
    y0 = int(m.group(1))
    y1 = m.group(2)
    if y1:
        y1 = int(y1) if len(y1) == 4 else int(str(y0)[:2] + y1)
    else:
        y1 = y0
    return (y0, y1)


def parse_span(label: str) -> str | None:
    """Management-span row label → management_span_band enum value."""
    t = label.strip().replace("–", "-").replace(" ", "")
    if t == "0":
        return "none"
    if t in SPAN_BOUNDS:
        return t
    m = re.match(r"^(\d+)\+$", t)
    if m and f"{m.group(1)}+" in SPAN_BOUNDS:
        return f"{m.group(1)}+"
    return None


# --- per-table panel-title → typed dimensions ------------------------------
_REGION = {
    "hovedstaden": "hovedstaden", "sjælland": "sjaelland", "sjaelland": "sjaelland",
    "syddanmark": "syddanmark", "midtjylland": "midtjylland", "nordjylland": "nordjylland",
}
_PUBLIC_POSITION = [  # (keyword in title, position_level)
    ("specialkonsulent", "specialist"),
    ("chefkonsulent", "chef_seniorkonsulent"),
    ("menig", "ingenior_akademiker"),
    ("almindelig", "ingenior_akademiker"),
    ("chef", "afdelingschef"),
]
_PRIVATE_POSITION = [
    ("topchef", "topchef"), ("funktionsdirekt", "topchef"),
    ("afdelingschef", "afdelingschef"), ("projektleder", "projektleder"),
]


def _is_alle(title: str) -> bool:
    return norm(title).replace("*", "").strip() in ("alle", "alleledere", "alle*")


def _base(sector: str, page: int, rec_type: str = "salary_observation") -> dict:
    return dict(
        id="", source_doc=SOURCE_DOC, source_page=page, sector=sector,
        record_type=rec_type, currency="DKK", union="IDA",
    )


def build_region(rt: dict, page: int) -> list[IdaSalaryRecord]:
    """Tabel 11: private engineers by region × cohort (sector FIX: private)."""
    out = []
    titles = rt["panel_titles"]
    for row in rt["rows"]:
        coh = parse_cohort(row["label"])
        if coh is None:
            continue
        for pi, m in enumerate(row["panels"]):
            if "mean" not in m:
                continue
            region = _REGION.get(norm(titles[pi]).strip())
            if region is None:  # 'Alle' aggregate panel → skip (redundant)
                continue
            n = m.get("count")
            out.append(IdaSalaryRecord(
                **_base("private", page), table_id=11, table_title="REGION",
                pay_concept="gross_monthly", data_period_month="2025-09",
                region=region, row_dimension="region × kandidatår",
                row_label=row["label"], segment_label=titles[pi],
                graduation_year_start=coh[0], graduation_year_end=coh[1],
                experience_years_min=2025 - coh[1] if coh[1] else None,
                experience_years_max=2025 - coh[0] if coh[0] else None,
                dimension_keys=["region"] + (["graduation_year"] if coh != (None, None) else []),
                specificity=1 + (coh != (None, None)),
                sample_size=n, reliability_tier=reliability_tier(n),
                measure=Measure(count=n, mean=m.get("mean")), rag_text="",
            ))
    return out


def build_gender(rt: dict, page: int) -> list[IdaSalaryRecord]:
    """Tabel 17: gender × leadership × cohort (the equal-pay comparison)."""
    out = []
    titles = rt["panel_titles"]
    # 6 panels = 3 leadership groups × (Kvinde, Mand), in order.
    group_by_panel = {0: ("leader", "yes"), 1: ("leader", "yes"),
                      2: ("leader", "no"), 3: ("leader", "no"),
                      4: ("non_leader", "no"), 5: ("non_leader", "no")}
    for row in rt["rows"]:
        coh = parse_cohort(row["label"])
        if coh is None:
            continue
        for pi, m in enumerate(row["panels"]):
            if "mean" not in m or pi not in group_by_panel:
                continue
            tnorm = norm(titles[pi])
            gender = "female" if "kvinde" in tnorm else ("male" if "mand" in tnorm else None)
            if gender is None:
                continue
            is_leader, manages = group_by_panel[pi]
            n = m.get("count")
            dims = ["gender", "is_leader", "manages_people"]
            if coh != (None, None):
                dims.append("graduation_year")
            out.append(IdaSalaryRecord(
                **_base("private", page), table_id=17,
                table_title="MÆND OG KVINDER, ledelse med/uden personaleansvar",
                pay_concept="gross_monthly", data_period_month="2025-09",
                gender=gender, is_leader=is_leader, manages_people=manages,
                row_dimension="køn × ledelse × kandidatår", row_label=row["label"],
                segment_label=titles[pi],
                graduation_year_start=coh[0], graduation_year_end=coh[1],
                experience_years_min=2025 - coh[1] if coh[1] else None,
                experience_years_max=2025 - coh[0] if coh[0] else None,
                dimension_keys=dims, specificity=len(dims),
                sample_size=n, reliability_tier=reliability_tier(n),
                measure=Measure(count=n, mean=m.get("mean")), rag_text="",
            ))
    return out


def build_public_position(rt: dict, page: int) -> list[IdaSalaryRecord]:
    """Tabel 27: public-state position level × cohort."""
    out = []
    titles = rt["panel_titles"]
    for row in rt["rows"]:
        coh = parse_cohort(row["label"])
        if coh is None:
            continue
        for pi, m in enumerate(row["panels"]):
            if "mean" not in m or _is_alle(titles[pi]):
                continue
            tnorm = norm(titles[pi])
            pos = next((p for kw, p in _PUBLIC_POSITION if kw in tnorm), None)
            if pos is None:
                continue
            manages = "yes" if "personaleansvar" in tnorm else "all"
            n = m.get("count")
            dims = ["position_level"] + (["graduation_year"] if coh != (None, None) else [])
            out.append(IdaSalaryRecord(
                **_base("public_state", page), table_id=27,
                table_title="STATEN: STILLINGSNIVEAU",
                pay_concept="gross_monthly", data_period_month="2025-05",
                position_level=pos, position_level_rank=POSITION_RANK.get(pos),
                manages_people=manages,
                row_dimension="stillingsniveau × kandidatår", row_label=row["label"],
                segment_label=titles[pi],
                graduation_year_start=coh[0], graduation_year_end=coh[1],
                experience_years_min=2025 - coh[1] if coh[1] else None,
                experience_years_max=2025 - coh[0] if coh[0] else None,
                dimension_keys=dims, specificity=len(dims),
                sample_size=n, reliability_tier=reliability_tier(n),
                measure=Measure(count=n, mean=m.get("mean")), rag_text="",
            ))
    return out


def build_management(rt: dict, page: int) -> list[IdaSalaryRecord]:
    """Tabel 14: position × management-span band (recovers management_span_band)."""
    out = []
    titles = rt["panel_titles"]
    for row in rt["rows"]:
        band = parse_span(row["label"])
        if band is None:
            continue
        lo, hi = SPAN_BOUNDS[band]
        for pi, m in enumerate(row["panels"]):
            if "mean" not in m or _is_alle(titles[pi]):
                continue
            tnorm = norm(titles[pi])
            pos = next((p for kw, p in _PRIVATE_POSITION if kw in tnorm), None)
            if pos is None:
                continue
            n = m.get("count")
            dims = ["position_level", "management_span_band", "manages_people", "is_leader"]
            out.append(IdaSalaryRecord(
                **_base("private", page), table_id=14, table_title="PERSONALEANSVAR",
                pay_concept="gross_monthly", data_period_month="2025-09",
                position_level=pos, position_level_rank=POSITION_RANK.get(pos),
                management_span_band=band, span_min=lo, span_max=hi,
                manages_people="no" if band == "none" else "yes",
                is_leader="leader",
                row_dimension="stillingstype × personaleansvar", row_label=row["label"],
                segment_label=titles[pi],
                dimension_keys=dims, specificity=len(dims),
                sample_size=n, reliability_tier=reliability_tier(n),
                measure=Measure(count=n, mean=m.get("mean")), rag_text="",
            ))
    return out


# Page 22 stacks tabel 14 (personaleansvar) and tabel 15 (virksomhedsstørrelse)
# on one page; the single stat-band detector can't separate them and reads
# garbage counts (the README documents these matrix tables as unreliable).
# management_span_band / stem_count_band are "captured, don't filter" in v1, so
# we deliberately omit page 22 rather than emit bad data. build_management is
# kept for a future dedicated parser.
_TABLES = [
    (18, build_region), (25, build_gender), (30, build_public_position),
]


def main() -> None:
    records: list[IdaSalaryRecord] = []
    with pdfplumber.open(IDA_PDF) as pdf:
        for page, builder in _TABLES:
            rt = extract_panels(pdf.pages[page - 1])
            if rt is None:
                print(f"  p{page}: no panel structure found — skipped")
                continue
            recs = builder(rt, page)
            print(f"  p{page} {builder.__name__}: {len(recs)} records "
                  f"({len(rt['panel_titles'])} panels: {rt['panel_titles']})")
            records += recs

    # stable ids
    for i, r in enumerate(records):
        r.id = f"ida-reextract-{r.table_id}-{i:04d}"

    OUT.write_text(
        "\n".join(r.model_dump_json(exclude_none=True) for r in records) + "\n",
        encoding="utf-8",
    )
    print(f"OK — {len(records)} re-extracted records → {OUT}")


if __name__ == "__main__":
    main()