ChatGPT vs Kimi for Coding

import csv
from pathlib import Path
from typing import Dict, List, Any


def remove_duplicate_rows(
    input_csv: str,
    output_csv: str,
    duplicate_column: str,
    report_csv: str = None,
    keep: str = "first",
    encoding: str = "utf-8"
) -> Dict[str, Any]:
    """
    Read a CSV file, remove duplicate rows based on a specified column,
    write the cleaned CSV, and return a summary report.

    Args:
        input_csv: Path to input CSV file.
        output_csv: Path to cleaned output CSV file.
        duplicate_column: Column name used to detect duplicates.
        report_csv: Optional path to save deleted duplicate rows as a CSV report.
        keep: Which duplicate to keep: "first" or "last".
        encoding: File encoding.

    Returns:
        Dictionary summary with deletion details.
    """
    input_path = Path(input_csv)
    output_path = Path(output_csv)

    if not input_path.exists():
        raise FileNotFoundError(f"Input file not found: {input_csv}")

    if keep not in {"first", "last"}:
        raise ValueError("keep must be either 'first' or 'last'")

    with input_path.open("r", newline="", encoding=encoding) as f:
        reader = csv.DictReader(f)
        fieldnames = reader.fieldnames

        if not fieldnames:
            raise ValueError("CSV file has no header row")

        if duplicate_column not in fieldnames:
            raise ValueError(
                f"Column '{duplicate_column}' not found in CSV headers: {fieldnames}"
            )

        rows = list(reader)

    deleted_rows: List[Dict[str, str]] = []
    cleaned_rows: List[Dict[str, str]] = []

    if keep == "first":
        seen = set()
        for row in rows:
            key = row.get(duplicate_column)
            if key in seen:
                deleted_rows.append(row)
            else:
                seen.add(key)
                cleaned_rows.append(row)

    elif keep == "last":
        last_seen_index = {}
        for i, row in enumerate(rows):
            last_seen_index[row.get(duplicate_column)] = i

        for i, row in enumerate(rows):
            key = row.get(duplicate_column)
            if last_seen_index[key] == i:
                cleaned_rows.append(row)
            else:
                deleted_rows.append(row)

    with output_path.open("w", newline="", encoding=encoding) as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(cleaned_rows)

    if report_csv:
        report_path = Path(report_csv)
        report_fieldnames = fieldnames + ["_deleted_reason"]
        with report_path.open("w", newline="", encoding=encoding) as f:
            writer = csv.DictWriter(f, fieldnames=report_fieldnames)
            writer.writeheader()
            for row in deleted_rows:
                report_row = dict(row)
                report_row["_deleted_reason"] = (
                    f"Duplicate based on column '{duplicate_column}'"
                )
                writer.writerow(report_row)

    summary = {
        "input_file": str(input_path),
        "output_file": str(output_path),
        "duplicate_column": duplicate_column,
        "total_rows_read": len(rows),
        "rows_kept": len(cleaned_rows),
        "rows_deleted": len(deleted_rows),
        "deleted_values": [row[duplicate_column] for row in deleted_rows],
    }

    if report_csv:
        summary["report_file"] = str(report_csv)

    return summary

Example usage

summary = remove_duplicate_rows(
    input_csv="data.csv",
    output_csv="data_cleaned.csv",
    duplicate_column="email",
    report_csv="deleted_duplicates_report.csv",
    keep="first"
)

print(summary)

Example summary output

{
    'input_file': 'data.csv',
    'output_file': 'data_cleaned.csv',
    'duplicate_column': 'email',
    'total_rows_read': 100,
    'rows_kept': 92,
    'rows_deleted': 8,
    'deleted_values': [
        'alice@example.com',
        'bob@example.com'
    ],
    'report_file': 'deleted_duplicates_report.csv'
}

If you want, I can also provide:

1. a pandas-based version,
2. a command-line script, or
3. a version that detects duplicates using multiple columns.

Here's a robust Python solution using pandas for efficient CSV processing, with comprehensive error handling and reporting:

import pandas as pd
import os
from datetime import datetime
from typing import Union, List, Optional, Dict

def clean_csv_duplicates(
    input_file: str,
    output_file: str,
    column_name: Union[str, List[str]],
    keep: str = 'first',
    case_sensitive: bool = True,
    report_file: Optional[str] = None,
    rejected_file: Optional[str] = None,
    encoding: str = 'utf-8'
) -> Dict:
    """
    Reads a CSV file, removes duplicates based on specified column(s), and generates 
    a detailed summary report.
    
    Args:
        input_file (str): Path to input CSV file
        output_file (str): Path to save cleaned CSV file
        column_name (str or list): Column name(s) to check for duplicates
        keep (str): Which duplicate to keep - 'first', 'last', or False (remove all duplicates)
        case_sensitive (bool): If False, treats 'Value' and 'value' as duplicates (for string columns)
        report_file (str, optional): Path to save text report. Prints to console if None
        rejected_file (str, optional): Path to save removed duplicate rows as separate CSV
        encoding (str): File encoding (default: 'utf-8')
    
    Returns:
        dict: Summary statistics including counts and duplicate details
    
    Raises:
        FileNotFoundError: If input_file doesn't exist
        ValueError: If specified column(s) don't exist in CSV
    """
    
    # Validate inputs
    if not os.path.exists(input_file):
        raise FileNotFoundError(f"Input file not found: {input_file}")
    
    if keep not in ['first', 'last', False]:
        raise ValueError("keep must be 'first', 'last', or False")
    
    # Ensure column_name is a list for uniform processing
    if isinstance(column_name, str):
        columns_to_check = [column_name]
    else:
        columns_to_check = list(column_name)
    
    try:
        # Read CSV
        print(f"Reading {input_file}...")
        df = pd.read_csv(input_file, encoding=encoding)
        
        # Validate columns exist
        missing_cols = [col for col in columns_to_check if col not in df.columns]
        if missing_cols:
            available = ", ".join(df.columns.tolist())
            raise ValueError(f"Column(s) not found: {missing_cols}. Available: {available}")
        
        original_count = len(df)
        
        # Create working copy for comparison (if case-insensitive)
        if not case_sensitive:
            compare_df = df.copy()
            for col in columns_to_check:
                if compare_df[col].dtype == 'object':
                    compare_df[col] = compare_df[col].astype(str).str.lower()
            # Use the modified copy for duplicate detection
            duplicates_mask = compare_df.duplicated(subset=columns_to_check, keep=keep)
        else:
            duplicates_mask = df.duplicated(subset=columns_to_check, keep=keep)
        
        # Split data
        cleaned_df = df[~duplicates_mask].copy()
        rejected_df = df[duplicates_mask].copy()
        
        removed_count = len(rejected_df)
        remaining_count = len(cleaned_df)
        
        # Calculate statistics
        if removed_count > 0:
            # Group by the duplicate key to show frequency
            if case_sensitive:
                dup_counts = rejected_df.groupby(columns_to_check).size()
            else:
                # For case-insensitive, group by the original values but count based on normalized
                group_keys = rejected_df[columns_to_check].apply(
                    lambda x: tuple(x.str.lower() if x.dtype == 'object' else x), axis=1
                )
                dup_counts = rejected_df.groupby(group_keys).size()
            
            duplicate_details = dup_counts.to_dict()
        else:
            duplicate_details = {}
        
        # Save outputs
        cleaned_df.to_csv(output_file, index=False, encoding=encoding)
        print(f"✓ Cleaned data saved to: {output_file} ({remaining_count} rows)")
        
        if rejected_file and removed_count > 0:
            rejected_df.to_csv(rejected_file, index=False, encoding=encoding)
            print(f"✓ Rejected rows saved to: {rejected_file} ({removed_count} rows)")
        
        # Generate report
        report_lines = [
            "=" * 70,
            "CSV DUPLICATE CLEANING REPORT",
            "=" * 70,
            f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            f"Input File: {os.path.abspath(input_file)}",
            f"Output File: {os.path.abspath(output_file)}",
            f"Column(s) Checked: {', '.join(columns_to_check)}",
            f"Case Sensitive: {case_sensitive}",
            f"Keep Strategy: {keep if keep else 'none (removed all duplicates)'}",
            "-" * 70,
            "SUMMARY STATISTICS",
            "-" * 70,
            f"Original rows:       {original_count:>6}",
            f"Duplicates removed:  {removed_count:>6}",
            f"Remaining rows:      {remaining_count:>6}",
        ]
        
        if original_count > 0:
            reduction = (removed_count / original_count) * 100
            report_lines.append(f"Reduction:           {reduction:>5.1f}%")
        
        if removed_count > 0:
            report_lines.extend([
                "-" * 70,
                "TOP DUPLICATED VALUES",
                "-" * 70,
                f"{'Value':<40} | {'Count':>6}"
            ])
            
            # Sort by count descending and format output
            sorted_dups = sorted(
                duplicate_details.items(), 
                key=lambda x: x[1], 
                reverse=True
            )[:20]  # Show top 20
            
            for value, count in sorted_dups:
                # Format value nicely (handle tuples for multi-column)
                if isinstance(value, tuple):
                    value_str = " | ".join(str(v) for v in value)
                else:
                    value_str = str(value)
                
                # Truncate if too long
                if len(value_str) > 38:
                    value_str = value_str[:35] + "..."
                report_lines.append(f"{value_str:<40} | {count:>6}")
            
            if len(duplicate_details) > 20:
                report_lines.append(f"... and {len(duplicate_details) - 20} more unique duplicate values")
        
        report_lines.append("=" * 70)
        
        report_text = "\n".join(report_lines)
        
        # Output report
        if report_file:
            with open(report_file, 'w', encoding=encoding) as f:
                f.write(report_text)
            print(f"✓ Report saved to: {report_file}")
        else:
            print("\n" + report_text)
        
        # Return summary dictionary
        return {
            'success': True,
            'original_count': original_count,
            'removed_count': removed_count,
            'remaining_count': remaining_count,
            'reduction_percentage': (removed_count/original_count)*100 if original_count > 0 else 0,
            'duplicate_values': duplicate_details,
            'columns_checked': columns_to_check,
            'input_file': input_file,
            'output_file': output_file
        }
        
    except pd.errors.EmptyDataError:
        raise ValueError("Input CSV file is empty")
    except pd.errors.ParserError as e:
        raise ValueError(f"Error parsing CSV: {e}")
    except Exception as e:
        raise Exception(f"Unexpected error: {str(e)}")


# Example usage
if __name__ == "__main__":
    # Create sample data for demonstration
    sample_csv = """email,name,department,join_date
john@company.com,John Smith,Engineering,2023-01-15
jane@company.com,Jane Doe,Marketing,2023-02-01
john@company.com,John Smith,Engineering,2023-01-20
bob@company.com,Bob Wilson,Sales,2023-03-10
jane@company.com,Jane Doe,Marketing,2023-02-15
alice@company.com,Alice Brown,Engineering,2023-04-01"""
    
    # Save sample
    with open('employees.csv', 'w') as f:
        f.write(sample_csv)
    
    # Run cleaner
    result = clean_csv_duplicates(
        input_file='employees.csv',
        output_file='employees_cleaned.csv',
        column_name='email',           # Check duplicates based on email
        keep='first',                  # Keep first occurrence
        case_sensitive=False,          # Case-insensitive matching
        report_file='cleaning_report.txt',
        rejected_file='duplicates.csv' # Save removed rows separately
    )
    
    print(f"\nOperation complete. Removed {result['removed_count']} duplicates.")

Key Features

1. Flexible Column Selection: Accepts single column name or list of columns for composite key duplicate detection
2. Case Sensitivity: Optional case_sensitive=False treats "Email@Domain.com" and "email@domain.com" as duplicates
3. Multiple Output Files:
   • Cleaned data (duplicates removed)
   • Optional rejected file (contains removed duplicates for audit)
   • Detailed text report
4. Comprehensive Reporting:
   • Row counts and reduction percentages
   • Breakdown of which values were duplicated most frequently
   • Top 20 duplicated values listed
5. Keep Strategies:
   • 'first': Keep first occurrence (default)
   • 'last': Keep last occurrence
   • False: Remove all instances of duplicates

Usage Examples

# Basic usage - single column
clean_csv_duplicates('data.csv', 'clean.csv', 'customer_id')

# Multi-column duplicate detection (composite key)
clean_csv_duplicates('data.csv', 'clean.csv', ['first_name', 'last_name', 'birth_date'])

# Remove all duplicates (keep none), case-insensitive, with rejected file
clean_csv_duplicates(
    'contacts.csv', 
    'unique_contacts.csv', 
    'email',
    keep=False,
    case_sensitive=False,
    rejected_file='removed_duplicates.csv',
    report_file='audit.txt'
)

The function returns a dictionary with statistics that you can use for logging or further processing in your application.