File size: 2,168 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 | """Detector for exact duplicate rows.
Flags rows that are byte-for-byte duplicates of an earlier row. Each duplicate
occurrence after the first is reported (on the first column) as a
``duplicate_row`` issue for review. It is detection-only: removing rows is a
destructive operation the safety constitution forbids auto-applying, so no
repairer is registered - duplicates are surfaced, never silently deleted.
The detector is pure: no LLM calls, no I/O, no side effects.
"""
from __future__ import annotations
from dataforge.detectors.base import Issue, Schema, Severity
from dataforge.table import TableLike, column_names, row_count
class DuplicateRowDetector:
"""Flags exact duplicate rows (detection-only; never auto-deletes).
Example:
>>> import pandas as pd
>>> df = pd.DataFrame({"a": ["1", "2", "1"], "b": ["x", "y", "x"]})
>>> issues = DuplicateRowDetector().detect(df)
>>> issues[0].row
2
"""
def detect(self, df: TableLike, schema: Schema | None = None) -> list[Issue]:
"""Detect exact duplicate rows."""
columns = column_names(df)
if not columns:
return []
first_column = columns[0]
seen: dict[tuple[str, ...], int] = {}
issues: list[Issue] = []
for row_idx in range(row_count(df)):
signature = tuple(str(df.at[row_idx, col]) for col in columns)
original = seen.get(signature)
if original is None:
seen[signature] = row_idx
continue
issues.append(
Issue(
row=row_idx,
column=first_column,
issue_type="duplicate_row",
severity=Severity.REVIEW,
confidence=0.95,
actual=signature[0],
reason=(
f"Row {row_idx} is an exact duplicate of row {original}; "
"review before removing (deletion is never auto-applied)."
),
)
)
return issues
|