File size: 6,847 Bytes
13fe504 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | """Value-focused checks against inferred (advisory) constraints.
This module is the single source of truth for "does a proposed correction value
satisfy the inferred constraints for its column". It is shared by:
* :class:`dataforge.verifier.smt.SMTVerifier` -- the schema-less verification
guard that rejects untrusted (e.g. LLM) corrections instead of structurally
auto-accepting them when no authoritative schema exists; and
* :class:`dataforge.repairers.contract.CorrectionContract` -- the per-issue spec
that tells a corrector what a valid value looks like and cheaply pre-filters
candidates before the verifier/constitution gates.
Keeping the logic here guarantees the contract and the verifier never disagree:
a value the contract accepts as well-formed will not be rejected by the guard
for a value-local reason.
Design: pure Python, conservative. Value-local checks (type / numeric domain /
regex) inspect only the proposed value. The functional-dependency check is
table-relative but only fires on a *unanimous* determinant group -- the safest
signal that a dependent correction is wrong. Anything the inferred schema cannot
speak to passes through unchallenged.
"""
from __future__ import annotations
import re
from dataforge.table import TableLike, cell_value, column_names, row_count
from dataforge.verifier.schema import DomainBound, Schema
_INT_RE = re.compile(r"^[+-]?\d+$")
_FLOAT_RE = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)$")
# A correction may legitimately extrapolate slightly past the observed range
# (the observed range is drawn from dirty data). The guard only rejects values
# that fall well outside, controlled by this multiple of the observed span.
_DOMAIN_PAD_FRACTION = 0.5
def parse_numeric(value: str) -> float | None:
"""Parse a finite float from a string, or return ``None``."""
try:
parsed = float(value)
except (TypeError, ValueError):
return None
if parsed != parsed or parsed in (float("inf"), float("-inf")):
return None
return parsed
def type_violation(value: str, column_type: str) -> str | None:
"""Return a reason if ``value`` does not parse as ``column_type``."""
column_type = column_type.strip().lower()
if column_type in {"int", "integer"} and not _INT_RE.fullmatch(value):
return f"value {value!r} is not a valid integer (inferred type int)"
if column_type in {"float", "decimal", "real"} and not _FLOAT_RE.fullmatch(value):
return f"value {value!r} is not a valid number (inferred type {column_type})"
return None
def domain_violation(value: str, bound: DomainBound) -> str | None:
"""Return a reason if numeric ``value`` is far outside the inferred bound."""
numeric = parse_numeric(value)
if numeric is None:
return None
low = bound.min_value
high = bound.max_value
if low is None or high is None:
return None
span = high - low
pad = (abs(high) * _DOMAIN_PAD_FRACTION if span == 0 else span * _DOMAIN_PAD_FRACTION) or 1.0
if numeric < low - pad or numeric > high + pad:
return f"value {numeric} is far outside the inferred numeric range [{low}, {high}]"
return None
def regex_violation(value: str, pattern: str) -> str | None:
"""Return a reason if ``value`` does not match ``pattern``."""
try:
matches = re.fullmatch(pattern, value) is not None
except re.error:
return None
if not matches:
return f"value {value!r} does not match the inferred pattern {pattern!r}"
return None
def value_local_violation(value: str, column: str, schema: Schema) -> str | None:
"""Return a reason if ``value`` violates a value-local inferred constraint.
Covers type, numeric domain, and regex -- everything that depends only on
the proposed value and the column's inferred constraints, not on other rows.
"""
stripped = str(value).strip()
column_type = schema.column_type(column) or "str"
reason = type_violation(stripped, column_type)
if reason is not None:
return reason
for bound in schema.domain_bounds_for(column):
reason = domain_violation(stripped, bound)
if reason is not None:
return reason
for rule in schema.regex_constraints_for(column):
reason = regex_violation(stripped, rule.pattern)
if reason is not None:
return reason
return None
def fd_consensus_violation(
df: TableLike,
row: int,
column: str,
value: str,
schema: Schema,
) -> str | None:
"""Reject a dependent value that contradicts a unanimous determinant group.
Only rejects when every *other* row sharing the determinant value agrees on
a single dependent value that differs from the proposal. Mixed or empty
groups pass through.
"""
stripped = str(value).strip()
columns = set(column_names(df))
total_rows = row_count(df)
for fd in schema.functional_dependencies:
if fd.dependent != column:
continue
if any(det not in columns for det in fd.determinant):
continue
determinant_key = tuple(str(cell_value(df, row, det)).strip() for det in fd.determinant)
if any(part == "" for part in determinant_key):
continue
peers: set[str] = set()
for other in range(total_rows):
if other == row:
continue
other_key = tuple(str(cell_value(df, other, det)).strip() for det in fd.determinant)
if other_key != determinant_key:
continue
dependent_value = str(cell_value(df, other, column)).strip()
if dependent_value:
peers.add(dependent_value)
if len(peers) == 1 and stripped not in peers:
consensus = next(iter(peers))
determinant_label = "+".join(fd.determinant)
return (
f"value {value!r} contradicts the inferred dependency "
f"{determinant_label} -> {column}: rows with {determinant_key} "
f"consistently show {consensus!r}"
)
return None
def inferred_value_violation(
df: TableLike,
row: int,
column: str,
value: str,
schema: Schema,
) -> str | None:
"""Return a rejection reason if ``value`` violates any inferred constraint.
Orchestrates the value-local checks and the table-relative FD consensus
check. Returns ``None`` when the value is acceptable under inferred rules.
"""
reason = value_local_violation(value, column, schema)
if reason is not None:
return reason
return fd_consensus_violation(df, row, column, value, schema)
|