Duplicate Record Finder (python, written by Claude Code)

Claude Code 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

[]

#!/usr/bin/env python3
"""
Duplicate Record Finder - Trial 1
Uses pandas + fuzzywuzzy + python-Levenshtein for duplicate detection.
Supports exact matching, fuzzy string matching with configurable thresholds,
and multiple comparison strategies.
"""

import argparse
import csv
import json
import os
import sys
from collections import defaultdict
from itertools import combinations

import pandas as pd
from fuzzywuzzy import fuzz, process


def generate_sample_dataset(output_path="sample_data.csv"):
    """Generate a sample CSV dataset with known duplicates for testing."""
    records = [
        {"id": "1", "first_name": "John", "last_name": "Smith", "email": "john.smith@email.com", "phone": "555-0101", "city": "New York"},
        {"id": "2", "first_name": "John", "last_name": "Smith", "email": "john.smith@email.com", "phone": "555-0101", "city": "New York"},
        {"id": "3", "first_name": "Jon", "last_name": "Smyth", "email": "jon.smyth@email.com", "phone": "555-0101", "city": "New York"},
        {"id": "4", "first_name": "Jane", "last_name": "Doe", "email": "jane.doe@email.com", "phone": "555-0202", "city": "Los Angeles"},
        {"id": "5", "first_name": "Jane", "last_name": "Doe", "email": "jane.doe@email.com", "phone": "555-0202", "city": "Los Angeles"},
        {"id": "6", "first_name": "Jayne", "last_name": "Doe", "email": "jayne.doe@email.com", "phone": "555-0203", "city": "Los Angeles"},
        {"id": "7", "first_name": "Robert", "last_name": "Johnson", "email": "r.johnson@email.com", "phone": "555-0303", "city": "Chicago"},
        {"id": "8", "first_name": "Bob", "last_name": "Johnson", "email": "bob.johnson@email.com", "phone": "555-0304", "city": "Chicago"},
        {"id": "9", "first_name": "Alice", "last_name": "Williams", "email": "alice.w@email.com", "phone": "555-0404", "city": "Houston"},
        {"id": "10", "first_name": "Alice", "last_name": "Willams", "email": "alice.w@email.com", "phone": "555-0404", "city": "Houston"},
        {"id": "11", "first_name": "Michael", "last_name": "Brown", "email": "m.brown@email.com", "phone": "555-0505", "city": "Phoenix"},
        {"id": "12", "first_name": "Emily", "last_name": "Davis", "email": "emily.d@email.com", "phone": "555-0606", "city": "Philadelphia"},
        {"id": "13", "first_name": "Emilie", "last_name": "Davis", "email": "emilie.davis@email.com", "phone": "555-0607", "city": "Philadelphia"},
        {"id": "14", "first_name": "David", "last_name": "Garcia", "email": "d.garcia@email.com", "phone": "555-0707", "city": "San Antonio"},
        {"id": "15", "first_name": "David", "last_name": "Garcia", "email": "d.garcia@email.com", "phone": "555-0707", "city": "San Antonio"},
    ]
    df = pd.DataFrame(records)
    df.to_csv(output_path, index=False)
    print(f"Sample dataset generated: {output_path} ({len(records)} records)")
    return output_path


def levenshtein_similarity(s1, s2):
    """Compute Levenshtein-based similarity ratio using fuzzywuzzy."""
    return fuzz.ratio(str(s1), str(s2)) / 100.0


def jaro_winkler_similarity(s1, s2):
    """Compute Jaro-Winkler-like similarity using fuzzywuzzy partial ratio."""
    return fuzz.partial_ratio(str(s1), str(s2)) / 100.0


def cosine_similarity_strings(s1, s2):
    """Compute cosine similarity between two strings using character n-grams."""
    s1, s2 = str(s1).lower(), str(s2).lower()
    if not s1 or not s2:
        return 0.0

    def get_ngrams(s, n=2):
        return [s[i:i + n] for i in range(len(s) - n + 1)]

    ngrams1 = get_ngrams(s1)
    ngrams2 = get_ngrams(s2)

    if not ngrams1 or not ngrams2:
        return 1.0 if s1 == s2 else 0.0

    freq1 = defaultdict(int)
    freq2 = defaultdict(int)
    for ng in ngrams1:
        freq1[ng] += 1
    for ng in ngrams2:
        freq2[ng] += 1

    all_ngrams = set(freq1.keys()) | set(freq2.keys())
    dot_product = sum(freq1.get(ng, 0) * freq2.get(ng, 0) for ng in all_ngrams)
    mag1 = sum(v ** 2 for v in freq1.values()) ** 0.5
    mag2 = sum(v ** 2 for v in freq2.values()) ** 0.5

    if mag1 == 0 or mag2 == 0:
        return 0.0
    return dot_product / (mag1 * mag2)


STRATEGIES = {
    "levenshtein": levenshtein_similarity,
    "jaro-winkler": jaro_winkler_similarity,
    "cosine": cosine_similarity_strings,
}


def compute_record_similarity(row1, row2, column_rules, default_strategy="levenshtein", default_threshold=0.8):
    """
    Compare two records according to column-specific matching rules.
    Returns overall similarity score and per-column details.
    """
    column_scores = {}
    weights_sum = 0.0
    weighted_score = 0.0

    for col in column_rules:
        rule = column_rules[col]
        strategy_name = rule.get("strategy", default_strategy)
        weight = rule.get("weight", 1.0)
        strategy_fn = STRATEGIES.get(strategy_name, levenshtein_similarity)

        val1 = str(row1.get(col, "")).strip()
        val2 = str(row2.get(col, "")).strip()

        if rule.get("exact", False):
            score = 1.0 if val1.lower() == val2.lower() else 0.0
        else:
            score = strategy_fn(val1, val2)

        column_scores[col] = {"score": round(score, 4), "strategy": strategy_name}
        weighted_score += score * weight
        weights_sum += weight

    overall = weighted_score / weights_sum if weights_sum > 0 else 0.0
    return round(overall, 4), column_scores


def find_exact_duplicates(df):
    """Find rows that are exact duplicates across all columns."""
    duplicate_mask = df.duplicated(keep=False)
    dup_df = df[duplicate_mask]
    groups = defaultdict(list)
    for idx, row in dup_df.iterrows():
        key = tuple(row.values)
        groups[key].append(idx)
    return [indices for indices in groups.values() if len(indices) > 1]


def find_fuzzy_duplicates(df, column_rules, threshold=0.8, default_strategy="levenshtein"):
    """Find near-duplicate records using fuzzy matching with configurable threshold."""
    records = df.to_dict("index")
    n = len(df)
    indices = list(df.index)

    parent = {i: i for i in indices}

    def find(x):
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    def union(a, b):
        ra, rb = find(a), find(b)
        if ra != rb:
            parent[ra] = rb

    pair_scores = {}

    for i in range(n):
        for j in range(i + 1, n):
            idx_i, idx_j = indices[i], indices[j]
            overall, col_scores = compute_record_similarity(
                records[idx_i], records[idx_j], column_rules, default_strategy, threshold
            )
            if overall >= threshold:
                union(idx_i, idx_j)
                pair_scores[(idx_i, idx_j)] = {"overall": overall, "columns": col_scores}

    groups = defaultdict(list)
    for idx in indices:
        groups[find(idx)].append(idx)

    duplicate_groups = []
    for root, members in groups.items():
        if len(members) > 1:
            group_pairs = {}
            for a, b in combinations(members, 2):
                key = (min(a, b), max(a, b))
                if key in pair_scores:
                    group_pairs[f"{key[0]}-{key[1]}"] = pair_scores[key]
            duplicate_groups.append({"indices": sorted(members), "pair_scores": group_pairs})

    return duplicate_groups


def build_column_rules(columns, strategy="levenshtein", exact_cols=None):
    """Build column-specific matching rules."""
    exact_cols = exact_cols or []
    rules = {}
    for col in columns:
        if col.lower() in ("id", "index"):
            continue
        rules[col] = {
            "strategy": strategy,
            "weight": 1.0,
            "exact": col in exact_cols,
        }
    return rules


def print_console_report(df, exact_groups, fuzzy_groups):
    """Print a formatted console report of duplicate findings."""
    print("\n" + "=" * 70)
    print("  DUPLICATE RECORD FINDER - REPORT")
    print("=" * 70)
    print(f"\nTotal records analyzed: {len(df)}")

    print(f"\n--- Exact Duplicates: {len(exact_groups)} group(s) ---")
    for i, group in enumerate(exact_groups, 1):
        print(f"\n  Group {i} ({len(group)} records):")
        for idx in group:
            row = df.iloc[idx] if isinstance(idx, int) else df.loc[idx]
            print(f"    Row {idx}: {dict(row)}")

    print(f"\n--- Fuzzy Duplicate Groups: {len(fuzzy_groups)} group(s) ---")
    for i, group in enumerate(fuzzy_groups, 1):
        print(f"\n  Group {i} ({len(group['indices'])} records):")
        for idx in group["indices"]:
            row = df.iloc[idx] if isinstance(idx, int) else df.loc[idx]
            print(f"    Row {idx}: {dict(row)}")
        for pair_key, scores in group["pair_scores"].items():
            print(f"    Pair {pair_key}: overall={scores['overall']}")
            for col, info in scores["columns"].items():
                print(f"      {col}: {info['score']} ({info['strategy']})")

    print("\n" + "=" * 70)


def generate_json_report(df, exact_groups, fuzzy_groups, output_path="duplicates_report.json"):
    """Generate a JSON report of all duplicate findings."""
    report = {
        "summary": {
            "total_records": len(df),
            "exact_duplicate_groups": len(exact_groups),
            "fuzzy_duplicate_groups": len(fuzzy_groups),
        },
        "exact_duplicates": [],
        "fuzzy_duplicates": [],
    }

    for group in exact_groups:
        records = []
        for idx in group:
            row = df.iloc[idx] if isinstance(idx, int) else df.loc[idx]
            records.append({"row_index": int(idx), "data": {k: str(v) for k, v in dict(row).items()}})
        report["exact_duplicates"].append({"records": records})

    for group in fuzzy_groups:
        records = []
        for idx in group["indices"]:
            row = df.iloc[idx] if isinstance(idx, int) else df.loc[idx]
            records.append({"row_index": int(idx), "data": {k: str(v) for k, v in dict(row).items()}})
        report["fuzzy_duplicates"].append({
            "records": records,
            "pair_scores": group["pair_scores"],
        })

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"\nJSON report saved to: {output_path}")
    return report


def main():
    parser = argparse.ArgumentParser(description="Duplicate Record Finder using pandas + fuzzywuzzy")
    parser.add_argument("--input", "-i", type=str, help="Path to input CSV file")
    parser.add_argument("--threshold", "-t", type=float, default=0.8, help="Similarity threshold (0-1, default: 0.8)")
    parser.add_argument("--strategy", "-s", choices=["levenshtein", "jaro-winkler", "cosine"],
                        default="levenshtein", help="Default comparison strategy")
    parser.add_argument("--exact-cols", nargs="*", default=[], help="Columns requiring exact match")
    parser.add_argument("--output", "-o", type=str, default="duplicates_report.json", help="Output JSON report path")
    args = parser.parse_args()

    if args.input:
        if not os.path.exists(args.input):
            print(f"Error: File '{args.input}' not found.")
            sys.exit(1)
        csv_path = args.input
    else:
        print("No input file specified. Generating sample dataset...")
        csv_path = generate_sample_dataset()

    print(f"Loading data from: {csv_path}")
    df = pd.read_csv(csv_path)
    df = df.fillna("")
    print(f"Loaded {len(df)} records with columns: {list(df.columns)}")

    column_rules = build_column_rules(df.columns, strategy=args.strategy, exact_cols=args.exact_cols)
    print(f"Using strategy: {args.strategy}, threshold: {args.threshold}")
    print(f"Column rules: {json.dumps(column_rules, indent=2)}")

    print("\nFinding exact duplicates...")
    exact_groups = find_exact_duplicates(df)

    print("Finding fuzzy duplicates...")
    fuzzy_groups = find_fuzzy_duplicates(df, column_rules, threshold=args.threshold, default_strategy=args.strategy)

    print_console_report(df, exact_groups, fuzzy_groups)
    generate_json_report(df, exact_groups, fuzzy_groups, output_path=args.output)


if __name__ == "__main__":
    main()

# Duplicate Record Finder - Python Trial 1

Identifies duplicate and near-duplicate records in CSV files using exact matching and fuzzy string matching with configurable similarity thresholds.

## Dependencies
- pandas==2.1.4
- fuzzywuzzy==0.18.0
- python-Levenshtein==0.23.0

## Installation
```bash
pip install -r requirements.txt
```

## Usage
```bash
# With sample data (auto-generated)
python finder.py

# With custom CSV
python finder.py --input data.csv

# With custom threshold and strategy
python finder.py --input data.csv --threshold 0.85 --strategy jaro-winkler

# With exact match columns
python finder.py --input data.csv --exact-cols email phone

# Custom output path
python finder.py --input data.csv --output results.json
```

## Strategies
- `levenshtein` - Levenshtein distance-based similarity
- `jaro-winkler` - Jaro-Winkler similarity (via partial ratio)
- `cosine` - Cosine similarity using character n-grams

## Output
- Console report with duplicate groups and similarity scores
- `duplicates_report.json` with detailed findings

pandas==2.1.4
fuzzywuzzy==0.18.0
python-Levenshtein==0.23.0