| """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
|
|
|
|
|
| _NUMERIC_ISSUE_TYPES: frozenset[IssueTypeLiteral] = frozenset({"decimal_shift", "outlier"})
|
|
|
|
|
| _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)
|
|
|
