validator.py

"""Sanitize, validate, and normalize MecCog challenge submissions.

Accepts ``.xlsx`` or ``.csv`` uploads and parses them with a header-driven,
lenient state machine ported from
``CrowdSourcedLLMEvals/evaluate_submissions.py`` (``_clean`` + ``parse_xlsx``,
lines 83-145) so the offline pipeline and this intake agree on interpretation.

Public API:
    parse(path)                       -> Parsed
    validate(parsed, hypothesis=None) -> Report
    to_canonical(parsed)              -> (xlsx_bytes, parsed_json_dict)
"""

from __future__ import annotations

import csv
import io
import re
import unicodedata
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from pathlib import Path
from urllib.parse import urlparse

import openpyxl

from hypotheses import Hypothesis

MAX_PAPERS = 15

# Canonical column order for the normalized re-emit (template row 1, cols A-N).
CANONICAL_HEADERS = [
    "Hypothesis",
    "DOI",
    "Paper type",
    "Paper ID",
    "Findings",
    "Code",
    "Relevance score",
    "Segment confidence",
    "Finding confidence",
    "Sample size",
    "Stat test",
    "P value",
    "Effect size",
    "Evidence type",
]

_PAPER_CODE = re.compile(r"^P\d+$", re.IGNORECASE)
_FINDING_CODE = re.compile(r"^P\d+\.F\d+$", re.IGNORECASE)


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------

@dataclass
class Finding:
    code: str | None
    text: str
    relevance: float | None = None
    relevance_raw: str | None = None


@dataclass
class Paper:
    code: str
    doi: str | None
    paper_type: str | None
    paper_id: str | None
    pmid: str | None
    findings: list[Finding] = field(default_factory=list)
    inferred_code: bool = False


@dataclass
class Parsed:
    hypothesis: str
    papers: list[Paper]
    source_name: str
    n_columns: int
    headers: list[str]


@dataclass
class Issue:
    """One validation finding, anchored to the spreadsheet location it concerns.

    ``anchor`` is ``"A2"`` (the hypothesis cell), ``"header"`` (the header row), a
    paper code (``"P1"``), a finding code (``"P1.F2"``), or ``None`` (whole file).
    The UI uses it to highlight the matching preview row; ``fix`` is a short,
    actionable hint shown alongside the message.
    """

    severity: str  # "error" | "warning"
    message: str
    fix: str | None = None
    anchor: str | None = None


@dataclass
class Report:
    issues: list[Issue] = field(default_factory=list)
    summary: dict = field(default_factory=dict)

    def add_error(self, message: str, fix: str | None = None, anchor: str | None = None) -> None:
        self.issues.append(Issue("error", message, fix, anchor))

    def add_warning(self, message: str, fix: str | None = None, anchor: str | None = None) -> None:
        self.issues.append(Issue("warning", message, fix, anchor))

    @property
    def errors(self) -> list[str]:
        return [i.message for i in self.issues if i.severity == "error"]

    @property
    def warnings(self) -> list[str]:
        return [i.message for i in self.issues if i.severity == "warning"]

    @property
    def ok(self) -> bool:
        return not any(i.severity == "error" for i in self.issues)


# ---------------------------------------------------------------------------
# Sanitization helpers (ported from evaluate_submissions._clean)
# ---------------------------------------------------------------------------

def _clean(value) -> str | None:
    """Strip whitespace and literal single-quote wrappers; map ''/None/'None' -> None."""
    if value is None:
        return None
    s = str(value).strip()
    if s.startswith("'") and s.endswith("'") and len(s) > 1:
        s = s[1:-1].strip()
    if s.lower() == "none" or s == "":
        return None
    return s


def normalize_text(s: str) -> str:
    s = unicodedata.normalize("NFKD", s or "")
    s = s.lower()
    s = re.sub(r"[*_`#]", " ", s)
    s = re.sub(r"\s+", " ", s)
    return s.strip()


def _to_float(value) -> tuple[float | None, bool]:
    """Return (parsed_float, was_present). was_present distinguishes blank from bad."""
    cleaned = _clean(value)
    if cleaned is None:
        return None, False
    try:
        return float(cleaned), True
    except ValueError:
        return None, True


# ---------------------------------------------------------------------------
# Loading: produce a uniform list[list[cell]] grid from xlsx or csv
# ---------------------------------------------------------------------------

def _load_grid(path: str | Path) -> tuple[list[list], str]:
    p = Path(path)
    ext = p.suffix.lower()
    if ext == ".xlsx":
        wb = openpyxl.load_workbook(p, read_only=True, data_only=True)
        ws = wb.worksheets[0]
        grid = [[ws.cell(r, c).value for c in range(1, ws.max_column + 1)]
                for r in range(1, ws.max_row + 1)]
        wb.close()
        return grid, ws.title or p.name
    if ext == ".csv":
        with open(p, newline="", encoding="utf-8-sig") as f:
            grid = [list(row) for row in csv.reader(f)]
        return grid, p.name
    raise ValueError(f"Unsupported file type '{ext}'. Upload a .xlsx or .csv file.")


def _cell(row: list, idx: int):
    return row[idx] if (idx is not None and 0 <= idx < len(row)) else None


# Canonical default column positions (0-indexed) and the header synonyms that map
# to each logical field. The parser locates columns by header name first, falling
# back to these positions when a header is missing or unrecognised.
_DEFAULT_COLS = {
    "hypothesis": 0, "doi": 1, "paper_type": 2, "paper_id": 3,
    "findings": 4, "code": 5, "relevance": 6,
}
_HEADER_SYNONYMS = {
    "hypothesis": "hypothesis",
    "doi": "doi",
    "paper type": "paper_type",
    "paper id": "paper_id",
    "findings": "findings", "finding": "findings",
    "code": "code",
    "relevance score": "relevance", "relevance": "relevance",
}


def _column_map(headers: list[str]) -> dict[str, int]:
    cols = dict(_DEFAULT_COLS)
    for i, h in enumerate(headers):
        key = _HEADER_SYNONYMS.get(normalize_text(h))
        if key:
            cols[key] = i
    return cols


# ---------------------------------------------------------------------------
# Parse: row state machine (ported from evaluate_submissions.parse_xlsx,
# generalized to a header-driven column map and an inline first-paper row 2)
# ---------------------------------------------------------------------------

def parse(path: str | Path) -> Parsed:
    grid, source_name = _load_grid(path)
    n_columns = max((len(r) for r in grid), default=0)
    headers = [_clean(c) or "" for c in (grid[0] if grid else [])]
    cols = _column_map(headers)

    # Row 2, "Hypothesis" column = hypothesis text. (grid 0-indexed; row 2 -> grid[1])
    hypothesis = ""
    if len(grid) >= 2:
        hypothesis = (_clean(_cell(grid[1], cols["hypothesis"])) or "") \
            .replace("\n", " ").replace("\r", " ").strip()

    papers: list[Paper] = []
    paper_counter = 0

    # Process from row 2 onward: some templates inline the first paper (P1) on the
    # same row as the hypothesis (DOI/code in B/F), others start papers on row 3.
    for row in grid[1:]:
        if not any(v is not None and str(v).strip() != "" for v in row):
            continue

        doi = _clean(_cell(row, cols["doi"]))
        paper_type = _clean(_cell(row, cols["paper_type"]))
        paper_id_raw = _cell(row, cols["paper_id"])
        paper_id = _clean(str(paper_id_raw)) if paper_id_raw is not None else None
        finding_text = _clean(_cell(row, cols["findings"]))
        code = _clean(_cell(row, cols["code"]))
        relevance, rel_present = _to_float(_cell(row, cols["relevance"]))
        rel_raw = _clean(_cell(row, cols["relevance"]))

        is_paper = bool(code and _PAPER_CODE.match(code))
        is_finding = bool(code and _FINDING_CODE.match(code))

        if is_paper:
            paper_counter += 1
            pmid = paper_id if (paper_type and "pmid" in paper_type.lower()) else None
            papers.append(Paper(code.upper(), doi, paper_type, paper_id, pmid))
        elif doi and not code:
            # Paper row with a missing code — infer one.
            paper_counter += 1
            pmid = paper_id if (paper_type and "pmid" in paper_type.lower()) else None
            papers.append(Paper(f"P{paper_counter}", doi, paper_type, paper_id, pmid,
                                inferred_code=True))
        elif is_finding:
            if not papers:
                paper_counter += 1
                papers.append(Paper(f"P{paper_counter}", doi, None, None, None,
                                    inferred_code=True))
            if finding_text:
                papers[-1].findings.append(
                    Finding(code.upper(), finding_text, relevance,
                            rel_raw if rel_present else None))
        elif finding_text and papers:
            # Finding text with no/garbled code — attach to current paper, flag later.
            papers[-1].findings.append(
                Finding(None, finding_text, relevance, rel_raw if rel_present else None))

    return Parsed(hypothesis, papers, source_name, n_columns, headers)


# ---------------------------------------------------------------------------
# Validate
# ---------------------------------------------------------------------------

def _similar(a: str, b: str) -> float:
    return SequenceMatcher(None, normalize_text(a), normalize_text(b)).ratio()


def validate(parsed: Parsed, hypothesis: Hypothesis | None = None) -> Report:
    rep = Report()

    # --- hypothesis (A2) ---
    if not parsed.hypothesis:
        rep.add_error(
            "Cell A2 (the hypothesis statement) is empty. Put the hypothesis text in row 2, column A.",
            fix="Type the hypothesis text into cell A2 (row 2, column A).",
            anchor="A2",
        )
    elif hypothesis is not None:
        sim = _similar(parsed.hypothesis, hypothesis.text)
        if sim < 0.6:
            rep.add_warning(
                f"The hypothesis in A2 ({parsed.hypothesis!r}) does not closely match the "
                f"selected hypothesis {hypothesis.code} ({hypothesis.text!r}). "
                "Double-check you selected the right mechanism segment.",
                fix=f"Either re-select the correct hypothesis in the dropdown, or set A2 to: {hypothesis.text!r}.",
                anchor="A2",
            )

    # --- headers ---
    if parsed.headers:
        for h in parsed.headers:
            if h and h != h.strip():
                rep.add_warning(
                    "One or more column headers have leading/trailing whitespace — they were trimmed.",
                    fix="Remove the extra spaces around your column headers in row 1.",
                    anchor="header",
                )
                break
        if parsed.headers[0] and normalize_text(parsed.headers[0]) != "hypothesis":
            rep.add_warning(
                f"Column A header is {parsed.headers[0]!r}; expected 'Hypothesis'.",
                fix="Rename the column A header (cell A1) to 'Hypothesis'.",
                anchor="header",
            )

    # --- papers ---
    papers = parsed.papers
    if not papers:
        rep.add_error(
            "No papers found. Add at least one paper row with a P# code (e.g. 'P1') and its DOI.",
            fix="Add a row with 'P1' in the Code column and the paper's DOI in column B.",
        )
    if len(papers) > MAX_PAPERS:
        rep.add_error(
            f"{len(papers)} papers found; the maximum is {MAX_PAPERS}. Trim to the {MAX_PAPERS} most relevant.",
            fix=f"Remove paper rows until at most {MAX_PAPERS} remain.",
        )

    seen_paper_codes: set[str] = set()
    for p in papers:
        loc = f"paper {p.code}"
        if p.inferred_code:
            rep.add_warning(
                f"A paper row was missing a P# code; inferred {p.code}. Add an explicit code to be safe.",
                fix=f"Put '{p.code}' in the Code column for this paper row.",
                anchor=p.code,
            )
        if p.code in seen_paper_codes:
            rep.add_error(
                f"Duplicate paper code {p.code}.",
                fix=f"Give each paper a unique P# code; renumber the second {p.code}.",
                anchor=p.code,
            )
        seen_paper_codes.add(p.code)

        if not p.doi and not p.paper_id:
            rep.add_error(
                f"{loc} has neither a DOI nor a Paper ID. A DOI (column B) is required.",
                fix="Add the paper's DOI in column B (e.g. '10.1000/xyz').",
                anchor=p.code,
            )
        if p.doi and "/" not in p.doi and not p.doi.lower().startswith("10."):
            rep.add_warning(
                f"{loc} DOI {p.doi!r} does not look like a DOI (expected '10.xxxx/...').",
                fix="Check column B holds a DOI like '10.1000/xyz', not a URL or title.",
                anchor=p.code,
            )

        if not p.findings:
            rep.add_error(
                f"{loc} has no findings. Add at least one P{p.code[1:]}.F# finding row beneath it.",
                fix=f"Add a row with '{p.code}.F1' in the Code column and finding text in column E.",
                anchor=p.code,
            )

        seen_finding_codes: set[str] = set()
        for f in p.findings:
            f_anchor = f.code or p.code
            if f.code is None:
                rep.add_warning(
                    f"{loc}: a finding row has no/garbled code; expected '{p.code}.F#'.",
                    fix=f"Put a code like '{p.code}.F1' in the Code column for this finding.",
                    anchor=f_anchor,
                )
            else:
                if not f.code.startswith(p.code.upper() + "."):
                    rep.add_warning(
                        f"{loc}: finding code {f.code} does not match its paper {p.code}.",
                        fix=f"Rename {f.code} to '{p.code}.F#' so it matches its paper.",
                        anchor=f_anchor,
                    )
                if f.code in seen_finding_codes:
                    rep.add_error(
                        f"{loc}: duplicate finding code {f.code}.",
                        fix=f"Give each finding a unique code; renumber the second {f.code}.",
                        anchor=f_anchor,
                    )
                seen_finding_codes.add(f.code)
            if f.relevance_raw is not None and f.relevance is None:
                rep.add_error(
                    f"{f.code or loc}: relevance {f.relevance_raw!r} is not a number.",
                    fix="Put a number between 0 and 1 in the Relevance score column (G).",
                    anchor=f_anchor,
                )
            elif f.relevance is not None and not (0.0 <= f.relevance <= 1.0):
                rep.add_error(
                    f"{f.code or loc}: relevance {f.relevance} is out of the 0–1 range.",
                    fix="Set the Relevance score (column G) to a value between 0 and 1.",
                    anchor=f_anchor,
                )

    n_findings = sum(len(p.findings) for p in papers)
    rep.summary = {
        "hypothesis": parsed.hypothesis,
        "slug": hypothesis.slug if hypothesis else None,
        "code": hypothesis.code if hypothesis else None,
        "n_papers": len(papers),
        "n_findings": n_findings,
        "n_columns": parsed.n_columns,
        "source_name": parsed.source_name,
    }
    return rep


# ---------------------------------------------------------------------------
# Normalize: re-emit a clean canonical XLSX + the pipeline-compatible JSON
# ---------------------------------------------------------------------------

def to_json(parsed: Parsed) -> dict:
    """Match the {hypothesis, papers:[{code, doi, pmid, findings:[...]}]} shape
    that the offline evaluate_submissions.py / prune scripts already consume,
    while preserving relevance + paper_id for richer downstream use."""
    return {
        "hypothesis": parsed.hypothesis,
        "papers": [
            {
                "code": p.code,
                "doi": p.doi,
                "pmid": p.pmid,
                "paper_type": p.paper_type,
                "paper_id": p.paper_id,
                "findings": [
                    {"code": f.code, "text": f.text, "relevance": f.relevance}
                    for f in p.findings
                ],
                # Plain string list kept for byte-compatibility with parse_xlsx consumers.
                "findings_text": [f.text for f in p.findings],
            }
            for p in parsed.papers
        ],
    }


def to_canonical_xlsx(parsed: Parsed) -> bytes:
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "Submission"
    ws.append(CANONICAL_HEADERS)
    ws.append([parsed.hypothesis] + [None] * (len(CANONICAL_HEADERS) - 1))
    for p in parsed.papers:
        ws.append([None, p.doi, p.paper_type, p.paper_id, None, p.code, None])
        for f in p.findings:
            ws.append([None, None, None, None, f.text, f.code, f.relevance])
    buf = io.BytesIO()
    wb.save(buf)
    return buf.getvalue()


def to_canonical(parsed: Parsed) -> tuple[bytes, dict]:
    return to_canonical_xlsx(parsed), to_json(parsed)


def preview_rows(parsed: Parsed) -> list[list[str]]:
    """Flat table for the preview: one row per paper and finding."""
    return [cells for _, cells in preview_table(parsed)]


def preview_table(parsed: Parsed) -> list[tuple[str | None, list[str]]]:
    """Flat preview rows paired with each row's anchor (paper/finding code), so
    the UI can highlight the exact rows that triggered an issue. Anchors match
    those set by ``validate``: ``p.code`` for paper rows, ``f.code or p.code`` for
    finding rows."""
    rows: list[tuple[str | None, list[str]]] = []
    for p in parsed.papers:
        rows.append((p.code, [p.code, p.doi or "", p.paper_id or "", "", ""]))
        for f in p.findings:
            rel = "" if f.relevance is None else f"{f.relevance:g}"
            rows.append((f.code or p.code,
                         ["", "", "", f.code or "?", f"{rel}  {f.text}".strip()]))
    return rows


PREVIEW_HEADERS = ["Paper", "DOI", "Paper ID", "Finding", "Relevance / text"]


def is_valid_url(s: str) -> bool:
    """True for a well-formed absolute http(s) URL with a host. Used to validate
    the optional write-up / wiki link before accepting a submission."""
    try:
        u = urlparse((s or "").strip())
    except ValueError:
        return False
    return u.scheme in ("http", "https") and bool(u.netloc)