lib/review.py

"""Human-in-the-loop review logic for the Live Parser output.

Pure functions over the entity list (no Streamlit imports) so the corrected
highlighted text, the structured summary, the JSON export and the override
stats all derive from one editable source of truth.

A "work" item mirrors a model entity plus review bookkeeping:
    {"_id", "type", "start", "end", "text", "conf",
     "origin": "model"|"added", "orig_type": <type or None>}
"""
from __future__ import annotations

import re


def seed_review(entities):
    """Copy model entities into an editable working list with stable ids."""
    work = []
    for i, e in enumerate(entities):
        work.append({
            "_id": i,
            "type": e["type"],
            "start": e["start"],
            "end": e["end"],
            "text": e["text"],
            "conf": e.get("conf", 1.0),
            "origin": "model",
            "orig_type": e["type"],   # frozen original label, for the relabel diff
        })
    return work


def _next_id(work):
    return max((e["_id"] for e in work), default=-1) + 1


def relabel(work, _id, new_type):
    """Change the label of the entity with id ``_id`` (in place)."""
    for e in work:
        if e["_id"] == _id:
            e["type"] = new_type
    return work


def delete(work, _id):
    """Remove the entity with id ``_id``."""
    return [e for e in work if e["_id"] != _id]


def add_entity(work, phrase, full_text, etype, hint=0):
    """Tag ``phrase`` as a new entity, locating it in ``full_text``.

    Returns (work, error). ``error`` is None on success. The phrase is located
    near ``hint`` (the clicked char offset) first, then anywhere, so repeated
    words resolve to the one the reviewer clicked.
    """
    phrase = (phrase or "").strip()
    if not phrase:
        return work, "Nothing to add — the selection was empty."
    idx = full_text.find(phrase, max(0, hint - len(phrase)))
    if idx < 0:
        idx = full_text.find(phrase)
    if idx < 0:
        return work, f"Couldn't find “{phrase}” in the CV text."
    start, end = idx, idx + len(phrase)
    for e in work:
        if e["start"] == start and e["end"] == end:
            return work, "That exact span is already tagged."
    work = work + [{
        "_id": _next_id(work),
        "type": etype, "start": start, "end": end, "text": phrase,
        "conf": 1.0, "origin": "added", "orig_type": None,
    }]
    return work, None


def diff_stats(original, corrected):
    """Compare model output (``original``) with human-corrected output.

    ``original`` is the seeded snapshot; ``corrected`` is the edited work list.
    """
    corr_spans = {(e["start"], e["end"]) for e in corrected}

    relabeled, added = [], []
    for e in corrected:
        if e.get("origin") == "added":
            added.append(e)
        elif e.get("orig_type") and e["type"] != e["orig_type"]:
            relabeled.append(e)

    deleted = [e for e in original if (e["start"], e["end"]) not in corr_spans]

    n_model = len(original)
    n_touched = len(relabeled) + len(deleted)   # model entities the human changed
    return {
        "n_model": n_model,
        "n_corrected": len(corrected),
        "relabeled": relabeled,
        "deleted": deleted,
        "added": added,
        "n_relabeled": len(relabeled),
        "n_deleted": len(deleted),
        "n_added": len(added),
        "n_touched": n_touched,
        "override_rate": (n_touched / n_model) if n_model else 0.0,
    }