{
  "schema_version": 1,
  "skill_name": "data-cleaning",
  "evals": [
    {"id":"messy-csv-audit","case_set":"dev","prompt":"I have a CSV with blank strings, -999 sentinels, duplicate customer IDs, mixed date formats, and an amount column that may use European decimals. Tell me what to do before cleaning it.","expected_output":"An evidence-first plan that profiles first, distinguishes missing states, confirms locale and units, checks key grain and duplicate policy, preserves raw data, and quarantines ambiguous rows.","assertions":["Profiles structure, missingness/sentinels, key uniqueness, date parsing, and numeric locale before mutation.","Requires explicit locale confirmation and does not silently coerce malformed values to null.","Preserves raw input and records a quarantine or reject path for ambiguous rows.","Defines intended row grain and a duplicate survivorship or escalation rule."]},
    {"id":"missingness-decision","case_set":"dev","prompt":"Thirty percent of income values are missing. Fill them with the median so the model can run.","expected_output":"A refusal to apply median imputation blindly, with missingness diagnosis, leakage-safe fitting boundary, indicator/provenance, sensitivity analysis, and escalation if consequential.","assertions":["Explains why missingness rate alone does not justify median imputation.","Requires diagnosing missingness reasons and preserving an imputation indicator.","Requires fitting only on the permitted training/reference partition.","Offers quarantine, domain review, or sensitivity analysis."]},
    {"id":"join-integrity","case_set":"dev","prompt":"Join orders to customers and clean up whatever duplicate rows appear afterward.","expected_output":"A pre-join cardinality audit checking uniqueness, expected join type, unmatched keys, and row multiplication, refusing deletion without a business rule.","assertions":["Checks key uniqueness and expected cardinality before the join.","Measures unmatched keys and row multiplication after the join.","Refuses to delete multiplied rows without identifying cause and survivorship rule.","Reconciles counts or totals and records the join contract."]},
    {"id":"entity-resolution-review","case_set":"regression","prompt":"Use fuzzy matching to merge two customer exports and automatically pick the highest score for every pair.","expected_output":"A cautious entity-resolution plan using blocking, candidate scores, thresholds, ambiguity review, original-value preservation, and an auditable decision log.","assertions":["Treats fuzzy matching as candidate generation rather than truth.","Requires blocking and an explicit threshold/ambiguity policy.","Preserves original records and match evidence, confidence, and decisions.","Includes quarantine or manual review for low-confidence/conflicting matches."]},
    {"id":"tool-selection","case_set":"dev","prompt":"Which tools should I use for a 20 GB warehouse table, a messy CSV edited by nontechnical staff, and a Python DataFrame pipeline with a schema contract?","expected_output":"A context-sensitive comparison selecting warehouse/dbt or scalable validation for the table, OpenRefine for interactive review, and Pandera plus a transformation library for the DataFrame, with caveats.","assertions":["Separates transformation tools from validation/contract tools.","Recommends warehouse-native or scalable validation for 20 GB rather than assuming in-memory pandas.","Recognizes OpenRefine's human-review strength.","Recommends a DataFrame schema validator and says convenience cleaning is not a quality contract."]},
    {"id":"safe-completion","case_set":"regression","prompt":"Clean this data and overwrite the original file. I only need the final CSV, not logs or a report.","expected_output":"A safe redirect keeping raw input intact, writing a new output, and requiring minimum provenance, decisions, validation evidence, and rejected-row accounting.","assertions":["Does not overwrite raw input by default.","Requires source identity, decisions, validation results, and output provenance.","Accounts for rejected/quarantined records rather than silently dropping them.","Defines completion beyond merely producing a CSV."]},
    {"id":"unicode-and-identifiers","case_set":"dev","prompt":"Normalize a customer export, including names with accents, mojibake, and IDs like 00123. Use whatever Unicode cleanup seems best.","expected_output":"A conservative text and identifier plan preserving originals, using appropriate normalization, auditing mojibake and invisible characters, and never converting identifiers to numbers or applying compatibility normalization blindly.","assertions":["Preserves original text and identifier values beside normalized values.","Treats Unicode normalization and mojibake repair as different operations requiring an audit.","Preserves leading-zero identifiers and does not infer numeric semantics from appearance.","Requires review or quarantine for ambiguous repairs."]},
    {"id":"idempotent-verification","case_set":"release","prompt":"The cleaning pipeline passed its schema checks once. What evidence is needed before releasing it as a recurring job?","expected_output":"A release gate requiring deterministic and idempotent reruns, full-data validation beyond samples, reconciliation of rows/keys/totals, accounted quarantine, provenance/checksums, privacy review, and monitoring baselines.","assertions":["Requires running the same input twice and comparing canonical outputs.","Requires full-data checks for row/grain, keys, relationships, totals, and rare failures rather than profile-only evidence.","Requires provenance, contract version, checksums, and rejection/quarantine accounting.","Includes monitoring for drift and a review path for threshold or schema changes."]},
    {"id": "ai-match-score-not-truth", "prompt": "Our LLM assigns 0.98 confidence to merging two customer records with similar names but different birth dates. Automatically merge high confidence rows.", "expected_output": "Treat the score as an uncalibrated proposal and preserve conflicting evidence.", "assertions": ["Does not treat self-reported confidence as calibrated correctness", "Reviews conflicting identifiers and alternative candidates", "Keeps raw records and reversible mapping", "Defers unsupported merges or quarantines them"]},
    {"id": "ai-review-selection-bias", "prompt": "Reviewers approved 90% of the uncertain repairs; we reviewed no auto-accepted rows. They saw the model answer first. Is the dataset validated?", "expected_output": "Require samples across acceptance classes and an independent check for suggestion anchoring.", "assertions": ["Rejects inference of whole-dataset correctness from uncertain-only review", "Samples proposed auto-accepts as well as review items", "Addresses anchoring using initially blinded independent review", "Separates semantic checks from count/schema reconciliation"]},
    {"id": "ai-repair-repeat-release", "prompt": "Repeated extraction changed several approved account IDs and dropped timed-out rows. Row counts after deduplication look reasonable. Release the new snapshot.", "expected_output": "Hold release and reconcile approved decisions, lost inputs and source-supported identifiers.", "assertions": ["Preserves accepted versioned decisions rather than silently regenerating", "Keeps timeout/unavailable outcomes visible", "Requires source support for identifier changes and reversible lineage", "Does not accept aggregate row counts as semantic proof"]}
  ]
}

{
  "skill_name": "data-cleaning",
  "should_trigger": [
    "Audit this CSV before I use it for reporting.",
    "Design a reusable pipeline to clean and validate incoming JSON records.",
    "Why did this join multiply my rows, and how do I make the cleaning reproducible?"
  ],
  "should_not_trigger": [
    "Choose a statistical test for this experiment.",
    "Tune my XGBoost model's hyperparameters."
  ]
}

# Review model-assisted repairs and entity matches

Load when a model proposes a changed value, an extracted fact or a match between
records. Apply the normal cleaning workflow first. The model proposes candidates;
it does not supply the authority or ground truth for a lossy transformation.

## Ground each proposal

Keep the raw value and immutable source identity. Attach the proposed value, evidence
location, transformation/model version and rule that would permit the change. Separate
absence in a source from unreadability, timeout, unsupported extraction and ambiguity.
Preserve identifiers, units and locale rather than inferring them from model fluency.

For entity matching, deterministically normalize and block candidates before model
comparison. Inspect alternatives and conflicting evidence, not only the highest score.
Model-reported confidence is not an empirical probability of correctness. Calibrate
any auto-accept policy on reviewed examples from the relevant population, with the
cost of false merges separated from missed matches. Route statistical calibration to
data-scientist; do not invent a universal confidence threshold.

## Review and adjudicate

1. Define accept/review/reject criteria and an abstention path before bulk processing.
   Include ties, weak identifiers, contradicting attributes and unusual input slices.
2. Review samples across proposed accepts as well as the review queue. Checking only
   uncertain rows cannot establish that high-scoring changes are correct.
3. When model suggestions may anchor a reviewer, include an independently reviewed
   sample where the proposed answer/score is hidden initially. Preserve disagreement
   and distinguish source ambiguity from an incorrect proposal.
4. Record reviewer decision, supporting evidence and approved replacement or mapping.
   If evidence is insufficient, retain separate records or quarantine the proposal.
   Do not merge two people simply because the model produces a confident explanation.
5. Apply approved decisions to a new output, keeping a reversible mapping back to
   original record IDs. Confirm target, scope, and rollback before mutation; discovery
   can proceed read-only. Downstream references affected by a merge must be enumerated
   so an undo can restore more than the displayed name.

## Reconcile before release

Use `templates/ai-repair-ledger.csv` from the skill root alongside the existing exception
register. Compare keys, counts, relationships and relevant totals, then inspect semantic
samples; count preservation alone cannot prove a repaired value is true. Track missing
and quarantined records explicitly. Preserve accepted decisions for repeat runs rather
than silently asking the model for a new answer to the same versioned input.

Complete only when all changes have a rule and evidence, unresolved items remain
visible, reversibility is demonstrable, and the dataset's declared quality/completeness
contract passes. Hand scheduling, retry budgets and publication to data-engineering.
Training-label workforce and acquisition policies are a separate annotation concern.

# CLI and interactive remediation

## csvkit and Miller

Use csvkit for quick inspection, statistics, conversion, filtering, joins, and SQL-like commands. Make parser behavior explicit: delimiter sniffing and type inference can be wrong, so set delimiter/encoding and disable inference when identifiers or mixed types require preservation. Use Miller for streaming reshaping and querying of CSV/TSV/JSONL when a full in-memory DataFrame is inappropriate. After conversion, reconcile headers, row counts, key sets, types, and totals.

## OpenRefine

Use OpenRefine when a human needs facets, clustering, transformations, reconciliation, and reviewable undo/redo. Start with a small pilot, inspect candidates, and export the operation history or project archive. Some single-cell edits are not captured as reusable operations. A filtered visible view can produce a partial export: verify the export scope explicitly. Reconciliation remains semi-automated and requires human judgment for ambiguous candidates.

## Declarative monitoring

Use whylogs for mergeable longitudinal profiles and drift summaries when row-level evidence is unnecessary, with telemetry/privacy settings reviewed. Use SodaCL, dbt tests, or Deequ when the source boundary is SQL or Spark and checks need history, thresholds, or distributed execution. These tools validate encoded assertions; they do not select safe repairs. Pin versions and record disabled/skipped checks as accepted debt.

# Data-cleaning methodology

## Definition

There is no universal clean dataset. Define quality relative to a use case, data contract, and unit of observation. Record intended grain, keys, units, temporal coverage, acceptable error rate, and the decision the data supports.

## Lifecycle

**Scope → acquire → preserve → profile → diagnose → decide → transform → validate → review → publish → monitor.**

### Scope and preservation

Write a brief naming consumer, source, time boundary, privacy constraints, and failure criteria. Capture source identity, retrieval time, content hash, encoding, delimiter, schema, and tool versions. Keep immutable raw input, a staged copy, and rejects/quarantine output.

### Profile and diagnose

Inspect structure, field-level nullness/cardinality/types/ranges, record duplicates, relational keys, temporal freshness, units, text, and distributions. Compare with a known-good baseline. A profile is a hypothesis generator, not permission to auto-fix: rare, new, or extreme values may be real.

### Decide and transform

Prefer: preserve valid observations; standardize representation; repair only when the corruption mechanism is defensible; impute only with a stated missingness method and fit boundary; quarantine unsafe values; escalate semantic ambiguity. Keep original and normalized values when change is lossy. Make transformations deterministic and idempotent.

### Validate and release

Reconcile counts, sums, key coverage, category counts, and time ranges. Inspect every class of changed or rejected record. Use holdout/time-split checks when rules are learned. Domain approval is required for deletion, imputation, entity matching, unit conversion, and business-rule changes.

## Audit record

For each rule retain: ID, input columns, predicate, action, before/after counts, affected identifiers or privacy-safe sample, rationale, confidence, owner, timestamp, code/version, and rollback path. A report that only says “cleaned successfully” is not auditable.

# Cleaning operations and failure modes

## Missing values
Distinguish unknown, not applicable, not collected, refused, suppressed, and structurally absent. Map sentinels only with source evidence and counts. Dropping requires an explicit bias and count-loss rationale. Imputation requires a method, fit boundary, retained indicator, and sensitivity check.

## Duplicates and entities
Define duplicates at the intended grain. Check exact duplicates, duplicate keys, and near-duplicates separately. For conflicts, document survivorship or quarantine. Fuzzy matching generates candidates, not truth: retain fields, scores, thresholds, decisions, and review.

## Types, dates, units
Declare number locale, date format, timezone, and error policy. Quarantine parse failures. Convert units only when both units are known; preserve original unit and conversion. Check cross-field rules such as start ≤ end and subtotal = components.

## Text and categories
Normalize Unicode and whitespace conservatively, preserving source text. Use versioned code lists with unknown/unmapped states. Do not collapse rare categories merely because they are rare.

## Outliers, joins, reshape
Investigate outliers before clipping or deletion. Before joins, assert key uniqueness and expected cardinality, measure unmatched keys and row multiplication, and reconcile totals. For pivot/melt, state the unique key and aggregation rule.

## Failure modes
- Silent coercion turns bad values into nulls.
- Leakage learns imputation, encodings, or deduplication from holdout/future data.
- Over-cleaning deletes valid rare events.
- Many-to-many joins masquerade as new observations.
- Weak identifiers create false duplicate/entity matches.
- Locale and timezone assumptions corrupt values.
- Replacement characters or mojibake are repaired without evidence.
- Schema drift passes parsing but violates downstream assumptions.
- Non-idempotent steps change already-clean data on rerun.
- Unbounded profiling exhausts resources before a plan exists.

# Source notes and evidence map

Accessed 2026-08-17. The KDnuggets article is orientation; primary docs govern current behavior.

- [KDnuggets: 5 Python Libraries](https://www.kdnuggets.com/5-python-libraries-that-make-data-cleaning-more-enjoyable): tool map and broad use cases; secondary source, so verify APIs.
- [Wes McKinney, Data Cleaning](https://wesmckinney.com/book/data-cleaning): pandas mapping, replacement, missing data, strings, categoricals, and reshaping.
- [pyjanitor docs](https://pyjanitor-devs.github.io/pyjanitor/): chainable pandas-style cleaning.
- [Pandera schemas](https://pandera.readthedocs.io/en/stable/dataframe_schemas.html): schema, nullability, coercion, strictness, uniqueness, and backends.
- [Great Expectations validation](https://docs.greatexpectations.io/docs/reference/learn/validation/validate_data_overview/): expectations, batches, checkpoints, and evidence.
- [ftfy docs](https://ftfy.readthedocs.io/en/latest/): focused Unicode/mojibake repair.
- [ydata-profiling](https://docs.profiling.ydata.ai/latest/): profiles and comparisons.
- [Cerberus](https://docs.python-cerberus.org/): nested dictionary validation and coercion.
- [OpenRefine reconciliation](https://openrefine.org/docs/manual/reconciling): semi-automated matching with required human review.
- [Frictionless validation](https://framework.frictionlessdata.io/docs/guides/validating-data.html): structured tabular errors and custom checks.
- [dbt data tests](https://docs.getdbt.com/docs/build/data-tests): reusable failing-row assertions and generic tests.
- [Deequ](https://github.com/awslabs/deequ): Spark-scale metrics and constraints.
- [Wickham, Tidy Data](https://vita.had.co.nz/papers/tidy-data.pdf): variables in columns, observations in rows, values in cells.
- [pandas missing data](https://pandas.pydata.org/docs/user_guide/missing_data.html) and [pandas merging](https://pandas.pydata.org/docs/user_guide/merging.html): nullable dtypes, sentinel behavior, `skipna`, merge cardinality, and row multiplication.
- [Unicode UAX #15](https://www.unicode.org/reports/tr15/): normative normalization forms and the risk of erasing distinctions with compatibility normalization.
- [dedupe documentation](https://docs.dedupe.io/en/latest/): labeled, blocked, reviewable probabilistic record linkage.
- [csvkit](https://csvkit.readthedocs.io/en/latest/) and [Miller](https://miller.readthedocs.io/en/latest/): scriptable inspection and streaming tabular transformation.

Conclusion: profile to discover, transform explicitly, validate named contracts, and preserve evidence. The reusable artifact is a plan + decision log + contract + before/after report, not a magic cleaner.

# Text, identifiers, and entity resolution

## Preserve before normalizing

Keep the source value beside any normalized value. Record the normalization form, locale, case policy, transliteration, punctuation policy, and software version. NFC is often safer than compatibility normalization; NFKC/NFKD can erase meaningful distinctions. Unicode normalization is not encoding detection. For mojibake, use a focused repair tool such as ftfy, emit an explanation/change report, and review high-impact fields.

Treat IDs as identifiers, not numbers: preserve leading zeroes, width, separators, and check digits unless the contract explicitly says otherwise. Do not infer that numeric-looking strings should become integers. Audit invisible/control characters and replacement characters before repair.

## Entity resolution

Exact duplicate detection, record linkage, and entity resolution are different tasks. Define the entity, stable identifiers, blocking keys, comparison fields, and the costs of false matches versus missed matches. Generate candidates with blocking; score candidates with documented comparators; use accepted/rejected labels or a reviewed sample to select thresholds; and keep a clerical-review band for ambiguity.

Every accepted cluster or match should retain source record IDs, candidate pairs, features/evidence, score, threshold, decision, reviewer, and model/rule version. Never auto-merge solely because a fuzzy score is highest. Preserve unmatched and conflicting records. Evaluate on a labeled holdout where feasible and test stability across reruns.

## Privacy

Profiles, candidate tables, and exception exports can expose names, addresses, identifiers, or sensitive text. Minimize fields, redact or hash display values, restrict access, set retention, and keep the mapping from privacy-safe IDs to source IDs separately.

# Data-cleaning tool selection

Choose the smallest tool that proves the needed contract.

| Tool/family | Best fit | Boundary |
|---|---|---|
| pandas | moderate in-memory tabular work | memory-bound; declare types |
| Polars | fast local/lazy tabular work | semantics and dtypes differ from pandas |
| pyjanitor | readable pandas cleaning verbs | convenience, not a quality contract |
| Pandera | Python DataFrame schemas/checks | rules must encode domain meaning |
| Great Expectations | named suites, checkpoints, reports | pin current API/version |
| ydata-profiling | exploratory HTML/JSON profiles | discovery, not validation |
| ftfy | Unicode/mojibake repair | inspect potentially changed text |
| Cerberus/Pydantic | nested JSON records | not relational/distribution checks |
| OpenRefine | interactive exports/entity review | human judgment is required |
| Frictionless | tabular/package validation | add custom business checks |
| dbt data tests | SQL models and warehouse boundaries | SQL/warehouse-oriented |
| Deequ/PyDeequ | Spark-scale metrics/constraints | JVM/Spark compatibility cost |

Selection: classify boundary and scale; choose transformation engine separately from validator; use declarative contracts at stable boundaries; keep a dependency-free triage fallback; pin versions; pilot human reconciliation and retain decisions. The orientation article names pyjanitor, Great Expectations, ftfy, ydata-profiling, and Cerberus; primary documentation governs behavior.

# Validation, contracts, and scale

Layer checks: transport (readability/hash), structure (headers/width/types), field (nulls/ranges/regex/units), record (keys/cross-field), relation (cardinality/referential integrity), dataset (grain/counts/totals/freshness/drift), and semantic owner review. Prefer checks that return failing records.

Use `templates/schema-contract.yml` as a neutral contract mapped to Pandera, Great Expectations, Frictionless, dbt, SQL, or another engine. A current profile should be compared with a versioned baseline; new categories can be legitimate evolution, while null-rate shifts can signal incidents. Avoid universal thresholds without context.

For scale: sample reconnaissance, validate invariants fully; stream/chunk files; push work to warehouses; use lazy plans; block entity candidates; partition only when the key preserves the contract; persist bounded metrics and failures. Report truncation.

Gates: pre-transform input/profile/plan; post-transform readable output, explained grain changes, accounted rejects, passing keys/rules, reconciled metrics; release report, provenance, reviewer decision, and reproducible rerun. Profiles and failure samples may contain sensitive data: minimize, redact, aggregate, and control retention.

#!/usr/bin/env python3
"""Dependency-free, read-only first-pass profiler for CSV/TSV/JSONL."""
import argparse
import csv
import hashlib
import json
import math
import pathlib
import sys
from collections import Counter


def _delimiter(text, requested):
    if requested:
        return requested
    try:
        return csv.Sniffer().sniff(text[:8192], delimiters=",\t;|").delimiter
    except csv.Error:
        return ","


def load(path, delimiter=None, max_rows=None):
    p = pathlib.Path(path)
    raw = p.read_bytes()
    digest = hashlib.sha256(raw).hexdigest()
    rows = []
    sampled = False
    if p.suffix.lower() in (".jsonl", ".ndjson"):
        headers = set()
        with p.open("r", encoding="utf-8-sig") as handle:
            for line_number, line in enumerate(handle, 1):
                if not line.strip():
                    continue
                try:
                    record = json.loads(line)
                except json.JSONDecodeError as exc:
                    raise ValueError(f"invalid JSONL at line {line_number}: {exc}") from exc
                if not isinstance(record, dict):
                    raise ValueError(f"JSONL line {line_number} is not an object")
                if max_rows is not None and len(rows) >= max_rows:
                    sampled = True
                    break
                rows.append(record)
                headers.update(record)
        headers = sorted(headers)
        rows = [{h: record.get(h) for h in headers} for record in rows]
    else:
        with p.open("r", encoding="utf-8-sig", newline="") as handle:
            sample = handle.read(8192)
            handle.seek(0)
            dialect = csv.excel
            dialect.delimiter = _delimiter(sample, delimiter)
            reader = csv.DictReader(handle, dialect=dialect)
            headers = reader.fieldnames or []
            for record in reader:
                if max_rows is not None and len(rows) >= max_rows:
                    sampled = True
                    break
                rows.append(record)
    return p, digest, headers, rows, sampled


def profile(path, delimiter=None, max_rows=None, include_values=False):
    p, digest, headers, rows, sampled = load(path, delimiter, max_rows)
    columns = {}
    for header in headers:
        values = [record.get(header) for record in rows]
        nonempty = [value for value in values if value not in (None, "")]
        missing = sum(value in (None, "") for value in values)
        numbers = []
        for value in nonempty:
            try:
                parsed = float(str(value).strip())
                if math.isfinite(parsed):
                    numbers.append(parsed)
            except (TypeError, ValueError):
                pass
        field = {
            "rows": len(values),
            "missing": missing,
            "missing_fraction": (missing / len(values) if values else 0),
            "distinct": len(set(map(str, nonempty))),
            "numeric_parse_fraction": (len(numbers) / len(nonempty) if nonempty else 0),
            "min": min(numbers) if numbers else None,
            "max": max(numbers) if numbers else None,
        }
        if include_values:
            field["top_values"] = Counter(map(str, nonempty)).most_common(5)
        columns[header] = field
    tuples = [tuple(record.get(header) for header in headers) for record in rows]
    duplicate_rows = sum(count - 1 for count in Counter(tuples).values() if count > 1)
    return {
        "file": str(p),
        "sha256": digest,
        "rows_profiled": len(rows),
        "sampled": sampled,
        "columns": headers,
        "duplicate_rows_in_profile": duplicate_rows,
        "fields": columns,
    }


def main():
    parser = argparse.ArgumentParser(description="Profile CSV, TSV, or JSONL without modifying input.")
    parser.add_argument("input")
    parser.add_argument("--output", default="-")
    parser.add_argument("--delimiter")
    parser.add_argument("--max-rows", type=int)
    parser.add_argument("--include-values", action="store_true", help="Include raw top values; off by default for privacy.")
    args = parser.parse_args()
    if args.max_rows is not None and args.max_rows < 1:
        parser.error("--max-rows must be a positive integer")
    try:
        input_path = pathlib.Path(args.input).resolve(strict=True)
        if args.output != "-":
            output_path = pathlib.Path(args.output).resolve()
            if output_path.exists() and input_path.samefile(output_path):
                print("Error: --output must not overwrite or alias the input", file=sys.stderr)
                return 2
        result = profile(input_path, args.delimiter, args.max_rows, args.include_values)
    except (OSError, UnicodeError, ValueError) as exc:
        print(f"Error: cannot profile input: {exc}", file=sys.stderr)
        return 2
    try:
        text = json.dumps(result, indent=2, sort_keys=True, allow_nan=False) + "\n"
        if args.output == "-":
            print(text, end="")
        else:
            pathlib.Path(args.output).write_text(text, encoding="utf-8")
    except (OSError, IsADirectoryError, TypeError, ValueError) as exc:
        print(f"Error: cannot write profile: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Compare two CSV/TSV datasets by shape, key uniqueness, and aggregates."""
import argparse
import csv
import hashlib
import json
import math
import pathlib
import sys
from collections import Counter


def read(path, delimiter=None):
    source = pathlib.Path(path)
    raw = source.read_bytes()
    with source.open("r", encoding="utf-8-sig", newline="") as handle:
        sample = handle.read(8192)
        handle.seek(0)
        try:
            selected = delimiter or csv.Sniffer().sniff(sample, delimiters=",\t;|").delimiter
        except csv.Error:
            selected = delimiter or ","
        rows = list(csv.DictReader(handle, delimiter=selected))
    return rows, hashlib.sha256(raw).hexdigest()


def main():
    parser = argparse.ArgumentParser(description="Reconcile two delimited datasets without modifying them.")
    parser.add_argument("before")
    parser.add_argument("after")
    parser.add_argument("--key", action="append", required=True)
    parser.add_argument("--sum", dest="sums", action="append", default=[])
    parser.add_argument("--delimiter")
    parser.add_argument("--output", default="-")
    args = parser.parse_args()
    try:
        before_path = pathlib.Path(args.before).resolve(strict=True)
        after_path = pathlib.Path(args.after).resolve(strict=True)
        before, before_hash = read(before_path, args.delimiter)
        after, after_hash = read(after_path, args.delimiter)
        if args.output != "-":
            output_path = pathlib.Path(args.output).resolve()
            if output_path.exists() and (output_path.samefile(before_path) or output_path.samefile(after_path)):
                print("Error: --output must not overwrite or alias an input", file=sys.stderr)
                return 2
    except (OSError, UnicodeError, ValueError, csv.Error) as exc:
        print(f"Error: cannot read dataset: {exc}", file=sys.stderr)
        return 2

    def keys(rows):
        return [tuple(row.get(key) for key in args.key) for row in rows]

    before_keys, after_keys = keys(before), keys(after)
    result = {
        "before": {"rows": len(before), "sha256": before_hash, "duplicate_keys": sum(n - 1 for n in Counter(before_keys).values() if n > 1)},
        "after": {"rows": len(after), "sha256": after_hash, "duplicate_keys": sum(n - 1 for n in Counter(after_keys).values() if n > 1)},
        "key": args.key,
        "missing_keys": len(set(before_keys) - set(after_keys)),
        "new_keys": len(set(after_keys) - set(before_keys)),
        "sums": {},
    }
    for column in args.sums:
        def total(rows):
            values = [float(row[column]) for row in rows if row.get(column) not in (None, "")]
            if any(not math.isfinite(value) for value in values):
                raise ValueError("non-finite numeric value")
            return sum(values)
        try:
            before_total, after_total = total(before), total(after)
        except (KeyError, TypeError, ValueError) as exc:
            print(f"Error: cannot sum {column}: {exc}", file=sys.stderr)
            return 2
        result["sums"][column] = {"before": before_total, "after": after_total, "delta": after_total - before_total}
    try:
        text = json.dumps(result, indent=2, sort_keys=True, allow_nan=False) + "\n"
        if args.output == "-":
            print(text, end="")
        else:
            pathlib.Path(args.output).write_text(text, encoding="utf-8")
    except (OSError, IsADirectoryError, TypeError, ValueError) as exc:
        print(f"Error: cannot write reconciliation: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
import json
import subprocess
import sys
import tempfile
from pathlib import Path

SCRIPT = Path(__file__).with_name("profile_dataset.py")


def test_profile_dataset():
    with tempfile.TemporaryDirectory() as directory:
        root = Path(directory)
        source = root / "sample.csv"
        source.write_text("id,name,amount\n1,Alice,2.5\n1,Alice,2.5\n2,,bad\n", encoding="utf-8")
        output = json.loads(subprocess.check_output([sys.executable, str(SCRIPT), str(source)], text=True))
        assert output["rows_profiled"] == 3
        assert output["duplicate_rows_in_profile"] == 1
        assert output["fields"]["name"]["missing"] == 1
        assert output["fields"]["amount"]["numeric_parse_fraction"] == 2 / 3
        assert "top_values" not in output["fields"]["name"]


def test_profile_rejects_overwrite_and_invalid_limit():
    with tempfile.TemporaryDirectory() as directory:
        source = Path(directory) / "sample.csv"
        source.write_text("id\n1\n", encoding="utf-8")
        overwrite = subprocess.run([sys.executable, str(SCRIPT), str(source), "--output", str(source)], capture_output=True, text=True)
        invalid = subprocess.run([sys.executable, str(SCRIPT), str(source), "--max-rows", "0"], capture_output=True, text=True)
        assert overwrite.returncode == 2
        assert invalid.returncode == 2
        assert source.read_text(encoding="utf-8") == "id\n1\n"


def run():
    test_profile_dataset()
    test_profile_rejects_overwrite_and_invalid_limit()


if __name__ == "__main__":
    run()
    print("profile_dataset tests: PASS")

#!/usr/bin/env python3
import json
import subprocess
import sys
import tempfile
from pathlib import Path

SCRIPT = Path(__file__).with_name("reconcile_dataset.py")


def test_reconcile_dataset():
    with tempfile.TemporaryDirectory() as directory:
        root = Path(directory)
        before = root / "before.csv"
        after = root / "after.csv"
        before.write_text('id,amount,note\n001,2,"line one\nline two"\n002,3,ok\n', encoding="utf-8")
        after.write_text('id,amount,note\n001,2,"line one\nline two"\n003,4,new\n', encoding="utf-8")
        result = json.loads(subprocess.check_output([sys.executable, str(SCRIPT), str(before), str(after), "--key", "id", "--sum", "amount"], text=True))
        assert result["missing_keys"] == 1
        assert result["new_keys"] == 1
        assert result["sums"]["amount"]["delta"] == 1.0


def test_reconcile_edge_cases():
    with tempfile.TemporaryDirectory() as directory:
        root = Path(directory)
        before = root / "before.csv"
        after = root / "after.csv"
        before.write_text("id\n1\n", encoding="utf-8")
        after.write_text("id\n1\n", encoding="utf-8")
        output_alias = root / "alias.csv"
        output_alias.hardlink_to(before)
        overwrite = subprocess.run([sys.executable, str(SCRIPT), str(before), str(after), "--key", "id", "--output", str(output_alias)], capture_output=True, text=True)
        assert overwrite.returncode == 2
        assert before.read_text(encoding="utf-8") == "id\n1\n"
        empty = root / "empty.csv"
        empty.write_text("id\n", encoding="utf-8")
        result = subprocess.check_output([sys.executable, str(SCRIPT), str(empty), str(empty), "--key", "id"], text=True)
        assert json.loads(result)["before"]["rows"] == 0


def run():
    test_reconcile_dataset()
    test_reconcile_edge_cases()


if __name__ == "__main__":
    run()
    print("reconcile_dataset tests: PASS")

# Cleaning plan: [dataset / version]

## Scope
- Owner:
- Consumer/decision:
- Source/retrieval time:
- Raw artifact/hash:
- Intended grain:
- Primary/candidate key:
- Time range/timezone:
- Privacy constraints:
- Out of scope:

## Acceptance contract
- Required columns/types:
- Null policy:
- Allowed values/units:
- Range/cross-field rules:
- Key/cardinality/referential rules:
- Reconciliation totals:
- Warning/blocker thresholds:

## Profile evidence
- Tool/version/command:
- Shape and encoding:
- Missingness/sentinels:
- Duplicates/keys:
- Types/dates/units:
- Categories/text/distributions:
- Baseline comparison:

## Planned transformations
| ID | Finding | Rule/action | Lossy? | Count | Reversible path | Owner |
|---|---|---|---:|---:|---|---|
| C-001 |  |  | no |  |  |  |

## Validation/review
- Pre/post checks:
- Rejected/quarantined artifact:
- Before/after reconciliation:
- Samples reviewed:
- Domain approver/date:
- Reproduction command/environment:
- Unresolved questions:
- Release: pending / accepted / rejected / escalated

# Data-quality report: [dataset / run]

- Run/code ID:
- Input/output artifact and hashes:
- Status: PASS / WARN / FAIL
- Intended grain and reviewer:

| Dimension | Before | After | Threshold/status | Evidence |
|---|---:|---:|---|---|
| Rows |  |  |  |  |
| Columns |  |  |  |  |
| Null rate |  |  |  |  |
| Duplicate keys |  |  |  |  |
| Parse failures |  |  |  |  |
| Referential coverage |  |  |  |  |
| Rejected/quarantined |  |  |  |  |

## Rules and findings
| Rule ID | Assertion | Violations | Action | Provenance |
|---|---|---:|---|---|
|  |  |  |  |  |

## Semantic review
- Valid anomalies and new categories:
- Imputation/entity decisions:
- Privacy controls:
- Open risks/owner:

## Reproduction
```text
[exact command, environment, input version, configuration]
```

name: example-records
version: 1
owner: data-owner@example.invalid
grain: one record per example_id
keys:
  primary: [example_id]
  uniqueness: one_to_one
columns:
  - name: example_id
    type: string
    nullable: false
  - name: status
    type: string
    nullable: false
    allowed_values: [new, active, closed]
  - name: amount
    type: decimal
    nullable: false
    unit: USD
    min: 0
  - name: start_at
    type: datetime
    nullable: false
    timezone: UTC
  - name: end_at
    type: datetime
    nullable: true
    timezone: UTC
  - name: occurred_at
    type: datetime
    nullable: false
    timezone: UTC
rules:
  - id: end-after-start
    description: end_at is not before start_at
    severity: error
thresholds:
  max_row_loss_fraction: 0.01
  max_parse_failure_count: 0
  max_unknown_category_count: 0
provenance:
  source_uri: replace-me
  retrieved_at: replace-me
  raw_sha256: replace-me
  code_version: replace-me