diff --git a/docs/findings/rrb-01/tools/README.md b/docs/findings/rrb-01/tools/README.md new file mode 100644 index 0000000..fc43a0d --- /dev/null +++ b/docs/findings/rrb-01/tools/README.md @@ -0,0 +1,59 @@ +# RRB-01 extraction toolkit + +Format tooling for the founder extraction (and later the AI extraction) under the +locked protocol `rrb-01-protocol-v0.3.md` (tag `rrb-01-protocol-lock`). It enforces +the schema's format only. It makes no chemistry judgments and never decides whether +a passage qualifies. + +## Files + +| File | Purpose | +| --- | --- | +| `rrb01_extract.py` | `template`, `validate` and `convert` commands | +| `test_rrb01_extract.py` | Tests, synthetic data only | +| `rrb01_extraction_entry.xlsx` | Blank entry workbook (sheet `records`; sheet `legend` explains fields) | + +Requires Python 3.9+, `openpyxl`; tests need `pytest`. + +## Use + +``` +python rrb01_extract.py validate founder_extraction.xlsx +python rrb01_extract.py convert founder_extraction.xlsx founder_extraction.jsonl +python -m pytest test_rrb01_extract.py +``` + +`convert` writes nothing unless every row passes, never overwrites an existing file, +and writes `.jsonl.report.json` with the SHA-256 of both the workbook and the JSONL +plus record counts. + +## What the converter guarantees + +- Text is copied exactly as typed: no trimming, case change or formula normalization. + Leading or trailing spaces produce a warning and are kept. +- Every row is exported, including `exclude` and `ambiguous`. +- Decisions, parents and reasons are never inferred or changed. +- `parent_normalized` and `target_normalized` are always `null`; normalization is a + separate step after the extraction is preserved. +- Deterministic output: fixed key order, records sorted by source, page, record number; + UTF-8; LF line endings. + +## Implementation notes (format decisions within the locked schema) + +These interpret the locked schema's wording; none changes what qualifies as a pair. + +1. `composition_type` ("for parent and target separately") is entered as two columns and + exported as `{"parent": ..., "target": ...}`. The parent value is `null` when no parent + is named. +2. `direction_basis`, `relationship_type` and `characterization` are required for + `include` records and optional otherwise. +3. An empty parent is accepted only with `exclude` / `no_named_parent`, as the record + trigger requires. A parent entered with `no_named_parent` is a warning, not an error. +4. `confounders` is an open list of snake_case tokens separated by `;`. +5. Cells must be text. A number, date or formula cell is an error, so the spreadsheet + cannot silently convert values. + +## Relationship between the two files + +The workbook is the entry interface; the JSONL is the scientific record. Both are +preserved and committed unchanged, and the report file links them by SHA-256. diff --git a/docs/findings/rrb-01/tools/rrb01_extract.py b/docs/findings/rrb-01/tools/rrb01_extract.py new file mode 100644 index 0000000..299a60b --- /dev/null +++ b/docs/findings/rrb-01/tools/rrb01_extract.py @@ -0,0 +1,403 @@ +#!/usr/bin/env python3 +"""RRB-01 extraction toolkit (format only; makes no chemistry judgments). + +Commands + template write a blank entry workbook + validate [--extractor founder] check every row, write nothing + convert [--extractor founder] + validate, then write JSONL + report + +Guarantees (RRB-01 protocol lock, extraction schema): + * Text is copied exactly as entered: no trimming, case changes or + formula normalization. Whitespace problems are reported, never fixed. + * Every row is exported, including excludes and ambiguous records. + * Decisions, parents and reasons are never inferred or altered. + * parent_normalized / target_normalized are always null here; a separate + normalization step fills them after the extraction is preserved. + * Output is deterministic (fixed key order, record order, UTF-8, LF) and + an existing output file is never overwritten. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from pathlib import Path + +SCHEMA_VERSION = "rrb01-extract-1" +SHEET = "records" + +# Spreadsheet columns, in order. composition_type is entered as two columns +# and exported as {"parent": ..., "target": ...}. +COLUMNS = [ + "record_id", "source", "location", "evidence", "cited_primary_study", + "parent_as_published", "target_as_published", + "parent_composition_type", "target_composition_type", + "direction_basis", "relationship_type", "characterization", + "confounders", "decision", "exclusion_reason", "ambiguity_note", +] + +SOURCES = {"HWA17", "NOE21"} +EXTRACTORS = {"founder", "ai"} +COMPOSITION_TYPES = { + "fixed_integer", "fixed_fractional", "fixed_doped", "unresolved_variable", + "variable_with_stated_value", "family_no_composition", + "spatially_heterogeneous", +} +RELATIONSHIP_TYPES = {"cross_chemistry_analog", "within_family_substitution"} +CHARACTERIZATIONS = { + "formula_or_structural_analog", "proposed_replacement", + "experimentally_evaluated_substitute", +} +DECISIONS = {"include", "exclude", "ambiguous"} +EXCLUSION_REASONS = { + "no_named_parent", "not_directed", "synthesis_route_only", + "morphology_or_surface_only", "spatially_heterogeneous", + "unresolved_variable", "family_no_composition", +} +RECORD_ID = re.compile(r"^(HWA17|NOE21)-([1-9][0-9]*)-([1-9][0-9]*)$") +CONFOUNDER_TOKEN = re.compile(r"^[a-z][a-z0-9_]*$") +MAX_EVIDENCE_WORDS = 40 + +# JSONL key order (locked schema order). +OUTPUT_KEYS = [ + "schema_version", "extractor", "record_id", "source", "location", + "evidence", "cited_primary_study", "parent_as_published", + "target_as_published", "composition_type", "direction_basis", + "relationship_type", "characterization", "confounders", "decision", + "exclusion_reason", "ambiguity_note", "parent_normalized", + "target_normalized", +] + + +class Problem: + def __init__(self, row, field, message, error=True): + self.row, self.field, self.message, self.error = row, field, message, error + + def __str__(self): + kind = "ERROR" if self.error else "WARNING" + return f"{kind} row {self.row} [{self.field}]: {self.message}" + + +def sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(1 << 20), b""): + h.update(chunk) + return h.hexdigest().upper() + + +# ---------------------------------------------------------------- template +def write_template(out: Path) -> None: + from openpyxl import Workbook + from openpyxl.styles import Font, PatternFill, Alignment + from openpyxl.worksheet.datavalidation import DataValidation + + if out.exists(): + sys.exit(f"Refusing to overwrite existing file: {out}") + wb = Workbook() + ws = wb.active + ws.title = SHEET + font = Font(name="Arial", size=10) + head = Font(name="Arial", size=10, bold=True) + fill = PatternFill("solid", start_color="FFF2CC") + ws.append(COLUMNS) + for c in ws[1]: + c.font = head + c.fill = PatternFill("solid", start_color="D9E1F2") + widths = {"evidence": 60, "location": 28, "direction_basis": 40, + "ambiguity_note": 40, "cited_primary_study": 24} + for i, name in enumerate(COLUMNS, start=1): + letter = ws.cell(row=1, column=i).column_letter + ws.column_dimensions[letter].width = widths.get(name, 22) + # Text format for every entry cell so Excel never converts values + # (e.g. "1/3" to a date, "12" to a number). + for r in range(2, 2001): + cell = ws.cell(row=r, column=i) + cell.number_format = "@" + cell.font = font + cell.fill = fill + cell.alignment = Alignment(wrap_text=name in widths, vertical="top") + ws.freeze_panes = "B2" + + def dropdown(col_name, values): + col = ws.cell(row=1, column=COLUMNS.index(col_name) + 1).column_letter + dv = DataValidation(type="list", formula1='"' + ",".join(sorted(values)) + '"', + allow_blank=True, showErrorMessage=True) + ws.add_data_validation(dv) + dv.add(f"{col}2:{col}2000") + + dropdown("source", SOURCES) + dropdown("parent_composition_type", COMPOSITION_TYPES) + dropdown("target_composition_type", COMPOSITION_TYPES) + dropdown("relationship_type", RELATIONSHIP_TYPES) + dropdown("characterization", CHARACTERIZATIONS) + dropdown("decision", DECISIONS) + dropdown("exclusion_reason", EXCLUSION_REASONS) + + leg = wb.create_sheet("legend") + rows = [ + ("RRB-01 founder extraction entry workbook", ""), + ("", ""), + ("Where to type", "Sheet 'records', yellow cells only, one row per record. " + "Every cell is formatted as text; keep it that way."), + ("Converter", "python rrb01_extract.py convert "), + ("Rules", "The locked protocol (rrb-01-protocol-v0.3.md) defines what creates a " + "record and every field. This workbook only enforces format."), + ("record_id", "--, n = record number on that page in reading order. " + "Example: NOE21-4-2"), + ("parent_as_published", "Leave empty only when the source names no parent; such a " + "record must be exclude / no_named_parent."), + ("composition types", "parent_composition_type is empty when the parent is empty."), + ("include records", "need direction_basis, relationship_type and characterization."), + ("exclude records", "need exclusion_reason; ambiguity_note stays empty."), + ("ambiguous records", "need ambiguity_note; exclusion_reason stays empty."), + ("confounders", "snake_case tokens separated by ';' e.g. morphology_change;coating"), + ("cited_primary_study", "Reference number and DOI as printed, or the word none."), + ("evidence", f"Quote exactly as printed, at most {MAX_EVIDENCE_WORDS} words."), + ("", ""), + ("Format example (synthetic, not from either source)", ""), + ("record_id", "NOE21-99-1"), + ("location", "p.99, section 9.9, paragraph 2"), + ("evidence", "AbX2 was replaced by AbY2 to improve stability."), + ("cited_primary_study", "[999] 10.0000/example"), + ("parent / target", "AbX2 / AbY2 (types: fixed_integer / fixed_integer)"), + ("direction_basis", "was replaced by"), + ("relationship / characterization", "within_family_substitution / proposed_replacement"), + ("decision", "include"), + ] + for a, b in rows: + leg.append([a, b]) + for row in leg.iter_rows(): + for c in row: + c.font = font + leg["A1"].font = Font(name="Arial", size=12, bold=True) + leg["A16"].font = head + leg.column_dimensions["A"].width = 34 + leg.column_dimensions["B"].width = 100 + wb.save(out) + print(f"Template written: {out}") + + +# ---------------------------------------------------------------- reading +def read_rows(path: Path): + """Yield (excel_row_number, {column: raw value}) for non-empty rows.""" + from openpyxl import load_workbook + + wb = load_workbook(path, data_only=False, read_only=True) + if SHEET not in wb.sheetnames: + sys.exit(f"Workbook has no sheet named '{SHEET}'") + ws = wb[SHEET] + it = ws.iter_rows(values_only=True) + header = list(next(it, []) or []) + header = [h for h in header if h is not None] + if header != COLUMNS: + sys.exit("Header row does not match the locked column list.\n" + f"Expected: {COLUMNS}\nFound: {header}") + for n, values in enumerate(it, start=2): + values = list(values[:len(COLUMNS)]) + [None] * (len(COLUMNS) - len(values)) + if all(v is None or v == "" for v in values): + continue + yield n, dict(zip(COLUMNS, values)) + + +# ---------------------------------------------------------------- validation +def validate_rows(rows, extractor: str): + problems, records, seen = [], [], {} + + for n, raw in rows: + p = [] + + def err(field, msg): + p.append(Problem(n, field, msg)) + + def warn(field, msg): + p.append(Problem(n, field, msg, error=False)) + + val = {} + for k, v in raw.items(): + if v is None or v == "": + val[k] = None + elif isinstance(v, str): + val[k] = v + if v != v.strip(): + warn(k, "leading or trailing whitespace (kept as entered)") + if v.startswith("="): + err(k, "value starts with '=' (spreadsheet formula); enter text") + else: + err(k, f"cell is {type(v).__name__}, not text; format the cell as " + "text and retype it") + val[k] = None + + def required(field): + if val[field] is None: + err(field, "required") + + def enum(field, allowed): + if val[field] is not None and val[field] not in allowed: + err(field, f"'{val[field]}' is not one of {sorted(allowed)}") + + for f in ("record_id", "source", "location", "evidence", + "cited_primary_study", "target_as_published", + "target_composition_type", "decision"): + required(f) + enum("source", SOURCES) + enum("parent_composition_type", COMPOSITION_TYPES) + enum("target_composition_type", COMPOSITION_TYPES) + enum("relationship_type", RELATIONSHIP_TYPES) + enum("characterization", CHARACTERIZATIONS) + enum("decision", DECISIONS) + enum("exclusion_reason", EXCLUSION_REASONS) + + rid = val["record_id"] + if rid is not None: + m = RECORD_ID.match(rid) + if not m: + err("record_id", f"'{rid}' does not match --") + elif val["source"] is not None and m.group(1) != val["source"]: + err("record_id", f"prefix {m.group(1)} does not match source {val['source']}") + if rid in seen: + err("record_id", f"duplicate of row {seen[rid]}") + else: + seen[rid] = n + + if val["evidence"] is not None: + words = len(val["evidence"].split()) + if words > MAX_EVIDENCE_WORDS: + err("evidence", f"{words} words; limit is {MAX_EVIDENCE_WORDS}") + + parent = val["parent_as_published"] + if parent is None and val["parent_composition_type"] is not None: + err("parent_composition_type", "must be empty when the parent is empty") + if parent is not None and val["parent_composition_type"] is None: + err("parent_composition_type", "required when a parent is entered") + if parent is None and not (val["decision"] == "exclude" + and val["exclusion_reason"] == "no_named_parent"): + err("parent_as_published", "empty parent requires decision exclude " + "with reason no_named_parent") + if parent is not None and val["exclusion_reason"] == "no_named_parent": + warn("exclusion_reason", "no_named_parent but a parent is entered; check") + + d = val["decision"] + if d == "include": + for f in ("direction_basis", "relationship_type", "characterization"): + required(f) + if d == "exclude": + required("exclusion_reason") + elif val["exclusion_reason"] is not None: + err("exclusion_reason", "only allowed when decision is exclude") + if d == "ambiguous": + required("ambiguity_note") + elif val["ambiguity_note"] is not None: + err("ambiguity_note", "only allowed when decision is ambiguous") + + confounders = [] + if val["confounders"] is not None: + confounders = val["confounders"].split(";") + for t in confounders: + if not CONFOUNDER_TOKEN.match(t): + err("confounders", f"'{t}' is not a snake_case token " + "(separate tokens with ';', no spaces)") + + problems.extend(p) + if not any(x.error for x in p): + records.append({ + "schema_version": SCHEMA_VERSION, + "extractor": extractor, + "record_id": rid, + "source": val["source"], + "location": val["location"], + "evidence": val["evidence"], + "cited_primary_study": val["cited_primary_study"], + "parent_as_published": parent, + "target_as_published": val["target_as_published"], + "composition_type": {"parent": val["parent_composition_type"], + "target": val["target_composition_type"]}, + "direction_basis": val["direction_basis"], + "relationship_type": val["relationship_type"], + "characterization": val["characterization"], + "confounders": confounders, + "decision": d, + "exclusion_reason": val["exclusion_reason"], + "ambiguity_note": val["ambiguity_note"], + "parent_normalized": None, + "target_normalized": None, + }) + return problems, records + + +def sort_key(rec): + m = RECORD_ID.match(rec["record_id"]) + return (m.group(1), int(m.group(2)), int(m.group(3))) + + +def to_jsonl(records) -> bytes: + lines = [] + for r in sorted(records, key=sort_key): + ordered = {k: r[k] for k in OUTPUT_KEYS} + lines.append(json.dumps(ordered, ensure_ascii=False, separators=(",", ":"))) + return ("\n".join(lines) + "\n").encode("utf-8") if lines else b"" + + +# ---------------------------------------------------------------- commands +def run_validate(path: Path, extractor: str): + problems, records = validate_rows(read_rows(path), extractor) + for p in problems: + print(p) + errors = sum(p.error for p in problems) + warnings = len(problems) - errors + print(f"\n{len(records)} valid record(s); {errors} error(s); {warnings} warning(s)") + return problems, records, errors + + +def run_convert(src: Path, out: Path, extractor: str): + if out.exists(): + sys.exit(f"Refusing to overwrite existing file: {out}") + _, records, errors = run_validate(src, extractor) + if errors: + sys.exit("Conversion stopped: fix every ERROR first. Nothing was written.") + data = to_jsonl(records) + out.write_bytes(data) + counts = {d: sum(r["decision"] == d for r in records) for d in sorted(DECISIONS)} + report = { + "schema_version": SCHEMA_VERSION, + "extractor": extractor, + "input_xlsx": src.name, + "input_xlsx_sha256": sha256_file(src), + "output_jsonl": out.name, + "output_jsonl_sha256": sha256_file(out), + "record_count": len(records), + "decision_counts": counts, + "source_counts": {s: sum(r["source"] == s for r in records) for s in sorted(SOURCES)}, + } + rep = out.with_suffix(out.suffix + ".report.json") + if rep.exists(): + sys.exit(f"JSONL written, but report already exists and was not overwritten: {rep}") + rep.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", newline="\n") + print(json.dumps(report, indent=2, ensure_ascii=False)) + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + sub = ap.add_subparsers(dest="cmd", required=True) + t = sub.add_parser("template"); t.add_argument("out", type=Path) + v = sub.add_parser("validate"); v.add_argument("src", type=Path) + v.add_argument("--extractor", choices=sorted(EXTRACTORS), default="founder") + c = sub.add_parser("convert"); c.add_argument("src", type=Path); c.add_argument("out", type=Path) + c.add_argument("--extractor", choices=sorted(EXTRACTORS), default="founder") + a = ap.parse_args(argv) + if a.cmd == "template": + write_template(a.out) + elif a.cmd == "validate": + _, _, errors = run_validate(a.src, a.extractor) + sys.exit(1 if errors else 0) + else: + run_convert(a.src, a.out, a.extractor) + + +if __name__ == "__main__": + main() diff --git a/docs/findings/rrb-01/tools/rrb01_extraction_entry.xlsx b/docs/findings/rrb-01/tools/rrb01_extraction_entry.xlsx new file mode 100644 index 0000000..ff9a22d Binary files /dev/null and b/docs/findings/rrb-01/tools/rrb01_extraction_entry.xlsx differ diff --git a/docs/findings/rrb-01/tools/test_rrb01_extract.py b/docs/findings/rrb-01/tools/test_rrb01_extract.py new file mode 100644 index 0000000..1b02978 --- /dev/null +++ b/docs/findings/rrb-01/tools/test_rrb01_extract.py @@ -0,0 +1,196 @@ +"""Tests for rrb01_extract.py. All data is synthetic (not from either source).""" +import json +from pathlib import Path + +import pytest +openpyxl = pytest.importorskip("openpyxl") +load_workbook = openpyxl.load_workbook + +import rrb01_extract as tool + +GOOD_INCLUDE = { + "record_id": "NOE21-3-1", "source": "NOE21", "location": "p.3, s2, para 1", + "evidence": "AbX2 was replaced by AbY2 \u2014 \u201cstable\u201d.", + "cited_primary_study": "[9] 10.0000/x", + "parent_as_published": "AbX0.9Q0.1O2", "target_as_published": "AbY2", + "parent_composition_type": "fixed_fractional", "target_composition_type": "fixed_integer", + "direction_basis": "was replaced by", "relationship_type": "within_family_substitution", + "characterization": "proposed_replacement", "confounders": "morphology_change;coating", + "decision": "include", "exclusion_reason": None, "ambiguity_note": None, +} +GOOD_NO_PARENT = { + "record_id": "HWA17-10-2", "source": "HWA17", "location": "p.10, para 3", + "evidence": "AbZ is an alternative to cobalt cathodes.", "cited_primary_study": "none", + "parent_as_published": None, "target_as_published": "AbZ", + "parent_composition_type": None, "target_composition_type": "fixed_integer", + "direction_basis": None, "relationship_type": None, "characterization": None, + "confounders": None, "decision": "exclude", "exclusion_reason": "no_named_parent", + "ambiguity_note": None, +} +GOOD_AMBIGUOUS = { + "record_id": "HWA17-2-1", "source": "HWA17", "location": "p.2, para 1", + "evidence": "AbQ and AbR were compared.", "cited_primary_study": "[3] 10.0000/y", + "parent_as_published": "AbQ", "target_as_published": "AbR", + "parent_composition_type": "fixed_integer", "target_composition_type": "fixed_integer", + "direction_basis": None, "relationship_type": None, "characterization": None, + "confounders": None, "decision": "ambiguous", "exclusion_reason": None, + "ambiguity_note": "direction unclear from wording", +} + + +_counter = [0] + + +def make_book(tmp_path, rows, name=None): + _counter[0] += 1 + path = tmp_path / (name or f"in{_counter[0]}.xlsx") + tool.write_template(path) + wb = load_workbook(path) + ws = wb[tool.SHEET] + for i, row in enumerate(rows, start=2): + for j, col in enumerate(tool.COLUMNS, start=1): + ws.cell(row=i, column=j).value = row.get(col) + wb.save(path) + return path + + +def check(tmp_path, rows): + path = make_book(tmp_path, rows) + return tool.validate_rows(tool.read_rows(path), "founder") + + +def errors_of(problems): + return [p for p in problems if p.error] + + +def test_valid_rows_pass_and_text_is_unchanged(tmp_path): + problems, recs = check(tmp_path, [GOOD_INCLUDE, GOOD_NO_PARENT, GOOD_AMBIGUOUS]) + assert not errors_of(problems) + by_id = {r["record_id"]: r for r in recs} + r = by_id["NOE21-3-1"] + assert r["evidence"] == GOOD_INCLUDE["evidence"] # unicode kept exactly + assert r["parent_as_published"] == "AbX0.9Q0.1O2" # not normalized + assert r["parent_normalized"] is None and r["target_normalized"] is None + assert r["confounders"] == ["morphology_change", "coating"] + assert r["composition_type"] == {"parent": "fixed_fractional", "target": "fixed_integer"} + assert by_id["HWA17-10-2"]["decision"] == "exclude" # every decision kept + assert by_id["HWA17-2-1"]["decision"] == "ambiguous" + + +def test_whitespace_is_kept_and_only_warned(tmp_path): + row = dict(GOOD_INCLUDE, evidence=" AbX2 was replaced by AbY2. ") + problems, recs = check(tmp_path, [row]) + assert not errors_of(problems) + assert any(not p.error and p.field == "evidence" for p in problems) + assert recs[0]["evidence"] == " AbX2 was replaced by AbY2. " + + +@pytest.mark.parametrize("field,value", [ + ("decision", "maybe"), ("source", "XYZ99"), ("relationship_type", "analog"), + ("exclusion_reason", "other"), ("target_composition_type", "fixed"), +]) +def test_invalid_enum_rejected(tmp_path, field, value): + row = dict(GOOD_NO_PARENT if field == "exclusion_reason" else GOOD_INCLUDE, **{field: value}) + problems, _ = check(tmp_path, [row]) + assert any(p.field == field for p in errors_of(problems)) + + +@pytest.mark.parametrize("field", ["evidence", "location", "record_id", "decision", + "target_as_published", "cited_primary_study"]) +def test_missing_required_rejected(tmp_path, field): + problems, _ = check(tmp_path, [dict(GOOD_INCLUDE, **{field: None})]) + assert errors_of(problems) + + +def test_duplicate_record_id_rejected(tmp_path): + problems, _ = check(tmp_path, [GOOD_INCLUDE, dict(GOOD_INCLUDE)]) + assert any("duplicate" in p.message for p in errors_of(problems)) + + +def test_record_id_prefix_must_match_source(tmp_path): + problems, _ = check(tmp_path, [dict(GOOD_INCLUDE, source="HWA17")]) + assert any(p.field == "record_id" for p in errors_of(problems)) + + +def test_empty_parent_requires_no_named_parent_exclusion(tmp_path): + row = dict(GOOD_INCLUDE, parent_as_published=None, parent_composition_type=None) + problems, _ = check(tmp_path, [row]) + assert any(p.field == "parent_as_published" for p in errors_of(problems)) + + +def test_include_needs_direction_relationship_characterization(tmp_path): + for f in ("direction_basis", "relationship_type", "characterization"): + problems, _ = check(tmp_path, [dict(GOOD_INCLUDE, **{f: None})]) + assert any(p.field == f for p in errors_of(problems)), f + + +def test_reason_and_note_only_with_matching_decision(tmp_path): + p1, _ = check(tmp_path, [dict(GOOD_INCLUDE, exclusion_reason="not_directed")]) + assert any(p.field == "exclusion_reason" for p in errors_of(p1)) + p2, _ = check(tmp_path, [dict(GOOD_INCLUDE, ambiguity_note="x")]) + assert any(p.field == "ambiguity_note" for p in errors_of(p2)) + + +def test_evidence_word_limit(tmp_path): + long = " ".join(["word"] * (tool.MAX_EVIDENCE_WORDS + 1)) + problems, _ = check(tmp_path, [dict(GOOD_INCLUDE, evidence=long)]) + assert any(p.field == "evidence" for p in errors_of(problems)) + + +def test_non_text_cell_rejected(tmp_path): + path = make_book(tmp_path, [GOOD_INCLUDE]) + wb = load_workbook(path) + ws = wb[tool.SHEET] + ws.cell(row=2, column=tool.COLUMNS.index("cited_primary_study") + 1).value = 12 + wb.save(path) + problems, _ = tool.validate_rows(tool.read_rows(path), "founder") + assert any(p.field == "cited_primary_study" for p in errors_of(problems)) + + +def test_formula_text_rejected(tmp_path): + problems, _ = check(tmp_path, [dict(GOOD_INCLUDE, target_as_published="=A1")]) + assert any(p.field == "target_as_published" for p in errors_of(problems)) + + +def test_bad_confounder_token_rejected(tmp_path): + problems, _ = check(tmp_path, [dict(GOOD_INCLUDE, confounders="morphology change")]) + assert any(p.field == "confounders" for p in errors_of(problems)) + + +def test_convert_is_deterministic_and_complete(tmp_path): + rows = [GOOD_INCLUDE, GOOD_NO_PARENT, GOOD_AMBIGUOUS] + src = make_book(tmp_path, rows) + out1, out2 = tmp_path / "a.jsonl", tmp_path / "b.jsonl" + tool.run_convert(src, out1, "founder") + tool.run_convert(src, out2, "founder") + assert out1.read_bytes() == out2.read_bytes() + lines = out1.read_bytes().decode("utf-8").splitlines() + assert len(lines) == 3 # every record exported + ids = [json.loads(x)["record_id"] for x in lines] + assert ids == ["HWA17-2-1", "HWA17-10-2", "NOE21-3-1"] # numeric page order + assert list(json.loads(lines[0]).keys()) == tool.OUTPUT_KEYS + assert b"\r\n" not in out1.read_bytes() + rep = json.loads((tmp_path / "a.jsonl.report.json").read_text(encoding="utf-8")) + assert rep["output_jsonl_sha256"] == tool.sha256_file(out1) + assert rep["record_count"] == 3 + + +def test_convert_refuses_errors_and_overwrite(tmp_path): + src = make_book(tmp_path, [dict(GOOD_INCLUDE, decision="maybe")]) + out = tmp_path / "o.jsonl" + with pytest.raises(SystemExit): + tool.run_convert(src, out, "founder") + assert not out.exists() # nothing written + good = make_book(tmp_path, [GOOD_INCLUDE], name="good.xlsx") + tool.run_convert(good, out, "founder") + with pytest.raises(SystemExit): + tool.run_convert(good, out, "founder") # never overwrites + + +def test_header_must_match(tmp_path): + path = make_book(tmp_path, [GOOD_INCLUDE]) + wb = load_workbook(path) + wb[tool.SHEET].cell(row=1, column=1).value = "id" + wb.save(path) + with pytest.raises(SystemExit): + list(tool.read_rows(path))