Clinical Reports

scripts/format_adverse_events.py

#!/usr/bin/env python3
"""Format bounded aggregate adverse-event counts into a review-only Markdown table."""

from __future__ import annotations

import argparse
import csv
import re
import sys
from dataclasses import dataclass
from pathlib import Path

sys.dont_write_bytecode = True

from _common import (  # noqa: E402
    MAX_CSV_BYTES,
    MAX_CSV_ROWS,
    ValidationError,
    load_json_object,
    local_input_path,
    local_output_path,
    require_bool,
    require_data_class,
    require_exact_keys,
    require_string,
)

TOOL = "format_adverse_events"
REQUIRED_COLUMNS = (
    "analysis_set",
    "treatment_group",
    "meddra_version",
    "system_organ_class",
    "preferred_term",
    "subjects_affected",
    "event_count",
    "denominator",
)
FORBIDDEN_HEADER_FRAGMENTS = {
    "patient",
    "subject_id",
    "participant",
    "case_id",
    "initial",
    "name",
    "email",
    "phone",
    "address",
    "narrative",
    "verbatim",
    "onset",
    "date",
    "mrn",
}
VERSION_RE = re.compile(r"^\d{1,2}\.\d$")
METADATA_FIELDS = {
    "schema_version",
    "artifact_kind",
    "draft_status",
    "safety_notice",
    "data_classification",
    "authorized_purpose",
    "authorization_verified",
    "local_only_handling_confirmed",
    "analysis_metadata",
    "input_csv",
    "required_columns",
    "prohibited_content",
    "provenance_manifest",
    "review",
}
ANALYSIS_METADATA_FIELDS = {
    "protocol_reference",
    "sap_reference",
    "data_cut_reference",
    "analysis_set",
    "counting_rule",
    "threshold_rule",
    "meddra_version",
    "meddra_language",
    "coding_source_reference",
}


@dataclass(frozen=True)
class AggregateRow:
    analysis_set: str
    treatment_group: str
    meddra_version: str
    system_organ_class: str
    preferred_term: str
    subjects_affected: int
    event_count: int
    denominator: int


def validate_aggregate_metadata(
    data: dict[str, object],
    *,
    input_name: str,
) -> dict[str, object]:
    """Require an authorized aggregate-only sidecar manifest."""
    require_exact_keys(data, METADATA_FIELDS, "metadata")
    if data.get("schema_version") != "2.0":
        raise ValidationError("metadata schema_version must be 2.0")
    if data.get("artifact_kind") != "clinical_trial_safety_aggregate_draft":
        raise ValidationError(
            "metadata artifact_kind must be clinical_trial_safety_aggregate_draft"
        )
    if (
        data.get("draft_status")
        != "BLOCKED_INCOMPLETE_NOT_AN_INDIVIDUAL_CASE_SAFETY_REPORT"
    ):
        raise ValidationError("metadata must preserve the blocked non-ICSR status")
    require_string(data.get("safety_notice"), "metadata.safety_notice", max_length=1000)
    if require_data_class(data.get("data_classification")) != "aggregate":
        raise ValidationError("AE formatter accepts aggregate data only")
    require_string(data.get("authorized_purpose"), "metadata.authorized_purpose")
    if not require_bool(
        data.get("authorization_verified"),
        "metadata.authorization_verified",
    ):
        raise ValidationError("metadata.authorization_verified must be true")
    if not require_bool(
        data.get("local_only_handling_confirmed"),
        "metadata.local_only_handling_confirmed",
    ):
        raise ValidationError("metadata.local_only_handling_confirmed must be true")
    require_string(data.get("provenance_manifest"), "metadata.provenance_manifest")
    if data.get("input_csv") != input_name:
        raise ValidationError("metadata.input_csv must equal the aggregate CSV filename")
    if data.get("required_columns") != list(REQUIRED_COLUMNS):
        raise ValidationError("metadata.required_columns does not match the formatter schema")
    prohibited = data.get("prohibited_content")
    if not isinstance(prohibited, list) or not prohibited:
        raise ValidationError("metadata.prohibited_content must be non-empty")

    analysis = require_exact_keys(
        data.get("analysis_metadata"),
        ANALYSIS_METADATA_FIELDS,
        "metadata.analysis_metadata",
    )
    for field in ANALYSIS_METADATA_FIELDS:
        require_string(
            analysis.get(field),
            f"metadata.analysis_metadata.{field}",
            max_length=500,
        )
    if not VERSION_RE.fullmatch(str(analysis["meddra_version"])):
        raise ValidationError("metadata MedDRA version must look like 29.0")

    review = require_exact_keys(
        data.get("review"),
        {
            "safety_coding_review",
            "statistical_review",
            "privacy_review",
            "regulatory_review",
            "submission_authorized",
        },
        "metadata.review",
    )
    if review.get("submission_authorized") is not False:
        raise ValidationError("metadata.review.submission_authorized must remain false")
    return data


def load_aggregate_metadata(
    raw_path: str,
    *,
    input_name: str,
) -> dict[str, object]:
    """Load a bounded aggregate metadata sidecar."""
    _, data = load_json_object(raw_path)
    return validate_aggregate_metadata(data, input_name=input_name)


def _label(value: str | None, field: str) -> str:
    if value is None:
        raise ValidationError(f"{field} is required")
    text = value.strip()
    if (
        not 1 <= len(text) <= 200
        or not text[0].isalnum()
        or any(char in "|`<>\r\n" or ord(char) < 32 for char in text)
    ):
        raise ValidationError(
            f"{field} must be a bounded label without markup or line breaks"
        )
    return text


def _count(value: str | None, field: str) -> int:
    if value is None or not re.fullmatch(r"\d{1,9}", value.strip()):
        raise ValidationError(f"{field} must be a non-negative integer")
    return int(value)


def load_aggregate_csv(raw_path: str) -> list[AggregateRow]:
    """Load and validate aggregate-only AE rows."""
    path = local_input_path(
        raw_path,
        suffixes={".csv"},
        max_bytes=MAX_CSV_BYTES,
    )
    try:
        path.read_text(encoding="utf-8-sig")
    except UnicodeDecodeError as exc:
        raise ValidationError("CSV input must be UTF-8") from exc
    handle = path.open("r", encoding="utf-8-sig", newline="")

    rows: list[AggregateRow] = []
    with handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames is None:
            raise ValidationError("CSV header is required")
        normalized_headers = [header.strip() for header in reader.fieldnames]
        lowered = {header.lower() for header in normalized_headers}
        for header in lowered:
            if any(fragment in header for fragment in FORBIDDEN_HEADER_FRAGMENTS):
                raise ValidationError(f"row-level or sensitive column is prohibited: {header}")
        if set(normalized_headers) != set(REQUIRED_COLUMNS):
            raise ValidationError(
                f"CSV columns must be exactly: {list(REQUIRED_COLUMNS)}"
            )

        for line_number, raw in enumerate(reader, start=2):
            if len(rows) >= MAX_CSV_ROWS:
                raise ValidationError(f"CSV exceeds {MAX_CSV_ROWS} data rows")
            if None in raw:
                raise ValidationError(f"row {line_number} contains extra columns")
            row = AggregateRow(
                analysis_set=_label(raw.get("analysis_set"), f"row {line_number} analysis_set"),
                treatment_group=_label(
                    raw.get("treatment_group"),
                    f"row {line_number} treatment_group",
                ),
                meddra_version=_label(
                    raw.get("meddra_version"),
                    f"row {line_number} meddra_version",
                ),
                system_organ_class=_label(
                    raw.get("system_organ_class"),
                    f"row {line_number} system_organ_class",
                ),
                preferred_term=_label(
                    raw.get("preferred_term"),
                    f"row {line_number} preferred_term",
                ),
                subjects_affected=_count(
                    raw.get("subjects_affected"),
                    f"row {line_number} subjects_affected",
                ),
                event_count=_count(
                    raw.get("event_count"),
                    f"row {line_number} event_count",
                ),
                denominator=_count(
                    raw.get("denominator"),
                    f"row {line_number} denominator",
                ),
            )
            if not VERSION_RE.fullmatch(row.meddra_version):
                raise ValidationError(
                    f"row {line_number} MedDRA version must look like 29.0"
                )
            if row.denominator == 0:
                raise ValidationError(f"row {line_number} denominator must be greater than zero")
            if row.subjects_affected > row.denominator:
                raise ValidationError(
                    f"row {line_number} subjects_affected exceeds denominator"
                )
            if row.event_count < row.subjects_affected:
                raise ValidationError(
                    f"row {line_number} event_count is less than subjects_affected"
                )
            rows.append(row)

    if not rows:
        raise ValidationError("CSV must contain at least one aggregate data row")

    group_metadata: dict[tuple[str, str], tuple[int, str]] = {}
    row_keys: set[tuple[str, str, str, str]] = set()
    for row in rows:
        group = (row.analysis_set, row.treatment_group)
        metadata = (row.denominator, row.meddra_version)
        if group in group_metadata and group_metadata[group] != metadata:
            raise ValidationError(
                f"inconsistent denominator or MedDRA version for group {group}"
            )
        group_metadata[group] = metadata
        row_key = (
            row.analysis_set,
            row.treatment_group,
            row.system_organ_class,
            row.preferred_term,
        )
        if row_key in row_keys:
            raise ValidationError(f"duplicate aggregate row: {row_key}")
        row_keys.add(row_key)
    return rows


def render_markdown(
    rows: list[AggregateRow],
    *,
    metadata: dict[str, object],
    expected_meddra_version: str | None = None,
) -> str:
    """Render rows without inferring missing cells or statistical comparisons."""
    versions = sorted({row.meddra_version for row in rows})
    if len(versions) != 1:
        raise ValidationError("one table may contain exactly one MedDRA version")
    analysis_metadata = metadata["analysis_metadata"]
    if not isinstance(analysis_metadata, dict):
        raise ValidationError("metadata.analysis_metadata must be an object")
    metadata_version = str(analysis_metadata["meddra_version"])
    if versions != [metadata_version]:
        raise ValidationError(
            "CSV MedDRA version does not match metadata.analysis_metadata.meddra_version"
        )
    if expected_meddra_version is not None:
        if not VERSION_RE.fullmatch(expected_meddra_version):
            raise ValidationError("--expected-meddra-version must look like 29.0")
        if versions != [expected_meddra_version]:
            raise ValidationError(
                f"input MedDRA versions {versions} do not match expected "
                f"{expected_meddra_version}"
            )

    analysis_sets = sorted({row.analysis_set for row in rows})
    if len(analysis_sets) != 1:
        raise ValidationError(
            "one output table may contain exactly one analysis_set"
        )
    if analysis_sets[0] != analysis_metadata["analysis_set"]:
        raise ValidationError(
            "CSV analysis_set does not match metadata.analysis_metadata.analysis_set"
        )
    groups = sorted({row.treatment_group for row in rows})
    denominators = {
        group: next(row.denominator for row in rows if row.treatment_group == group)
        for group in groups
    }
    lookup = {
        (row.system_organ_class, row.preferred_term, row.treatment_group): row
        for row in rows
    }
    terms = sorted({(row.system_organ_class, row.preferred_term) for row in rows})

    headers = ["System organ class", "Preferred term"]
    for group in groups:
        headers.extend(
            [
                f"{group}: subjects n/N (%)",
                f"{group}: events",
            ]
        )
    lines = [
        "# Draft aggregate adverse-event table",
        "",
        "> DRAFT — aggregate display only. Not an ICSR, reportability decision, "
        "clinical interpretation, filing, or submission. Qualified safety and "
        "statistical review is required.",
        "",
        f"- Analysis set: {analysis_sets[0]}",
        f"- MedDRA version(s), as supplied and not terminology-validated: {', '.join(versions)}",
        f"- MedDRA language, as supplied: {analysis_metadata['meddra_language']}",
        f"- Counting rule, as supplied: {analysis_metadata['counting_rule']}",
        f"- Threshold rule, as supplied: {analysis_metadata['threshold_rule']}",
        "- Missing group/term cells are shown as an em dash and are not assumed to be zero.",
        "- Subjects and events are distinct; event counts may exceed subject counts.",
        "",
        "| " + " | ".join(headers) + " |",
        "| " + " | ".join(["---"] * len(headers)) + " |",
    ]
    for soc, term in terms:
        cells = [soc, term]
        for group in groups:
            row = lookup.get((soc, term, group))
            if row is None:
                cells.extend(["—", "—"])
                continue
            percent = row.subjects_affected / row.denominator * 100
            cells.extend(
                [
                    f"{row.subjects_affected}/{row.denominator} ({percent:.1f}%)",
                    str(row.event_count),
                ]
            )
        lines.append("| " + " | ".join(cells) + " |")
    lines.extend(
        [
            "",
            "Denominators by treatment group: "
            + "; ".join(f"{group} N={denominators[group]}" for group in groups)
            + ".",
            "",
            "No MedDRA semantic validation, deduplication, inference, causal assessment, "
            "or between-group testing was performed.",
        ]
    )
    return "\n".join(lines) + "\n"


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "Format aggregate-only AE CSV rows as review-only Markdown. "
            "Rejects row-level identifiers and performs no safety assessment."
        )
    )
    parser.add_argument("input_file", help="Local aggregate CSV")
    parser.add_argument(
        "--metadata",
        required=True,
        help="Authorized aggregate safety metadata sidecar (.json)",
    )
    parser.add_argument("-o", "--output", help="Optional local Markdown output")
    parser.add_argument("--expected-meddra-version")
    parser.add_argument("--overwrite", action="store_true")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        rows = load_aggregate_csv(args.input_file)
        input_name = Path(args.input_file).name
        metadata = load_aggregate_metadata(
            args.metadata,
            input_name=input_name,
        )
        rendered = render_markdown(
            rows,
            metadata=metadata,
            expected_meddra_version=args.expected_meddra_version,
        )
        if args.output:
            output = local_output_path(
                args.output,
                suffixes={".md"},
                overwrite=args.overwrite,
            )
            output.write_text(rendered, encoding="utf-8")
        else:
            print(rendered, end="")
        return 0
    except (OSError, ValidationError, csv.Error) as exc:
        print(f"{TOOL}: BLOCKED_INVALID_INPUT: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())