File size: 5,749 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 | """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
# Candidate date formats, ordered. ISO first.
_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."""
# 1. Whitespace/case normalization that lands on the dominant shape.
trimmed = value.strip()
if value_shape(trimmed) == dominant_shape and trimmed != value:
return trimmed
# 2. Date reformat: dominant format is a date, value parses unambiguously.
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
# 3. Leading-zero padding of fixed-width numeric codes.
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))
# Unambiguous only when every successful parse yields the same target.
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
|