How to Clean Messy CSV Files with Python and Pandas

A messy CSV file can do more damage than cause a few Python errors.

Duplicate customers can inflate campaign numbers. Invalid email addresses can increase bounce rates. Incorrect dates can break reports. Missing identifiers can cause CRM imports to fail. Currency values stored as text can produce inaccurate revenue calculations.

The business cost can be significant. According to an IBM analysis on the cost of poor data quality, more than one-quarter of organizations estimate that poor data quality costs them over $5 million annually. Seven percent report losses of $25 million or more. Gartner also reports that 59% of organizations do not measure data quality, which means many teams cannot clearly see how much bad data is affecting their operations.

This guide shows how to clean a messy CSV file with Python and pandas using a practical business workflow.

Instead of silently deleting questionable records, we will:

Inspect the source file before loading it
Define clear business rules
Preserve identifiers as text
Standardize names, categories, emails, dates, and currency
Detect exact and business-key duplicates
Separate accepted and rejected records
Record why each row was rejected
Generate a measurable data quality report
Validate the final output before importing it into another system

The result is not simply a cleaner spreadsheet. It is an auditable CSV cleaning pipeline that can support CRM migrations, ecommerce reporting, customer outreach, finance operations, and business automation.

Quick Answer

To clean a messy CSV file with Python, load the file with explicit settings, inspect its structure, standardize column names and text values, convert numbers and dates carefully, validate important fields, identify duplicates, and separate invalid records instead of deleting them without explanation.

A reliable workflow should create at least three outputs:

A clean CSV containing approved records
A rejected-records CSV containing failed rows and rejection reasons
A data quality report showing what was fixed, rejected, or left for review

This approach gives analysts and business teams confidence that the final file is safe to use.

Why Messy CSV Files Create Real Business Problems

CSV files are commonly used to move information between CRMs, ecommerce platforms, accounting systems, analytics tools, advertising platforms, and internal applications.

The format is convenient, but CSV files do not enforce a schema. A file does not automatically know that:

A customer ID must be unique
A phone number should remain text
A negative order total may be invalid
A date must not be in the future
Gold, GOLD, and gold represent the same membership tier
A blank purchase amount is different from a purchase amount of zero

This is why a file can open successfully and still contain unreliable data.

Consider a marketing team preparing a customer list for a CRM migration. The raw export contains 10,000 rows, but some records have:

Missing customer IDs
Duplicate email addresses
Invalid phone numbers
Currency symbols inside spending columns
Dates written in several formats
Old placeholder values such as N/A
Extra spaces before and after names
Invalid membership categories
Negative spending totals
Duplicate customers with slightly different capitalization

Uploading this file without validation could create duplicate contacts, inaccurate segmentation, failed automation, and unreliable revenue reports.

If your broader goal is to automate repetitive data operations, Digital Exclude’s guide to AI automation for small businesses explains where structured automation can create practical value.

What We Will Build

In this tutorial, we will clean a synthetic customer CSV file and produce four files:

Output file	Purpose
`clean_customers.csv`	Approved records ready for import
`clean_customers_audit.csv`	Approved records with source-row and warning details
`rejected_customers.csv`	Original rejected records with rejection reasons
`data_quality_report.csv`	Summary of quality issues and results

Separating these outputs is safer than using dropna() and drop_duplicates() throughout the script without keeping an audit trail.

Example Business Rules

Before writing cleaning code, define what valid data means for your business.

The rules below are designed for a customer import.

Column	Business rule	Action when invalid
`customer_id`	Required and unique	Reject record
`full_name`	Required	Reject record
`age`	Optional, but must be between 18 and 120	Clear value and add warning
`email_address`	Optional, but must have a basic valid structure	Clear value and add warning
`join_date`	Required, valid, and not in the future	Reject record
`city`	Optional	Replace missing value with `Unknown`
`membership`	Bronze, Silver, or Gold	Set to `unassigned` and add warning
`total_spend`	Required numeric value of zero or more	Reject record
`phone`	Optional and stored as text	Normalize punctuation only

These decisions will vary by project.

For example, an invalid email may not justify deleting a customer from a CRM. The customer may still have a valid phone number and purchase history. An invalid invoice amount, however, may make a finance record unsafe to import.

Data cleaning should reflect business meaning, not just convenient pandas commands.

Set Up Python and pandas

Install pandas if it is not already available:

pip install pandas

Then import the required modules:

from pathlib import Path
import csv
import re
import unicodedata

import pandas as pd

The examples follow current pandas 3.x guidance. In pandas 3.0, Copy-on-Write is the only mode, and chained assignments do not update the original DataFrame. Use direct column assignments or .loc[] rather than code such as df["age"].fillna(0, inplace=True). Review the official pandas Copy-on-Write documentation when updating older scripts.

Use this:

df["age"] = df["age"].fillna(0)

Avoid this:

df["age"].fillna(0, inplace=True)

Step 1: Inspect the CSV Before Loading It

Many CSV problems occur before pandas creates a DataFrame.

A file may use:

UTF-8, Windows-1252, or another encoding
Commas, semicolons, tabs, or pipes as delimiters
Quoted values containing commas
Line breaks inside quoted fields
An unexpected number of columns
A byte-order mark
Malformed rows

Opening the file with default settings and hoping pandas guesses correctly is not a dependable production workflow.

Detect the Encoding and Delimiter

The following helper tries several common encodings and uses Python’s CSV sniffer to estimate the delimiter:

def detect_text_settings(path: str | Path) -> tuple[str, str]:
    path = Path(path)

    encoding = None
    sample = ""

    for candidate in ("utf-8-sig", "utf-8", "cp1252", "latin-1"):
        try:
            with path.open("r", encoding=candidate, newline="") as file:
                sample = file.read(8192)

            encoding = candidate
            break
        except UnicodeDecodeError:
            continue

    if encoding is None:
        raise UnicodeError(
            "The CSV could not be decoded with the configured encodings."
        )

    try:
        delimiter = csv.Sniffer().sniff(
            sample,
            delimiters=",;\t|",
        ).delimiter
    except csv.Error:
        delimiter = ","

    return encoding, delimiter

Why `newline=""` Matters

Python’s official csv module documentation recommends opening CSV files with newline="". This allows the CSV parser to handle newlines correctly, including newlines that appear inside quoted fields.

Do Not Trust Automatic Detection Blindly

Delimiter detection is useful, but it is still an estimate.

For an important migration, inspect a sample of the raw file and confirm:

The detected delimiter is correct
The header row is correct
Quoted commas remain inside one field
Special characters display properly
The expected number of columns is present

Step 2: Load Every Column as Text First

A common mistake is allowing pandas to infer identifiers as numbers.

Customer IDs, ZIP codes, account numbers, SKUs, and phone numbers are labels. They are not quantities.

For example, this customer ID:

may become:

if a spreadsheet or parser treats it as a number.

The safest starting point for a mixed business CSV is often to load every column as a string and perform controlled conversions afterward.

path = Path("messy_customers.csv")

encoding, delimiter = detect_text_settings(path)

df = pd.read_csv(
    path,
    sep=delimiter,
    encoding=encoding,
    dtype="string",
    na_values=[
        "",
        "NA",
        "N/A",
        "n/a",
        "unknown",
        "Unknown",
        "not a date",
        "-",
    ],
    keep_default_na=True,
    on_bad_lines="error",
)

The official pandas.read_csv() documentation includes options for delimiters, encodings, missing-value markers, data types, malformed-line behavior, and chunked reading. These options should be chosen deliberately when the file supports business operations.

Why Use `on_bad_lines="error"`?

It may be tempting to use:

on_bad_lines="skip"

That can make the file load, but it can also remove records without resolving why they were malformed.

For a business import, failing loudly is usually safer. Investigate malformed lines, repair the export, or create a separate preprocessing step that quarantines them.

Step 3: Add Source-Row Tracking

When a stakeholder asks why a customer was rejected, you should be able to identify the original row.

df.insert(0, "source_row", range(2, len(df) + 2))

The count begins at 2 because row 1 normally contains the CSV header.

Now every accepted, rejected, or warned record can be traced back to its source location.

Step 4: Clean and Validate the Column Names

Messy column names make code harder to maintain.

Examples include:

 Customer ID
Email Address
JOIN DATE
Total-Spend

Convert them into consistent snake_case names:

def clean_header(name: str) -> str:
    return re.sub(
        r"[^a-z0-9]+",
        "_",
        name.strip().casefold(),
    ).strip("_")


df.columns = [clean_header(column) for column in df.columns]

The resulting columns become:

source_row
customer_id
full_name
age
email_address
join_date
city
membership
total_spend
phone

Confirm the Expected Schema

Do not continue if a required column is missing.

required_columns = {
    "customer_id",
    "full_name",
    "age",
    "email_address",
    "join_date",
    "city",
    "membership",
    "total_spend",
    "phone",
}

missing_columns = required_columns.difference(df.columns)

if missing_columns:
    raise ValueError(
        f"Missing required columns: {sorted(missing_columns)}"
    )

This catches changed exports early. For example, a source system may rename customer_id to contact_id without notifying the team.

Step 5: Preserve the Original Values

Create a copy before changing the cell values:

source_snapshot = df.copy(deep=True)

This snapshot will be used when exporting rejected records.

The clean file should contain normalized values. The rejected file should preserve what the source system originally provided so someone can investigate the problem accurately.

Step 6: Normalize Text and Unicode

Customer data can contain invisible spaces, repeated spaces, full-width characters, and inconsistent Unicode representations.

Create a helper:

def normalize_unicode(value):
    if pd.isna(value):
        return pd.NA

    return unicodedata.normalize("NFKC", str(value))

Apply it to text columns:

text_columns = [
    "customer_id",
    "full_name",
    "email_address",
    "join_date",
    "city",
    "membership",
    "total_spend",
    "phone",
]

for column in text_columns:
    df[column] = (
        df[column]
        .map(normalize_unicode)
        .astype("string")
        .str.strip()
        .str.replace(r"\s+", " ", regex=True)
    )

Then apply field-specific normalization:

df["customer_id"] = df["customer_id"].str.upper()
df["email_address"] = df["email_address"].str.casefold()
df["membership"] = df["membership"].str.casefold()
df["city"] = df["city"].str.title()

Be Careful With Automatic Title Case

Applying .str.title() to every person or company name can produce incorrect results.

Examples include:

McDonald
iPhone Repair Center
de la Cruz
O'NEILL
Brand names with custom capitalization

For sensitive master data, collapse whitespace but preserve the original name. Use an approved mapping table when exact canonical capitalization matters.

Step 7: Convert Age Without Inventing Information

Convert age values to numbers:

df["age"] = pd.to_numeric(
    df["age"],
    errors="coerce",
).astype("Float64")

Identify impossible values:

invalid_age = (
    df["age"].notna()
    & ~df["age"].between(18, 120)
)

df.loc[invalid_age, "age"] = pd.NA
df["age"] = df["age"].astype("Int64")

In this business scenario, age is optional. An invalid age is cleared and recorded as a warning rather than causing the entire customer record to be rejected.

Should Missing Age Be Filled With the Median?

Not automatically.

Median imputation may be useful for statistical modeling, but it changes the meaning of a system-of-record export. A customer whose age is unknown is not necessarily the median age.

Keep the value missing unless the downstream analysis has a documented reason for imputation. If you do impute values for modeling, create a separate indicator such as:

df["age_was_imputed"] = df["age"].isna()

This lets analysts distinguish reported values from estimated ones.

Step 8: Parse Mixed Dates Safely

Dates such as 03/04/2025 are ambiguous.

In the United States, this usually means March 4. In many other countries, it means April 3.

The pandas documentation warns that dayfirst=True is a preference, not strict enforcement. The safest approach is to use formats that are known to be produced by the source systems. Review the official pandas.to_datetime() documentation before relying on automatic mixed-format parsing.

Define the formats accepted by this export:

KNOWN_DATE_FORMATS = (
    "%Y-%m-%d",
    "%m/%d/%Y",
    "%Y/%m/%d",
    "%d-%m-%Y",
)

Create a parser:

def parse_known_dates(series: pd.Series) -> pd.Series:
    parsed = pd.Series(
        pd.NaT,
        index=series.index,
        dtype="datetime64[ns]",
    )

    for date_format in KNOWN_DATE_FORMATS:
        mask = parsed.isna() & series.notna()

        if not mask.any():
            break

        parsed.loc[mask] = pd.to_datetime(
            series.loc[mask],
            format=date_format,
            errors="coerce",
        )

    return parsed

Apply it:

df["join_date"] = parse_known_dates(df["join_date"])

Validate the result:

today = pd.Timestamp.today().normalize()

invalid_or_missing_date = df["join_date"].isna()

future_date = (
    df["join_date"].notna()
    & df["join_date"].gt(today)
)

A future joining date is rejected because it violates the business rule for this customer history file.

Step 9: Clean Currency Values

Currency columns frequently include:

Dollar signs
Rupee symbols
Commas
Spaces
Text placeholders
Parentheses for negative amounts
Different decimal conventions

For values such as $1,250.50 or ₹75,000, remove formatting characters:

spend_text = df["total_spend"].astype("string")

spend_clean = spend_text.str.replace(
    r"[^\d.\-]",
    "",
    regex=True,
)

df["total_spend"] = pd.to_numeric(
    spend_clean,
    errors="coerce",
)

Create validation masks:

invalid_spend = df["total_spend"].isna()

negative_spend = (
    df["total_spend"].notna()
    & df["total_spend"].lt(0)
)

Money Requires a Defined Locale

The code above assumes a period is the decimal separator.

A value such as:

1.500,00

may mean 1500.00 in parts of Europe, but a generic regular expression could interpret it incorrectly.

When your CSV contains multiple currencies or regional formats, add columns such as:

currency_code
source_locale

Then parse each format using explicit locale rules.

Use Decimal or Integer Cents for Accounting

Floating-point values are usually acceptable for general analysis, but financial ledger calculations should use fixed-precision decimal values or integer cents.

Data cleaning should not quietly introduce rounding risk into accounting workflows.

Step 10: Validate Email Addresses

Normalize emails first:

df["email_address"] = (
    df["email_address"]
    .astype("string")
    .str.strip()
    .str.casefold()
)

Apply basic syntax validation:

email_pattern = r"^[^@\s]+@[^@\s]+\.[A-Za-z]{2,}$"

invalid_email = (
    df["email_address"].notna()
    & ~df["email_address"].str.fullmatch(
        email_pattern,
        na=False,
    )
)

df.loc[invalid_email, "email_address"] = pd.NA

This only checks the basic structure.

It does not prove that:

The mailbox exists
The domain accepts email
The customer owns the address
The address can receive your campaign

Deliverability verification is a separate process.

For this workflow, an invalid email is cleared and recorded as a warning. The customer remains in the approved file because other contact and purchase information may still be valid.

Step 11: Normalize Phone Numbers Without Reconstructing Them

Phone numbers must be loaded as text.

Never attempt to recover a phone number that has already been converted into scientific notation. For example, removing non-digit characters from 9.994973e+09 does not reliably restore the original number.

Normalize punctuation only:

def normalize_phone(value):
    if pd.isna(value):
        return pd.NA

    cleaned = re.sub(r"[^\d+]", "", str(value))
    cleaned = re.sub(r"(?<!^)\+", "", cleaned)

    return cleaned if cleaned else pd.NA


df["phone"] = (
    df["phone"]
    .map(normalize_phone)
    .astype("string")
)

Do not assume a country code based only on the number of digits. Proper international validation requires country context and a dedicated phone-number library.

Step 12: Standardize Membership Categories

Create a mapping for known variations:

membership_map = {
    "gld": "gold",
    "slvr": "silver",
    "brz": "bronze",
}

df["membership"] = df["membership"].replace(
    membership_map
)

Define the approved categories:

ALLOWED_MEMBERSHIPS = {
    "bronze",
    "silver",
    "gold",
}

Identify invalid and missing categories:

invalid_membership = (
    df["membership"].notna()
    & ~df["membership"].isin(ALLOWED_MEMBERSHIPS)
)

missing_membership = df["membership"].isna()

Replace them with a transparent fallback:

df.loc[
    invalid_membership | missing_membership,
    "membership",
] = "unassigned"

This is better than forcing platinum into gold without a documented business decision.

Step 13: Detect Exact and Business-Key Duplicates

There are several kinds of duplicate data.

Exact Duplicates

Every business field is identical:

business_columns = [
    column
    for column in df.columns
    if column != "source_row"
]

exact_duplicate = df.duplicated(
    subset=business_columns,
    keep="first",
)

Duplicate Customer IDs

Two records may have the same ID but different values:

duplicate_id = (
    df["customer_id"].notna()
    & df["customer_id"].duplicated(keep=False)
)

In this workflow, every row involved in a duplicate customer ID is rejected for review.

That is safer than keeping the first row simply because it appears first in the file.

A different business may use a rule such as:

Keep the row with the newest updated_at
Keep the row from the trusted master system
Merge non-conflicting values
Send all conflicts to a data steward

The rule should be explicit.

Probable Entity Duplicates

These records may refer to the same customer even though the IDs differ:

C1001, John Smith, JOHN@EXAMPLE.COM
C1844, John Smith, john@example.com

Probable duplicates can be identified using normalized emails, phone numbers, addresses, or fuzzy matching.

Do not automatically merge probable duplicates unless the confidence rule has been approved. Incorrect merges can be more damaging than duplicates.

Step 14: Handle Missing Values Based on Business Meaning

Different missing values require different actions.

missing_id = df["customer_id"].isna()
missing_name = df["full_name"].isna()

For this import:

Missing customer ID causes rejection
Missing full name causes rejection
Missing city becomes Unknown
Missing membership becomes unassigned
Missing age remains missing
Missing email remains missing
Missing total spend causes rejection
Missing join date causes rejection

Fill the city:

df["city"] = df["city"].fillna("Unknown")

Do not replace missing spending with zero unless the business confirms that blank means no spending.

These values are not equivalent:

0.00 = The customer spent nothing
blank = The spending amount is unknown

Step 15: Create Rejection and Warning Reasons

Some problems invalidate the record. Others allow the record to continue with a warning.

Create two columns:

rejection_reasons = pd.Series(
    "",
    index=df.index,
    dtype="string",
)

warning_reasons = pd.Series(
    "",
    index=df.index,
    dtype="string",
)

Use a helper to add multiple reasons to one row:

def append_reason(
    reason_series: pd.Series,
    mask: pd.Series,
    reason: str,
) -> None:
    current = reason_series.loc[mask]

    reason_series.loc[mask] = (
        current.where(
            current.eq(""),
            current + "|",
        )
        + reason
    )

Add critical rejection reasons:

append_reason(
    rejection_reasons,
    exact_duplicate,
    "exact_duplicate",
)

append_reason(
    rejection_reasons,
    missing_id,
    "missing_customer_id",
)

append_reason(
    rejection_reasons,
    duplicate_id,
    "duplicate_customer_id",
)

append_reason(
    rejection_reasons,
    missing_name,
    "missing_full_name",
)

append_reason(
    rejection_reasons,
    invalid_or_missing_date,
    "invalid_or_missing_join_date",
)

append_reason(
    rejection_reasons,
    future_date,
    "future_join_date",
)

append_reason(
    rejection_reasons,
    invalid_spend,
    "invalid_or_missing_total_spend",
)

append_reason(
    rejection_reasons,
    negative_spend,
    "negative_total_spend",
)

Add non-critical warnings:

append_reason(
    warning_reasons,
    invalid_age,
    "invalid_age_set_to_missing",
)

append_reason(
    warning_reasons,
    invalid_email,
    "invalid_email_set_to_missing",
)

append_reason(
    warning_reasons,
    invalid_membership,
    "invalid_membership_set_to_unassigned",
)

append_reason(
    warning_reasons,
    missing_membership,
    "missing_membership_set_to_unassigned",
)

Attach both columns:

df["rejection_reasons"] = rejection_reasons
df["warning_reasons"] = warning_reasons

Step 16: Separate Clean and Rejected Records

Create two DataFrames:

rejected = df[
    df["rejection_reasons"].ne("")
].copy()

clean = df[
    df["rejection_reasons"].eq("")
].copy()

Restore the original values for rejected records:

rejected_original = source_snapshot.loc[
    rejected.index
].copy()

rejected_original["rejection_reasons"] = (
    rejected["rejection_reasons"]
)

rejected_original["warning_reasons"] = (
    rejected["warning_reasons"]
)

This gives the review team the unmodified source values and a clear explanation of what failed.

Step 17: Run Final Validation Checks

Never assume that the transformation code worked because it completed without an exception.

Validate the accepted records:

final_memberships = {
    "bronze",
    "silver",
    "gold",
    "unassigned",
}

assert len(df) == len(clean) + len(rejected)

assert clean["customer_id"].notna().all()
assert clean["customer_id"].is_unique
assert clean["full_name"].notna().all()

assert clean["join_date"].notna().all()
assert clean["join_date"].le(today).all()

assert clean["total_spend"].notna().all()
assert clean["total_spend"].ge(0).all()

assert clean["membership"].isin(
    final_memberships
).all()

assert (
    clean["email_address"]
    .dropna()
    .str.fullmatch(email_pattern)
    .all()
)

print("All validation checks passed.")

These assertions turn business rules into executable checks.

If a future code change breaks one of the rules, the script stops before exporting an unreliable file.

Step 18: Generate a Data Quality Report

A quality report allows stakeholders to see what happened.

quality_metrics = {
    "input_rows": len(df),
    "accepted_rows": len(clean),
    "rejected_rows": len(rejected),
    "exact_duplicate_rows": int(
        exact_duplicate.sum()
    ),
    "rows_with_duplicate_customer_id": int(
        duplicate_id.sum()
    ),
    "invalid_email_values_cleared": int(
        invalid_email.sum()
    ),
    "invalid_age_values_cleared": int(
        invalid_age.sum()
    ),
    "membership_values_unassigned": int(
        (
            invalid_membership
            | missing_membership
        ).sum()
    ),
}

quality_report = pd.DataFrame(
    quality_metrics.items(),
    columns=["metric", "value"],
)

For the sample data, the report may look like this:

Metric	Value
Input rows	12
Accepted rows	6
Rejected rows	6
Exact duplicate rows	1
Rows with duplicate customer IDs	2
Invalid emails cleared	1
Invalid ages cleared	1
Membership values set to unassigned	2

This is more useful than saying, “The data has been cleaned.”

It shows exactly what changed.

Step 19: Export the Results

Prepare a clean import file without audit columns:

clean_for_import = clean.drop(
    columns=[
        "source_row",
        "rejection_reasons",
        "warning_reasons",
    ]
)

Export all outputs:

clean_for_import.to_csv(
    "clean_customers.csv",
    index=False,
    date_format="%Y-%m-%d",
    encoding="utf-8",
)

clean.to_csv(
    "clean_customers_audit.csv",
    index=False,
    date_format="%Y-%m-%d",
    encoding="utf-8",
)

rejected_original.to_csv(
    "rejected_customers.csv",
    index=False,
    encoding="utf-8",
)

quality_report.to_csv(
    "data_quality_report.csv",
    index=False,
    encoding="utf-8",
)

Use utf-8-sig instead of utf-8 when stakeholders frequently open the file in older spreadsheet environments that handle the UTF-8 byte-order mark more reliably.

Always keep the original file unchanged.

Complete Reusable CSV Cleaning Script

The following script combines the full workflow:

from pathlib import Path
import csv
import re
import unicodedata

import pandas as pd


INPUT_FILE = Path("messy_customers.csv")

ALLOWED_MEMBERSHIPS = {
    "bronze",
    "silver",
    "gold",
}

KNOWN_DATE_FORMATS = (
    "%Y-%m-%d",
    "%m/%d/%Y",
    "%Y/%m/%d",
    "%d-%m-%Y",
)


def detect_text_settings(
    path: str | Path,
) -> tuple[str, str]:
    path = Path(path)

    encoding = None
    sample = ""

    for candidate in (
        "utf-8-sig",
        "utf-8",
        "cp1252",
        "latin-1",
    ):
        try:
            with path.open(
                "r",
                encoding=candidate,
                newline="",
            ) as file:
                sample = file.read(8192)

            encoding = candidate
            break

        except UnicodeDecodeError:
            continue

    if encoding is None:
        raise UnicodeError(
            "Could not decode the CSV."
        )

    try:
        delimiter = csv.Sniffer().sniff(
            sample,
            delimiters=",;\t|",
        ).delimiter
    except csv.Error:
        delimiter = ","

    return encoding, delimiter


def clean_header(name: str) -> str:
    return re.sub(
        r"[^a-z0-9]+",
        "_",
        name.strip().casefold(),
    ).strip("_")


def normalize_unicode(value):
    if pd.isna(value):
        return pd.NA

    return unicodedata.normalize(
        "NFKC",
        str(value),
    )


def parse_known_dates(
    series: pd.Series,
) -> pd.Series:
    parsed = pd.Series(
        pd.NaT,
        index=series.index,
        dtype="datetime64[ns]",
    )

    for date_format in KNOWN_DATE_FORMATS:
        mask = parsed.isna() & series.notna()

        if not mask.any():
            break

        parsed.loc[mask] = pd.to_datetime(
            series.loc[mask],
            format=date_format,
            errors="coerce",
        )

    return parsed


def normalize_phone(value):
    if pd.isna(value):
        return pd.NA

    cleaned = re.sub(
        r"[^\d+]",
        "",
        str(value),
    )

    cleaned = re.sub(
        r"(?<!^)\+",
        "",
        cleaned,
    )

    return cleaned if cleaned else pd.NA


def append_reason(
    reason_series: pd.Series,
    mask: pd.Series,
    reason: str,
) -> None:
    current = reason_series.loc[mask]

    reason_series.loc[mask] = (
        current.where(
            current.eq(""),
            current + "|",
        )
        + reason
    )


encoding, delimiter = detect_text_settings(
    INPUT_FILE
)

df = pd.read_csv(
    INPUT_FILE,
    sep=delimiter,
    encoding=encoding,
    dtype="string",
    na_values=[
        "",
        "NA",
        "N/A",
        "n/a",
        "unknown",
        "Unknown",
        "not a date",
        "-",
    ],
    keep_default_na=True,
    on_bad_lines="error",
)

df.insert(
    0,
    "source_row",
    range(2, len(df) + 2),
)

df.columns = [
    clean_header(column)
    for column in df.columns
]

required_columns = {
    "customer_id",
    "full_name",
    "age",
    "email_address",
    "join_date",
    "city",
    "membership",
    "total_spend",
    "phone",
}

missing_columns = (
    required_columns.difference(df.columns)
)

if missing_columns:
    raise ValueError(
        f"Missing required columns: "
        f"{sorted(missing_columns)}"
    )

source_snapshot = df.copy(deep=True)

text_columns = [
    "customer_id",
    "full_name",
    "email_address",
    "join_date",
    "city",
    "membership",
    "total_spend",
    "phone",
]

for column in text_columns:
    df[column] = (
        df[column]
        .map(normalize_unicode)
        .astype("string")
        .str.strip()
        .str.replace(
            r"\s+",
            " ",
            regex=True,
        )
    )

df["customer_id"] = (
    df["customer_id"].str.upper()
)

df["email_address"] = (
    df["email_address"].str.casefold()
)

df["membership"] = (
    df["membership"].str.casefold()
)

df["city"] = df["city"].str.title()
df["city"] = df["city"].fillna("Unknown")

df["age"] = pd.to_numeric(
    df["age"],
    errors="coerce",
).astype("Float64")

invalid_age = (
    df["age"].notna()
    & ~df["age"].between(18, 120)
)

df.loc[invalid_age, "age"] = pd.NA
df["age"] = df["age"].astype("Int64")

df["join_date"] = parse_known_dates(
    df["join_date"]
)

today = pd.Timestamp.today().normalize()

invalid_or_missing_date = (
    df["join_date"].isna()
)

future_date = (
    df["join_date"].notna()
    & df["join_date"].gt(today)
)

spend_clean = (
    df["total_spend"]
    .astype("string")
    .str.replace(
        r"[^\d.\-]",
        "",
        regex=True,
    )
)

df["total_spend"] = pd.to_numeric(
    spend_clean,
    errors="coerce",
)

invalid_spend = df["total_spend"].isna()

negative_spend = (
    df["total_spend"].notna()
    & df["total_spend"].lt(0)
)

email_pattern = (
    r"^[^@\s]+@[^@\s]+\.[A-Za-z]{2,}$"
)

invalid_email = (
    df["email_address"].notna()
    & ~df["email_address"].str.fullmatch(
        email_pattern,
        na=False,
    )
)

df.loc[
    invalid_email,
    "email_address",
] = pd.NA

df["phone"] = (
    df["phone"]
    .map(normalize_phone)
    .astype("string")
)

membership_map = {
    "gld": "gold",
    "slvr": "silver",
    "brz": "bronze",
}

df["membership"] = (
    df["membership"].replace(
        membership_map
    )
)

invalid_membership = (
    df["membership"].notna()
    & ~df["membership"].isin(
        ALLOWED_MEMBERSHIPS
    )
)

missing_membership = (
    df["membership"].isna()
)

df.loc[
    invalid_membership
    | missing_membership,
    "membership",
] = "unassigned"

business_columns = [
    column
    for column in df.columns
    if column != "source_row"
]

exact_duplicate = df.duplicated(
    subset=business_columns,
    keep="first",
)

missing_id = df["customer_id"].isna()

duplicate_id = (
    df["customer_id"].notna()
    & df["customer_id"].duplicated(
        keep=False
    )
)

missing_name = df["full_name"].isna()

rejection_reasons = pd.Series(
    "",
    index=df.index,
    dtype="string",
)

warning_reasons = pd.Series(
    "",
    index=df.index,
    dtype="string",
)

append_reason(
    rejection_reasons,
    exact_duplicate,
    "exact_duplicate",
)

append_reason(
    rejection_reasons,
    missing_id,
    "missing_customer_id",
)

append_reason(
    rejection_reasons,
    duplicate_id,
    "duplicate_customer_id",
)

append_reason(
    rejection_reasons,
    missing_name,
    "missing_full_name",
)

append_reason(
    rejection_reasons,
    invalid_or_missing_date,
    "invalid_or_missing_join_date",
)

append_reason(
    rejection_reasons,
    future_date,
    "future_join_date",
)

append_reason(
    rejection_reasons,
    invalid_spend,
    "invalid_or_missing_total_spend",
)

append_reason(
    rejection_reasons,
    negative_spend,
    "negative_total_spend",
)

append_reason(
    warning_reasons,
    invalid_age,
    "invalid_age_set_to_missing",
)

append_reason(
    warning_reasons,
    invalid_email,
    "invalid_email_set_to_missing",
)

append_reason(
    warning_reasons,
    invalid_membership,
    "invalid_membership_set_to_unassigned",
)

append_reason(
    warning_reasons,
    missing_membership,
    "missing_membership_set_to_unassigned",
)

df["rejection_reasons"] = (
    rejection_reasons
)

df["warning_reasons"] = warning_reasons

rejected = df[
    df["rejection_reasons"].ne("")
].copy()

clean = df[
    df["rejection_reasons"].eq("")
].copy()

rejected_original = (
    source_snapshot.loc[
        rejected.index
    ].copy()
)

rejected_original[
    "rejection_reasons"
] = rejected["rejection_reasons"]

rejected_original[
    "warning_reasons"
] = rejected["warning_reasons"]

final_memberships = {
    "bronze",
    "silver",
    "gold",
    "unassigned",
}

assert len(df) == len(clean) + len(rejected)

assert clean["customer_id"].notna().all()
assert clean["customer_id"].is_unique
assert clean["full_name"].notna().all()

assert clean["join_date"].notna().all()
assert clean["join_date"].le(today).all()

assert clean["total_spend"].notna().all()
assert clean["total_spend"].ge(0).all()

assert clean["membership"].isin(
    final_memberships
).all()

assert (
    clean["email_address"]
    .dropna()
    .str.fullmatch(email_pattern)
    .all()
)

quality_metrics = {
    "input_rows": len(df),
    "accepted_rows": len(clean),
    "rejected_rows": len(rejected),
    "exact_duplicate_rows": int(
        exact_duplicate.sum()
    ),
    "rows_with_duplicate_customer_id": int(
        duplicate_id.sum()
    ),
    "invalid_email_values_cleared": int(
        invalid_email.sum()
    ),
    "invalid_age_values_cleared": int(
        invalid_age.sum()
    ),
    "membership_values_unassigned": int(
        (
            invalid_membership
            | missing_membership
        ).sum()
    ),
}

quality_report = pd.DataFrame(
    quality_metrics.items(),
    columns=["metric", "value"],
)

clean_for_import = clean.drop(
    columns=[
        "source_row",
        "rejection_reasons",
        "warning_reasons",
    ]
)

clean_for_import.to_csv(
    "clean_customers.csv",
    index=False,
    date_format="%Y-%m-%d",
    encoding="utf-8",
)

clean.to_csv(
    "clean_customers_audit.csv",
    index=False,
    date_format="%Y-%m-%d",
    encoding="utf-8",
)

rejected_original.to_csv(
    "rejected_customers.csv",
    index=False,
    encoding="utf-8",
)

quality_report.to_csv(
    "data_quality_report.csv",
    index=False,
    encoding="utf-8",
)

print("All validation checks passed.")
print(f"Accepted records: {len(clean)}")
print(f"Rejected records: {len(rejected)}")

How to Clean a Large CSV File in Chunks

Loading a multi-gigabyte CSV into memory may not be practical.

Use the chunksize argument:

reader = pd.read_csv(
    "large_customers.csv",
    dtype="string",
    chunksize=100_000,
)

Process each chunk:

first_chunk = True

for chunk in reader:
    cleaned_chunk = clean_customer_chunk(
        chunk
    )

    cleaned_chunk.to_csv(
        "clean_large_customers.csv",
        mode="w" if first_chunk else "a",
        header=first_chunk,
        index=False,
    )

    first_chunk = False

Cross-Chunk Duplicates Need Extra Logic

Calling drop_duplicates() inside each chunk only detects duplicates within that chunk.

To detect repeated IDs across the whole file, keep a persistent set:

seen_customer_ids = set()

for chunk in reader:
    duplicate_across_chunks = (
        chunk["customer_id"].isin(
            seen_customer_ids
        )
    )

    new_ids = (
        chunk.loc[
            ~duplicate_across_chunks,
            "customer_id",
        ]
        .dropna()
        .tolist()
    )

    seen_customer_ids.update(new_ids)

For extremely large datasets, use a database, disk-backed store, or data-processing platform rather than keeping millions of IDs in Python memory.

How to Clean a CSV Without pandas

For a small dependency-free script, Python’s built-in csv module can handle row-based cleaning.

import csv

unwanted_values = {
    "NA",
    "N/A",
    "-",
    "#NAME?",
}

with open(
    "input.csv",
    "r",
    encoding="utf-8",
    newline="",
) as source_file, open(
    "cleaned.csv",
    "w",
    encoding="utf-8",
    newline="",
) as output_file:

    reader = csv.DictReader(source_file)

    if reader.fieldnames is None:
        raise ValueError(
            "The CSV does not contain a header."
        )

    writer = csv.DictWriter(
        output_file,
        fieldnames=reader.fieldnames,
    )

    writer.writeheader()

    for row in reader:
        cleaned_row = {}

        for key, value in row.items():
            value = (value or "").strip()

            if value in unwanted_values:
                value = ""

            cleaned_row[key.strip()] = value

        if any(cleaned_row.values()):
            writer.writerow(cleaned_row)

This works well for:

Trimming cells
Removing blank rows
Replacing placeholders
Streaming small transformations
Avoiding additional dependencies

Pandas is usually better when you need data profiling, type conversion, complex validation, duplicate analysis, grouping, or quality reporting.

Can AI Help Clean CSV Files?

AI coding tools can help generate cleaning functions, explain pandas errors, create test cases, and suggest validation rules.

They should not decide business rules without human review.

For example, an AI assistant may suggest filling every missing purchase value with zero. That code may run successfully while changing the meaning of the data.

Use AI for:

Drafting repetitive transformation code
Explaining parser errors
Creating unit-test ideas
Reviewing a cleaning function
Identifying possible edge cases
Documenting the pipeline

Keep humans responsible for:

Required fields
Duplicate resolution
Financial rules
Privacy requirements
Rejection policies
Final approval

Digital Exclude’s vibe coding guide explains why AI-generated code still requires testing and human judgment. Developers evaluating coding agents can also read the guide to OpenAI Codex features, uses, and limitations.

Never paste confidential customer files, credentials, payment records, or private company exports into an AI tool unless your organization has approved that use. For related security practices, review Digital Exclude’s guide to cloud security risks and best practices.

Common CSV Cleaning Mistakes

Treating Missing Values as Zero

Unknown and zero are different business facts.

Dropping Every Row With a Missing Cell

A missing optional email should not necessarily remove a customer with valid purchase data.

Loading IDs and Phone Numbers as Numbers

This can remove leading zeros or convert values into scientific notation.

Trusting Automatic Date Parsing

Ambiguous dates can be interpreted incorrectly without producing an obvious error.

Removing All Duplicate IDs Automatically

Duplicate IDs may require conflict resolution, not arbitrary deletion.

Correcting Data Without Preserving the Original

An audit trail is essential when business users need to understand what changed.

Skipping Malformed Rows Silently

A script that “works” by skipping records may create an incomplete import.

Using Regex as Proof That an Email Exists

Syntax validation does not confirm mailbox ownership or deliverability.

Filling Missing Values Without Documenting the Decision

Imputation should be tied to a specific analytical or operational purpose.

Exporting Without Final Assertions

A successful to_csv() call does not mean the data is valid.

Final Recommendation

A good CSV cleaning process should be repeatable, explainable, and safe.

The strongest workflow is:

Inspect the source file
Load identifiers as text
Preserve the original values
Define business rules
Normalize values carefully
Separate warnings from critical failures
Quarantine rejected records
Validate the accepted output
Generate measurable quality metrics
Export without changing the original file

The most important lesson is that data cleaning is not only about making values look consistent. It is about deciding whether each record is trustworthy enough for its intended business use.

A customer import, financial report, machine learning dataset, and email campaign may require different rules. Build the pipeline around the decision the data will support.

For more practical technology and software guides, explore the Digital Exclude Apps and Software section. Readers interested in more advanced Python workflows can also review the guide to building a multi-agent system in Python.

Frequently Asked Questions

1. What is the best way to clean a messy CSV file with Python?

The best approach is to inspect the file first, load important identifiers as strings, standardize columns and text, convert dates and numbers using explicit rules, validate required fields, identify duplicates, and separate rejected records from approved records. Do not silently delete data without recording why it was removed.

2. Should I drop or fill missing values in pandas?

The decision depends on what the column means. Drop a record when a critical required field is missing and the record cannot be used safely. Keep optional values missing when the information is genuinely unknown. Impute values only when there is a documented analytical reason, and record which values were estimated.

3. How can I clean a CSV file that is too large for memory?

Use pd.read_csv() with the chunksize argument and process the file in smaller batches. Remember that duplicates may appear in separate chunks, so cross-chunk validation requires persistent state, a database, or another disk-backed system.

4. Can I clean a CSV file without pandas?

Yes. Python’s built-in csv module can trim whitespace, remove empty rows, replace placeholder values, and stream records without loading the whole file into memory. Pandas is more convenient for data types, duplicate detection, validation, aggregation, and quality reporting.

5. How do I know whether my cleaned CSV is correct?

Define business rules before cleaning and convert those rules into assertions. Check required IDs, uniqueness, allowed categories, valid dates, non-negative amounts, accepted row counts, and rejected row counts. Also compare the input count with the total accepted and rejected records to confirm that no rows disappeared unexpectedly.

Written by