fix: positional writes for duplicate row labels, UTC date bounds, honest join-key scoring

src/freshdata/enterprise/join_assist.py

@@ -14,10 +14,12 @@
 
 from __future__ import annotations
 
+import math
 from dataclasses import dataclass, field
 from itertools import product
 from typing import Any
 
+import numpy as np
 import pandas as pd
 
 from ..render import html as H
@@ -32,6 +34,39 @@ def _similarity(a: str, b: str) -> float:
     return round(0.5 * jaro_winkler(a, b) + 0.5 * levenshtein_similarity(a, b), 4)
 
 
+def _is_missing(value: Any) -> bool:
+    """True for None / NaN / NaT / ``pd.NA`` / ``""`` — a key carrying no evidence."""
+    if value is None:
+        return True
+    if isinstance(value, str):
+        return value == ""
+    try:
+        return bool(pd.isna(value))
+    except (TypeError, ValueError):  # array-like cell: not a missing scalar
+        return False
+
+
+def _key_text(value: Any) -> str:
+    """Comparable text for a key value; integral floats render without ``.0``.
+
+    A numeric key column holding a missing value is promoted to float64, so
+    ``101`` and ``101.0`` must compare equal (#273). Only finite, integral
+    floats below 2**53 (exactly representable) are rewritten.
+    """
+    if isinstance(value, (float, np.floating)):
+        f = float(value)
+        if math.isfinite(f) and f.is_integer() and abs(f) < 2**53:
+            return str(int(f))
+    return str(value)
+
+
+def _field_similarity(a: Any, b: Any) -> float:
+    """Similarity of two key cells; a missing value on either side scores 0 (#272)."""
+    if _is_missing(a) or _is_missing(b):
+        return 0.0
+    return _similarity(_key_text(a), _key_text(b))
+
+
 @dataclass(frozen=True)
 class JoinCandidate:
     """One suggested (left, right) match with per-field evidence."""
@@ -161,9 +196,13 @@ def _html_sections(self) -> list[str]:
         return [cards, exact, tbl, dl]
 
 
+def _key_values(series: pd.Series) -> set[str]:
+    return {_key_text(v) for v in series.tolist() if not _is_missing(v)}
+
+
 def _exact_key_overlap(left: pd.DataFrame, right: pd.DataFrame, col: str) -> float:
-    lv = set(left[col].dropna().astype(str))
-    rv = set(right[col].dropna().astype(str))
+    lv = _key_values(left[col])
+    rv = _key_values(right[col])
     if not lv or not rv:
         return 0.0
     return len(lv & rv) / min(len(lv), len(rv))
@@ -203,53 +242,63 @@ def suggest_join_keys(
         raise ValueError("none of the `on` columns are present in both frames")
     blocking = [c for c in (exact_within or []) if c in left.columns and c in right.columns]
 
-    exact_keys = [
-        {"column": c, "overlap": round(_exact_key_overlap(left, right, c), 4),
-         "recommended": _exact_key_overlap(left, right, c) >= 0.95}
-        for c in on
-    ]
-
-    # Build blocks (exact match on blocking columns); one block if none given.
-    def block_key(row: pd.Series) -> str:
-        return "|".join(str(row[c]) for c in blocking) if blocking else "*"
-
-    left_blocks: dict[str, list[Any]] = {}
-    for idx, row in left.iterrows():
-        left_blocks.setdefault(block_key(row), []).append(idx)
-    right_blocks: dict[str, list[Any]] = {}
-    for idx, row in right.iterrows():
-        right_blocks.setdefault(block_key(row), []).append(idx)
-
-    candidates: list[JoinCandidate] = []
+    exact_keys = []
+    for c in on:
+        overlap = _exact_key_overlap(left, right, c)
+        exact_keys.append(
+            {"column": c, "overlap": round(overlap, 4), "recommended": overlap >= 0.95})
+
+    # Rows are addressed by *position* throughout: row labels need not be unique
+    # (e.g. after pd.concat), and a label lookup would return a Series (#231).
+    # The original labels are only used in the reported candidates.
+    def block_keys(frame: pd.DataFrame) -> list[str]:
+        """Blocking key per row (exact match on blocking columns); "*" if none."""
+        if not blocking:
+            return ["*"] * len(frame)
+        cols = [frame[c].tolist() for c in blocking]
+        return ["|".join(_key_text(v) for v in vals) for vals in zip(*cols)]
+
+    left_blocks: dict[str, list[int]] = {}
+    for pos, key in enumerate(block_keys(left)):
+        left_blocks.setdefault(key, []).append(pos)
+    right_blocks: dict[str, list[int]] = {}
+    for pos, key in enumerate(block_keys(right)):
+        right_blocks.setdefault(key, []).append(pos)
+
+    left_vals = {c: left[c].tolist() for c in on}
+    right_vals = {c: right[c].tolist() for c in on}
+
+    candidates: list[tuple[int, JoinCandidate]] = []
     pairs = 0
     truncated = False
-    # best-per-left tracking for ambiguity detection
-    per_left: dict[Any, list[JoinCandidate]] = {}
+    # best-per-left tracking for ambiguity detection (keyed by left position)
+    per_left: dict[int, list[JoinCandidate]] = {}
 
-    for bkey, l_idxs in left_blocks.items():
-        r_idxs = right_blocks.get(bkey, [])
-        for li, ri in product(l_idxs, r_idxs):
+    for bkey, l_pos in left_blocks.items():
+        r_pos = right_blocks.get(bkey, [])
+        for lp, rp in product(l_pos, r_pos):
             if pairs >= max_pairs:
                 truncated = True
                 break
             pairs += 1
             fscores = {
-                c: _similarity(str(left.at[li, c]), str(right.at[ri, c])) for c in on
+                c: _field_similarity(left_vals[c][lp], right_vals[c][rp]) for c in on
             }
             score = round(sum(fscores.values()) / len(on), 4)
             if score >= review_threshold:
-                cand = JoinCandidate(li, ri, score, "review", fscores, bkey)
-                candidates.append(cand)
-                per_left.setdefault(li, []).append(cand)
+                cand = JoinCandidate(
+                    left.index[lp], right.index[rp], score, "review", fscores, bkey)
+                candidates.append((lp, cand))
+                per_left.setdefault(lp, []).append(cand)
         if truncated:
             break
 
     # Resolve status: confident vs ambiguous vs review.
     resolved: list[JoinCandidate] = []
     from dataclasses import replace
 
-    for cand in candidates:
-        siblings = [c for c in per_left[cand.left_index] if c.score >= threshold]
+    for lp, cand in candidates:
+        siblings = [c for c in per_left[lp] if c.score >= threshold]
         if cand.score >= threshold:
             if len([c for c in siblings
                     if abs(c.score - max(s.score for s in siblings)) <= 0.05]) > 1:

src/freshdata/fieldcheck.py

@@ -24,7 +24,7 @@ class to an action. The default policy is non-destructive: nothing is deleted,
 
 import re
 from collections.abc import Callable, Collection, Mapping, Sequence
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any
 
 import pandas as pd
@@ -350,6 +350,15 @@ def _date_bound(value: float | str | None) -> pd.Timestamp | None:
     return None if pd.isna(ts) else ts
 
 
+def _as_utc(ts: pd.Timestamp) -> pd.Timestamp:
+    """``ts`` in UTC; a naive timestamp is taken to already be UTC.
+
+    Lets tz-aware values be compared with naive bounds (and vice versa) instead
+    of raising ``TypeError: Cannot compare tz-naive and tz-aware`` (#233).
+    """
+    return ts.tz_localize("UTC") if ts.tzinfo is None else ts.tz_convert("UTC")
+
+
 def _parse_numeric(s: str) -> float | None:
     if is_plain_number(s):
         return float(str(s).strip().replace(",", ""))
@@ -442,12 +451,12 @@ def issue(classification: str, reason: str, rule: str, *,
                 "date_parse",
             )
         lo, hi = _date_bound(spec.min_value), _date_bound(spec.max_value)
-        if lo is not None and ts < lo:
+        if lo is not None and _as_utc(ts) < _as_utc(lo):
             return issue(
                 "domain_mismatch",
                 f"{col}={ts.date()} is before the configured minimum {lo.date()}",
                 "min_value")
-        if hi is not None and ts > hi:
+        if hi is not None and _as_utc(ts) > _as_utc(hi):
             return issue(
                 "domain_mismatch",
                 f"{col}={ts.date()} is after the configured maximum {hi.date()}",
@@ -594,13 +603,15 @@ def _suspect_rows(series: pd.Series, spec: FieldSpec) -> pd.Index:
 
             with warnings.catch_warnings():
                 warnings.simplefilter("ignore")
-                parsed_dt = pd.to_datetime(strs, errors="coerce")
+                # utc=True: same naive-as-UTC semantics as the per-cell check,
+                # and mixed offsets parse instead of raising
+                parsed_dt = pd.to_datetime(strs, errors="coerce", utc=True)
             fine = parsed_dt.notna()
             lo_d, hi_d = _date_bound(spec.min_value), _date_bound(spec.max_value)
             if lo_d is not None:
-                fine &= parsed_dt >= lo_d
+                fine &= parsed_dt >= _as_utc(lo_d)
             if hi_d is not None:
-                fine &= parsed_dt <= hi_d
+                fine &= parsed_dt <= _as_utc(hi_d)
         except (ValueError, TypeError):  # pragma: no cover - exotic payloads
             fine = pd.Series(False, index=series.index)
         return series.index[must_flag | (checkable & ~fine.fillna(False))]
@@ -700,7 +711,43 @@ def _validate_column(
     clean_config: TextCleanConfig | None,
     sensitive: bool = False,
 ) -> None:
-    """Run every per-column check for one column, appending to ``report``."""
+    """Run every per-column check for one column, appending to ``report``.
+
+    The checks run on a positional (``reset_index(drop=True)``) copy, so a
+    duplicated row label (common after ``pd.concat``) can never turn a cell
+    lookup into a Series (#231). Positions are mapped back to the original
+    row labels in the issues and the normalization audit.
+    """
+    labels = series.index
+    positional = FieldValidationReport(n_rows=report.n_rows)
+    _validate_column_by_position(
+        col, series.reset_index(drop=True), spec, policy, positional,
+        rare_threshold=rare_threshold, outlier_fence=outlier_fence,
+        clean_config=clean_config, sensitive=sensitive,
+    )
+    report.issues.extend(
+        issue if issue.row is None else replace(issue, row=labels[issue.row])
+        for issue in positional.issues
+    )
+    report.normalized_cells.extend(
+        {**cell, "row": labels[cell["row"]]} for cell in positional.normalized_cells
+    )
+    report.inferred_types.update(positional.inferred_types)
+
+
+def _validate_column_by_position(
+    col: str,
+    series: pd.Series,
+    spec: FieldSpec | None,
+    policy: RemediationPolicy,
+    report: FieldValidationReport,
+    *,
+    rare_threshold: float,
+    outlier_fence: float,
+    clean_config: TextCleanConfig | None,
+    sensitive: bool = False,
+) -> None:
+    """Body of :func:`_validate_column`; ``series`` must have a unique index."""
     # --- text normalization (audited, never in-place) -------------------------
     cleaned_col: dict = {}
     transforms_col: dict = {}

src/freshdata/textclean.py

@@ -331,19 +331,25 @@ def clean_text(
         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():
+        # Changes are collected by *position* and written back with iloc: row
+        # labels need not be unique (e.g. after pd.concat), and a label-keyed
+        # write would copy one row's cleaned value over its namesakes (#231).
+        positions: list[int] = []
+        cleaned_values: list[str] = []
+        for pos, (idx, val) in enumerate(series.items()):
             if not isinstance(val, str):
                 continue
             result = clean_text_value(val, cfg)
             if result.changed:
-                cleaned_values[idx] = result.cleaned
+                positions.append(pos)
+                cleaned_values.append(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)
+        if positions:
+            new_col = series.copy()
+            new_col.iloc[positions] = cleaned_values
+            out[col] = new_col
     return out, report