140 lines
5.9 KiB
Python
140 lines
5.9 KiB
Python
"""Validation helpers for externally owned case worksheet rubrics.
|
|
|
|
The clinical team owns scoring criteria. This module only validates the
|
|
machine-readable scaffold that lets those criteria live outside application
|
|
code.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
SCHEMA_VERSION = "vignette.case_worksheet_rubric.v1"
|
|
VALID_STATUSES = {"scaffold_only", "draft", "approved"}
|
|
EXPECTED_CONTENT_OWNER = "clinical_team"
|
|
|
|
|
|
def load_rubric(path: Path) -> dict[str, Any]:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict):
|
|
raise ValueError("case worksheet rubric must be a JSON object")
|
|
return data
|
|
|
|
|
|
def validate_rubric(
|
|
rubric: Mapping[str, Any],
|
|
*,
|
|
expected_item_keys: Mapping[str, set[str]] | None = None,
|
|
) -> dict[str, Any]:
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
|
|
schema_version = str(rubric.get("schema_version") or "")
|
|
status = str(rubric.get("status") or "")
|
|
scoring_enabled = bool(rubric.get("scoring_enabled"))
|
|
sections = rubric.get("sections")
|
|
|
|
if schema_version != SCHEMA_VERSION:
|
|
errors.append("schema_version must be vignette.case_worksheet_rubric.v1")
|
|
if status not in VALID_STATUSES:
|
|
errors.append("status must be one of scaffold_only, draft, approved")
|
|
if str(rubric.get("content_owner") or "") != EXPECTED_CONTENT_OWNER:
|
|
errors.append("content_owner must be clinical_team")
|
|
if scoring_enabled and status != "approved":
|
|
errors.append("scoring_enabled requires status=approved")
|
|
if status == "approved":
|
|
approval = rubric.get("approval")
|
|
if not isinstance(approval, Mapping):
|
|
errors.append("approved rubric requires approval metadata")
|
|
else:
|
|
if not str(approval.get("clinical_reviewer") or ""):
|
|
errors.append("approved rubric requires approval.clinical_reviewer")
|
|
if not str(approval.get("approved_at") or ""):
|
|
errors.append("approved rubric requires approval.approved_at")
|
|
|
|
section_count = 0
|
|
item_count = 0
|
|
section_item_keys: dict[str, set[str]] = {}
|
|
if not isinstance(sections, list) or not sections:
|
|
errors.append("sections must be a non-empty list")
|
|
else:
|
|
seen_sections: set[str] = set()
|
|
for section in sections:
|
|
if not isinstance(section, Mapping):
|
|
errors.append("each section must be an object")
|
|
continue
|
|
section_key = str(section.get("key") or "")
|
|
if not section_key:
|
|
errors.append("section.key is required")
|
|
continue
|
|
if section_key in seen_sections:
|
|
errors.append(f"duplicate section key: {section_key}")
|
|
seen_sections.add(section_key)
|
|
section_count += 1
|
|
items = section.get("items")
|
|
if not isinstance(items, list) or not items:
|
|
errors.append(f"{section_key}: items must be a non-empty list")
|
|
continue
|
|
seen_items: set[str] = set()
|
|
for item in items:
|
|
if not isinstance(item, Mapping):
|
|
errors.append(f"{section_key}: each item must be an object")
|
|
continue
|
|
item_key = str(item.get("key") or "")
|
|
if not item_key:
|
|
errors.append(f"{section_key}: item.key is required")
|
|
continue
|
|
if item_key in seen_items:
|
|
errors.append(f"{section_key}: duplicate item key: {item_key}")
|
|
seen_items.add(item_key)
|
|
item_count += 1
|
|
criteria = item.get("criteria")
|
|
score_scale = item.get("score_scale")
|
|
if scoring_enabled:
|
|
if not isinstance(criteria, list) or not criteria:
|
|
errors.append(f"{section_key}.{item_key}: scoring requires non-empty criteria")
|
|
if not _valid_score_scale(score_scale):
|
|
errors.append(f"{section_key}.{item_key}: scoring requires a valid score_scale")
|
|
elif not criteria:
|
|
warnings.append(f"{section_key}.{item_key}: criteria pending clinical team input")
|
|
section_item_keys[section_key] = seen_items
|
|
|
|
if expected_item_keys is not None:
|
|
expected_sections = set(expected_item_keys)
|
|
actual_sections = set(section_item_keys)
|
|
for missing_section in sorted(expected_sections - actual_sections):
|
|
errors.append(f"missing worksheet section: {missing_section}")
|
|
for extra_section in sorted(actual_sections - expected_sections):
|
|
errors.append(f"unexpected worksheet section: {extra_section}")
|
|
for section_key in sorted(expected_sections & actual_sections):
|
|
missing_items = expected_item_keys[section_key] - section_item_keys[section_key]
|
|
extra_items = section_item_keys[section_key] - expected_item_keys[section_key]
|
|
for item_key in sorted(missing_items):
|
|
errors.append(f"{section_key}: missing worksheet item: {item_key}")
|
|
for item_key in sorted(extra_items):
|
|
errors.append(f"{section_key}: unexpected worksheet item: {item_key}")
|
|
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"rubric_id": str(rubric.get("rubric_id") or ""),
|
|
"status": status,
|
|
"scoring_enabled": scoring_enabled,
|
|
"sections_total": section_count,
|
|
"items_total": item_count,
|
|
"passed": not errors,
|
|
"errors": errors,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def _valid_score_scale(value: object) -> bool:
|
|
if not isinstance(value, Mapping):
|
|
return False
|
|
minimum = value.get("min")
|
|
maximum = value.get("max")
|
|
anchors = value.get("anchors")
|
|
if not isinstance(minimum, int) or not isinstance(maximum, int) or minimum >= maximum:
|
|
return False
|
|
return isinstance(anchors, list) and len(anchors) >= 2
|