numeric_core.py

"""Pure deterministic claim/evidence number matching."""
import re
from dataclasses import dataclass

@dataclass(frozen=True)
class Num:
    kind: str
    op: str
    lo: float
    hi: float | None
    unit: str | None

_UNIT_NORM = {
    "mg": 1.0, "g": 1000.0, "ug": 1e-3, "mcg": 1e-3,
    "ml": 1.0, "l": 1000.0, "iu": 1.0, "ui": 1.0,
    "mmol": 1.0, "nmol": 1e-3, "umol": 1e-6,
}

_DAY_NORM = {
    "day": 1, "days": 1, "jour": 1, "jours": 1,
    "week": 7, "weeks": 7, "semaine": 7, "semaines": 7,
    "month": 30, "months": 30, "mois": 30,
}

_WORD_NUMBER_NORM = {
    "one": 1, "two": 2, "three": 3, "four": 4,
    "five": 5, "six": 6, "seven": 7, "eight": 8,
    "nine": 9, "ten": 10, "eleven": 11, "twelve": 12,
    "un": 1, "une": 1, "deux": 2, "trois": 3,
    "quatre": 4, "cinq": 5, "six": 6, "sept": 7,
    "huit": 8, "neuf": 9, "dix": 10, "onze": 11,
    "douze": 12,
}

_NUMBER = (
    r"-?(?:\d+(?:[.,]\d+)?|[.,]\d+)"
    r"(?:[eE][+-]?\d+)?"
)

_STUDY_COUNT_SAFE_UNITS = {
    "one": 1,
    "two": 2,
    "three": 3,
    "four": 4,
    "five": 5,
    "six": 6,
    "seven": 7,
    "eight": 8,
    "nine": 9,
}

_STUDY_COUNT_SAFE_DIRECT = {
    **_STUDY_COUNT_SAFE_UNITS,
    "ten": 10,
    "eleven": 11,
    "twelve": 12,
    "thirteen": 13,
    "fourteen": 14,
    "fifteen": 15,
    "sixteen": 16,
    "seventeen": 17,
    "eighteen": 18,
    "nineteen": 19,
    "un": 1,
    "une": 1,
    "deux": 2,
    "trois": 3,
    "quatre": 4,
    "cinq": 5,
    "six": 6,
    "sept": 7,
    "huit": 8,
    "neuf": 9,
    "dix": 10,
    "onze": 11,
    "douze": 12,
}

_STUDY_COUNT_SAFE_TENS = {
    "twenty": 20,
    "thirty": 30,
    "forty": 40,
    "fifty": 50,
    "sixty": 60,
    "seventy": 70,
    "eighty": 80,
    "ninety": 90,
}

_STUDY_COUNT_SAFE_WORD_TOKEN = (
    r"(?:"
    r"one|two|three|four|five|six|seven|eight|nine|"
    r"ten|eleven|twelve|thirteen|fourteen|fifteen|"
    r"sixteen|seventeen|eighteen|nineteen|"
    r"(?:twenty|thirty|forty|fifty|sixty|seventy|"
    r"eighty|ninety)"
    r"(?:[\s-]+(?:one|two|three|four|five|six|"
    r"seven|eight|nine))?|"
    r"un|une|deux|trois|quatre|cinq|six|sept|"
    r"huit|neuf|dix|onze|douze"
    r")"
)

# Liste fermée : les mots arbitraires comme "hundred",
# "analyses", "comparisons", "across", "of" ou "year"
# ne peuvent pas être consommés comme qualificatifs.
_STUDY_COUNT_SAFE_QUALIFIER = (
    r"(?:"
    r"animal|human|clinical|preclinical|experimental|"
    r"randomized|randomised|controlled|observational|"
    r"prospective|retrospective|eligible|included|"
    r"published|previous|independent|comparative|"
    r"animales?|animaux|humaines?|cliniques?|"
    r"precliniques?|précliniques?|experimentales?|"
    r"expérimentales?|randomisees?|randomisées?|"
    r"controlees?|contrôlées?|observationnelles?|"
    r"prospectives?|retrospectives?|rétrospectives?|"
    r"eligibles?|éligibles?|incluses?|publiees?|"
    r"publiées?|anterieures?|antérieures?|"
    r"independantes?|indépendantes?|comparatives?"
    r")"
)


def _parse_safe_study_count_word(
    value: str,
) -> float:
    normalized = re.sub(
        r"[\s-]+",
        " ",
        str(value or "").strip().casefold(),
    )

    if normalized in _STUDY_COUNT_SAFE_DIRECT:
        return float(
            _STUDY_COUNT_SAFE_DIRECT[normalized]
        )

    parts = normalized.split()

    if (
        len(parts) == 1
        and parts[0] in _STUDY_COUNT_SAFE_TENS
    ):
        return float(
            _STUDY_COUNT_SAFE_TENS[parts[0]]
        )

    if (
        len(parts) == 2
        and parts[0] in _STUDY_COUNT_SAFE_TENS
        and parts[1] in _STUDY_COUNT_SAFE_UNITS
    ):
        return float(
            _STUDY_COUNT_SAFE_TENS[parts[0]]
            + _STUDY_COUNT_SAFE_UNITS[parts[1]]
        )

    raise ValueError(
        f"Unsupported safe study count: {value!r}"
    )


_PATTS = [
    (
        "p",
        re.compile(
            rf"\bp\s*([<>=]=?)\s*({_NUMBER})",
            re.I,
        ),
    ),
    (
        "ci",
        re.compile(
            rf"(?:(?:95\s*%?\s*(?:CI|IC))|"
            rf"(?:(?:CI|IC)\s*95\s*%?))"
            rf"\s*[,;:=]?\s*[\[\(]?\s*"
            rf"({_NUMBER})\s*%?"
            rf"\s*(?:[-\u2013]|\bto\b|\bà\b|,\s+)\s*"
            rf"({_NUMBER})\s*%?",
            re.I,
        ),
    ),
    (
        "dose",
        re.compile(
            rf"({_NUMBER})\s*"
            r"(mg|g|ug|mcg|IU|UI|ml|l|mmol|nmol|umol)\b",
            re.I,
        ),
    ),
    (
        "percent",
        re.compile(
            rf"({_NUMBER})\s*%(?!\s*(?:CI|IC))",
            re.I,
        ),
    ),
    (
        "n",
        re.compile(
            r"\bn\s*=\s*(\d[\d ,]*)",
            re.I,
        ),
    ),
    (
        "sample",
        re.compile(
            r"\b(\d[\d ,]*)\s+"
            r"(?:(?:[A-Za-zÀ-ÿ-]+)\s+){0,2}"
            r"(?:participants?|patients?|subjects?|students?|"
            r"volunteers?|adults?|children|individuals?|women|men|"
            r"étudiants?|etudiants?|sujets?|volontaires?|adultes?|"
            r"enfants?|femmes?|hommes?)\b",
            re.I,
        ),
    ),
    (
        "publication_count_word_safe",
        re.compile(
            rf"(?<![A-Za-zÀ-ÿ-])"
            rf"({_STUDY_COUNT_SAFE_WORD_TOKEN})"
            rf"(?:\s+{_STUDY_COUNT_SAFE_QUALIFIER}){{0,2}}"
            r"\s+"
            r"(?:papers?|articles?|publications?)\b",
            re.I,
        ),
    ),
    (
        "publication_count",
        re.compile(
            r"\b(\d+)\s+"
            r"(?:papers?|articles?|publications?)\b",
            re.I,
        ),
    ),
    (
        "study_count_word_safe",
        re.compile(
            rf"(?<![A-Za-zÀ-ÿ-])"
            rf"({_STUDY_COUNT_SAFE_WORD_TOKEN})"
            rf"(?:\s+{_STUDY_COUNT_SAFE_QUALIFIER}){{0,2}}"
            r"\s+"
            r"(?:studies|trials|experiments|études|etudes|"
            r"essais|expériences|experiences)\b",
            re.I,
        ),
    ),
    (
        "study_count",
        re.compile(
            r"\b(\d+)\s+"
            r"(?:studies|trials|experiments|études|etudes|"
            r"essais|expériences|experiences)\b",
            re.I,
        ),
    ),
    (
        "duration_word",
        re.compile(
            r"\b(one|two|three|four|five|six|seven|eight|nine|"
            r"ten|eleven|twelve|un|une|deux|trois|quatre|cinq|"
            r"six|sept|huit|neuf|dix|onze|douze)"
            r"(?:\s+|-)\s*"
            r"(weeks?|days?|months?|semaines?|jours?|mois)\b",
            re.I,
        ),
    ),
    (
        "duration",
        re.compile(
            rf"({_NUMBER})\s*"
            r"(weeks?|days?|months?|semaines?|jours?|mois)\b",
            re.I,
        ),
    ),
    (
        "effect_size",
        re.compile(
            rf"\b(?:SMD|MD|"
            rf"standardized\s+mean\s+difference|"
            rf"standardised\s+mean\s+difference|"
            rf"mean\s+difference)\b"
            rf"\s*(?:[,;:=]|\bis\b|\bwas\b|\bof\b)?"
            rf"\s*({_NUMBER})",
            re.I,
        ),
    ),
    (
        "ratio",
        re.compile(
            rf"\b(?:OR|RR|HR|"
            rf"hazard\s+ratio|odds\s+ratio|"
            rf"risk\s+ratio|relative\s+risk)\b"
            rf"\s*(?:[,;:=]|\bis\b|\bwas\b|\bof\b)?"
            rf"\s*({_NUMBER})",
            re.I,
        ),
    ),
]

_GENERIC_NUMBER = re.compile(
    rf"(?<![\w.,])({_NUMBER})(?![\w]|[.,]\d)",
    re.I,
)

_EFFECT_SIZE_CUE = re.compile(
    r"\b(?:"
    r"effect\s+sizes?|"
    r"effect\s+estimates?|"
    r"standardized\s+mean\s+differences?|"
    r"standardised\s+mean\s+differences?|"
    r"hedges(?:['’]s)?\s+g|"
    r"mean\s+g|"
    r"SMD|MD|OR|RR|HR|"
    r"tailles?\s+d['’]effet|"
    r"estimations?\s+de\s+l['’]effet"
    r")\b",
    re.I,
)

_PARENTHETICAL_CI_AFTER_ESTIMATE = re.compile(
    rf"(?P<estimate>{_NUMBER})\s*"
    rf"(?P<interval>[\[(]\s*"
    rf"(?P<lo>{_NUMBER})\s*,\s*"
    rf"(?P<hi>{_NUMBER})\s*[\])])",
    re.I,
)

_SENTENCE_BOUNDARY = re.compile(
    r"(?<!\d)[.!?](?!\d)|\n"
)


def _to_float(value: str) -> float:
    return float(
        str(value).strip().replace(",", ".")
    )


def _overlaps(
    span: tuple[int, int],
    occupied: list[tuple[int, int]],
) -> bool:
    start, end = span

    return any(
        start < previous_end
        and previous_start < end
        for previous_start, previous_end
        in occupied
    )




_STUDY_COUNT_ELLIPSIS_RE = re.compile(
    r"(?P<prefix>\bfrom\s+(?:the\s+)?"
    r"(?:\d+|[A-Za-z]+(?:[- ][A-Za-z]+)?)\s+"
    r"(?:included\s+)?"
    r"(?:papers?|stud(?:y|ies)|trials?|experiments?)"
    r"\s*,\s*)"
    r"(?P<count>\d+)"
    r"(?P<suffix>\s+with\s+\d+\s+comparisons?\s+"
    r"were\s+meta[- ]analy[sz]ed\b)",
    re.I,
)


def _expand_elliptical_study_counts(
    text: str,
) -> str:
    """Expand only strongly anchored study-count ellipses."""
    return _STUDY_COUNT_ELLIPSIS_RE.sub(
        lambda match: (
            f"{match.group('prefix')}"
            f"{match.group('count')} studies"
            f"{match.group('suffix')}"
        ),
        str(text or ""),
    )


def extract(text: str) -> list[Num]:
    text = _expand_elliptical_study_counts(str(text or ""))

    source = str(text or "")
    found = []
    occupied = []

    # Certains abstracts rapportent une taille d'effet suivie
    # directement de ses bornes, sans écrire explicitement
    # "95% CI", par exemple: 1.19 (0.74, 1.64).
    #
    # La paire parenthétique n'est classée comme intervalle
    # que si la même phrase contient un marqueur explicite
    # de taille/estimation d'effet et si l'estimation est
    # comprise entre les deux bornes. Le span occupé exclut
    # l'estimation, afin qu'elle reste extraite comme scalaire
    # ou ratio par les règles existantes.
    for match in _PARENTHETICAL_CI_AFTER_ESTIMATE.finditer(
        source
    ):
        context_start = 0

        for boundary in _SENTENCE_BOUNDARY.finditer(
            source,
            0,
            match.start("estimate"),
        ):
            context_start = boundary.end()

        context = source[
            context_start:match.start("estimate")
        ]

        if not _EFFECT_SIZE_CUE.search(context):
            continue

        try:
            estimate = _to_float(
                match.group("estimate")
            )
            lo = _to_float(match.group("lo"))
            hi = _to_float(match.group("hi"))
        except ValueError:
            continue

        if not (
            lo < hi
            and lo <= estimate <= hi
        ):
            continue

        span = match.span("interval")

        if _overlaps(span, occupied):
            continue

        found.append(
            (
                span,
                Num(
                    "ci",
                    "=",
                    lo,
                    hi,
                    None,
                ),
            )
        )
        occupied.append(span)

    for kind, rx in _PATTS:
        for match in rx.finditer(source):
            groups = match.groups()

            try:
                if kind == "p":
                    number = Num(
                        "p",
                        groups[0],
                        _to_float(groups[1]),
                        None,
                        None,
                    )

                elif kind == "ci":
                    number = Num(
                        "ci",
                        "=",
                        _to_float(groups[0]),
                        _to_float(groups[1]),
                        None,
                    )

                elif kind == "dose":
                    unit = groups[1].lower()
                    number = Num(
                        "dose",
                        "=",
                        _to_float(groups[0])
                        * _UNIT_NORM.get(unit, 1.0),
                        None,
                        unit,
                    )

                elif kind == "percent":
                    number = Num(
                        "percent",
                        "=",
                        _to_float(groups[0]),
                        None,
                        None,
                    )

                elif kind in {"n", "sample"}:
                    value = (
                        groups[0]
                        .replace(",", "")
                        .replace(" ", "")
                    )
                    number = Num(
                        "n",
                        "=",
                        float(value),
                        None,
                        None,
                    )

                elif kind == "publication_count_word_safe":
                    number = Num(
                        "publication_count",
                        "=",
                        _parse_safe_study_count_word(
                            groups[0]
                        ),
                        None,
                        None,
                    )

                elif kind == "publication_count":
                    number = Num(
                        "publication_count",
                        "=",
                        float(groups[0]),
                        None,
                        None,
                    )

                elif kind == "study_count_word_safe":
                    number = Num(
                        "study_count",
                        "=",
                        _parse_safe_study_count_word(
                            groups[0]
                        ),
                        None,
                        None,
                    )

                elif kind == "study_count":
                    number = Num(
                        "study_count",
                        "=",
                        float(groups[0]),
                        None,
                        None,
                    )

                elif kind == "duration_word":
                    number = Num(
                        "duration",
                        "=",
                        float(
                            _WORD_NUMBER_NORM[
                                groups[0].lower()
                            ]
                        )
                        * _DAY_NORM.get(
                            groups[1].lower(),
                            1.0,
                        ),
                        None,
                        "days",
                    )

                elif kind == "duration":
                    number = Num(
                        "duration",
                        "=",
                        _to_float(groups[0])
                        * _DAY_NORM.get(
                            groups[1].lower(),
                            1.0,
                        ),
                        None,
                        "days",
                    )

                elif kind in {
                    "effect_size",
                    "ratio",
                }:
                    number = Num(
                        kind,
                        "=",
                        _to_float(groups[0]),
                        None,
                        None,
                    )

                else:
                    continue

            except (ValueError, IndexError):
                continue

            span = match.span()

            if _overlaps(span, occupied):
                continue

            found.append((span, number))
            occupied.append(span)

    for match in _GENERIC_NUMBER.finditer(
        source
    ):
        span = match.span()

        if _overlaps(span, occupied):
            continue

        try:
            value = _to_float(
                match.group(1)
            )
        except ValueError:
            continue

        found.append(
            (
                span,
                Num(
                    "scalar",
                    "=",
                    value,
                    None,
                    None,
                ),
            )
        )
        occupied.append(span)

    found.sort(
        key=lambda item: item[0][0]
    )

    return [
        number
        for _, number in found
    ]


def _match_one(c: Num, ev: list[Num], rel_tol: float = 0.02) -> str:
    # "scalar" est le type historique des estimations
    # additives non libellées dans les abstracts.
    # "effect_size" est le type explicite de SMD/MD.
    #
    # Ces deux représentations additives sont compatibles,
    # mais restent strictement séparées des vrais ratios
    # OR/RR/HR, des pourcentages, p-values et intervalles.
    compatible_kinds = (
        {"scalar", "effect_size"}
        if c.kind in {"scalar", "effect_size"}
        else {c.kind}
    )

    cands = [
        e
        for e in ev
        if e.kind in compatible_kinds
    ]
    if not cands:
        return "UNSUPPORTED"
    for e in cands:
        if c.kind == "p":
            if abs(e.lo - c.lo) <= 1e-9:
                return "MATCH"
        elif c.kind == "ci":
            if (c.hi is not None and e.hi is not None
                    and abs(e.lo - c.lo) <= rel_tol * max(abs(c.lo), 1e-9)
                    and abs(e.hi - c.hi) <= rel_tol * max(abs(c.hi), 1e-9)):
                return "MATCH"
        else:
            if abs(e.lo - c.lo) <= rel_tol * max(abs(c.lo), 1e-9):
                return "MATCH"
    return "MISMATCH"


def validate_numeric(
    claim_en: str,
    evidence_en: str,
    rel_tol: float = 0.02,
) -> dict:
    claim_text = str(claim_en or "")
    evidence_text = str(evidence_en or "")

    c_nums = extract(claim_text)
    e_nums = extract(evidence_text)

    digit_present = bool(
        re.search(r"\d", claim_text)
    )

    results = [
        {
            "kind": c.kind,
            "value": c.lo,
            "verdict": _match_one(
                c,
                e_nums,
                rel_tol,
            ),
        }
        for c in c_nums
    ]

    unparsed_numeric = (
        digit_present
        and not c_nums
    )

    numeric_ok = (
        all(
            result["verdict"] == "MATCH"
            for result in results
        )
        if c_nums
        else not digit_present
    )

    return {
        "numeric_ok": numeric_ok,
        "has_numbers": bool(c_nums),
        "unparsed_numeric":
            unparsed_numeric,
        "detail": results,
    }