extraction/numeric.py

"""Deterministic numeric fact extraction with exact source spans.

This module never validates or corrects a value. Every extracted number starts
with status=pending and must pass the dedicated numeric validator later.
"""
from __future__ import annotations

import re
from dataclasses import dataclass


@dataclass(frozen=True)
class NumericMatch:
    kind: str
    raw_text: str
    value: float | str | None
    unit: str | None
    start: int
    end: int


_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
    ("confidence_interval", re.compile(r"(?:95\s*%?\s*(?:CI|IC))\s*[:=\[(]?\s*(-?\d+(?:\.\d+)?)\s*(?:[-–,]|to)\s*(-?\d+(?:\.\d+)?)", re.I)),
    ("p_value", re.compile(r"\bp\s*([<>=]=?)\s*(0?\.\d+|\d+(?:\.\d+)?(?:e-?\d+)?)", re.I)),
    ("effect_estimate", re.compile(r"\b(OR|RR|HR|SMD|MD)\s*[=:]?\s*(-?\d+(?:\.\d+)?)", re.I)),
    ("dose", re.compile(r"\b(\d+(?:\.\d+)?)\s*(mg|g|µg|ug|mcg|IU|UI|mL|ml|L|mmol|µmol|umol|nmol)(?:\s*/\s*(?:kg|day|d|jour))?\b", re.I)),
    ("percentage", re.compile(r"\b(\d+(?:\.\d+)?)\s*%(?!\s*(?:CI|IC))", re.I)),
    ("sample_size", re.compile(r"\bn\s*=\s*(\d[\d ,]*)", re.I)),
    ("sample_size", re.compile(r"\b(\d[\d ,]*)\s+(?:(?:[A-Za-zÀ-ÿ-]+)\s+){0,2}(?:participants?|patients?|subjects?|students?|volunteers?|adults?|children|individuals?|women|men)\b", re.I)),
    ("duration", re.compile(r"\b(\d+(?:\.\d+)?)\s*(hours?|days?|weeks?|months?|years?)\b", re.I)),
    ("age", re.compile(r"\b(?:aged?|age)\s*(\d+(?:\.\d+)?)\s*(?:years?|yrs?)\b", re.I)),
    ("frequency", re.compile(r"\b(once|twice|three times|four times)\s+(?:a|per)\s+(day|week)\b", re.I)),
]


def _number(value: str) -> float | str:
    cleaned = value.replace(",", "").replace(" ", "")
    try:
        return float(cleaned)
    except ValueError:
        return value


def extract_numeric_matches(text: str) -> list[NumericMatch]:
    matches: list[NumericMatch] = []
    occupied: list[tuple[int, int]] = []

    for kind, pattern in _PATTERNS:
        for match in pattern.finditer(text):
            span = match.span()
            if any(span[0] < end and span[1] > start for start, end in occupied):
                continue

            groups = match.groups()
            raw = match.group(0)
            value: float | str | None = None
            unit: str | None = None

            if kind == "confidence_interval":
                value = f"{groups[0]} to {groups[1]}"
                unit = "95% CI"
            elif kind == "p_value":
                value = _number(groups[1])
                unit = f"p{groups[0]}"
            elif kind == "effect_estimate":
                value = _number(groups[1])
                unit = groups[0].upper()
            elif kind == "dose":
                value = _number(groups[0])
                unit = groups[1]
            elif kind in {"percentage", "sample_size", "duration", "age"}:
                value = _number(groups[0])
                unit = groups[1] if len(groups) > 1 and groups[1] else {
                    "percentage": "%",
                    "sample_size": "participants",
                    "age": "years",
                }.get(kind)
            elif kind == "frequency":
                value = groups[0]
                unit = groups[1]

            matches.append(NumericMatch(kind, raw, value, unit, span[0], span[1]))
            occupied.append(span)

    return sorted(matches, key=lambda item: (item.start, item.end, item.kind))


def to_numeric_facts(text: str) -> list[dict]:
    return [
        {
            "kind": item.kind,
            "raw_text": item.raw_text,
            "value": item.value,
            "unit": item.unit,
            "context": text,
            "status": "pending",
            "correction": None,
        }
        for item in extract_numeric_matches(text)
    ]