"""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)