fix(quality): count detected duplicates and distinct PII columns in quality-debt dimensions

src/freshdata/quality.py

@@ -171,8 +171,17 @@ def _score_debt(
     miss = report.missing_after / cells
     out["missingness"] = (miss, f"{report.missing_after:,} missing cell(s) remain")
 
-    dup = report.duplicates_removed / max(1, report.rows_before)
-    out["duplicates"] = (dup, f"{report.duplicates_removed:,} duplicate row(s) removed")
+    # Duplicate removal is opt-in (drop_duplicates=False by default), so the
+    # removed count alone is usually 0. Like the other dimensions, score the
+    # debt left in the cleaned output, but never below what was removed.
+    dup_removed = report.duplicates_removed
+    try:
+        dup_remaining = int(cleaned.duplicated().sum())
+    except TypeError:  # unhashable cells (lists, dicts): detection impossible
+        dup_remaining = 0
+    n_dup = max(dup_removed, dup_remaining)
+    out["duplicates"] = (n_dup / max(1, report.rows_before),
+                         f"{n_dup:,} duplicate row(s) detected ({dup_removed:,} removed)")
 
     outl = report.outliers_handled / rows
     out["outlier_spikes"] = (outl, f"{report.outliers_handled:,} outlier(s) flagged")
@@ -204,7 +213,9 @@ def _score_debt(
         from .enterprise.privacy import detect_pii  # noqa: PLC0415
 
         scan = detect_pii(cleaned)
-        n_pii = len(getattr(scan, "entities", []) or [])
+        # One scan entity is recorded per matching cell; count distinct
+        # columns (findings without a column key under "").
+        n_pii = len([col for col in scan.by_column() if col])
         out["pii_risk"] = (min(1.0, n_pii / max(1, report.cols_after)),
                            f"{n_pii} potential PII column(s)")
     except Exception:

tests/test_quality_debt_dimensions.py

@@ -0,0 +1,88 @@
+"""Regression tests for the duplicates and pii_risk quality-debt dimensions."""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import freshdata as fd
+from freshdata.enterprise import privacy
+from freshdata.quality import DebtItem, QualityDebtGate
+
+
+def _item(gate: QualityDebtGate, dimension: str) -> DebtItem:
+    return next(i for i in gate.items if i.dimension == dimension)
+
+
+def _half_duplicated() -> pd.DataFrame:
+    return pd.DataFrame({"a": list(range(50)) * 2, "b": list("xy" * 25) * 2})
+
+
+def test_duplicates_detected_without_removal() -> None:
+    """#264: the default drop_duplicates=False must still register debt."""
+    cleaned, gate = fd.evaluate_quality_debt(
+        _half_duplicated(), debt_policy="fail", ledger=None, verbose=False
+    )
+    assert len(cleaned) == 100  # nothing removed by default
+    dup = _item(gate, "duplicates")
+    assert dup.score == pytest.approx(0.5)
+    assert dup.over
+    assert dup.detail == "50 duplicate row(s) detected (0 removed)"
+    assert gate.status == "fail"
+
+
+def test_duplicates_removed_still_scored() -> None:
+    cleaned, gate = fd.evaluate_quality_debt(
+        _half_duplicated(), debt_policy="fail", ledger=None, verbose=False, drop_duplicates=True
+    )
+    assert len(cleaned) == 50
+    dup = _item(gate, "duplicates")
+    assert dup.score == pytest.approx(0.5)
+    assert dup.over
+    assert dup.detail == "50 duplicate row(s) detected (50 removed)"
+
+
+def test_no_duplicates_scores_zero() -> None:
+    df = pd.DataFrame({"a": range(20), "b": list("xy" * 10)})
+    _, gate = fd.evaluate_quality_debt(df, ledger=None, verbose=False)
+    dup = _item(gate, "duplicates")
+    assert dup.score == 0.0
+    assert not dup.over
+    assert dup.detail == "0 duplicate row(s) detected (0 removed)"
+
+
+def test_pii_risk_counts_columns_not_matches() -> None:
+    """#286: ten matching cells in one column are one PII column."""
+    df = pd.DataFrame({"email": [f"user{i}@x.com" for i in range(10)], "v": range(10)})
+    _, gate = fd.evaluate_quality_debt(df, debt_policy="warn", ledger=None, verbose=False)
+    pii = _item(gate, "pii_risk")
+    assert pii.detail == "1 potential PII column(s)"
+    assert pii.score == pytest.approx(0.5)
+
+
+def _entity(text: str, metadata: dict) -> privacy.PIIEntity:
+    return privacy.PIIEntity(
+        entity_type="EMAIL_ADDRESS",
+        start=0,
+        end=len(text),
+        text=text,
+        score=1.0,
+        source="regex",
+        metadata=metadata,
+    )
+
+
+def test_pii_risk_ignores_findings_without_column(monkeypatch: pytest.MonkeyPatch) -> None:
+    entities = [
+        _entity("x", {"column": "email"}),
+        _entity("y", {"column": "email"}),
+        _entity("z", {}),
+    ]
+    monkeypatch.setattr(
+        privacy, "detect_pii", lambda _df: privacy.PIIScanReport(entities=entities)
+    )
+    df = pd.DataFrame({"email": ["a", "b"], "v": [1, 2], "w": [3, 4], "z": [5, 6]})
+    _, gate = fd.evaluate_quality_debt(df, ledger=None, verbose=False)
+    pii = _item(gate, "pii_risk")
+    assert pii.detail == "1 potential PII column(s)"
+    assert pii.score == pytest.approx(0.25)