term_fields.py

"""Map a grader CSV of term → field of law onto the current vocabulary.

CSV field names are messy (case, slashes, dropped ``Law``, a typo). Resolve
them onto the 26 canonical names already on the template, drop anything that
cannot be resolved or has no source analysis, then replace each dropped term
with an unused CSV term in the same concept class, preferring the field of
law that currently has the fewest selected terms.
"""

from __future__ import annotations

import csv
import re
from collections import Counter
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path

_FIELD_ALIASES = {
    "company law": "business company law",
    "personal injury tort": "personal injury tort law",
    "debtor creditor": "debtor creditor law",
}


@dataclass(frozen=True)
class CsvFieldRow:
    """One row from the term → field CSV."""

    term: str
    raw_field: str
    exclude: bool


@dataclass(frozen=True)
class TermAction:
    """One keep / drop / add / discard decision for the reassignment report."""

    term: str
    action: str
    previous_fol: str
    new_fol: str
    fol_changed: str
    concept_class: str
    reason: str


@dataclass(frozen=True)
class CsvTermSelection:
    """The keep / drop / fill decision for a CSV-driven grading deck."""

    pairs: tuple[tuple[str, str], ...]
    kept: tuple[str, ...]
    dropped: tuple[str, ...]
    added: tuple[str, ...]
    actions: tuple[TermAction, ...] = ()

    @property
    def by_term(self) -> dict[str, str]:
        return dict(self.pairs)


def normalize_field_name(raw: str) -> str:
    """Fold case, slashes, punctuation, and the injutry typo for matching."""
    text = str(raw).strip().lower().replace("injutry", "injury")
    text = re.sub(r"[/_]+", " ", text)
    text = re.sub(r"[^a-z0-9]+", " ", text)
    return " ".join(text.split())


def resolve_csv_field(raw: str, vocabulary: Sequence[str]) -> str | None:
    """Return the canonical vocabulary name, or None if it is out of set."""
    text = str(raw or "").strip()
    if not text or text.lower() == "nan":
        return None
    index = {normalize_field_name(name): name for name in vocabulary}
    key = _FIELD_ALIASES.get(normalize_field_name(text), normalize_field_name(text))
    return index.get(key)


def read_term_fields_csv(path: Path) -> tuple[CsvFieldRow, ...]:
    """Load ``term, field of law[, note]`` rows in file order."""
    with path.open(encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        return tuple(_row_from_csv(row) for row in reader if _csv_term(row))


def seed_terms_from_template(terms: Sequence[str]) -> list[str]:
    """First-seen term order from a grading template."""
    return list(dict.fromkeys(str(term).strip() for term in terms))


_REPORT_COLUMNS = (
    "Term",
    "Action",
    "Previous field of law",
    "New field of law",
    "Field of law changed",
    "Concept class",
    "Reason",
)
_ACTION_LABELS = {
    "kept": "Kept",
    "dropped": "Dropped",
    "added": "Added",
    "discarded": "Discarded",
}
_CHANGED_LABELS = {"true": "Yes", "false": "No"}
_REPORT_COLUMN_WIDTHS = (28, 12, 32, 32, 22, 32, 72)


def write_reassignment_report(actions: Sequence[TermAction], path: Path) -> Path:
    """Write a sortable Excel report with a frozen header row."""
    from openpyxl import Workbook
    from openpyxl.styles import Alignment, Font
    from openpyxl.utils import get_column_letter
    from openpyxl.worksheet.table import Table, TableStyleInfo

    output = path.with_suffix(".xlsx")
    output.parent.mkdir(parents=True, exist_ok=True)
    workbook = Workbook()
    sheet = workbook.active
    assert sheet is not None
    sheet.title = "Reassignment"
    sheet.append(list(_REPORT_COLUMNS))
    for row in actions:
        sheet.append(
            [
                row.term,
                _ACTION_LABELS.get(row.action, row.action),
                row.previous_fol,
                row.new_fol,
                _CHANGED_LABELS.get(row.fol_changed, row.fol_changed),
                row.concept_class,
                row.reason,
            ]
        )
    last_row = max(sheet.max_row, 1)
    last_col = get_column_letter(len(_REPORT_COLUMNS))
    header_font = Font(bold=True)
    wrap = Alignment(vertical="top", wrap_text=True)
    for cell in sheet[1]:
        cell.font = header_font
        cell.alignment = Alignment(vertical="center", wrap_text=True)
    for row in sheet.iter_rows(min_row=2, max_row=last_row):
        for cell in row:
            cell.alignment = wrap
    for index, width in enumerate(_REPORT_COLUMN_WIDTHS, start=1):
        sheet.column_dimensions[get_column_letter(index)].width = width
    sheet.freeze_panes = "A2"
    if last_row > 1:
        table = Table(displayName="FoLReassignment", ref=f"A1:{last_col}{last_row}")
        table.tableStyleInfo = TableStyleInfo(
            name="TableStyleMedium2",
            showRowStripes=True,
        )
        sheet.add_table(table)
    workbook.save(output)
    return output


def choose_terms_from_csv(
    seed_terms: Sequence[str],
    rows: Sequence[CsvFieldRow],
    *,
    vocabulary: Sequence[str],
    source_pairs: set[tuple[str, str]],
    concept_by_term: Mapping[str, str] | None = None,
    previous_fols: Mapping[str, str] | None = None,
    target_count: int = 80,
) -> CsvTermSelection:
    """Keep in-vocab seed terms, then replace each drop by concept and scarcest FoL."""
    by_term = {row.term: row for row in rows}
    csv_index = {row.term: index for index, row in enumerate(rows)}
    concepts = dict(concept_by_term or {})
    previous = dict(previous_fols or {})
    kept = _usable_pairs(seed_terms, by_term, vocabulary, source_pairs)
    kept_terms = {term for term, _ in kept}
    dropped = tuple(term for term in seed_terms if term not in kept_terms)
    selected = list(kept)
    used = set(kept_terms)
    added: list[tuple[str, str]] = []
    discarded_first: dict[str, TermAction] = {}
    added_actions: list[TermAction] = []

    for dropped_term in dropped:
        if len(selected) >= target_count:
            break
        picked = _pick_replacement(
            rows,
            used=used,
            dropped_term=dropped_term,
            vocabulary=vocabulary,
            source_pairs=source_pairs,
            concepts=concepts,
            csv_index=csv_index,
            selected_fields=Counter(field for _, field in selected),
        )
        if picked is None:
            continue
        term, field, pool, same_concept = picked
        selected.append((term, field))
        added.append((term, field))
        used.add(term)
        added_actions.append(
            _added_action(
                term,
                field,
                dropped_term,
                same_concept,
                concepts,
            )
        )
        _record_discards(
            discarded_first,
            pool=pool,
            picked=term,
            dropped_term=dropped_term,
            concepts=concepts,
        )

    actions = (
        tuple(
            _seed_action(
                term,
                kept=term in kept_terms,
                by_term=by_term,
                vocabulary=vocabulary,
                source_pairs=source_pairs,
                concepts=concepts,
                previous=previous,
            )
            for term in seed_terms
        )
        + tuple(added_actions)
        + tuple(
            row
            for term, row in discarded_first.items()
            if term not in {added_term for added_term, _ in added}
        )
    )
    return CsvTermSelection(
        pairs=tuple(selected),
        kept=tuple(term for term, _ in kept),
        dropped=dropped,
        added=tuple(term for term, _ in added),
        actions=actions,
    )


def _header_key(key: str | None) -> str:
    return str(key or "").strip().lstrip("\ufeff").lower()


def _csv_term(row: dict[str, str | None]) -> str:
    for key, value in row.items():
        if _header_key(key) == "term":
            return str(value or "").strip()
    return ""


def _csv_field(row: dict[str, str | None]) -> str:
    for key, value in row.items():
        if "field" in _header_key(key):
            return str(value or "").strip()
    return ""


def _csv_exclude(row: dict[str, str | None]) -> bool:
    for key, value in row.items():
        lowered = _header_key(key)
        if lowered == "term" or "field" in lowered:
            continue
        if "exclude" in str(value or "").lower():
            return True
    return False


def _row_from_csv(row: dict[str, str | None]) -> CsvFieldRow:
    return CsvFieldRow(
        term=_csv_term(row),
        raw_field=_csv_field(row),
        exclude=_csv_exclude(row),
    )


def _usable_field(
    row: CsvFieldRow,
    vocabulary: Sequence[str],
    source_pairs: set[tuple[str, str]],
) -> str | None:
    if row.exclude:
        return None
    field = resolve_csv_field(row.raw_field, vocabulary)
    if field is None:
        return None
    if (row.term, field) not in source_pairs:
        return None
    return field


def _usable_pairs(
    terms: Sequence[str],
    by_term: Mapping[str, CsvFieldRow],
    vocabulary: Sequence[str],
    source_pairs: set[tuple[str, str]],
) -> list[tuple[str, str]]:
    pairs: list[tuple[str, str]] = []
    for term in terms:
        row = by_term.get(term)
        if row is None:
            continue
        field = _usable_field(row, vocabulary, source_pairs)
        if field is None:
            continue
        pairs.append((term, field))
    return pairs


def _eligible_unused(
    rows: Sequence[CsvFieldRow],
    *,
    used: set[str],
    vocabulary: Sequence[str],
    source_pairs: set[tuple[str, str]],
) -> list[tuple[str, str]]:
    eligible: list[tuple[str, str]] = []
    for row in rows:
        if row.term in used:
            continue
        field = _usable_field(row, vocabulary, source_pairs)
        if field is None:
            continue
        eligible.append((row.term, field))
    return eligible


def _pick_replacement(
    rows: Sequence[CsvFieldRow],
    *,
    used: set[str],
    dropped_term: str,
    vocabulary: Sequence[str],
    source_pairs: set[tuple[str, str]],
    concepts: Mapping[str, str],
    csv_index: Mapping[str, int],
    selected_fields: Counter[str],
) -> tuple[str, str, list[tuple[str, str]], bool] | None:
    eligible = _eligible_unused(
        rows, used=used, vocabulary=vocabulary, source_pairs=source_pairs
    )
    if not eligible:
        return None
    dropped_concept = concepts.get(dropped_term, "")
    same = [pair for pair in eligible if concepts.get(pair[0], "") == dropped_concept]
    pool = same if same else eligible
    term, field = min(
        pool,
        key=lambda pair: (selected_fields[pair[1]], csv_index.get(pair[0], 10**9)),
    )
    return term, field, pool, bool(same)


def _csv_new_fol(row: CsvFieldRow | None, vocabulary: Sequence[str]) -> str:
    if row is None:
        return ""
    resolved = resolve_csv_field(row.raw_field, vocabulary)
    return resolved or row.raw_field


def _fol_changed(previous: str, new: str) -> str:
    if not previous:
        return ""
    return "true" if previous != new else "false"


def _drop_reason(
    row: CsvFieldRow | None,
    vocabulary: Sequence[str],
    source_pairs: set[tuple[str, str]],
) -> str:
    if row is None:
        return "dropped: not in CSV"
    if row.exclude:
        return "dropped: excluded"
    field = resolve_csv_field(row.raw_field, vocabulary)
    if field is None:
        return f"dropped: out of vocabulary ({row.raw_field or 'empty'})"
    if (row.term, field) not in source_pairs:
        return f"dropped: no source analysis for {field}"
    return "dropped"


def _seed_action(
    term: str,
    *,
    kept: bool,
    by_term: Mapping[str, CsvFieldRow],
    vocabulary: Sequence[str],
    source_pairs: set[tuple[str, str]],
    concepts: Mapping[str, str],
    previous: Mapping[str, str],
) -> TermAction:
    row = by_term.get(term)
    new_fol = _csv_new_fol(row, vocabulary)
    previous_fol = previous.get(term, "")
    if kept:
        reason = f"kept: CSV FoL {new_fol} is in vocabulary"
    else:
        reason = _drop_reason(row, vocabulary, source_pairs)
    return TermAction(
        term=term,
        action="kept" if kept else "dropped",
        previous_fol=previous_fol,
        new_fol=new_fol,
        fol_changed=_fol_changed(previous_fol, new_fol),
        concept_class=concepts.get(term, ""),
        reason=reason,
    )


def _added_action(
    term: str,
    field: str,
    dropped_term: str,
    same_concept: bool,
    concepts: Mapping[str, str],
) -> TermAction:
    if same_concept:
        reason = (
            f"added: same concept as {dropped_term} "
            f"({concepts.get(dropped_term, '')}); scarcest FoL in selected set"
        )
    else:
        reason = (
            f"added: no same-concept replacement for {dropped_term}; "
            "scarcest FoL among remaining eligible terms"
        )
    return TermAction(
        term=term,
        action="added",
        previous_fol="",
        new_fol=field,
        fol_changed="",
        concept_class=concepts.get(term, ""),
        reason=reason,
    )


def _record_discards(
    discarded_first: dict[str, TermAction],
    *,
    pool: list[tuple[str, str]],
    picked: str,
    dropped_term: str,
    concepts: Mapping[str, str],
) -> None:
    for term, field in pool:
        if term == picked or term in discarded_first:
            continue
        discarded_first[term] = TermAction(
            term=term,
            action="discarded",
            previous_fol="",
            new_fol=field,
            fol_changed="",
            concept_class=concepts.get(term, ""),
            reason=(
                f"discarded: considered for {dropped_term} but {picked} was "
                "chosen (scarcest FoL / CSV order)"
            ),
        )