File size: 7,450 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 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | """Per-issue correction contracts.
A :class:`CorrectionContract` binds a detected :class:`~dataforge.detectors.base.Issue`
to the constraints a valid correction for it must satisfy. It is the bridge
between detection and (LLM) correction:
* ``describe()`` turns the issue + inferred/declared constraints into a precise,
evidence-light target specification a corrector can be prompted with.
* ``validate(value)`` cheaply checks a candidate value against that target
before the SMT verifier and constitution gates are consulted, and gives the
corrector an actionable rejection reason for a retry.
The contract is intentionally value-local: it reuses
:mod:`dataforge.verifier.inferred` for type / domain / regex checks so it can
never disagree with the schema-less verification guard. Table-relative
constraints (functional dependencies, uniqueness) remain the verifier's job;
the contract never claims a candidate is fully valid, only that it is well
formed enough to be worth verifying.
"""
from __future__ import annotations
from pydantic import BaseModel, Field
from dataforge.detectors.base import Issue, IssueTypeLiteral
from dataforge.verifier.inferred import value_local_violation
from dataforge.verifier.schema import Schema
# Issue classes whose correct value is necessarily numeric.
_NUMERIC_ISSUE_TYPES: frozenset[IssueTypeLiteral] = frozenset({"decimal_shift", "outlier"})
# Row-level issue classes that are not single-cell value corrections.
_NON_CELL_ISSUE_TYPES: frozenset[IssueTypeLiteral] = frozenset({"duplicate_row"})
class ContractResult(BaseModel):
"""Outcome of validating a candidate correction against a contract."""
ok: bool
reason: str = Field(min_length=1)
model_config = {"frozen": True}
class CorrectionContract(BaseModel):
"""What a valid correction for a specific issue must look like."""
issue: Issue
constraints: Schema
model_config = {"frozen": True, "arbitrary_types_allowed": True}
@property
def column(self) -> str:
"""Column the correction targets."""
return self.issue.column
@property
def is_cell_correction(self) -> bool:
"""Whether this issue is a single-cell value correction at all."""
return self.issue.issue_type not in _NON_CELL_ISSUE_TYPES
def _requires_numeric(self) -> bool:
if self.issue.issue_type in _NUMERIC_ISSUE_TYPES:
return True
column_type = (self.constraints.column_type(self.column) or "str").strip().lower()
return column_type in {"int", "integer", "float", "decimal", "real"}
def check(self, value: str) -> ContractResult:
"""Validate a candidate correction value against the contract.
Rejects definitively wrong candidates (a no-op equal to the dirty value,
an empty value, a value violating inferred type/domain/regex, or a value
for a non-cell issue). A passing result means "well formed enough to send
to the verifier", not "guaranteed correct".
"""
if not self.is_cell_correction:
return ContractResult(
ok=False,
reason=(
f"Issue type '{self.issue.issue_type}' is a row-level operation, "
"not a single-cell value correction."
),
)
candidate = str(value).strip()
if not candidate:
return ContractResult(
ok=False,
reason="A correction must be a non-empty value.",
)
if candidate == str(self.issue.actual).strip():
return ContractResult(
ok=False,
reason="A correction must differ from the current (erroneous) value.",
)
if self._requires_numeric():
numeric_reason = value_local_violation(
candidate, self.column, _numeric_only_schema(self.constraints, self.column)
)
if numeric_reason is not None:
return ContractResult(ok=False, reason=numeric_reason)
violation = value_local_violation(candidate, self.column, self.constraints)
if violation is not None:
return ContractResult(ok=False, reason=violation)
return ContractResult(
ok=True,
reason="Candidate satisfies the value-local correction contract.",
)
def describe(self) -> str:
"""Return a precise, corrector-facing specification of a valid value."""
requirements: list[str] = [
"be different from the current value",
"be a single non-empty value",
]
column_type = (self.constraints.column_type(self.column) or "str").strip().lower()
if self._requires_numeric() or column_type in {
"int",
"integer",
"float",
"decimal",
"real",
}:
if column_type in {"int", "integer"}:
requirements.append("be a valid integer")
else:
requirements.append("be a valid number")
for bound in self.constraints.domain_bounds_for(self.column):
if bound.min_value is not None and bound.max_value is not None:
requirements.append(
f"lie near the observed numeric range [{bound.min_value}, {bound.max_value}]"
)
for rule in self.constraints.regex_constraints_for(self.column):
requirements.append(f"match the pattern {rule.pattern}")
requirement_lines = "\n".join(f" - {item}" for item in requirements)
expected_hint = ""
if self.issue.expected is not None:
expected_hint = (
f"\nA deterministic analysis suggests the value may be "
f"{self.issue.expected!r}; use it only if it is correct."
)
return (
f"Correct the value at row {self.issue.row}, column '{self.column}'.\n"
f"Current value: {self.issue.actual!r}\n"
f"Detected issue: {self.issue.issue_type} - {self.issue.reason}\n"
f"A valid correction must:\n{requirement_lines}{expected_hint}\n"
f"Respond with only the corrected value and nothing else."
)
def _numeric_only_schema(schema: Schema, column: str) -> Schema:
"""Return a schema asserting ``column`` is numeric, for the numeric pre-check.
Used when the issue class implies a numeric value (e.g. outlier) even if the
inferred column type was left as ``str`` because of dirty cells.
"""
if (schema.column_type(column) or "str").strip().lower() in {
"int",
"integer",
"float",
"decimal",
"real",
}:
return schema
columns = dict(schema.columns)
columns[column] = "float"
return Schema(
columns=columns,
functional_dependencies=schema.functional_dependencies,
regex_constraints=schema.regex_constraints,
domain_bounds=schema.domain_bounds,
)
def build_correction_contract(issue: Issue, constraints: Schema) -> CorrectionContract:
"""Build a correction contract for an issue against inferred/declared rules."""
return CorrectionContract(issue=issue, constraints=constraints)
|