normalize_scopes.py

#!/usr/bin/env python3
"""Replace ``scope`` column values using ``scope_normalization.json``.

Same pattern as :mod:`actions_normalization.normalize_actions`, but for the ``scope``
column. Canonical labels and raw spellings were derived from distinct non-empty
``scope`` values in the original cohort exports (historical note; this script reads
``datasets/normalized`` by default, not ``datasets/raw``).

Default ``--in-dir`` and ``--out-dir`` are both ``datasets/normalized`` under the
project root (read normalized CSVs, rewrite ``scope`` to canonical form, write back
to the same folder layout). Use a different ``--out-dir`` to avoid overwriting.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Dict, Tuple

_SCRIPT_DIR = Path(__file__).resolve().parent
_ROOT = _SCRIPT_DIR
for _ in range(8):
    if (_ROOT / "utils" / "csv_utils.py").is_file():
        break
    if _ROOT.parent == _ROOT:
        raise RuntimeError(
            f"Cannot find project root (utils/csv_utils.py) starting from {_SCRIPT_DIR}"
        )
    _ROOT = _ROOT.parent
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))

from utils.csv_utils import normalize_key, read_log_csv_rows, write_log_csv_dict_rows  # noqa: E402


def resolve_path(p: str | Path) -> Path:
    return Path(p).expanduser().resolve()


def load_mapping(path: Path) -> Dict[str, str]:
    """Canonical label -> list of raw values; build raw (exact) -> canonical."""
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError(f"Mapping must be a JSON object: {path}")
    variant_to_canonical: Dict[str, str] = {}
    for canonical, variants in data.items():
        if not isinstance(variants, list):
            raise ValueError(f"Value for {canonical!r} must be a list in {path}")
        for v in variants:
            s = str(v)
            if s in variant_to_canonical:
                raise ValueError(
                    f"Duplicate raw scope {s!r} maps to both "
                    f"{variant_to_canonical[s]!r} and {canonical!r}"
                )
            variant_to_canonical[s] = str(canonical).strip()
    # Idempotence: after rewriting to canonical, those strings must map to themselves.
    for canonical in data.keys():
        c = str(canonical).strip()
        variant_to_canonical.setdefault(c, c)
    return variant_to_canonical


def find_scope_column(headers: list[str]) -> str | None:
    for h in headers:
        if normalize_key(h) == "scope":
            return h
    return None


def normalize_file(
    csv_path: Path,
    out_path: Path,
    variant_to_canonical: Dict[str, str],
) -> Tuple[int, int, int]:
    """Returns (rows_out, cells_changed, unmapped_nonempty)."""
    headers, rows, _, _, delimiter = read_log_csv_rows(csv_path, encoding="utf-8-sig")
    scope_col = find_scope_column(headers)
    if scope_col is None:
        raise ValueError(f"No 'scope' column in {csv_path}")

    changed = 0
    unmapped = 0
    out_rows: list[Dict[str, str]] = []
    for row in rows:
        raw = row.get(scope_col, "")
        if raw in variant_to_canonical:
            canon = variant_to_canonical[raw]
            out_row = {**row, scope_col: canon}
            if raw != canon:
                changed += 1
        else:
            out_row = dict(row)
            if raw.strip():
                unmapped += 1
        out_rows.append(out_row)

    write_log_csv_dict_rows(out_path, headers, out_rows, delimiter=delimiter)

    return len(out_rows), changed, unmapped


def main() -> None:
    default_norm = _ROOT / "datasets" / "normalized"
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument(
        "--in-dir",
        default=str(default_norm),
        help="Directory of input CSV files (default: <project>/datasets/normalized).",
    )
    parser.add_argument(
        "--out-dir",
        default=str(default_norm),
        help="Directory for output CSVs (default: same as --in-dir).",
    )
    parser.add_argument(
        "--mapping",
        default=str(_SCRIPT_DIR / "scope_normalization.json"),
        help="JSON file: canonical scope name -> list of raw spellings.",
    )
    args = parser.parse_args()

    in_dir = resolve_path(args.in_dir)
    out_dir = resolve_path(args.out_dir)
    mapping_path = resolve_path(args.mapping)

    if not in_dir.is_dir():
        raise SystemExit(f"Input directory does not exist: {in_dir}")
    if not mapping_path.is_file():
        raise SystemExit(f"Mapping file not found: {mapping_path}")

    variant_to_canonical = load_mapping(mapping_path)
    csv_files = sorted(in_dir.glob("*.csv"))
    if not csv_files:
        raise SystemExit(f"No *.csv files in {in_dir}")

    total_rows = 0
    total_changed = 0
    total_unmapped = 0
    print("Scope normalization:")
    for csv_path in csv_files:
        out_path = out_dir / csv_path.name
        try:
            n, chg, unm = normalize_file(csv_path, out_path, variant_to_canonical)
        except ValueError as e:
            raise SystemExit(str(e)) from e
        total_rows += n
        total_changed += chg
        total_unmapped += unm
        print(
            f"  {csv_path.name}: {n} rows -> {out_path} "
            f"(scope strings changed casing/spelling: {chg}, non-empty unmapped: {unm})"
        )
    print(
        f"Done. {len(csv_files)} file(s), {total_rows} rows, "
        f"{total_changed} scope cells rewritten to canonical form, "
        f"{total_unmapped} rows with non-empty unmapped scopes."
    )


if __name__ == "__main__":
    main()