shared_resources/validate_submission.py

#!/usr/bin/env python3
# -----------------------------------------------------------------------------
# Papers & Findings submission validator — MecCog / APOE4 challenge
# Author: Abigail Djossou
# Date:   2026-07-03
# -----------------------------------------------------------------------------
"""
Validator for MecCog "papers and findings" submission spreadsheets.

Based on the official spreadsheet description. Checks structure and per-column rules, and prints
a clear report telling the submitter exactly what to fix if the sheet is not valid.

Usage:
    python validate_submission.py submission.xlsx
    python validate_submission.py submission.xlsx --strict   # warnings -> errors

The spreadsheet layout (official):
  Row 1  : column labels (must all be present, correct columns)
  Row 2  : hypothesis being considered (must be present)
  Row 3+ : blocks, one per paper/source. Each block is:
             - a PAPER row  : DOI, source type, PubMed ID, ID = P1, P2, ...
             - one FINDING row per finding : ID = P1.F1, P1.F2, ...

Official columns:
  B DOI | C Paper/source type | D PubMed ID | E Paper/Finding ID |
  F Finding description | G Finding quote | H Finding summary |
  I Finding relevance (0..1) | J Experimental system | K Data location |
  L Effect size (int%) | M P value (<1.0) | N Sample size (int>0)

Messages are written for challenge participants: they say which cell is wrong
and what the expected format is, without referring to internal templates.
"""

import argparse
import re
import sys
from openpyxl import load_workbook
from openpyxl.utils import get_column_letter, column_index_from_string

# ---- official spec ------------------------------------------------------

ALLOWED_SOURCE_TYPES = {
    "pubmed published", "pubmed preprint", "web article", "database", "other",
}

# canonical field -> list of header aliases (lowercased, stripped) we accept
FIELD_ALIASES = {
    "doi":            ["doi"],
    "source_type":    ["paper/source type", "paper type", "source type", "paper/source", "type"],
    "pmid":           ["pubmed id", "pmid", "pubmed"],
    "id":             ["paper or source id", "paper/finding id", "finding id", "id", "code"],
    "finding_desc":   ["finding description", "findings", "finding"],
    "quote":          ["finding quote", "quote"],
    "summary":        ["finding summary", "summary"],
    "relevance":      ["finding relevance", "relevance"],
    "exp_system":     ["experimental system", "exp system"],
    "data_location":  ["data location", "data source", "data location in the source", "location"],
    "effect_size":    ["effect size"],
    "p_value":        ["p value", "p-value", "pvalue"],
    "sample_size":    ["sample size", "n"],
}

# official column letters (fallback if header mapping fails)
OFFICIAL_COLS = {
    "doi": "B", "source_type": "C", "pmid": "D", "id": "E",
    "finding_desc": "F", "quote": "G", "summary": "H", "relevance": "I",
    "exp_system": "J", "data_location": "K", "effect_size": "L",
    "p_value": "M", "sample_size": "N",
}

# fields required on a FINDING row
FINDING_REQUIRED = ["finding_desc", "relevance", "exp_system", "data_location"]


class Report:
    def __init__(self):
        self.errors = []
        self.warnings = []

    def err(self, row, col, msg):
        self.errors.append((row, col, msg))

    def warn(self, row, col, msg):
        self.warnings.append((row, col, msg))

    def dump(self, strict=False):
        errs = list(self.errors)
        warns = list(self.warnings)
        if strict:
            errs += warns
            warns = []
        print(f"\n{'='*66}")
        if not errs and not warns:
            print("SUBMISSION VALID — no problems found.")
            print('='*66)
            print("\nYour spreadsheet passed all checks. You can submit it.\n")
            return True
        status = "NOT VALID" if errs else "VALID (with suggestions)"
        print(f"SUBMISSION {status} — "
              f"{len(errs)} thing(s) to fix, {len(warns)} suggestion(s)")
        print('='*66)
        if errs:
            print("\nMUST FIX (the submission will be rejected until these are corrected):")
            for row, col, msg in sorted(errs, key=lambda x: (x[0] or 0, x[1] or "")):
                loc = f"row {row}" + (f", column {col}" if col else "")
                print(f"  • [{loc}] {msg}")
        if warns:
            print("\nPLEASE CHECK (allowed, but worth reviewing):")
            for row, col, msg in sorted(warns, key=lambda x: (x[0] or 0, x[1] or "")):
                loc = f"row {row}" + (f", column {col}" if col else "")
                print(f"  • [{loc}] {msg}")
        print()
        return len(errs) == 0


def norm(s):
    return re.sub(r"\s+", " ", str(s).strip().lower()) if s is not None else ""


def map_columns(ws, rpt):
    """Map canonical field -> column letter, by header name, falling back to
    official letters. Returns (colmap, header_ok)."""
    header_row = 1
    headers = {}
    headers_raw = {}
    for c in range(1, ws.max_column + 1):
        v = ws.cell(row=header_row, column=c).value
        if v is not None and str(v).strip():
            letter = get_column_letter(c)
            headers[letter] = norm(v)
            headers_raw[letter] = str(v).strip()

    colmap = {}
    matched_by_header = {}
    used_headers = False
    matched_letters = set()
    for field, aliases in FIELD_ALIASES.items():
        found = None
        for letter, htext in headers.items():
            if htext in aliases:
                found = letter
                break
        if found:
            colmap[field] = found
            matched_by_header[field] = True
            matched_letters.add(found)
            used_headers = True
        else:
            colmap[field] = OFFICIAL_COLS.get(field)
            matched_by_header[field] = False

    # ---- header conformance check (reported up front) ----
    header_ok = True
    OFFICIAL_LABELS = {
        "doi": "DOI", "source_type": "Paper/source type", "pmid": "PubMed ID",
        "id": "Paper/Finding ID", "finding_desc": "Finding description",
        "quote": "Finding quote", "summary": "Finding summary",
        "relevance": "Finding relevance", "exp_system": "Experimental system",
        "data_location": "Data location", "effect_size": "Effect size",
        "p_value": "P value", "sample_size": "Sample size",
    }
    if not used_headers:
        rpt.err(1, None, "The column headers in row 1 were not recognised. "
                         "Please use the official submission template so the "
                         "columns are: " + ", ".join(OFFICIAL_LABELS.values()) + ".")
        header_ok = False
    else:
        missing = [OFFICIAL_LABELS[f] for f in
                   ("doi", "source_type", "pmid", "id", "finding_desc", "quote",
                    "summary", "relevance", "exp_system", "data_location",
                    "effect_size", "p_value", "sample_size")
                   if not matched_by_header.get(f)]
        if missing:
            header_ok = False
            rpt.err(1, None, "These required columns are missing from row 1: "
                    + ", ".join(missing) + ". Please add them (use the official "
                    "submission template) so every column is present with its "
                    "exact heading.")

        # extra columns that aren't part of the official layout -> warn, ignored
        extra = []
        for letter, htext in headers.items():
            if letter == "A":
                continue  # column A carries the hypothesis (row 2), no data header
            if letter not in matched_letters and htext:
                extra.append((letter, htext))
        for letter, htext in sorted(extra):
            shown = headers_raw.get(letter, htext)
            rpt.warn(1, letter, f"Column '{shown}' is not an official column "
                                f"and will be ignored. If this was meant to be "
                                f"one of the required columns, please rename it "
                                f"to the exact official heading.")
    return colmap, header_ok


def cell(ws, row, letter):
    if not letter:
        return None
    return ws.cell(row=row, column=column_index_from_string(letter)).value


def is_paper_id(v):
    return bool(re.fullmatch(r"P\d+", str(v).strip())) if v is not None else False


def is_finding_id(v):
    return bool(re.fullmatch(r"P\d+\.F\d+", str(v).strip())) if v is not None else False


def validate(path, strict=False):
    rpt = Report()
    wb = load_workbook(path, data_only=True)
    ws = wb.active
    cm, header_ok = map_columns(ws, rpt)

    # ---- row 2: hypothesis present ----
    hyp = None
    for letter in ("A", cm.get("doi")):
        pass
    # hypothesis is expected in row 2, column A (or first non-empty cell)
    hyp_val = ws.cell(row=2, column=1).value
    if not hyp_val or not str(hyp_val).strip():
        # try any cell in row 2
        row2 = [ws.cell(row=2, column=c).value for c in range(1, ws.max_column + 1)]
        if not any(v and str(v).strip() for v in row2):
            rpt.err(2, "A", "Hypothesis being considered is missing (row 2).")

    # ---- iterate blocks from row 3 ----
    expected_paper_n = 1
    current_paper = None            # e.g. "P1"
    finding_counters = {}           # paper -> next finding index expected
    seen_dois = {}

    r = 3
    max_r = ws.max_row
    while r <= max_r:
        idv = cell(ws, r, cm["id"])
        doi = cell(ws, r, cm["doi"])
        # skip fully empty rows
        rowvals = [ws.cell(row=r, column=c).value for c in range(1, ws.max_column + 1)]
        if not any(v is not None and str(v).strip() for v in rowvals):
            r += 1
            continue

        if is_paper_id(idv):
            # ---- PAPER row ----
            pid = str(idv).strip()
            num = int(pid[1:])
            if num != expected_paper_n:
                rpt.err(r, cm["id"], f"Paper ID '{pid}' out of sequence; "
                                     f"expected 'P{expected_paper_n}'.")
            expected_paper_n = num + 1
            current_paper = pid
            finding_counters[pid] = 1

            # DOI required
            if not doi or not str(doi).strip():
                rpt.err(r, cm["doi"], f"{pid}: DOI is missing.")
            else:
                d = str(doi).strip().lower()
                if not re.match(r"10\.\d{4,9}/\S+", d):
                    rpt.warn(r, cm["doi"], f"{pid}: DOI '{doi}' does not look "
                                           f"like a standard DOI (10.xxxx/...).")
                if d in seen_dois:
                    rpt.err(r, cm["doi"], f"{pid}: this DOI is already used by "
                                          f"{seen_dois[d]}. Each source must have "
                                          f"a unique DOI.")
                else:
                    seen_dois[d] = pid

            # source type
            st = cell(ws, r, cm["source_type"])
            if not st or not str(st).strip():
                rpt.err(r, cm["source_type"], f"{pid}: Paper/source type missing.")
            elif norm(st) not in ALLOWED_SOURCE_TYPES:
                rpt.err(r, cm["source_type"],
                        f"{pid}: source type '{st}' is not allowed. Use one of: "
                        f"PubMed published, PubMed preprint, Web article, "
                        f"Database, Other.")

            # PMID (required for PubMed sources; recommended generally)
            pmid = cell(ws, r, cm["pmid"])
            st_norm = norm(st)
            if st_norm.startswith("pubmed"):
                if not pmid or not str(pmid).strip():
                    rpt.err(r, cm["pmid"], f"{pid}: PubMed ID required for "
                                           f"PubMed sources.")
                elif not re.fullmatch(r"\d+", str(pmid).strip()):
                    rpt.err(r, cm["pmid"], f"{pid}: PubMed ID '{pmid}' must be "
                                           f"digits only.")

        elif is_finding_id(idv):
            # ---- FINDING row ----
            fid = str(idv).strip()
            fpaper, findex = fid.split(".")
            if current_paper is None:
                rpt.err(r, cm["id"], f"Finding '{fid}' appears before any paper row.")
            elif fpaper != current_paper:
                rpt.err(r, cm["id"], f"Finding '{fid}' does not belong to current "
                                     f"paper '{current_paper}'.")
            else:
                exp = finding_counters.get(current_paper, 1)
                if int(findex[1:]) != exp:
                    rpt.err(r, cm["id"], f"Finding '{fid}' out of sequence; "
                                         f"expected '{current_paper}.F{exp}'.")
                finding_counters[current_paper] = int(findex[1:]) + 1

            # required finding fields (must be non-empty)
            READABLE = {
                "finding_desc": "Finding description",
                "relevance": "Finding relevance",
                "exp_system": "Experimental system",
                "data_location": "Data location",
            }
            for fld in FINDING_REQUIRED:
                v = cell(ws, r, cm[fld])
                if v is None or not str(v).strip():
                    rpt.err(r, cm[fld], f"{fid}: '{READABLE[fld]}' is required "
                            f"but empty. Please fill it in.")

            # quote and summary must contain a value OR the text 'N/A' (never blank)
            for fld, label in (("quote", "Finding quote"),
                               ("summary", "Finding summary")):
                v = cell(ws, r, cm[fld])
                if v is None or not str(v).strip():
                    rpt.err(r, cm[fld], f"{fid}: '{label}' must not be empty. "
                            f"Enter the {label.lower()}, or 'N/A' if there is none.")

            # relevance 0..1
            rel = cell(ws, r, cm["relevance"])
            if rel is not None and str(rel).strip():
                try:
                    rv = float(rel)
                    if not (0.0 <= rv <= 1.0):
                        rpt.err(r, cm["relevance"], f"{fid}: relevance must be a "
                                f"number between 0 and 1 (found {rv}).")
                except ValueError:
                    rpt.err(r, cm["relevance"], f"{fid}: relevance must be a "
                            f"number between 0 and 1 (found '{rel}').")

            # data location precision (warn if whole-figure only)
            loc = cell(ws, r, cm["data_location"])
            if loc and str(loc).strip():
                locs = str(loc).strip()
                if re.fullmatch(r"(fig(ure)?\s*\d+)", locs, flags=re.I):
                    rpt.warn(r, cm["data_location"],
                             f"{fid}: location '{locs}' may be imprecise "
                             f"(panel not specified, e.g. Fig4C).")

            # effect size: whole number followed by %, or N/A (must not be blank)
            es = cell(ws, r, cm.get("effect_size"))
            if es is None or not str(es).strip():
                rpt.err(r, cm.get("effect_size"), f"{fid}: 'Effect size' must "
                        f"not be empty. Enter a whole number + '%' (e.g. 300%), "
                        f"or 'N/A' if not available/applicable.")
            elif norm(es) != "n/a" and not re.fullmatch(r"\d+%", str(es).strip()):
                rpt.err(r, cm.get("effect_size"),
                        f"{fid}: effect size '{es}' must be a whole number "
                        f"followed by '%' (e.g. 300%), or 'N/A'.")

            # p value: number < 1.0, or N/A (must not be blank)
            pv = cell(ws, r, cm.get("p_value"))
            if pv is None or not str(pv).strip():
                rpt.err(r, cm.get("p_value"), f"{fid}: 'P value' must not be "
                        f"empty. Enter a number less than 1.0, or 'N/A' if not "
                        f"available/applicable.")
            elif norm(pv) != "n/a":
                m = re.search(r"[-+]?\d*\.?\d+(e[-+]?\d+)?", str(pv), flags=re.I)
                if not m:
                    rpt.err(r, cm.get("p_value"),
                            f"{fid}: p value '{pv}' must be a number less than "
                            f"1.0, or 'N/A'.")
                else:
                    try:
                        if float(m.group()) >= 1.0:
                            rpt.err(r, cm.get("p_value"),
                                    f"{fid}: p value '{pv}' must be less than 1.0.")
                    except ValueError:
                        rpt.err(r, cm.get("p_value"),
                                f"{fid}: p value '{pv}' must be a number less "
                                f"than 1.0, or 'N/A'.")

            # sample size: whole number > 0, or N/A (must not be blank)
            ss = cell(ws, r, cm.get("sample_size"))
            if ss is None or not str(ss).strip():
                rpt.err(r, cm.get("sample_size"), f"{fid}: 'Sample size' must "
                        f"not be empty. Enter a whole number greater than 0, or "
                        f"'N/A' if not applicable.")
            elif norm(ss) != "n/a":
                if not re.fullmatch(r"\d+", str(ss).strip()) or int(ss) <= 0:
                    rpt.err(r, cm.get("sample_size"),
                            f"{fid}: sample size '{ss}' must be a whole number "
                            f"greater than 0, or 'N/A'.")

        else:
            # a non-empty row whose ID cell isn't a valid P#/P#.F# code
            if idv is not None and str(idv).strip():
                rpt.err(r, cm["id"], f"ID '{idv}' is neither a paper ID (P1) "
                                     f"nor a finding ID (P1.F1).")
            else:
                rpt.warn(r, cm["id"], "Non-empty row with no ID in the ID column.")
        r += 1

    if expected_paper_n == 1:
        rpt.err(None, None, "No paper/source blocks found (expected P1, P2, ...).")

    return rpt.dump(strict=strict)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("file", help="submission .xlsx to validate")
    ap.add_argument("--strict", action="store_true",
                    help="treat warnings as errors")
    args = ap.parse_args()
    ok = validate(args.file, strict=args.strict)
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()