Merge messy monthly CSVs with Python and verify the row counts

2026-11-10来源 dev.to BeginnersEnglish

If you export a CSV every month — from a shop, a CRM, or a reports tool — you eventually want one file that holds all of them, and a simple way to see that no rows disappeared on the way in.

This tutorial walks through a small Python script that:

  1. Reads every .csv file in a folder
  2. Writes one merged CSV
  3. Prints a check of the form rows in = rows out, including a per-file count

It uses the Python standard library only (csv, argparse, and pathlib). You do not need pandas.

The sample rows below are SAMPLE – FICTIONAL. They are made-up demonstration data, not a real shop or a real customer. Every order ID starts with SAMPLE-, and every row is labelled SAMPLE – FICTIONAL.

This script does one job. It does not tidy dates, parse currency, rename columns, or drop duplicates. If a row is messy, it is copied as it is. That is intentional: the check is only about counts, not about cleaning.

What you need

  • Python 3.9 or newer
  • A terminal
  • A folder of .csv files (we will create three short samples first)

Check your Python version:

python3 --version 

On Windows, try python --version if python3 is not recognized.

The sample files

Create a folder called shop_exports and save the three files below inside it. You can also type them by hand; they are short on purpose.

shop_exports/SAMPLE_orders_2026-08.csv:

order_id,date,product,qty,amount,sample_label SAMPLE-2608-001,2026-08-02,Cedar Notebook,1,12.00,SAMPLE – FICTIONAL SAMPLE-2608-002,2026-08-05,Brass Keyring,2,17.00,SAMPLE – FICTIONAL SAMPLE-2608-003,2026-08-07,Linen Pouch,1,18.00,SAMPLE – FICTIONAL SAMPLE-2608-003,2026-08-07,Linen Pouch,1,18.00,SAMPLE – FICTIONAL 

The last row is an exact copy of SAMPLE-2608-003. The script will keep both copies and tell you that it found one exact duplicate.

shop_exports/SAMPLE_orders_2026-09.csv:

order_id,date,product,qty,amount,sample_label SAMPLE-2609-001,2026-09-01,Oak Coaster,2,12.00,SAMPLE – FICTIONAL SAMPLE-2609-002,2026-09-03,Cedar Notebook,1,12.00,SAMPLE – FICTIONAL SAMPLE-2609-003,2026-09-05,Linen Pouch,2,36.00,SAMPLE – FICTIONAL SAMPLE-2609-004,2026-09-07,Brass Keyring,1,8.50,SAMPLE – FICTIONAL SAMPLE-2609-005,2026-09-09,Cedar Notebook,3,36.00,SAMPLE – FICTIONAL 

shop_exports/SAMPLE_orders_2026-10.csv:

order_id,date,product,qty,amount,sample_label SAMPLE-2610-001,2026-10-02,Cedar Notebook,1,12.00,SAMPLE – FICTIONAL SAMPLE-2610-002,2026-10-05,Linen Pouch,1,18.00,SAMPLE – FICTIONAL SAMPLE-2610-003,2026-10-08,Oak Coaster,3,18.00,SAMPLE – FICTIONAL 

Expected totals: 4 + 5 + 3 = 12 data rows in, 12 data rows out, and 1 exact duplicate (kept).

Step 1: List the CSV files

We only look at the top level of the folder, and only at files that end in .csv. A README in the same folder is ignored.

from pathlib import Path def list_csv_files(input_dir: Path) -> list[Path]: return sorted( path for path in input_dir.iterdir() if path.is_file() and path.suffix.lower() == ".csv" ) 

Sorting by name keeps the monthly files in a stable order (2026-08, then 2026-09, then 2026-10).

Step 2: Read each file and skip blank rows

csv.DictReader uses the first row as headers. We open the file with newline="" (the documented way to let the csv module handle line endings) and encoding="utf-8-sig" so a UTF-8 file exported from Excel, which may start with a BOM, still reads cleanly.

A completely blank data row is skipped, so it is not counted on either side of the check. A row with even one non-empty cell is kept.

import csv from pathlib import Path def is_blank_row(values) -> bool: return all(not (value or "").strip() for value in values) def read_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]: with path.open(newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) if not reader.fieldnames: raise ValueError(f"{path.name}: no header row found") headers = list(reader.fieldnames) rows = [] for raw in reader: if is_blank_row(raw.values()): continue rows.append({key: (raw.get(key) or "") for key in headers}) return headers, rows 

Step 3: Stack the rows

If every file has the same headers, this is a straight concatenation. If one file has an extra column, we keep it: the output header is the union of the source headers, in the order they first appear. The tool does not treat order_id and Order ID as the same column. That would be a cleaning step, and this script does not clean.

def union_headers(header_lists: list[list[str]]) -> list[str]: seen: set[str] = set() headers: list[str] = [] for header_list in header_lists: for name in header_list: if name not in seen: seen.add(name) headers.append(name) return headers 

Each output row then uses row.get(column, "") so a missing column becomes an empty cell, not an error.

Step 4: Count exact duplicates, but keep them

An exact duplicate means every source column has the same value as an earlier row. The first copy is the original; later copies are counted. They are still written to the merged file. The check is “did any rows vanish?”, not “please delete repeats for me.”

def count_exact_duplicates(rows: list[dict[str, str]], headers: list[str]) -> int: seen: set[tuple[str, ...]] = set() duplicates = 0 for row in rows: key = tuple(row.get(header, "") for header in headers) if key in seen: duplicates += 1 else: seen.add(key) return duplicates 

A trailing space in one cell makes the row different. That is a limitation worth knowing about.

Step 5: Write the merged file and print the check

Write the stacked rows with csv.DictWriter. Then print one line per source file, the two totals, the duplicate count, and OK or MISMATCH.

On the sample folder the printed check should look like this:

File Rows in -------------------------- -------- SAMPLE_orders_2026-08.csv 4 SAMPLE_orders_2026-09.csv 5 SAMPLE_orders_2026-10.csv 3 -------------------------- -------- TOTAL in 12 TOTAL out 12 Exact duplicates 1 (kept; not removed) Row-count check: 12 in = 12 out OK 

If the two totals ever disagree, the script should exit with a non-zero status so you notice it. In normal use they should match, because every non-blank data row that was read is written back out.

The complete script

Save the following as merge_csvs.py in the same directory as the shop_exports folder.

# merge_csvs.py """Merge every CSV in a folder into one file and check that row counts add up.""" from __future__ import annotations import argparse import csv import sys from pathlib import Path def list_csv_files(input_dir: Path) -> list[Path]: if not input_dir.is_dir(): raise FileNotFoundError(f"Input folder does not exist: {input_dir}") return sorted( path for path in input_dir.iterdir() if path.is_file() and path.suffix.lower() == ".csv" ) def is_blank_row(values) -> bool: return all(not (value or "").strip() for value in values) def read_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]: with path.open(newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) if not reader.fieldnames: raise ValueError(f"{path.name}: no header row found") headers = list(reader.fieldnames) rows = [] for raw in reader: if is_blank_row(raw.values()): continue rows.append({key: (raw.get(key) or "") for key in headers}) return headers, rows def union_headers(header_lists: list[list[str]]) -> list[str]: seen: set[str] = set() headers: list[str] = [] for header_list in header_lists: for name in header_list: if name not in seen: seen.add(name) headers.append(name) return headers def count_exact_duplicates(rows: list[dict[str, str]], headers: list[str]) -> int: seen: set[tuple[str, ...]] = set() duplicates = 0 for row in rows: key = tuple(row.get(header, "") for header in headers) if key in seen: duplicates += 1 else: seen.add(key) return duplicates def format_check(file_counts: list[tuple[str, int]], rows_out: int, duplicates: int) -> str: rows_in = sum(count for _, count in file_counts) name_width = max((len(name) for name, _ in file_counts), default=4) name_width = max(name_width, len("TOTAL in"), len("File")) count_width = 8 lines = [ f"{'File':<{name_width}} {'Rows in':>{count_width}}", f"{'-' * name_width} {'-' * count_width}", ] for name, count in file_counts: lines.append(f"{name:<{name_width}} {count:>{count_width}d}") lines.append(f"{'-' * name_width} {'-' * count_width}") lines.append(f"{'TOTAL in':<{name_width}} {rows_in:>{count_width}d}") lines.append(f"{'TOTAL out':<{name_width}} {rows_out:>{count_width}d}") lines.append( f"{'Exact duplicates':<{name_width}} {duplicates:>{count_width}d}" " (kept; not removed)" ) status = "OK" if rows_in == rows_out else "MISMATCH" lines.append(f"Row-count check: {rows_in} in = {rows_out} out {status}") return "\n".join(lines) + "\n" def merge_folder(input_dir: Path) -> tuple[list[str], list[dict[str, str]], str, bool]: paths = list_csv_files(input_dir) if not paths: raise FileNotFoundError(f"No .csv files were found in {input_dir}") header_lists: list[list[str]] = [] file_counts: list[tuple[str, int]] = [] stacked: list[tuple[str, list[dict[str, str]], list[str]]] = [] for path in paths: headers, rows = read_csv(path) header_lists.append(headers) file_counts.append((path.name, len(rows))) stacked.append((path.name, rows, headers)) output_headers = union_headers(header_lists) merged: list[dict[str, str]] = [] for _filename, rows, _headers in stacked: for row in rows: merged.append({name: row.get(name, "") for name in output_headers}) duplicates = count_exact_duplicates(merged, output_headers) check = format_check(file_counts, len(merged), duplicates) ok = sum(count for _, count in file_counts) == len(merged) return output_headers, merged, check, ok def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Merge CSV files in a folder and print a row-count check." ) parser.add_argument("--input", required=True, type=Path, help="Folder of .csv files") parser.add_argument("--output", required=True, type=Path, help="Merged CSV path") args = parser.parse_args(argv) try: headers, rows, check, ok = merge_folder(args.input) except (FileNotFoundError, ValueError, OSError) as exc: print(f"Error: {exc}", file=sys.stderr) return 1 args.output.parent.mkdir(parents=True, exist_ok=True) with args.output.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=headers) writer.writeheader() writer.writerows(rows) sys.stdout.write(check) print(f"Wrote: {args.output}") return 0 if ok else 2 if __name__ == "__main__": sys.exit(main()) 

Run it:

python3 merge_csvs.py --input shop_exports --output merged.csv 

You should see the 12-in / 12-out check above, and merged.csv should contain 12 data rows plus a header. Open the file and confirm that SAMPLE-2608-003 appears twice.

Limits (please read these)

  • Only .csv files are read. Excel workbooks (.xlsx) are ignored.
  • The default dialect is comma-separated. Semicolon- or tab-separated files need a different csv dialect, which this script does not set.
  • Files are read as UTF-8 (a leading BOM is accepted) and written as UTF-8 without a BOM.
  • Column names are not normalized. If one month says order_id and the next says Order ID, the merged file will have both columns.
  • Dates and amounts are copied as text. 08/31/2026, 31.08.2026, $12.00, and USD 12.00 are left unchanged.
  • “Exact duplicate” means every column matches. Extra spaces, a different date format, or a re-ordered column will hide a repeat.
  • This is not a hosted service. You run the script on your own computer. Nothing is uploaded.

If the printed check ever says MISMATCH, something unexpected happened (for example the output path could not be fully written). Do not trust the merged file until the totals agree.

The free tool

If you'd rather not copy the script by hand, the same idea is packaged as a free, MIT-licensed command-line tool with tests and sample files: csv-merge-check on GitHub.

Disclosure: this article was written with AI assistance. The complete script was run on the sample files shown above before publishing.

原文:https://www.a8q9.cn/en/article/merge-messy-monthly-csvs-with-python-and-verify-the-row-counts-8bf1af.html

相关文章