| """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+)$")
|
|
|
|
|
|
|
|
|
| _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)
|
|
|
