Duplicate Record Finder (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Duplicate Record Finder

Write a program that identifies duplicate and near-duplicate records in tabular datasets using exact matching, fuzzy string matching, and configurable similarity thresholds.

FUNCTIONAL REQUIREMENTS:
- Accept a CSV file path as a command-line argument
- Support exact duplicate detection: find rows where all specified columns match exactly
- Support fuzzy duplicate detection: find rows where string columns are similar above a configurable threshold (0.0 to 1.0) using edit distance or similar string similarity metric
- Accept a --columns flag to specify which columns to compare (default: all columns)
- Accept a --threshold flag for fuzzy matching sensitivity (default: 0.85)
- Group duplicates into clusters and assign each cluster an ID
- For each duplicate cluster, designate the first occurrence as the "primary" record and subsequent ones as duplicates
- Compute similarity scores for each pair within a cluster
- Print a summary to console: total records, number of duplicate clusters, total duplicate records, breakdown by similarity range
- Save a detailed report as JSON with --output flag (default: duplicates_report.json) containing each cluster with its records, similarity scores, and matching columns
- Export a deduplicated CSV (keeping only primary records) via --deduplicate flag
- If no input file is given, generate a sample CSV with 500 records including exact duplicates, near-duplicates (typos, case differences, abbreviations), and unique records, then analyze it
- Handle large files efficiently by using blocking/indexing strategies to avoid comparing every pair

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# Duplicate Record Finder (Python)

Finds exact and near-duplicate records in CSV data using exact matching + fuzzy similarity with clustering.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies

- Direct: none
- Transitive: none

Pinned as standard-library-only in `requirements.txt`.

## Run

With input:

```bash
python src/main.py /path/to/data.csv --columns name,email,city --threshold 0.85 --output duplicates_report.json --deduplicate deduplicated.csv
```

No input (generates 500-record sample with exact + fuzzy duplicates):

```bash
python src/main.py
```

## Features

- Exact duplicate detection
- Fuzzy duplicate detection with normalized Levenshtein similarity
- Blocking/indexing strategy to avoid all-pairs comparisons
- Cluster IDs + primary record assignment + pairwise similarity scoring
- JSON report and optional deduplicated CSV export

# No external dependencies required.
# Direct dependencies: none
# Transitive dependencies: none

#!/usr/bin/env python3
import argparse
import csv
import json
from collections import defaultdict
from pathlib import Path
from typing import Any


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Duplicate Record Finder")
    parser.add_argument("input_file", nargs="?", help="CSV input file")
    parser.add_argument("--columns", help="Comma-separated columns to compare (default: all)")
    parser.add_argument("--threshold", type=float, default=0.85, help="Fuzzy similarity threshold")
    parser.add_argument("--output", default="duplicates_report.json", help="JSON report output")
    parser.add_argument("--deduplicate", nargs="?", const="deduplicated.csv", help="Export deduplicated CSV")
    return parser.parse_args()


def normalize_text(value: Any) -> str:
    return str(value or "").lower().replace(".", "").strip()


def levenshtein(a: str, b: str) -> int:
    if a == b:
        return 0
    if not a:
        return len(b)
    if not b:
        return len(a)
    dp = [[0] * (len(b) + 1) for _ in range(len(a) + 1)]
    for i in range(len(a) + 1):
        dp[i][0] = i
    for j in range(len(b) + 1):
        dp[0][j] = j
    for i in range(1, len(a) + 1):
        for j in range(1, len(b) + 1):
            cost = 0 if a[i - 1] == b[j - 1] else 1
            dp[i][j] = min(dp[i - 1][j] + 1, dp[i][j - 1] + 1, dp[i - 1][j - 1] + cost)
    return dp[len(a)][len(b)]


def similarity(a: Any, b: Any) -> float:
    x = normalize_text(a)
    y = normalize_text(b)
    m = max(len(x), len(y))
    if m == 0:
        return 1.0
    return 1.0 - (levenshtein(x, y) / m)


def record_similarity(r1: dict[str, Any], r2: dict[str, Any], columns: list[str]) -> float:
    if not columns:
        return 1.0
    scores = [similarity(r1.get(c), r2.get(c)) for c in columns]
    return sum(scores) / len(scores)


class DSU:
    def __init__(self, n: int) -> None:
        self.parent = list(range(n))
        self.rank = [0] * n

    def find(self, x: int) -> int:
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])
        return self.parent[x]

    def union(self, a: int, b: int) -> None:
        ra = self.find(a)
        rb = self.find(b)
        if ra == rb:
            return
        if self.rank[ra] < self.rank[rb]:
            ra, rb = rb, ra
        self.parent[rb] = ra
        if self.rank[ra] == self.rank[rb]:
            self.rank[ra] += 1


def load_csv(path: Path) -> tuple[list[str], list[dict[str, Any]]]:
    with path.open("r", newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        headers = reader.fieldnames or []
        rows = []
        for i, row in enumerate(reader):
            row = dict(row)
            row["__index"] = i
            rows.append(row)
    return headers, rows


def blocking_key(row: dict[str, Any], columns: list[str]) -> str:
    parts = []
    for c in columns:
        token = "".join(ch for ch in normalize_text(row.get(c, "")) if ch.isalnum())
        parts.append(token[:4])
    return "|".join(parts)


def find_duplicates(rows: list[dict[str, Any]], columns: list[str], threshold: float) -> tuple[list[list[int]], list[dict[str, Any]]]:
    dsu = DSU(len(rows))
    pair_scores: list[dict[str, Any]] = []

    exact_groups: dict[str, list[int]] = defaultdict(list)
    for i, r in enumerate(rows):
        key = "\u0001".join(str(r.get(c, "")) for c in columns)
        exact_groups[key].append(i)
    for members in exact_groups.values():
        if len(members) <= 1:
            continue
        for m in members[1:]:
            dsu.union(members[0], m)
        for i in range(len(members)):
            for j in range(i + 1, len(members)):
                pair_scores.append({"i": members[i], "j": members[j], "score": 1.0, "type": "exact"})

    blocks: dict[str, list[int]] = defaultdict(list)
    for i, r in enumerate(rows):
        blocks[blocking_key(r, columns)].append(i)

    for bucket in blocks.values():
        if len(bucket) < 2:
            continue
        for x in range(len(bucket)):
            for y in range(x + 1, len(bucket)):
                i = bucket[x]
                j = bucket[y]
                score = record_similarity(rows[i], rows[j], columns)
                if score >= threshold:
                    dsu.union(i, j)
                    pair_scores.append({"i": i, "j": j, "score": score, "type": "fuzzy"})

    groups: dict[int, list[int]] = defaultdict(list)
    for i in range(len(rows)):
        groups[dsu.find(i)].append(i)
    clusters = [sorted(members) for members in groups.values() if len(members) > 1]
    return clusters, pair_scores


def build_report(
    rows: list[dict[str, Any]],
    headers: list[str],
    columns: list[str],
    clusters: list[list[int]],
    pair_scores: list[dict[str, Any]],
    threshold: float,
) -> dict[str, Any]:
    clusters_out = []
    for cid, members in enumerate(clusters, start=1):
        records = []
        for i, idx in enumerate(members):
            rec = {h: rows[idx].get(h, "") for h in headers}
            records.append(
                {
                    "row_index": idx,
                    "role": "primary" if i == 0 else "duplicate",
                    "data": rec,
                }
            )
        pairs = []
        for i in range(len(members)):
            for j in range(i + 1, len(members)):
                a = members[i]
                b = members[j]
                found = next((p for p in pair_scores if (p["i"], p["j"]) in {(a, b), (b, a)}), None)
                score = found["score"] if found else record_similarity(rows[a], rows[b], columns)
                pairs.append({"row_a": a, "row_b": b, "similarity": score})

        clusters_out.append(
            {
                "cluster_id": f"C{cid:04d}",
                "primary_row_index": members[0],
                "size": len(members),
                "matching_columns": columns,
                "records": records,
                "pairwise_similarity": pairs,
            }
        )

    all_scores = [p["similarity"] for c in clusters_out for p in c["pairwise_similarity"]]
    breakdown = {
        "0.85-0.90": sum(1 for s in all_scores if 0.85 <= s < 0.90),
        "0.90-0.95": sum(1 for s in all_scores if 0.90 <= s < 0.95),
        "0.95-1.00": sum(1 for s in all_scores if 0.95 <= s <= 1.00),
    }

    return {
        "metadata": {
            "total_records": len(rows),
            "threshold": threshold,
            "compared_columns": columns,
        },
        "summary": {
            "duplicate_clusters": len(clusters_out),
            "total_duplicate_records": sum(c["size"] - 1 for c in clusters_out),
            "similarity_breakdown": breakdown,
        },
        "clusters": clusters_out,
    }


def write_deduplicated(path: Path, headers: list[str], rows: list[dict[str, Any]], clusters: list[list[int]]) -> None:
    duplicate_indices = set()
    for c in clusters:
        duplicate_indices.update(c[1:])
    with path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=headers)
        writer.writeheader()
        for i, row in enumerate(rows):
            if i in duplicate_indices:
                continue
            writer.writerow({h: row.get(h, "") for h in headers})


def generate_sample(path: Path) -> None:
    first_names = ["Alice", "Bob", "Carol", "David", "Eva", "Frank", "Grace", "Helen"]
    last_names = ["Smith", "Johnson", "Brown", "Wilson", "Taylor", "Miller", "Davis", "Moore"]
    cities = ["Austin", "Boston", "Chicago", "Denver", "Seattle"]
    rows: list[dict[str, str]] = []
    for i in range(450):
        fn = first_names[i % len(first_names)]
        ln = last_names[(i * 3) % len(last_names)]
        city = cities[i % len(cities)]
        rows.append(
            {
                "id": f"R{i+1:04d}",
                "name": f"{fn} {ln}",
                "email": f"{fn.lower()}.{ln.lower()}{i}@example.com",
                "city": city,
                "phone": f"555-{1000 + i}",
            }
        )
    for i in range(25):
        dup = dict(rows[i])
        dup["id"] = f"DUPX{i}"
        rows.append(dup)
    for i in range(25):
        base = dict(rows[100 + i])
        base["id"] = f"DUPF{i}"
        base["name"] = base["name"].replace("Smith", "Smiht").replace("David", "Davd")
        base["city"] = base["city"].lower()
        base["email"] = base["email"].replace("@example.com", "@example.co")
        rows.append(base)

    with path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=["id", "name", "email", "city", "phone"])
        writer.writeheader()
        writer.writerows(rows)


def print_summary(report: dict[str, Any], output_path: Path, dedup_path: Path | None) -> None:
    print("Duplicate Record Finder")
    print("=======================")
    print(f"Total records         : {report['metadata']['total_records']}")
    print(f"Duplicate clusters    : {report['summary']['duplicate_clusters']}")
    print(f"Total duplicate rows  : {report['summary']['total_duplicate_records']}")
    print("Similarity breakdown  :")
    for k, v in report["summary"]["similarity_breakdown"].items():
        print(f"  {k}: {v}")
    print(f"JSON report saved     : {output_path}")
    if dedup_path:
        print(f"Deduplicated CSV saved: {dedup_path}")


def main() -> int:
    args = parse_args()
    threshold = min(1.0, max(0.0, args.threshold))
    output_path = Path(args.output).resolve()

    if args.input_file:
        input_path = Path(args.input_file).resolve()
        if not input_path.exists():
            print(f"Input file not found: {input_path}")
            return 1
    else:
        input_path = Path("sample_duplicates.csv").resolve()
        generate_sample(input_path)
        print(f"No input provided. Generated sample dataset: {input_path}")

    headers, rows = load_csv(input_path)
    if not headers or not rows:
        print("No rows found.")
        return 1

    if args.columns:
        columns = [c.strip() for c in args.columns.split(",") if c.strip() in headers]
    else:
        columns = list(headers)
    if not columns:
        print("No valid comparison columns found.")
        return 1

    clusters, pair_scores = find_duplicates(rows, columns, threshold)
    report = build_report(rows, headers, columns, clusters, pair_scores, threshold)
    output_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")

    dedup_path = Path(args.deduplicate).resolve() if args.deduplicate else None
    if dedup_path:
        write_deduplicated(dedup_path, headers, rows, clusters)

    print_summary(report, output_path, dedup_path)
    return 0


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