| """Repairer for format-violation issues via safe, invertible canonicalization.
|
|
|
| This repairer only proposes a fix when it can canonicalize a minority-format
|
| value to the column's dominant format through a *safe, well-understood*
|
| transform: date reformatting (when the parse is unambiguous), leading-zero
|
| padding of fixed-width numeric codes, or whitespace/case normalization. When
|
| the correct canonical value cannot be derived unambiguously, it abstains
|
| (returns ``None``) - the issue stays detection-only rather than risking a wrong
|
| fix. Every proposal still passes the SMT verifier and safety constitution.
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| from collections import Counter
|
| from datetime import datetime
|
|
|
| from dataforge.detectors.base import Issue, Schema
|
| from dataforge.detectors.format_violation import value_shape
|
| from dataforge.repairers.base import ProposedFix, RetryContext
|
| from dataforge.table import TableLike, cell_value, column_values
|
| from dataforge.transactions.txn import CellFix
|
|
|
|
|
| _DATE_FORMATS = (
|
| "%Y-%m-%d",
|
| "%Y/%m/%d",
|
| "%m/%d/%Y",
|
| "%d/%m/%Y",
|
| "%m-%d-%Y",
|
| "%d-%m-%Y",
|
| "%Y.%m.%d",
|
| "%d.%m.%Y",
|
| )
|
|
|
|
|
| class FormatViolationRepairer:
|
| """Canonicalize minority-format values to the column's dominant format."""
|
|
|
| def propose(
|
| self,
|
| issue: Issue,
|
| df: TableLike,
|
| schema: Schema | None,
|
| retry_context: RetryContext | None = None,
|
| ) -> ProposedFix | None:
|
| """Propose a canonicalization fix, or abstain when ambiguous."""
|
| del retry_context
|
| if issue.issue_type != "format_violation":
|
| return None
|
|
|
| old_value = cell_value(df, issue.row, issue.column)
|
| if old_value.strip() == "":
|
| return None
|
|
|
| dominant_shape, dominant_examples = self._dominant_profile(df, issue.column)
|
| if dominant_shape is None:
|
| return None
|
|
|
| new_value = self._canonicalize(old_value, dominant_shape, dominant_examples)
|
| if new_value is None or new_value == old_value:
|
| return None
|
|
|
| return ProposedFix(
|
| fix=CellFix(
|
| row=issue.row,
|
| column=issue.column,
|
| old_value=old_value,
|
| new_value=new_value,
|
| detector_id="format_violation",
|
| operation="update",
|
| ),
|
| reason=(f"Canonicalized '{old_value}' to dominant column format -> '{new_value}'."),
|
| confidence=issue.confidence,
|
| provenance="deterministic",
|
| )
|
|
|
| @staticmethod
|
| def _dominant_profile(df: TableLike, column: str) -> tuple[str | None, list[str]]:
|
| """Return the dominant shape and example values carrying that shape."""
|
| values = [str(v).strip() for v in column_values(df, column) if str(v).strip()]
|
| if not values:
|
| return None, []
|
| shapes = Counter(value_shape(v) for v in values)
|
| dominant_shape, _ = shapes.most_common(1)[0]
|
| examples = [v for v in values if value_shape(v) == dominant_shape]
|
| return dominant_shape, examples
|
|
|
| def _canonicalize(
|
| self, value: str, dominant_shape: str, dominant_examples: list[str]
|
| ) -> str | None:
|
| """Return the canonicalized value, or None when it cannot be derived safely."""
|
|
|
| trimmed = value.strip()
|
| if value_shape(trimmed) == dominant_shape and trimmed != value:
|
| return trimmed
|
|
|
|
|
| target_fmt = self._dominant_date_format(dominant_examples)
|
| if target_fmt is not None:
|
| reformatted = self._reformat_date(value, target_fmt)
|
| if reformatted is not None:
|
| return reformatted
|
|
|
|
|
| if set(dominant_shape) == {"9"} and trimmed.isdigit():
|
| target_len = len(dominant_examples[0]) if dominant_examples else len(dominant_shape)
|
| if all(len(ex) == target_len for ex in dominant_examples) and len(trimmed) < target_len:
|
| return trimmed.zfill(target_len)
|
|
|
| return None
|
|
|
| @staticmethod
|
| def _dominant_date_format(examples: list[str]) -> str | None:
|
| """Infer the single date format that parses all dominant examples."""
|
| for fmt in _DATE_FORMATS:
|
| if all(_parses(example, fmt) for example in examples[:25]):
|
| return fmt
|
| return None
|
|
|
| @staticmethod
|
| def _reformat_date(value: str, target_fmt: str) -> str | None:
|
| """Parse value with non-target formats; reformat only if unambiguous."""
|
| candidate = value.strip()
|
| parsed: set[str] = set()
|
| for fmt in _DATE_FORMATS:
|
| dt = _try_parse(candidate, fmt)
|
| if dt is not None:
|
| parsed.add(dt.strftime(target_fmt))
|
|
|
| if len(parsed) == 1:
|
| result = next(iter(parsed))
|
| return result if result != candidate else None
|
| return None
|
|
|
|
|
| def _parses(value: str, fmt: str) -> bool:
|
| """Return whether value parses under fmt as a valid date."""
|
| return _try_parse(value, fmt) is not None
|
|
|
|
|
| def _try_parse(value: str, fmt: str) -> datetime | None:
|
| """Try to parse a date; return the datetime or None."""
|
| try:
|
| return datetime.strptime(value.strip(), fmt)
|
| except (ValueError, TypeError):
|
| return None
|
|
|
