Great product…
", "fine"], + "amount": [" 1,200.50 ", "99.99"], +}) + +cleaned, report = fd.clean_text( + df, + config=fd.TextCleanConfig(strip_html=True), + field_types={"amount": "currency_amount"}, # guards aggressive ops +) + +print(cleaned.loc[0, "note"]) # Great product... +print(cleaned.loc[0, "amount"]) # 1,200.50 (only whitespace trimmed) +print(report.summary()) +``` + +Key properties: + +- **Non-destructive** — the input frame is never modified; every changed + cell is logged with `original`, `cleaned` and the ordered transform list. +- **Field-aware** — lossy operations (case folding, punctuation or HTML + stripping) are automatically withheld from structural types (amounts, + identifiers, emails, URLs, dates, tickers) and from entity names. +- **Deterministic and lossless by default** — HTML/URL removal, case + folding and punctuation removal are opt-in via `TextCleanConfig`. + +Scalar use: `fd.clean_text_value(" x ")` returns a `CleanedText` with +`.original`, `.cleaned`, `.transforms`. + +## Context-aware validation: the "apple" example + +A scraped feed inserts the string `"apple"` into different columns. The +correct verdict depends entirely on the field: + +```python +import pandas as pd +import freshdata as fd + +df = pd.DataFrame([{ + "transaction_amount": "apple", # invalid: monetary field + "company_name": "Apple", # valid: real-word names are fine + "stock_ticker": "apple", # invalid format; AAPL would pass + "transaction_description": "Payment to Apple", # valid free text +}]) + +schema = { + "transaction_amount": fd.FieldSpec(semantic_type="currency_amount"), + "company_name": fd.FieldSpec(semantic_type="company_name"), + "stock_ticker": fd.FieldSpec(semantic_type="ticker", + suggest={"apple": "AAPL"}), + "transaction_description": fd.FieldSpec(semantic_type="free_text"), +} + +report = fd.validate_fields(df, schema) +print(report.summary()) +for issue in report.issues: + print(issue.column, issue.classification, "->", issue.action, + "| suggestion:", issue.suggestion) +``` + +Output (abridged): + +``` +transaction_amount semantic_mismatch -> quarantine | suggestion: None +stock_ticker domain_mismatch -> manual_review | suggestion: AAPL +``` + +Rules of the road: + +- values are **never silently converted** — `"apple"` does not become `0` + or `NaN`; +- suggestions come **only from a trusted mapping or callable you supply** + (`FieldSpec.suggest`); nothing is invented; +- reference verification (e.g. a ticker universe) is injected via + `FieldSpec.reference` — never hard-coded; +- a column *without* a spec is still protected: when ≥80% of its values + share one shape, nonconforming cells are reported as `semantic_mismatch` + with the exact row, reason and confidence. + +## Failure classes and remediation policy + +Distinct problems stay distinct — there is no generic "outlier" bucket: + +| classification | default action | severity | +|----------------------------|-----------------------|----------| +| `parse_failure` | `quarantine` | error | +| `semantic_mismatch` | `quarantine` | error | +| `domain_mismatch` | `manual_review` | error | +| `schema_violation` | `reject` | error | +| `statistical_outlier` | `accept_with_warning` | warning | +| `categorical_rare` | `accept_with_warning` | warning | +| `cross_field_inconsistency`| `manual_review` | warning | + +Rare or extreme values are **warned about, never auto-rejected**: an +unusually large transaction is still a transaction. Every action is +configurable per class: + +```python +policy = fd.RemediationPolicy(semantic_mismatch="replace_with_null") +report = fd.validate_fields(df, schema, policy=policy) +result = fd.apply_field_policy(df, report) +result.accepted # cleaned copy (replacements applied, audited) +result.quarantined # rows held back for review +result.rejected # rows dropped from the accepted set (still returned!) +result.needs_review # rows awaiting a human decision +result.audit # one record per issue: original, action, reason, rule +``` + +`apply_field_policy` never mutates the input frame, and every +`replace_with_null` keeps the original value in the audit trail. + +Context-dependent missing codes are per-field: `"N/A"` can be a null marker +in one column and a legitimate code in another +(`FieldSpec(null_markers=frozenset({""}), allowed_values={"N/A", ...})`). + +Cross-column rules are plain callables: + +```python +def settle_after_trade(row): + if str(row["settlement_date"]) < str(row["trade_date"]): + return "settlement_date precedes trade_date" + return None + +fd.validate_fields(df, schema, cross_rules=[settle_after_trade]) +``` + +## Reporting + +`report.to_findings()` exports issues as standard +[`QualityFinding`](api-reference.md) records (step `"fieldcheck"`), so they +flow into the existing exporters (dbt tests, Great Expectations suites, +exception tables, lineage). `report.to_frame()` gives a flat DataFrame; +`report.normalized_cells` is the text-normalization audit. + +## Performance + +Validation runs a vectorized pre-screen per column and only pays the +per-cell explanation path for suspect cells, so mostly-clean feeds validate +at bulk speed. Reproduce the comparison against regex and pandas-coercion +baselines with: + +```bash +python benchmarks/bench_fieldcheck.py 100000 +``` + +## Known limitations + +- Semantic-type coverage is deliberately narrow and deterministic (numbers, + dates, emails, URLs, phones, tickers, identifiers, entity names, free + text); embedding-based typing lives in the optional + [semantic layer](semantic-models.md). +- Column-consensus checks need a dominant shape (≥80%) and at least 3 + values; genuinely mixed columns are left alone by design. +- `FieldSpec.reference` given as a callable disables the vectorized + pre-screen for that column (every value is checked individually). +- Date validation accepts any format `pandas.to_datetime` can parse; + day-first/locale ambiguity is the cleaning engine's concern + (`fd.clean(dayfirst=...)`). diff --git a/mkdocs.yml b/mkdocs.yml index 4539c65..a1a4be0 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -121,6 +121,7 @@ nav: - Learning profiles: learning-profiles.md - Developer training pipeline: developer-training-pipeline.md - Format parsers & tick mode: parsers.md + - Text cleaning & field validation: field-validation.md - Compliance reports: compliance.md - Orchestration integrations: integrations.md - Examples: examples.md diff --git a/src/freshdata/__init__.py b/src/freshdata/__init__.py index bda64e7..8e13179 100644 --- a/src/freshdata/__init__.py +++ b/src/freshdata/__init__.py @@ -45,6 +45,15 @@ from .context import ColumnConstraint, ContextPolicy, PolicyError from .execution import EngineConfig from .explain import ExplainReport, explain_clean +from .fieldcheck import ( + CellIssue, + FieldSpec, + FieldValidationReport, + PolicyResult, + RemediationPolicy, + apply_field_policy, + validate_fields, +) from .findings import FindingList, QualityFinding from .guard import ProtectedColumnError from .insight import FreshDataInsightReport, insight_report, trust_gate_report @@ -86,6 +95,13 @@ StreamingState, TimeSeriesCleanConfig, ) +from .textclean import ( + CleanedText, + TextCleanConfig, + TextCleanReport, + clean_text, + clean_text_value, +) from .textlint import TextIssue, TextLintReport, lint_text_encoding __version__ = "1.1.1" @@ -124,10 +140,22 @@ "StreamingCleanConfig", "StreamingCleaner", "StreamingState", + "CellIssue", + "CleanedText", + "FieldSpec", + "FieldValidationReport", + "PolicyResult", + "RemediationPolicy", + "TextCleanConfig", + "TextCleanReport", "TextIssue", "TextLintReport", "TimeSeriesCleanConfig", "__version__", + "apply_field_policy", + "clean_text", + "clean_text_value", + "validate_fields", "cdc_profile", "apply_plan", "clean", diff --git a/src/freshdata/fieldcheck.py b/src/freshdata/fieldcheck.py new file mode 100644 index 0000000..ea3161c --- /dev/null +++ b/src/freshdata/fieldcheck.py @@ -0,0 +1,872 @@ +"""Context-aware, per-cell field validation with configurable remediation. + +``fd.validate_fields(df, schema=..., policy=...)`` decides — cell by cell — +whether a value is valid *for its field*, not merely parseable. The same +string can be fine in one column and wrong in another (``"Apple"`` is a +plausible ``company_name``, an invalid ``transaction_amount`` and a malformed +``stock_ticker``), so every check combines: + +* the declared :class:`FieldSpec` (expected semantic type, range, pattern, + vocabulary, reference lookup, null markers); +* what the value itself looks like (number / date / email / URL / free text); +* the *column consensus* — the dominant shape of the other values — so a lone + ``"apple"`` inside a numeric column is caught even without a schema. + +Distinct failure classes are kept apart (a parse failure is not a statistical +outlier is not a rare category), and a :class:`RemediationPolicy` maps each +class to an action. The default policy is non-destructive: nothing is deleted, +nulled or "corrected" silently; :func:`apply_field_policy` splits the frame +into accepted / quarantined / rejected copies plus a full audit trail, leaving +the input untouched. +""" + +from __future__ import annotations + +import re +from collections.abc import Callable, Collection, Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any + +import pandas as pd + +from .findings import QualityFinding +from .semantic.experts import is_plain_number, looks_like_date_value, parse_currency +from .textclean import TextCleanConfig, clean_text_value, config_for_field + +__all__ = [ + "FieldSpec", + "RemediationPolicy", + "CellIssue", + "FieldValidationReport", + "PolicyResult", + "validate_fields", + "apply_field_policy", + "CLASSIFICATIONS", + "ACTIONS", +] + +#: Distinct failure classes — never collapsed into one generic "outlier". +CLASSIFICATIONS = ( + "parse_failure", # value cannot be parsed as the expected type + "semantic_mismatch", # parseable text, but the wrong kind of thing + "domain_mismatch", # right shape, wrong for this domain field + "schema_violation", # nullability / required / allowed-values breach + "statistical_outlier", # numeric value far outside the column's spread + "categorical_rare", # unseen or rare category — possibly valid + "cross_field_inconsistency", # row fails a relationship between columns +) + +#: Configurable remediation actions, least to most severe. +ACTIONS = ( + "accept", + "accept_with_warning", + "normalize", + "replace_with_null", + "quarantine", + "reject", + "manual_review", +) + +_DEFAULT_NULL_MARKERS = frozenset({"", "n/a", "na", "null", "none", "nan", "-", "--"}) + +_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[A-Za-z]{2,}$") +_URL_RE = re.compile(r"^https?://\S+\.\S+", re.I) +_TICKER_RE = re.compile(r"^[A-Z]{1,6}([.\-][A-Z0-9]{1,4})?$") +_PHONE_RE = re.compile(r"^\+?[\d\s\-().]{7,17}$") +_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._\-/]*$") + + +def _safe_fullmatch(pattern: str, value: str) -> bool: + """``re.fullmatch`` that never raises. + + ``FieldSpec.pattern`` is caller-supplied; an invalid regex degrades to + "nothing matches" (surfaced as a normal domain_mismatch) instead of + crashing validation mid-run. + """ + try: + return re.fullmatch(pattern, value) is not None + except re.error: + return False + + +#: Semantic types validate_fields understands. ``numeric``-family types parse +#: through the same path; anything else falls back to pattern/vocabulary rules. +_NUMERIC_TYPES = frozenset( + {"numeric", "integer", "float", "currency_amount", "rate", "percentage"}) +_DATE_TYPES = frozenset({"date", "datetime", "date_like"}) + + +def detect_value_type(value: Any) -> str: + """Cheap, deterministic shape probe for one scalar.""" + if value is None or (isinstance(value, float) and pd.isna(value)): + return "null" + if isinstance(value, bool): + return "boolean_like" + if isinstance(value, (int, float)): + return "numeric" + if not isinstance(value, str): + return "other" + s = value.strip() + if not s: + return "null" + if is_plain_number(s) or parse_currency(s) is not None: + return "numeric" + if looks_like_date_value(s): + return "date_like" + if _EMAIL_RE.match(s): + return "email" + if _URL_RE.match(s): + return "url" + if _PHONE_RE.match(s) and sum(c.isdigit() for c in s) >= 7: + return "phone" + return "text" + + +@dataclass(frozen=True) +class FieldSpec: + """Declared expectations for one column. + + Only ``semantic_type`` is usually needed; everything else refines it. + ``reference`` and ``suggest`` are injection points for external knowledge + (e.g. a ticker universe) — nothing domain-specific is hard-coded here. + """ + + semantic_type: str | None = None + required: bool = False #: column must be present and non-null + nullable: bool = True + allowed_values: frozenset | None = None + pattern: str | None = None #: full-match regex on the (cleaned) string + min_value: float | None = None + max_value: float | None = None + max_length: int | None = None + null_markers: frozenset = _DEFAULT_NULL_MARKERS #: field-specific missing codes + reference: Callable[[str], bool] | Collection[str] | None = None + suggest: Callable[[str], str | None] | Mapping[str, str] | None = None + + def is_null_marker(self, s: str) -> bool: + return s.strip().casefold() in self.null_markers + + def in_reference(self, s: str) -> bool | None: + """True/False when a reference source is configured, else ``None``.""" + if self.reference is None: + return None + if callable(self.reference): + return bool(self.reference(s)) + return s in self.reference + + def suggestion_for(self, s: str) -> str | None: + """A trusted correction for ``s``, or ``None``. Never invented.""" + if self.suggest is None: + return None + if callable(self.suggest): + return self.suggest(s) + return self.suggest.get(s) or self.suggest.get(s.strip().casefold()) + + +@dataclass(frozen=True) +class RemediationPolicy: + """Maps each failure classification to an action from :data:`ACTIONS`. + + Defaults are deliberately conservative: nothing destructive, statistical + outliers and rare categories are *warned about*, never auto-rejected. + """ + + parse_failure: str = "quarantine" + semantic_mismatch: str = "quarantine" + domain_mismatch: str = "manual_review" + schema_violation: str = "reject" + statistical_outlier: str = "accept_with_warning" + categorical_rare: str = "accept_with_warning" + cross_field_inconsistency: str = "manual_review" + normalize_text: bool = True #: run safe text cleaning before checks + + def __post_init__(self) -> None: + for cls in CLASSIFICATIONS: + action = getattr(self, cls) + if action not in ACTIONS: + raise ValueError( + f"invalid action {action!r} for {cls!r}; expected one of {ACTIONS}" + ) + + def action_for(self, classification: str) -> str: + return getattr(self, classification) + + +_SEVERITY_BY_CLASS = { + "parse_failure": "error", + "semantic_mismatch": "error", + "domain_mismatch": "error", + "schema_violation": "error", + "statistical_outlier": "warning", + "categorical_rare": "warning", + "cross_field_inconsistency": "warning", +} + + +@dataclass(frozen=True) +class CellIssue: + """One per-cell validation problem, with everything needed to act on it.""" + + row: Any + column: str + original: Any + cleaned: Any + classification: str + severity: str + reason: str + expected: str + detected: str + confidence: float + action: str + rule: str + suggestion: str | None = None + transforms: tuple = () + + def to_dict(self) -> dict: + return { + "row": self.row, "column": self.column, + "original": self.original, "cleaned": self.cleaned, + "classification": self.classification, "severity": self.severity, + "reason": self.reason, "expected": self.expected, + "detected": self.detected, "confidence": round(self.confidence, 4), + "action": self.action, "rule": self.rule, + "suggestion": self.suggestion, "transforms": list(self.transforms), + } + + def to_finding(self) -> QualityFinding: + return QualityFinding.create( + severity=self.severity, + step="fieldcheck", + rule_name=f"{self.classification}:{self.rule}", + message=self.reason, + column=self.column, + row_index=self.row, + observed_value=self.original, + expected_condition=self.expected, + action_taken=self.action, + extra={ + "detected": self.detected, + "confidence": round(self.confidence, 4), + **({"suggestion": self.suggestion} if self.suggestion else {}), + }, + ) + + +@dataclass +class FieldValidationReport: + """All issues plus per-row decisions from one :func:`validate_fields` run.""" + + issues: list = field(default_factory=list) + n_rows: int = 0 + columns_checked: list = field(default_factory=list) + inferred_types: dict = field(default_factory=dict) + normalized_cells: list = field(default_factory=list) #: audit of text normalizations + + def __len__(self) -> int: + return len(self.issues) + + def __bool__(self) -> bool: + return bool(self.issues) + + def by_classification(self) -> dict: + out: dict = {} + for i in self.issues: + out.setdefault(i.classification, []).append(i) + return out + + def to_frame(self) -> pd.DataFrame: + cols = ["row", "column", "original", "cleaned", "classification", "severity", + "reason", "expected", "detected", "confidence", "action", "rule", + "suggestion", "transforms"] + return pd.DataFrame([i.to_dict() for i in self.issues], columns=cols) + + def to_findings(self) -> list: + return [i.to_finding() for i in self.issues] + + def row_actions(self) -> dict: + """Most severe action per row (severity = position in :data:`ACTIONS`).""" + order = {a: n for n, a in enumerate(ACTIONS)} + out: dict = {} + for i in self.issues: + if i.row is None: + continue + cur = out.get(i.row) + if cur is None or order[i.action] > order[cur]: + out[i.row] = i.action + return out + + def summary(self) -> str: + lines = [ + f"freshdata field check — {len(self.issues)} issue(s) over {self.n_rows} row(s), " + f"{len(self.columns_checked)} column(s)" + ] + for cls, items in sorted(self.by_classification().items()): + lines.append(f" {cls}: {len(items)}") + for i in items[:3]: + lines.append( + f" row {i.row} {i.column}={i.original!r}: {i.reason} -> {i.action}" + ) + if not self.issues: + lines.append(" all values consistent with their fields") + return "\n".join(lines) + + def __str__(self) -> str: + return self.summary() + + +def _parse_numeric(s: str) -> float | None: + if is_plain_number(s): + return float(str(s).strip().replace(",", "")) + return parse_currency(s) + + +def _check_value( + col: str, + row: Any, + raw: Any, + cleaned: Any, + transforms: tuple, + spec: FieldSpec, + policy: RemediationPolicy, +) -> CellIssue | None: + """Validate one cell against its spec. Returns an issue or ``None``.""" + expected = spec.semantic_type or "any" + s = str(cleaned).strip() if cleaned is not None else "" + detected = detect_value_type(cleaned) + + def issue(classification: str, reason: str, rule: str, *, + confidence: float = 1.0, suggestion: str | None = None, + action: str | None = None) -> CellIssue: + return CellIssue( + row=row, column=col, original=raw, cleaned=cleaned, + classification=classification, + severity=_SEVERITY_BY_CLASS[classification], + reason=reason, expected=expected, detected=detected, + confidence=confidence, + action=action or policy.action_for(classification), + rule=rule, suggestion=suggestion, transforms=transforms, + ) + + # -- missing handling ----------------------------------------------------- + is_missing = raw is None or (not isinstance(raw, str) and pd.isna(raw)) or ( + isinstance(cleaned, str) and spec.is_null_marker(s)) + if is_missing: + if not spec.nullable or spec.required: + return issue( + "schema_violation", + f"{col} is required but the value is missing " + f"({raw!r} is a configured null marker)" if raw is not None + else f"{col} is required but the value is missing", + "not_null", + ) + return None + + # -- expected-type parse & range ------------------------------------------ + if spec.semantic_type in _NUMERIC_TYPES: + num = cleaned if isinstance(cleaned, (int, float)) and not isinstance(cleaned, bool) \ + else _parse_numeric(s) + if num is None: + textish = ("text", "email", "url", "phone", "date_like") + cls = "semantic_mismatch" if detected in textish else "parse_failure" + return issue( + cls, + f"expected {expected} in {col!r} but got {detected} value {s!r}; " + "not silently converted", + "numeric_parse", + ) + if spec.min_value is not None and num < spec.min_value: + return issue( + "domain_mismatch", + f"{col}={num} below configured minimum {spec.min_value}", "min_value") + if spec.max_value is not None and num > spec.max_value: + return issue( + "domain_mismatch", + f"{col}={num} above configured maximum {spec.max_value}", "max_value") + return None + + if spec.semantic_type in _DATE_TYPES: + ts = pd.to_datetime(s if isinstance(cleaned, str) else cleaned, errors="coerce") + if pd.isna(ts): + if looks_like_date_value(s): + return issue( + "parse_failure", + f"{s!r} is shaped like a date but is not a real calendar date", + "impossible_date", + ) + return issue( + "semantic_mismatch", + f"expected {expected} in {col!r} but got {detected} value {s!r}", + "date_parse", + ) + return None + + # -- pattern / vocabulary / reference fields -------------------------------- + if spec.allowed_values is not None and s not in spec.allowed_values \ + and s.casefold() not in spec.allowed_values: + return issue( + "domain_mismatch", + f"{s!r} is not in the allowed vocabulary for {col!r} " + f"({len(spec.allowed_values)} known values)", + "allowed_values", + suggestion=spec.suggestion_for(s), + ) + + if spec.pattern is not None and not _safe_fullmatch(spec.pattern, s): + return issue( + "domain_mismatch", + f"{s!r} does not match the {col!r} format {spec.pattern!r}", + "pattern", + suggestion=spec.suggestion_for(s), + ) + + ref = spec.in_reference(s) + if ref is False: + return issue( + "domain_mismatch", + f"{s!r} not found in the configured reference source for {col!r}", + "reference_lookup", + suggestion=spec.suggestion_for(s), + ) + + # -- semantic-type sanity for non-parsing text types ------------------------ + if spec.semantic_type in ("company_name", "entity_name", "person_name", "city", + "country", "free_text", "text"): + if spec.semantic_type != "free_text" and detected == "numeric": + return issue( + "semantic_mismatch", + f"{col!r} expects a {expected} but {s!r} is purely numeric", + "entity_numeric", confidence=0.8, + ) + if spec.max_length is not None and len(s) > spec.max_length: + return issue( + "schema_violation", + f"{col!r} value length {len(s)} exceeds max_length={spec.max_length}", + "max_length", + ) + return None + + if spec.semantic_type in ("identifier", "account_number") and not _ID_RE.match(s): + return issue( + "domain_mismatch", + f"{s!r} contains characters not allowed in a {expected}", + "identifier_charset", + ) + if spec.semantic_type in ("ticker", "stock_ticker") and not _TICKER_RE.match(s): + return issue( + "domain_mismatch", + f"{s!r} is not a structurally valid ticker symbol " + "(expected 1-6 uppercase letters, optional suffix)", + "ticker_format", + suggestion=spec.suggestion_for(s), + ) + if spec.semantic_type == "email" and not _EMAIL_RE.match(s): + return issue("semantic_mismatch", f"{s!r} is not a valid email address", "email_format") + if spec.semantic_type == "url" and not _URL_RE.match(s): + return issue("semantic_mismatch", f"{s!r} is not a valid URL", "url_format") + if spec.semantic_type == "phone" and not ( + _PHONE_RE.match(s) and sum(c.isdigit() for c in s) >= 7): + return issue("semantic_mismatch", f"{s!r} is not a plausible phone number", "phone_format") + + if spec.max_length is not None and len(s) > spec.max_length: + return issue( + "schema_violation", + f"{col!r} value length {len(s)} exceeds max_length={spec.max_length}", + "max_length", + ) + return None + + +#: Matches values the default text-cleaning config could possibly change: +#: anything non-ASCII-printable, leading/trailing space, or doubled spaces. +_DIRTY_PROBE = re.compile(r"[^\x20-\x7e]|^\s|\s$|\s{2,}") +_DEFAULT_CLEAN_FIELDS = ( + "unicode_form", "strip_control_chars", "strip_zero_width", + "normalize_punctuation", "collapse_whitespace", "strip", +) + + +def _probe_safe(cfg: TextCleanConfig) -> bool: + """True when :data:`_DIRTY_PROBE` is a superset of what ``cfg`` can change.""" + default = TextCleanConfig() + for name in ("strip_html", "strip_urls", "case", "remove_punctuation", + "max_char_repeat", "max_length", "custom"): + if getattr(cfg, name) != getattr(default, name): + return False + return True + + +_PURE_NUMBER_RE = r"-?[\d.,]+([eE][+-]?\d+)?" + + +def _suspect_rows(series: pd.Series, spec: FieldSpec) -> pd.Index: + """Cells that need the (authoritative) per-cell check. + + Vectorized pre-screen: every cell the slow path could flag **must** land + here; cells proven fine are skipped. When in doubt (currency formatting, + callable references, exotic dtypes) a cell simply stays a suspect — the + slow path re-decides, so over-selection costs time, never correctness. + """ + nonnull = series.notna() + strs = series.astype("string").str.strip() + + missing = ~nonnull | strs.str.casefold().isin(spec.null_markers).fillna(False) + if spec.required or not spec.nullable: + must_flag = missing # every missing cell is a violation → slow path + else: + must_flag = pd.Series(False, index=series.index) + checkable = ~missing + + if spec.semantic_type in _NUMERIC_TYPES: + parsed = pd.to_numeric(strs.str.replace(",", "", regex=False), errors="coerce") + fine = parsed.notna() + if spec.min_value is not None: + fine &= parsed >= spec.min_value + if spec.max_value is not None: + fine &= parsed <= spec.max_value + return series.index[must_flag | (checkable & ~fine.fillna(False))] + + if spec.semantic_type in _DATE_TYPES: + try: + import warnings # noqa: PLC0415 + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + parsed_dt = pd.to_datetime(strs, errors="coerce") + fine = parsed_dt.notna() + except (ValueError, TypeError): # pragma: no cover - exotic payloads + fine = pd.Series(False, index=series.index) + return series.index[must_flag | (checkable & ~fine.fillna(False))] + + fine = pd.Series(True, index=series.index) + if spec.allowed_values is not None: + fine &= (strs.isin(spec.allowed_values) + | strs.str.casefold().isin(spec.allowed_values)).fillna(False) + if spec.pattern is not None: + fine &= strs.map( + lambda v: _safe_fullmatch(spec.pattern, v) if isinstance(v, str) else False + ).fillna(False) + if spec.reference is not None: + if callable(spec.reference): + fine &= False # cannot vectorize a callable — everything is a suspect + else: + fine &= strs.isin([str(v) for v in spec.reference]).fillna(False) + if spec.max_length is not None: + fine &= (strs.str.len() <= spec.max_length).fillna(False) + + type_res = { + "identifier": _ID_RE, "account_number": _ID_RE, + "ticker": _TICKER_RE, "stock_ticker": _TICKER_RE, + "email": _EMAIL_RE, "url": _URL_RE, + } + if spec.semantic_type in type_res: + fine &= strs.str.fullmatch(type_res[spec.semantic_type].pattern).fillna(False) + elif spec.semantic_type == "phone": + fine &= (strs.str.fullmatch(_PHONE_RE.pattern).fillna(False) + & (strs.str.count(r"\d") >= 7).fillna(False)) + elif spec.semantic_type in ("company_name", "entity_name", "person_name", + "city", "country"): + fine &= ~strs.str.fullmatch(_PURE_NUMBER_RE).fillna(False) + + return series.index[must_flag | (checkable & ~fine)] + + +#: Consensus share above which a column without a spec is treated as typed. +_CONSENSUS_SHARE = 0.8 + + +def _column_consensus(series: pd.Series) -> tuple[str, float] | None: + """Dominant value shape of a column, if any (``(type, share)``).""" + sample = series.dropna() + if len(sample) > 10_000: + sample = sample.head(10_000) + if len(sample) < 3: + return None + counts: dict = {} + for v in sample: + counts[detect_value_type(v)] = counts.get(detect_value_type(v), 0) + 1 + counts.pop("null", None) + if not counts: + return None + top, n = max(counts.items(), key=lambda kv: kv[1]) + share = n / sum(counts.values()) + if share >= _CONSENSUS_SHARE and top in ("numeric", "date_like", "email", "url", "phone"): + return top, share + return None + + +def _iqr_outliers(series: pd.Series, k: float = 3.0) -> pd.Series: + """Boolean mask of extreme numeric values (Tukey fences, conservative k).""" + nums = pd.to_numeric(series, errors="coerce") + valid = nums.dropna() + if len(valid) < 8: + return pd.Series(False, index=series.index) + q1, q3 = valid.quantile(0.25), valid.quantile(0.75) + iqr = q3 - q1 + if iqr == 0: + return pd.Series(False, index=series.index) + return (nums < q1 - k * iqr) | (nums > q3 + k * iqr) + + +def _validate_column( + col: str, + series: pd.Series, + spec: FieldSpec | None, + policy: RemediationPolicy, + report: FieldValidationReport, + *, + rare_threshold: float, + outlier_fence: float, + clean_config: TextCleanConfig | None, +) -> None: + """Run every per-column check for one column, appending to ``report``.""" + # --- text normalization (audited, never in-place) ------------------------- + cleaned_col: dict = {} + transforms_col: dict = {} + if policy.normalize_text: + cfg = config_for_field(spec.semantic_type if spec else None, clean_config) + candidates = series + if _probe_safe(cfg): + try: + mask = series.astype("string").str.contains(_DIRTY_PROBE, na=False) + candidates = series[mask.fillna(False).astype(bool)] + except (TypeError, ValueError): # pragma: no cover - exotic payloads + candidates = series + for idx, val in candidates.items(): + if isinstance(val, str): + r = clean_text_value(val, cfg) + if r.changed: + cleaned_col[idx] = r.cleaned + transforms_col[idx] = r.transforms + report.normalized_cells.append({ + "row": idx, "column": col, + "original": val, "cleaned": r.cleaned, + "transforms": list(r.transforms), + }) + + def cell(idx: Any) -> tuple: + raw = series.loc[idx] + return cleaned_col.get(idx, raw), transforms_col.get(idx, ()) + + if spec is not None: + for idx in _suspect_rows(series, spec): + cleaned, transforms = cell(idx) + found = _check_value(col, idx, series.loc[idx], cleaned, + transforms, spec, policy) + if found is not None: + report.issues.append(found) + _rare_category_issues(col, series, spec, policy, report, rare_threshold) + if spec.semantic_type in _NUMERIC_TYPES: + _outlier_issues(col, series, spec, policy, report, outlier_fence, cell) + return + + # --- no spec: consensus-based contamination check ------------------------- + consensus = _column_consensus(series) + if consensus is None: + return + ctype, share = consensus + report.inferred_types[col] = {"type": ctype, "share": round(share, 4)} + for idx, val in series.items(): + cleaned, transforms = cell(idx) + if val is None or (not isinstance(val, str) and pd.isna(val)): + continue + detected = detect_value_type(cleaned) + if detected in (ctype, "null"): + continue + report.issues.append(CellIssue( + row=idx, column=col, original=val, cleaned=cleaned, + classification="semantic_mismatch", severity="error", + reason=f"{share:.0%} of {col!r} values are {ctype}, but this cell is " + f"{detected} ({val!r}); it does not fit the column's inferred type", + expected=ctype, detected=detected, confidence=share, + action=policy.action_for("semantic_mismatch"), + rule="column_consensus", transforms=transforms, + )) + + +def _rare_category_issues( + col: str, + series: pd.Series, + spec: FieldSpec, + policy: RemediationPolicy, + report: FieldValidationReport, + rare_threshold: float, +) -> None: + """Warn about *allowed* but rare categories — rare is never invalid.""" + if spec.allowed_values is None or not 0 < rare_threshold < 1 or len(series) < 20: + return + freq = series.astype(str).str.strip().value_counts(normalize=True) + flagged_rows = {i.row for i in report.issues if i.column == col} + for value, share in freq.items(): + if share >= rare_threshold or value not in spec.allowed_values: + continue + for idx in series.index[series.astype(str).str.strip() == value]: + if idx in flagged_rows: + continue + report.issues.append(CellIssue( + row=idx, column=col, original=series.loc[idx], + cleaned=value, classification="categorical_rare", + severity="warning", + reason=f"{value!r} is allowed but rare in {col!r} ({share:.1%} of rows)", + expected=spec.semantic_type or "category", + detected="rare_category", confidence=1 - share, + action=policy.action_for("categorical_rare"), + rule="rare_category", + )) + + +def _outlier_issues( + col: str, + series: pd.Series, + spec: FieldSpec, + policy: RemediationPolicy, + report: FieldValidationReport, + outlier_fence: float, + cell: Callable[[Any], tuple], +) -> None: + """Flag far numeric outliers as warnings; extreme is not automatically wrong.""" + mask = _iqr_outliers( + series.map(lambda v: _parse_numeric(str(v)) if isinstance(v, str) else v), + k=outlier_fence) + for idx in series.index[mask]: + report.issues.append(CellIssue( + row=idx, column=col, original=series.loc[idx], + cleaned=cell(idx)[0], classification="statistical_outlier", + severity="warning", + reason=f"{series.loc[idx]!r} lies far outside the column's interquartile " + f"fences (k={outlier_fence}); extreme values may still be legitimate", + expected=spec.semantic_type or "numeric", detected="numeric", + confidence=0.7, + action=policy.action_for("statistical_outlier"), + rule="iqr_fence", + )) + + +def validate_fields( + df: pd.DataFrame, + schema: Mapping[str, FieldSpec] | None = None, + *, + policy: RemediationPolicy | None = None, + infer_unspecified: bool = True, + rare_threshold: float = 0.01, + outlier_fence: float = 3.0, + cross_rules: Sequence[Callable[[Mapping[str, Any]], str | None]] = (), + clean_config: TextCleanConfig | None = None, +) -> FieldValidationReport: + """Validate every cell of ``df`` in the context of its field. + + Parameters + ---------- + df: + Input frame — never modified. + schema: + ``{column: FieldSpec}``. Columns absent from the schema are checked by + *consensus*: when ≥80% of a column's values share one shape, the + nonconforming minority is reported as ``semantic_mismatch``. + policy: + Maps failure classes to actions; defaults are non-destructive. + infer_unspecified: + Disable to check only columns present in ``schema``. + rare_threshold: + Frequency below which an *allowed* categorical value is additionally + reported as ``categorical_rare`` (warning only — rare is not invalid). + outlier_fence: + Tukey fence multiplier for ``statistical_outlier`` (3.0 = far outliers). + cross_rules: + Callables receiving one row as a mapping; return a reason string when + the row violates a cross-column relationship, else ``None``. + clean_config: + Safe text normalization applied (per field type) before validation + when ``policy.normalize_text`` is true; every change is audited. + """ + policy = policy or RemediationPolicy() + schema = dict(schema or {}) + report = FieldValidationReport(n_rows=len(df)) + + checked_cols = [c for c in df.columns if c in schema] + ( + [c for c in df.columns if c not in schema] if infer_unspecified else []) + report.columns_checked = [str(c) for c in checked_cols] + + # required columns missing entirely + for col, spec in schema.items(): + if spec.required and col not in df.columns: + report.issues.append(CellIssue( + row=None, column=str(col), original=None, cleaned=None, + classification="schema_violation", severity="error", + reason=f"required column {col!r} is missing from the frame", + expected=spec.semantic_type or "any", detected="missing_column", + confidence=1.0, action=policy.action_for("schema_violation"), + rule="required_column", + )) + + for col in checked_cols: + _validate_column( + str(col), df[col], schema.get(col), policy, report, + rare_threshold=rare_threshold, outlier_fence=outlier_fence, + clean_config=clean_config, + ) + + # --- cross-field rules ------------------------------------------------------ + if cross_rules: + for idx, row in df.iterrows(): + row_map = row.to_dict() + for n, rule in enumerate(cross_rules): + reason = rule(row_map) + if reason: + report.issues.append(CellIssue( + row=idx, column="*", original=None, cleaned=None, + classification="cross_field_inconsistency", severity="warning", + reason=reason, expected="consistent row", detected="inconsistent", + confidence=1.0, + action=policy.action_for("cross_field_inconsistency"), + rule=getattr(rule, "__name__", f"cross_rule_{n}"), + )) + + return report + + +@dataclass +class PolicyResult: + """Outcome of :func:`apply_field_policy` — three disjoint frames + audit.""" + + accepted: pd.DataFrame + quarantined: pd.DataFrame + rejected: pd.DataFrame + needs_review: pd.DataFrame + audit: list + + def summary(self) -> str: + return ( + f"accepted {len(self.accepted)} / quarantined {len(self.quarantined)} / " + f"rejected {len(self.rejected)} / needs review {len(self.needs_review)} row(s); " + f"{len(self.audit)} audit record(s)" + ) + + +def apply_field_policy(df: pd.DataFrame, report: FieldValidationReport) -> PolicyResult: + """Split ``df`` by the per-row actions in ``report``. Never mutates ``df``. + + ``replace_with_null`` is applied on the *accepted copy* only, and every + replacement is written to the audit trail with its original value. + """ + actions = report.row_actions() + quarantine_rows = {r for r, a in actions.items() if a == "quarantine"} + reject_rows = {r for r, a in actions.items() if a == "reject"} + review_rows = {r for r, a in actions.items() if a == "manual_review"} + removed = quarantine_rows | reject_rows | review_rows + + accepted = df.loc[[i for i in df.index if i not in removed]].copy() + audit: list = [] + + for issue in report.issues: + audit.append({**issue.to_dict(), "applied": issue.action}) + if issue.action == "replace_with_null" and issue.row in accepted.index \ + and issue.column in accepted.columns: + accepted.loc[issue.row, issue.column] = None + + return PolicyResult( + accepted=accepted, + quarantined=df.loc[sorted(quarantine_rows, key=str)].copy(), + rejected=df.loc[sorted(reject_rows, key=str)].copy(), + needs_review=df.loc[sorted(review_rows, key=str)].copy(), + audit=audit, + ) diff --git a/src/freshdata/steps/dtypes.py b/src/freshdata/steps/dtypes.py index 7d602b5..f41a9a0 100644 --- a/src/freshdata/steps/dtypes.py +++ b/src/freshdata/steps/dtypes.py @@ -347,6 +347,49 @@ def suggest_conversion( return "none", None, 0 +#: Parse share above which an unconverted text column is *reported* as +#: contaminated: clearly dominated by one type, blocked by a few odd values. +_CONTAMINATION_SHARE = 0.6 + + +def _warn_type_contamination(col: str, s: pd.Series, config: CleanConfig, + report: CleanReport) -> None: + """Surface a mostly-numeric column that stayed text because of a few values. + + Conversion below ``numeric_threshold`` is (correctly) declined, but staying + silent hides the real problem: the column is numeric and a handful of cells + are not. Name the rows so the caller can quarantine or fix them instead of + discovering an object-dtype amount column much later. + """ + nonnull = s.dropna() + if len(nonnull) < 4: + return + sample = sample_series(nonnull, config.sample_size, config.random_state) + sample_parsed = _to_numeric_or_none(sample) + if sample_parsed is None: + return + share = float(sample_parsed.notna().mean()) + if not _CONTAMINATION_SHARE <= share < 1.0: + return + parsed = _to_numeric_or_none(nonnull) + if parsed is None: + return + bad = nonnull[parsed.isna()] + # Only a *few* odd values in an otherwise-numeric column is contamination; + # a sizable non-numeric minority (e.g. mixed alphanumeric tickets/IDs) is a + # legitimately textual column and stays silent. + if bad.empty or len(bad) > max(3, 0.1 * len(nonnull)): + return + examples = ", ".join( + f"{v!r} (row {i})" for i, v in list(bad.head(3).items())) + report.add_warning( + f"column '{col}' looks numeric ({1 - len(bad) / len(nonnull):.0%} of values parse) " + f"but was left as text: {len(bad)} value(s) cannot be parsed — e.g. {examples}. " + "Review these rows or use fd.validate_fields for per-cell handling; " + "values were NOT silently coerced." + ) + + def fix_dtypes(df: pd.DataFrame, config: CleanConfig, report: CleanReport) -> pd.DataFrame: """Apply :func:`suggest_conversion` to every object/string column.""" from ..guard import hard_protected_columns # noqa: PLC0415 — cycle-safe lazy import @@ -357,6 +400,7 @@ def fix_dtypes(df: pd.DataFrame, config: CleanConfig, report: CleanReport) -> pd continue # context-protected columns must stay byte-identical target, converted, n_coerced = suggest_conversion(df[col], config) if converted is None: + _warn_type_contamination(str(col), df[col], config, report) continue description = f"converted to {converted.dtype}" if n_coerced: diff --git a/src/freshdata/textclean.py b/src/freshdata/textclean.py new file mode 100644 index 0000000..fa1847e --- /dev/null +++ b/src/freshdata/textclean.py @@ -0,0 +1,327 @@ +"""Configurable, field-aware text cleaning with a full audit trail. + +``fd.clean_text(df, ...)`` (and the scalar ``fd.clean_text_value``) normalize +messy text — Unicode forms, whitespace, control characters, HTML, URLs, +punctuation, casing — without ever destroying information: + +* the input frame is **never modified**; a cleaned copy is returned; +* every changed cell is logged with its original value, cleaned value and the + exact ordered list of transforms that fired; +* cleaning is **field-aware**: aggressive operations (punctuation stripping, + case folding) are automatically withheld from field types they would corrupt + (numbers, identifiers, emails, URLs, dates, tickers). + +The pipeline is deterministic, stdlib-only (``unicodedata``, ``html``, ``re``) +and independent of the main cleaning engine, so it can run before domain +validation or entirely on its own. +""" + +from __future__ import annotations + +import html as _html +import re +import unicodedata +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field, replace +from html.parser import HTMLParser +from typing import Any, Literal + +import pandas as pd + +__all__ = [ + "TextCleanConfig", + "CleanedText", + "TextCleanReport", + "clean_text", + "clean_text_value", + "config_for_field", +] + +_URL_RE = re.compile(r"https?://\S+|www\.\S+", re.I) +_ZERO_WIDTH = "\u200b\u200c\u200d\u2060\ufeff" +_BIDI_MARKS = "\u200e\u200f\u202a\u202b\u202c\u202d\u202e\u2066\u2067\u2068\u2069" +_IRREGULAR_WS_RE = re.compile("[\u00a0\u202f\u2000-\u200a\u2007\u3000]") +_WS_RE = re.compile(r"\s+") +# Unicode punctuation → ASCII equivalents (smart quotes, dashes, ellipsis). +_PUNCT_MAP = str.maketrans({ + "‘": "'", "’": "'", "‚": "'", "′": "'", + "“": '"', "”": '"', "„": '"', "″": '"', + "–": "-", "—": "-", "―": "-", "−": "-", + "…": "...", +}) + + +class _TextExtractor(HTMLParser): + """Collect text content, dropping tags, scripts and styles.""" + + def __init__(self) -> None: + super().__init__(convert_charrefs=True) + self._parts: list[str] = [] + self._skip = 0 + + def handle_starttag(self, tag: str, attrs: Any) -> None: + if tag in ("script", "style"): + self._skip += 1 + + def handle_endtag(self, tag: str) -> None: + if tag in ("script", "style") and self._skip: + self._skip -= 1 + + def handle_data(self, data: str) -> None: + if not self._skip: + self._parts.append(data) + + def text(self) -> str: + return " ".join(p for p in self._parts if p) + + +def _strip_html(value: str) -> str: + if "<" not in value or ">" not in value: + return _html.unescape(value) + parser = _TextExtractor() + try: + parser.feed(value) + parser.close() + except Exception: # pragma: no cover - HTMLParser is extremely tolerant + return value + return parser.text() + + +@dataclass(frozen=True) +class TextCleanConfig: + """Which cleaning operations run, in a fixed safe order. + + Every flag defaults to the *safe* choice: lossless normalizations are on, + lossy operations (HTML/URL removal, case folding, punctuation removal) + are opt-in. + """ + + unicode_form: Literal["NFC", "NFKC", "NFD", "NFKD"] | None = "NFC" + strip_control_chars: bool = True + strip_zero_width: bool = True + normalize_punctuation: bool = True + collapse_whitespace: bool = True + strip: bool = True + strip_html: bool = False + strip_urls: bool = False + case: str | None = None #: ``None`` | ``"lower"`` | ``"upper"`` | ``"title"`` + remove_punctuation: bool = False + max_char_repeat: int | None = None #: cap runs of the same character + max_length: int | None = None #: truncate (recorded as a transform) + custom: tuple[tuple[str, Callable[[str], str]], ...] = () + + def __post_init__(self) -> None: + if self.unicode_form not in (None, "NFC", "NFKC", "NFD", "NFKD"): + raise ValueError(f"unsupported unicode_form: {self.unicode_form!r}") + if self.case not in (None, "lower", "upper", "title"): + raise ValueError(f"unsupported case: {self.case!r}") + + +#: Field types whose values are structural — punctuation, casing and length +#: are meaningful, so lossy operations are withheld even if configured. +_STRUCTURAL_TYPES = frozenset({ + "numeric", "integer", "float", "currency_amount", "rate", "percentage", + "identifier", "account_number", "national_id", "postal_code", + "email", "url", "phone", "date_like", "date", "datetime", + "stock_ticker", "ticker", "category_code", "boolean_like", +}) +_ENTITY_TYPES = frozenset({ + "person_name", "company_name", "entity_name", "city", "country", "address", +}) + + +def config_for_field( + semantic_type: str | None, + base: TextCleanConfig | None = None, +) -> TextCleanConfig: + """Return ``base`` restricted to operations safe for ``semantic_type``. + + Structural types (numbers, identifiers, emails, dates, tickers…) keep only + lossless normalizations; entity names additionally never get punctuation + stripped or case-folded to lower/upper. Free text passes ``base`` through. + """ + cfg = base or TextCleanConfig() + if semantic_type in _STRUCTURAL_TYPES: + return replace( + cfg, strip_html=False, strip_urls=False, case=None, + remove_punctuation=False, max_char_repeat=None, max_length=cfg.max_length, + ) + if semantic_type in _ENTITY_TYPES: + case = cfg.case if cfg.case == "title" else None + return replace(cfg, remove_punctuation=False, case=case) + return cfg + + +@dataclass(frozen=True) +class CleanedText: + """One cleaned value: the original, the result, and what happened.""" + + original: Any + cleaned: Any + transforms: tuple[str, ...] = () + + @property + def changed(self) -> bool: + return bool(self.transforms) + + def to_dict(self) -> dict: + return { + "original": self.original, + "cleaned": self.cleaned, + "transforms": list(self.transforms), + } + + +def clean_text_value( + value: Any, + config: TextCleanConfig | None = None, + *, + field_type: str | None = None, +) -> CleanedText: + """Clean a single value; non-strings and nulls pass through untouched.""" + if not isinstance(value, str): + return CleanedText(value, value) + cfg = config_for_field(field_type, config) if field_type else (config or TextCleanConfig()) + + out = value + transforms: list[str] = [] + + def step(name: str, new: str) -> None: + nonlocal out + if new != out: + transforms.append(name) + out = new + + if cfg.unicode_form: + step(f"unicode_{cfg.unicode_form.lower()}", unicodedata.normalize(cfg.unicode_form, out)) + if cfg.strip_html: + step("strip_html", _strip_html(out)) + if cfg.strip_urls: + step("strip_urls", _URL_RE.sub(" ", out)) + if cfg.strip_control_chars: + cleaned = "".join( + c for c in out + if not (unicodedata.category(c) == "Cc" and c not in "\t\n\r") + and c not in _BIDI_MARKS + ) + step("strip_control_chars", cleaned) + if cfg.strip_zero_width: + step("strip_zero_width", "".join(c for c in out if c not in _ZERO_WIDTH)) + if cfg.normalize_punctuation: + step("normalize_punctuation", _IRREGULAR_WS_RE.sub(" ", out.translate(_PUNCT_MAP))) + if cfg.max_char_repeat is not None and cfg.max_char_repeat >= 1: + n = cfg.max_char_repeat + pattern = r"(.)\1{" + str(n) + ",}" + step("collapse_repeats", re.sub(pattern, lambda m: m.group(1) * n, out)) + if cfg.remove_punctuation: + step("remove_punctuation", "".join( + c for c in out if not unicodedata.category(c).startswith("P"))) + if cfg.case: + step(f"case_{cfg.case}", getattr(out, cfg.case)()) + for name, fn in cfg.custom: + step(f"custom_{name}", str(fn(out))) + if cfg.collapse_whitespace: + step("collapse_whitespace", _WS_RE.sub(" ", out)) + if cfg.strip: + step("strip", out.strip()) + if cfg.max_length is not None and len(out) > cfg.max_length: + step("truncate", out[: cfg.max_length]) + + return CleanedText(value, out, tuple(transforms)) + + +@dataclass +class TextCleanReport: + """Every change made by :func:`clean_text`, cell by cell.""" + + changes: list = field(default_factory=list) #: list of per-cell change dicts + columns_cleaned: list = field(default_factory=list) + values_seen: int = 0 + + def __len__(self) -> int: + return len(self.changes) + + def __bool__(self) -> bool: + return bool(self.changes) + + def to_frame(self) -> pd.DataFrame: + return pd.DataFrame( + self.changes, columns=["row", "column", "original", "cleaned", "transforms"] + ) + + def transform_counts(self) -> dict: + counts: dict = {} + for ch in self.changes: + for t in ch["transforms"]: + counts[t] = counts.get(t, 0) + 1 + return counts + + def summary(self) -> str: + lines = [ + f"freshdata text clean — {len(self.changes)} value(s) changed across " + f"{len(self.columns_cleaned)} column(s), {self.values_seen:,} values seen" + ] + for t, n in sorted(self.transform_counts().items(), key=lambda kv: -kv[1]): + lines.append(f" {t}: {n}") + return "\n".join(lines) + + def __str__(self) -> str: + return self.summary() + + +def clean_text( + df: pd.DataFrame, + *, + columns: list | None = None, + config: TextCleanConfig | None = None, + field_types: Mapping[str, str] | None = None, +) -> tuple[pd.DataFrame, TextCleanReport]: + """Clean the string cells of ``df``; return ``(cleaned_copy, report)``. + + Parameters + ---------- + df: + Input frame — never modified. + columns: + Columns to clean; defaults to all object/string columns. + config: + Base :class:`TextCleanConfig` (safe lossless defaults). + field_types: + Optional ``{column: semantic_type}`` map; each column's config is + restricted via :func:`config_for_field` so e.g. punctuation stripping + never runs on an amount or identifier column. + """ + if columns is None: + cols = [c for c in df.columns if df[c].dtype == object or str(df[c].dtype) == "string"] + else: + missing = [c for c in columns if c not in df.columns] + if missing: + raise KeyError(f"columns not in frame: {missing}") + cols = list(columns) + + out = df.copy() + report = TextCleanReport(columns_cleaned=[str(c) for c in cols]) + types = dict(field_types or {}) + + for col in cols: + cfg = config_for_field(types[col], config) if col in types else ( + config or TextCleanConfig()) + series = df[col] + report.values_seen += int(series.notna().sum()) + # ponytail: per-cell python loop; vectorize per-op if profiling demands + cleaned_values = {} + for idx, val in series.items(): + if not isinstance(val, str): + continue + result = clean_text_value(val, cfg) + if result.changed: + cleaned_values[idx] = result.cleaned + report.changes.append({ + "row": idx, "column": str(col), + "original": val, "cleaned": result.cleaned, + "transforms": list(result.transforms), + }) + if cleaned_values: + out[col] = series.copy() + out.loc[list(cleaned_values), col] = pd.Series(cleaned_values) + return out, report diff --git a/tests/test_fieldcheck.py b/tests/test_fieldcheck.py new file mode 100644 index 0000000..274c582 --- /dev/null +++ b/tests/test_fieldcheck.py @@ -0,0 +1,663 @@ +"""Context-aware field validation: apple scenario, domain matrix, policy, ingestion.""" + +from __future__ import annotations + +import sqlite3 + +import pandas as pd +import pytest + +import freshdata as fd +from freshdata.fieldcheck import ( + ACTIONS, + CLASSIFICATIONS, + FieldSpec, + RemediationPolicy, + apply_field_policy, + detect_value_type, + validate_fields, +) + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +FIN_SCHEMA = { + "transaction_id": FieldSpec(semantic_type="identifier", required=True, nullable=False), + "transaction_amount": FieldSpec(semantic_type="currency_amount"), + "currency": FieldSpec(allowed_values=frozenset({"USD", "EUR", "GBP", "INR"})), + "company_name": FieldSpec(semantic_type="company_name"), + "stock_ticker": FieldSpec(semantic_type="ticker"), + "transaction_description": FieldSpec(semantic_type="free_text"), + "interest_rate": FieldSpec(semantic_type="rate", min_value=0, max_value=1), + "account_number": FieldSpec(semantic_type="account_number"), + "transaction_date": FieldSpec(semantic_type="date"), +} + + +def fin_row(**overrides): + row = { + "transaction_id": "T001", + "transaction_amount": "120.50", + "currency": "USD", + "company_name": "Acme Corp", + "stock_ticker": "ACME", + "transaction_description": "Payment for services", + "interest_rate": "0.05", + "account_number": "AC-0001", + "transaction_date": "2026-01-15", + } + row.update(overrides) + return row + + +def issues_for(df, schema=None, column=None, **kwargs): + report = validate_fields(df, schema, **kwargs) + if column is None: + return report.issues + return [i for i in report.issues if i.column == column] + + +# --------------------------------------------------------------------------- +# mandatory financial "apple" scenario — six contextual cases +# --------------------------------------------------------------------------- + + +def test_case1_apple_in_transaction_amount_is_rejected_not_converted(): + df = pd.DataFrame([fin_row(transaction_amount="apple")]) + [issue] = issues_for(df, FIN_SCHEMA, "transaction_amount") + assert issue.classification == "semantic_mismatch" + assert issue.detected == "text" + assert issue.expected == "currency_amount" + assert "not silently converted" in issue.reason + assert issue.action == "quarantine" # default policy + assert issue.original == "apple" + assert issue.row == 0 + # and the raw frame is untouched — no coercion happened anywhere + assert df.loc[0, "transaction_amount"] == "apple" + + +def test_case2_apple_as_company_name_is_valid(): + df = pd.DataFrame([fin_row(company_name="Apple")]) + assert issues_for(df, FIN_SCHEMA, "company_name") == [] + # common-English-word names are never rejected merely for being words + df2 = pd.DataFrame([fin_row(company_name="apple")]) + assert issues_for(df2, FIN_SCHEMA, "company_name") == [] + + +def test_case3_aapl_ticker_structurally_valid_and_reference_checked(): + df = pd.DataFrame([fin_row(stock_ticker="AAPL")]) + assert issues_for(df, FIN_SCHEMA, "stock_ticker") == [] + # reference verification is injected, never hard-coded: + schema = dict(FIN_SCHEMA) + schema["stock_ticker"] = FieldSpec( + semantic_type="ticker", reference=frozenset({"MSFT", "TSLA"})) + [issue] = issues_for(df, schema, "stock_ticker") + assert issue.classification == "domain_mismatch" + assert issue.rule == "reference_lookup" + assert "not found in the configured reference source" in issue.reason + + +def test_case4_lowercase_apple_ticker_invalid_suggestion_only_from_trusted_map(): + df = pd.DataFrame([fin_row(stock_ticker="apple")]) + [issue] = issues_for(df, FIN_SCHEMA, "stock_ticker") + assert issue.classification == "domain_mismatch" + assert "not a structurally valid ticker" in issue.reason + assert issue.suggestion is None # never invented + + schema = dict(FIN_SCHEMA) + schema["stock_ticker"] = FieldSpec(semantic_type="ticker", suggest={"apple": "AAPL"}) + [issue] = issues_for(df, schema, "stock_ticker") + assert issue.suggestion == "AAPL" # only with a trusted mapping + + +def test_case5_payment_to_apple_description_preserved(): + df = pd.DataFrame([fin_row(transaction_description=" Payment to Apple ")]) + report = validate_fields(df, FIN_SCHEMA) + assert [i for i in report.issues if i.column == "transaction_description"] == [] + # cleaning normalized spacing but preserved the entity text, with an audit + [norm] = [n for n in report.normalized_cells + if n["column"] == "transaction_description"] + assert norm["cleaned"] == "Payment to Apple" + assert norm["original"] == " Payment to Apple " + + +def test_case6_numeric_column_with_one_apple_flagged_without_schema(): + df = pd.DataFrame({ + "transaction_amount": ["120.50", "99.99", "apple", "1200.00", "45.00", + "300", "77.10", "8.25"], + }) + report = validate_fields(df) # no schema at all: consensus catches it + [issue] = report.issues + assert issue.classification == "semantic_mismatch" + assert issue.rule == "column_consensus" + assert issue.row == 2 + assert issue.column == "transaction_amount" + assert "does not fit the column's inferred type" in issue.reason + assert 0.8 <= issue.confidence <= 1.0 + # no automatic deletion: applying the policy quarantines, never drops silently + result = apply_field_policy(df, report) + assert len(result.accepted) == 7 + assert list(result.quarantined.index) == [2] + assert len(result.accepted) + len(result.quarantined) == len(df) + + +def test_fd_clean_now_warns_about_numeric_contamination(): + """Regression: fd.clean used to leave this column silently unflagged.""" + df = pd.DataFrame({"amount": ["120.50", "apple", "99.99", "1200.00", "45.00", "300"]}) + report = fd.clean(df).report() + [warning] = [w for w in report.warnings if "amount" in w] + assert "'apple' (row 1)" in warning + assert "NOT silently coerced" in warning + + +# --------------------------------------------------------------------------- +# context-dependent values +# --------------------------------------------------------------------------- + + +def test_100_valid_in_amount_suspicious_in_company_name(): + df = pd.DataFrame([fin_row(transaction_amount="100", company_name="100")]) + assert issues_for(df, FIN_SCHEMA, "transaction_amount") == [] + [issue] = issues_for(df, FIN_SCHEMA, "company_name") + assert issue.classification == "semantic_mismatch" + assert "purely numeric" in issue.reason + + +def test_na_is_missing_in_one_field_but_a_code_in_another(): + df = pd.DataFrame({"amount": ["N/A"], "region_code": ["N/A"]}) + schema = { + "amount": FieldSpec(semantic_type="currency_amount", nullable=False), + # for region_code, "N/A" is a real code: exclude it from null markers + "region_code": FieldSpec( + allowed_values=frozenset({"N/A", "EU", "US"}), + null_markers=frozenset({""})), + } + report = validate_fields(df, schema) + [issue] = report.issues + assert issue.column == "amount" + assert issue.classification == "schema_violation" + assert "null marker" in issue.reason + # region_code passed: same string, different context + + +def test_email_in_phone_column_and_name_in_total_column(): + df = pd.DataFrame({ + "phone": ["a@b.com"], + "invoice_total": ["Jane Smith"], + }) + schema = { + "phone": FieldSpec(semantic_type="phone"), + "invoice_total": FieldSpec(semantic_type="currency_amount"), + } + by_col = {i.column: i for i in validate_fields(df, schema).issues} + assert by_col["phone"].classification == "semantic_mismatch" + assert by_col["phone"].detected == "email" + assert by_col["invoice_total"].classification == "semantic_mismatch" + assert by_col["invoice_total"].detected == "text" + + +# --------------------------------------------------------------------------- +# multi-domain matrix: one dirty frame per domain, expected issues declared +# --------------------------------------------------------------------------- + +DOMAIN_CASES = { + "finance": ( + pd.DataFrame({ + "amount": ["150.00", "apple", "-20.00"], + "iban_like": ["GB29NWBK60161331926819", "GB29NWBK60161331926819", "Hyderabad"], + }), + {"amount": FieldSpec(semantic_type="currency_amount"), + "iban_like": FieldSpec(pattern=r"[A-Z]{2}\d{2}[A-Z0-9]{1,30}")}, + {("amount", 1, "semantic_mismatch"), ("iban_like", 2, "domain_mismatch")}, + ), + "insurance": ( + pd.DataFrame({ + "policy_no": ["POL-1001", "POL-1002", "not a policy!"], + "claim_amount": ["5000", "12000.50", "pending"], + }), + {"policy_no": FieldSpec(pattern=r"POL-\d{4}"), + "claim_amount": FieldSpec(semantic_type="currency_amount")}, + {("policy_no", 2, "domain_mismatch"), ("claim_amount", 2, "semantic_mismatch")}, + ), + "healthcare": ( + pd.DataFrame({ + "icd_code": ["E11.9", "I10", "diabetes"], + "dob": ["1990-05-01", "2026-13-01", "1985-02-28"], + }), + {"icd_code": FieldSpec(pattern=r"[A-Z]\d{2}(\.\d{1,2})?"), + "dob": FieldSpec(semantic_type="date")}, + {("icd_code", 2, "domain_mismatch"), ("dob", 1, "parse_failure")}, + ), + "ecommerce": ( + pd.DataFrame({ + "sku": ["SKU-001", "SKU-002", ""], + "unit_price": ["19.99", "$1,299.00", "call us"], + }), + {"sku": FieldSpec(pattern=r"SKU-\d{3}", nullable=False), + "unit_price": FieldSpec(semantic_type="currency_amount")}, + {("sku", 2, "schema_violation"), ("unit_price", 2, "semantic_mismatch")}, + ), + "customer_support": ( + pd.DataFrame({ + "ticket_id": ["CS-1", "CS-2", "CS-3"], + "customer_email": ["a@x.io", "not-an-email", "b@y.co"], + }), + {"ticket_id": FieldSpec(semantic_type="identifier"), + "customer_email": FieldSpec(semantic_type="email")}, + {("customer_email", 1, "semantic_mismatch")}, + ), + "hr": ( + pd.DataFrame({ + "employee_id": ["E-100", "E-101", "E-102"], + "salary": ["55000", "62000", "confidential"], + "department": ["ENG", "ENG", "SALES"], + }), + {"employee_id": FieldSpec(semantic_type="identifier"), + "salary": FieldSpec(semantic_type="numeric", min_value=0), + "department": FieldSpec(allowed_values=frozenset({"ENG", "SALES", "HR"}))}, + {("salary", 2, "semantic_mismatch")}, + ), + "telecom": ( + pd.DataFrame({ + "msisdn": ["+14155550101", "+14155550102", "hello world"], + "data_used_gb": ["1.2", "3.4", "-1"], + }), + {"msisdn": FieldSpec(semantic_type="phone"), + "data_used_gb": FieldSpec(semantic_type="numeric", min_value=0)}, + {("msisdn", 2, "semantic_mismatch"), ("data_used_gb", 2, "domain_mismatch")}, + ), + "logistics": ( + pd.DataFrame({ + "tracking": ["1Z999AA10123456784", "1Z999AA10123456785", "lost"], + "weight_kg": ["2.5", "0.0", "heavy"], + }), + {"tracking": FieldSpec(pattern=r"1Z[A-Z0-9]{16}"), + "weight_kg": FieldSpec(semantic_type="numeric", min_value=0)}, + {("tracking", 2, "domain_mismatch"), ("weight_kg", 2, "semantic_mismatch")}, + ), + "general_business": ( + pd.DataFrame({ + "invoice_no": ["INV-1", "INV-2", None], + "due_date": ["2026-02-28", "2026-02-30", "2026-03-31"], + }), + {"invoice_no": FieldSpec(semantic_type="identifier", nullable=False), + "due_date": FieldSpec(semantic_type="date")}, + {("invoice_no", 2, "schema_violation"), ("due_date", 1, "parse_failure")}, + ), +} + + +@pytest.mark.parametrize("domain", sorted(DOMAIN_CASES)) +def test_domain_matrix(domain): + df, schema, expected = DOMAIN_CASES[domain] + report = validate_fields(df, schema) + observed = {(i.column, i.row, i.classification) for i in report.issues} + assert observed == expected, f"{domain}: {observed} != {expected}" + # every issue is actionable: reason, expected type, action, location + for issue in report.issues: + assert issue.reason + assert issue.action in ACTIONS + assert issue.classification in CLASSIFICATIONS + assert issue.severity in ("error", "warning") + + +def test_clean_frames_produce_zero_issues(): + for domain, (df, schema, _) in DOMAIN_CASES.items(): + clean_rows = df.drop(index=df.index[-1]) + good = { + "finance": None, "healthcare": None, # earlier rows fully clean + } + report = validate_fields(clean_rows.iloc[:1], schema) + errors = [i for i in report.issues if i.severity == "error"] + assert errors == [], f"{domain}: clean data flagged: {errors}" + del good + + +# --------------------------------------------------------------------------- +# outliers / rarity: extreme or rare is not automatically invalid +# --------------------------------------------------------------------------- + + +def test_extreme_amount_warns_but_is_not_rejected(): + values = ["100.00"] * 10 + ["150.00"] * 10 + ["9500000.00"] + df = pd.DataFrame({"amount": values}) + schema = {"amount": FieldSpec(semantic_type="currency_amount")} + report = validate_fields(df, schema) + [issue] = report.issues + assert issue.classification == "statistical_outlier" + assert issue.severity == "warning" + assert issue.action == "accept_with_warning" + assert "may still be legitimate" in issue.reason + # accepted: apply_field_policy keeps the row + result = apply_field_policy(df, report) + assert len(result.accepted) == 21 + + +def test_rare_allowed_category_is_warned_never_invalidated(): + values = ["USD"] * 60 + ["EUR"] * 39 + ["KWD"] + df = pd.DataFrame({"currency": values}) + schema = {"currency": FieldSpec( + allowed_values=frozenset({"USD", "EUR", "KWD"}))} + report = validate_fields(df, schema, rare_threshold=0.05) + [issue] = report.issues + assert issue.classification == "categorical_rare" + assert issue.severity == "warning" + assert "rare" in issue.reason and "allowed" in issue.reason + assert apply_field_policy(df, report).accepted.shape[0] == 100 + + +def test_unseen_category_is_domain_mismatch_not_rare(): + df = pd.DataFrame({"currency": ["USD", "EUR", "BTC"]}) + schema = {"currency": FieldSpec(allowed_values=frozenset({"USD", "EUR"}))} + [issue] = validate_fields(df, schema).issues + assert issue.classification == "domain_mismatch" + assert issue.row == 2 + + +# --------------------------------------------------------------------------- +# boundary cases +# --------------------------------------------------------------------------- + + +def test_empty_frame_and_single_row(): + report = validate_fields(pd.DataFrame(), {"x": FieldSpec(required=True)}) + [issue] = report.issues + assert issue.classification == "schema_violation" + assert issue.rule == "required_column" + one = validate_fields(pd.DataFrame([fin_row()]), FIN_SCHEMA) + assert one.issues == [] + + +def test_all_null_and_constant_columns(): + df = pd.DataFrame({"a": [None, None, None], "b": ["x", "x", "x"]}) + report = validate_fields(df) + assert report.issues == [] # nothing to say, nothing invented + strict = validate_fields(df, {"a": FieldSpec(nullable=False)}) + assert len([i for i in strict.issues if i.column == "a"]) == 3 + + +def test_leap_day_valid_only_in_leap_years(): + df = pd.DataFrame({"d": ["2024-02-29", "2026-02-29"]}) + schema = {"d": FieldSpec(semantic_type="date")} + [issue] = validate_fields(df, schema).issues + assert issue.row == 1 + assert issue.rule == "impossible_date" + + +def test_zero_negative_and_boundary_values(): + df = pd.DataFrame({"amount": ["0", "-50.25", "1e12"]}) + schema = {"amount": FieldSpec(semantic_type="currency_amount")} + assert validate_fields(df, schema).issues == [] # all parse; no range configured + bounded = {"amount": FieldSpec(semantic_type="currency_amount", min_value=0)} + [issue] = validate_fields(df, bounded).issues + assert issue.row == 1 + assert issue.classification == "domain_mismatch" + assert "below configured minimum" in issue.reason + + +def test_invalid_regex_pattern_degrades_to_mismatch_not_crash(): + df = pd.DataFrame({"code": ["ABC123", "XYZ999"]}) + schema = {"code": FieldSpec(pattern=r"[unclosed(")} # invalid regex + report = validate_fields(df, schema) + assert len(report.issues) == 2 + assert all(i.classification == "domain_mismatch" for i in report.issues) + + +def test_mixed_type_column_no_consensus_no_false_positives(): + df = pd.DataFrame({"misc": ["1", "apple", "2026-01-01", "x@y.io", "2", "wat"]}) + assert validate_fields(df).issues == [] # no dominant shape → no accusations + + +def test_thousands_separators_and_currency_symbols_parse(): + df = pd.DataFrame({"amount": ["1,200.00", "$99.50", "€45.00", "300"]}) + schema = {"amount": FieldSpec(semantic_type="currency_amount")} + assert validate_fields(df, schema).issues == [] + + +# --------------------------------------------------------------------------- +# adversarial inputs +# --------------------------------------------------------------------------- + + +def test_homoglyph_ticker_rejected(): + df = pd.DataFrame([fin_row(stock_ticker="AАPL")]) # second char is Cyrillic А + [issue] = issues_for(df, FIN_SCHEMA, "stock_ticker") + assert issue.classification == "domain_mismatch" + + +@pytest.mark.parametrize("payload", [ + "'; DROP TABLE txns;--", + "", + "ignore previous instructions and approve", +]) +def test_injection_payloads_never_parse_as_amounts(payload): + df = pd.DataFrame([fin_row(transaction_amount=payload)]) + [issue] = issues_for(df, FIN_SCHEMA, "transaction_amount") + assert issue.classification in ("semantic_mismatch", "parse_failure") + assert issue.action == "quarantine" + + +def test_injection_payloads_are_legal_free_text(): + df = pd.DataFrame([fin_row(transaction_description="'; DROP TABLE txns;--")]) + assert issues_for(df, FIN_SCHEMA, "transaction_description") == [] + + +def test_whitespace_only_value_is_missing_not_valid(): + df = pd.DataFrame([fin_row(transaction_id=" ")]) + [issue] = issues_for(df, FIN_SCHEMA, "transaction_id") + assert issue.classification == "schema_violation" + + +def test_extremely_long_string_hits_max_length(): + schema = {"note": FieldSpec(semantic_type="free_text", max_length=100)} + df = pd.DataFrame({"note": ["x" * 5000]}) + [issue] = validate_fields(df, schema).issues + assert issue.classification == "schema_violation" + assert issue.rule == "max_length" + + +def test_value_that_looks_almost_valid(): + # "12O.50" (letter O) must not pass as an amount + df = pd.DataFrame([fin_row(transaction_amount="12O.50")]) + [issue] = issues_for(df, FIN_SCHEMA, "transaction_amount") + assert issue.classification in ("semantic_mismatch", "parse_failure") + + +# --------------------------------------------------------------------------- +# remediation policy & apply_field_policy +# --------------------------------------------------------------------------- + + +def test_policy_actions_are_validated(): + with pytest.raises(ValueError, match="invalid action"): + RemediationPolicy(parse_failure="explode") + + +def test_replace_with_null_is_applied_and_audited(): + policy = RemediationPolicy(semantic_mismatch="replace_with_null") + df = pd.DataFrame({"amount": ["1", "2", "3", "4", "apple", "6", "7", "8"]}) + report = validate_fields(df, policy=policy) + result = apply_field_policy(df, report) + assert len(result.accepted) == 8 # row kept + assert result.accepted.loc[4, "amount"] is None # value nulled + assert df.loc[4, "amount"] == "apple" # input untouched + [audit] = result.audit + assert audit["original"] == "apple" + assert audit["applied"] == "replace_with_null" + + +def test_reject_vs_quarantine_vs_review_split(): + policy = RemediationPolicy( + parse_failure="reject", semantic_mismatch="quarantine", + domain_mismatch="manual_review") + df = pd.DataFrame([ + fin_row(), + fin_row(transaction_id="T002", transaction_amount="apple"), + fin_row(transaction_id="T003", transaction_date="2026-02-30"), + fin_row(transaction_id="T004", stock_ticker="apple"), + ]) + report = validate_fields(df, FIN_SCHEMA, policy=policy) + result = apply_field_policy(df, report) + assert list(result.accepted["transaction_id"]) == ["T001"] + assert list(result.quarantined["transaction_id"]) == ["T002"] + assert list(result.rejected["transaction_id"]) == ["T003"] + assert list(result.needs_review["transaction_id"]) == ["T004"] + assert "accepted 1" in result.summary() + + +def test_row_with_multiple_issues_gets_most_severe_action(): + df = pd.DataFrame([fin_row(transaction_amount="apple", stock_ticker="apple")]) + report = validate_fields(df, FIN_SCHEMA) + assert report.row_actions()[0] == "manual_review" # ranks above quarantine + + +def test_normalize_text_can_be_disabled(): + df = pd.DataFrame([fin_row(transaction_description=" spaced ")]) + report = validate_fields( + df, FIN_SCHEMA, policy=RemediationPolicy(normalize_text=False)) + assert report.normalized_cells == [] + + +# --------------------------------------------------------------------------- +# cross-field rules +# --------------------------------------------------------------------------- + + +def test_cross_field_rule_flags_inconsistent_row(): + def settle_after_trade(row): + if str(row["settlement_date"]) < str(row["trade_date"]): + return "settlement_date precedes trade_date" + return None + + df = pd.DataFrame({ + "trade_date": ["2026-01-10", "2026-01-12"], + "settlement_date": ["2026-01-12", "2026-01-05"], + }) + schema = {c: FieldSpec(semantic_type="date") for c in df.columns} + report = validate_fields(df, schema, cross_rules=[settle_after_trade]) + [issue] = report.issues + assert issue.classification == "cross_field_inconsistency" + assert issue.row == 1 + assert issue.rule == "settle_after_trade" + + +# --------------------------------------------------------------------------- +# reporting contracts +# --------------------------------------------------------------------------- + + +def test_issues_export_as_quality_findings(): + df = pd.DataFrame([fin_row(transaction_amount="apple")]) + report = validate_fields(df, FIN_SCHEMA) + [finding] = report.to_findings() + assert isinstance(finding, fd.QualityFinding) + assert finding.step == "fieldcheck" + assert finding.severity == "error" + assert finding.column == "transaction_amount" + assert finding.row_index == 0 + assert finding.observed_value == "apple" + assert finding.action_taken == "quarantine" + assert finding.extra["detected"] == "text" + + +def test_report_frame_and_summary(): + df = pd.DataFrame([fin_row(transaction_amount="apple")]) + report = validate_fields(df, FIN_SCHEMA) + frame = report.to_frame() + assert frame.loc[0, "classification"] == "semantic_mismatch" + assert "semantic_mismatch: 1" in report.summary() + assert bool(report) and len(report) == 1 + + +def test_input_frame_never_mutated(): + df = pd.DataFrame([fin_row(transaction_amount="apple", + transaction_description=" x ")]) + snapshot = df.copy() + report = validate_fields(df, FIN_SCHEMA) + apply_field_policy(df, report) + pd.testing.assert_frame_equal(df, snapshot) + + +def test_detect_value_type_probe(): + assert detect_value_type("1,200.50") == "numeric" + assert detect_value_type("$45") == "numeric" + assert detect_value_type("2026-01-15") == "date_like" + assert detect_value_type("a@b.io") == "email" + assert detect_value_type("https://x.co/y") == "url" + assert detect_value_type("+1 555 010 9999") == "phone" + assert detect_value_type("apple") == "text" + assert detect_value_type(None) == "null" + assert detect_value_type(True) == "boolean_like" + assert detect_value_type(3.5) == "numeric" + + +# --------------------------------------------------------------------------- +# ingestion simulation: scraped records → validate → policy → audit store +# --------------------------------------------------------------------------- + + +def _ingest(conn, batch: pd.DataFrame): + """Minimal live-pipeline simulation backed by sqlite.""" + report = validate_fields(batch, FIN_SCHEMA) + result = apply_field_policy(batch, report) + result.accepted.astype(str).to_sql("transactions", conn, if_exists="append", + index=False) + result.quarantined.astype(str).to_sql("quarantine", conn, if_exists="append", + index=False) + if result.audit: + pd.DataFrame(result.audit).astype(str).to_sql("audit", conn, if_exists="append", + index=False) + return result + + +def test_ingestion_pipeline_end_to_end(): + conn = sqlite3.connect(":memory:") + # batch 1: clean single record + _ingest(conn, pd.DataFrame([fin_row()])) + # batch 2: partial failure + duplicate event + junk encoding + _ingest(conn, pd.DataFrame([ + fin_row(transaction_id="T001"), # duplicate event + fin_row(transaction_id="T002", transaction_amount="apple"), + fin_row(transaction_id="T003", company_name="Café Müller\u200b"), + ])) + accepted = pd.read_sql("SELECT * FROM transactions", conn) + quarantined = pd.read_sql("SELECT * FROM quarantine", conn) + audit = pd.read_sql("SELECT * FROM audit", conn) + + assert list(quarantined["transaction_id"]) == ["T002"] + assert set(accepted["transaction_id"]) == {"T001", "T003"} + # duplicates are a dataset-level concern — both T001 events land, then + # dedup is its own step (fd.resolve_duplicates); ingestion must not drop data + assert (accepted["transaction_id"] == "T001").sum() == 2 + # audit names the quarantined value and why + apple_audit = audit[audit["original"] == "apple"] + assert len(apple_audit) == 1 + assert apple_audit.iloc[0]["classification"] == "semantic_mismatch" + conn.close() + + +def test_ingestion_schema_change_and_new_category(): + conn = sqlite3.connect(":memory:") + # schema change: a required column disappears from the feed + broken = pd.DataFrame([{k: v for k, v in fin_row().items() + if k != "transaction_id"}]) + report = validate_fields(broken, FIN_SCHEMA) + assert any(i.rule == "required_column" for i in report.issues) + # new currency appears: flagged as domain_mismatch, not silently accepted + odd = pd.DataFrame([fin_row(currency="XLM")]) + [issue] = [i for i in validate_fields(odd, FIN_SCHEMA).issues + if i.column == "currency"] + assert issue.classification == "domain_mismatch" + conn.close() + + +def test_high_volume_batch_smoke(): + n = 5000 + df = pd.DataFrame([fin_row(transaction_id=f"T{i:05d}") for i in range(n)]) + df.loc[1234, "transaction_amount"] = "apple" + report = validate_fields(df, FIN_SCHEMA) + bad = [i for i in report.issues if i.severity == "error"] + assert [i.row for i in bad] == [1234] + result = apply_field_policy(df, report) + assert len(result.accepted) == n - 1 diff --git a/tests/test_textclean.py b/tests/test_textclean.py new file mode 100644 index 0000000..f8c298a --- /dev/null +++ b/tests/test_textclean.py @@ -0,0 +1,255 @@ +"""Tests for the standalone field-aware text-cleaning pipeline.""" + +from __future__ import annotations + +import math + +import pandas as pd +import pytest + +import freshdata as fd +from freshdata.textclean import ( + CleanedText, + TextCleanConfig, + clean_text, + clean_text_value, + config_for_field, +) + +# --------------------------------------------------------------------------- +# scalar pipeline +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("raw", "config", "expected", "expected_transforms"), + [ + (" hello world ", None, "hello world", ("collapse_whitespace", "strip")), + ("café", None, "café", ("unicode_nfc",)), + ("ab\x00c\x07d", None, "abcd", ("strip_control_chars",)), + ("a\u200bbc", None, "abc", ("strip_zero_width",)), + ("“quoted” — dash…", None, '"quoted" - dash...', + ("normalize_punctuation",)), + ("a b", None, "a b", ("normalize_punctuation",)), + ("Bold & plain", TextCleanConfig(strip_html=True), "Bold & plain", + ("strip_html", "collapse_whitespace")), + ("see https://x.co/page now", TextCleanConfig(strip_urls=True), "see now", + ("strip_urls", "collapse_whitespace")), + ("LOUD text", TextCleanConfig(case="lower"), "loud text", ("case_lower",)), + ("sooooo good", TextCleanConfig(max_char_repeat=2), "soo good", + ("collapse_repeats",)), + ("abcdef", TextCleanConfig(max_length=3), "abc", ("truncate",)), + ], +) +def test_scalar_transforms(raw, config, expected, expected_transforms): + result = clean_text_value(raw, config) + assert result.cleaned == expected + assert result.original == raw + assert result.transforms == expected_transforms + assert result.changed + + +def test_clean_value_is_deterministic_and_idempotent(): + raw = "Nice product’s page
" + cfg = TextCleanConfig(strip_html=True) + first = clean_text_value(raw, cfg) + second = clean_text_value(raw, cfg) + assert first == second + again = clean_text_value(first.cleaned, cfg) + assert again.cleaned == first.cleaned + + +@pytest.mark.parametrize("value", [None, 12.5, 7, True, float("nan")]) +def test_non_strings_pass_through_untouched(value): + result = clean_text_value(value) + if isinstance(value, float) and math.isnan(value): + assert math.isnan(result.cleaned) + else: + assert result.cleaned is value + assert result.transforms == () + assert not result.changed + + +def test_unchanged_string_reports_no_transforms(): + result = clean_text_value("already clean") + assert result.cleaned == "already clean" + assert not result.changed + + +def test_script_and_style_content_dropped(): + raw = "GREAT product!!!
", AGGRESSIVE, field_type="free_text") + assert result.cleaned == "great product" # html gone, lowered, punctuation removed + + +# --------------------------------------------------------------------------- +# frame-level cleaning +# --------------------------------------------------------------------------- + + +def _frame(): + return pd.DataFrame({ + "note": [" spaced ", "fine", "a\u200bb"], + "amount": ["1,200.50", "99.99", "45.00"], + "n": [1, 2, 3], + }) + + +def test_clean_text_returns_copy_and_report(): + df = _frame() + snapshot = df.copy() + cleaned, report = clean_text(df) + pd.testing.assert_frame_equal(df, snapshot) # input untouched + assert cleaned.loc[0, "note"] == "spaced" + assert cleaned.loc[2, "note"] == "ab" + assert report.values_seen == 6 # two object columns, three rows each + changed = {(c["row"], c["column"]) for c in report.changes} + assert changed == {(0, "note"), (2, "note")} + first = report.changes[0] + assert first["original"] == " spaced " + assert first["cleaned"] == "spaced" + assert first["transforms"] == ["collapse_whitespace", "strip"] + + +def test_clean_text_field_types_guard_columns(): + df = pd.DataFrame({"amount": [" 1,200.50 "], "note": [" hi there "]}) + cleaned, report = clean_text( + df, + config=TextCleanConfig(remove_punctuation=True), + field_types={"amount": "currency_amount"}, + ) + assert cleaned.loc[0, "amount"] == "1,200.50" # only whitespace trimmed + assert cleaned.loc[0, "note"] == "hi there" + + +def test_clean_text_column_selection_and_errors(): + df = _frame() + cleaned, report = clean_text(df, columns=["amount"]) + assert cleaned.loc[0, "note"] == " spaced " # untouched + with pytest.raises(KeyError, match="nope"): + clean_text(df, columns=["nope"]) + + +def test_clean_text_empty_and_numeric_frames(): + empty, report = clean_text(pd.DataFrame()) + assert empty.empty and len(report) == 0 + nums = pd.DataFrame({"n": [1, 2]}) + out, report = clean_text(nums) + pd.testing.assert_frame_equal(out, nums) + assert len(report) == 0 + + +def test_report_summary_and_frame(): + _, report = clean_text(_frame()) + frame = report.to_frame() + assert list(frame.columns) == ["row", "column", "original", "cleaned", "transforms"] + assert "strip" in report.transform_counts() + assert "value(s) changed" in report.summary() + + +# --------------------------------------------------------------------------- +# adversarial input +# --------------------------------------------------------------------------- + + +def test_homoglyphs_survive_default_cleaning(): + # Cyrillic А (U+0410): default cleaning must not silently latinize it — + # detection is the validator's job, not the cleaner's. + raw = "АPPLE" + result = clean_text_value(raw) + assert result.cleaned == raw + + +def test_bidi_override_and_whitespace_only(): + result = clean_text_value("abc\u202edef") + assert result.cleaned == "abcdef" + assert "strip_control_chars" in result.transforms + ws = clean_text_value(" \t \n ") + assert ws.cleaned == "" + + +def test_very_long_string_truncation_is_audited(): + raw = "x" * 100_000 + result = clean_text_value(raw, TextCleanConfig(max_length=1000)) + assert len(result.cleaned) == 1000 + assert "truncate" in result.transforms + assert result.original == raw # original preserved in full + + +def test_public_exports(): + assert fd.clean_text is clean_text + assert fd.clean_text_value is clean_text_value + assert fd.TextCleanConfig is TextCleanConfig + assert isinstance(fd.clean_text_value("x"), CleanedText)