Merges multiple CSV files covering the same underlying data into a single dataset, even when the files use different column names. Exact (case-insensitive) column matching runs first; difflib.SequenceMatcher fuzzy matching handles variations like vendor_name / vendor_nm / Vendor Name. A _source column tracks file origin in the output. Optional deduplication removes repeated records.
Data from different systems, vendors, or reporting periods often lands in files with inconsistent headers: vendor_name, vendor_nm, Vendor Name, and contractor may all refer to the same field. Manually aligning and stacking those files is tedious and does not scale when new files arrive on a recurring basis.
Government contract data is a common example: quarterly exports from agency procurement systems evolve column names as forms change. Combining them for year-over-year analysis requires either manual header reconciliation in Excel — which is error-prone and not reproducible — or a programmatic approach that handles name variations automatically and traces each record back to its source file.
This tool automates the column alignment step and produces one clean, traceable dataset:
- Which files contain each column, and which columns are missing from some files?
- Which column in file B corresponds to the canonical column from file A?
- After stacking, which records came from which file?
- What is the null rate per column across the consolidated output?
| Feature | Description |
|---|---|
| Fuzzy column mapping | difflib.SequenceMatcher from the Python standard library — no external dependency; adjustable threshold slider (0.50–1.00) |
| Schema coverage matrix | Color-coded YES/— table showing which columns are present in which files |
| Column coverage chart | Horizontal bar chart of the percentage of files that contain each column |
| Auto-detected mapping table | Per-file view of which source column was mapped to each canonical column; unmapped columns highlighted in red |
| Dropped columns report | Lists every column that could not be mapped and was excluded from the output |
| Null quality report | Per-column null count and null % in the consolidated output with bar chart |
| Deduplication | Toggle on/off; key on a specific column or full row; before/after counts shown in the header |
| Export | Consolidated CSV, column mapping reference CSV, and dropped columns report CSV |
| Skill | Implementation |
|---|---|
| Algorithm design | difflib.SequenceMatcher fuzzy matching with configurable threshold; exact-first precedence; one-to-one mapping enforcement |
| Data structures | ConsolidationResult dataclass (7 fields); full_mapping nested dict {canonical: {file: original_or_None}} |
| Schema alignment | First-file-as-canonical pattern; build_full_mapping() → mapping_to_rename_dicts() → stack_dataframes() pipeline |
| Pandas | pd.concat for vertical stacking; drop_duplicates with subset; isna().sum() for null profiling |
| Caching strategy | @st.cache_data at two levels: file load and full consolidation run; consolidation keyed by JSON serialization |
| Plotly visualization | Heatmap-style coverage matrix; column coverage bar chart; source-file pie chart; null % bar chart |
| Test coverage | 70 tests across 9 classes in 3 files; edge cases: empty file list, missing canonical columns, None mappings, full-row dedup vs. key-based dedup |
| Compliance framing | _source column for row-level provenance; dropped columns report for audit; null summary for data quality sign-off |
multi-file-consolidator/
├── app/main.py ← Streamlit UI: 5 tabs, sidebar, header metrics
│
├── src/
│ ├── schema.py ← infer_schema; coverage_matrix; column_coverage_pct;
│ │ suggest_mapping; build_full_mapping; mapping_to_rename_dicts;
│ │ SIMILARITY_THRESHOLD = 0.70
│ ├── consolidator.py ← ConsolidationResult dataclass; SOURCE_COL = "_source";
│ │ apply_renames; stack_dataframes; deduplicate; consolidate
│ └── reporter.py ← consolidation_summary; source_breakdown;
│ dropped_columns_report; null_summary
│
├── data/
│ ├── q1_contracts.csv ← Canonical schema (9 cols, ~60 rows)
│ ├── q2_contracts.csv ← Same domain, some columns renamed
│ ├── q3_contracts.csv ← Additional naming variations
│ └── q4_contracts.csv ← Additional naming variations
│
├── tests/
│ ├── test_schema.py ← 30 tests: 6 classes covering schema and mapping
│ ├── test_consolidator.py ← 22 tests: 4 classes covering the consolidation pipeline
│ └── test_reporter.py ← 18 tests: 4 classes covering summary functions
│
└── docs/
├── ARCHITECTURE.md ← Module reference, algorithm, design decisions
├── TESTING.md ← Test inventory, per-class breakdown, CI
└── DATA_DICTIONARY.md ← Input/output schemas, public API reference
file_schemas (dict[file: list[col]])
│
▼ build_full_mapping(file_schemas, threshold)
│ └─ suggest_mapping(canonical_cols, target_cols, threshold) per file
│ ├─ Exact (case-insensitive) match → map immediately
│ └─ Fuzzy: SequenceMatcher.ratio() → map if ≥ threshold
│
▼ full_mapping: {canonical: {file: original_col_or_None}}
│
▼ mapping_to_rename_dicts(full_mapping)
│
▼ rename_dicts: {file: {original_col: canonical_col}}
│
▼ consolidate(named_dfs, rename_dicts, target_cols, ...)
├─ apply_renames(df, rename_map) per file
├─ stack_dataframes(renamed, target_cols) → stacked + dropped
└─ deduplicate(stacked, dedup_col) → final ConsolidationResult
| Field | Type | Description |
|---|---|---|
consolidated |
DataFrame |
Final stacked + deduped output with _source column |
source_counts |
dict[str, int] |
Row count contributed by each source file (pre-dedup) |
cols_kept |
list[str] |
Canonical columns in the output |
cols_dropped |
dict[str, list[str]] |
Columns excluded per file (not mapped to any canonical column) |
rows_before_dedup |
int |
Raw row count before deduplication |
rows_after_dedup |
int |
Final row count |
dedup_removed |
int |
Rows removed by deduplication |
multi-file-consolidator/
├── .github/
│ └── workflows/
│ └── tests.yml # CI — runs pytest on push and pull_request
├── .streamlit/
│ └── config.toml # Light theme configuration
├── app/
│ └── main.py # Streamlit entry point
├── data/
│ ├── q1_contracts.csv # Canonical quarterly contract snapshot
│ ├── q2_contracts.csv # Q2 — naming variations
│ ├── q3_contracts.csv # Q3 — naming variations
│ └── q4_contracts.csv # Q4 — naming variations
├── docs/
│ ├── ARCHITECTURE.md # Design decisions and module reference
│ ├── DATA_DICTIONARY.md # Input/output schemas and API reference
│ ├── ENGINEERING_DECISIONS.md # Six annotated architecture decisions with trade-offs
│ └── TESTING.md # Test inventory and coverage details
├── scripts/
│ └── generate_sample_data.py # Sample data generator
├── screenshots/
│ ├── 01_overview.png # Application overview with schema coverage matrix
│ ├── 02_core_feature.png # Column mapping and fuzzy match results
│ └── 03_results.png # Consolidated data output and quality report
├── src/
│ ├── __init__.py # Public API exports
│ ├── consolidator.py # ConsolidationResult, consolidate pipeline
│ ├── reporter.py # Summary and report functions
│ └── schema.py # Fuzzy matching and schema inspection
├── tests/
│ ├── test_consolidator.py # Consolidation pipeline tests
│ ├── test_reporter.py # Reporter function tests
│ └── test_schema.py # Schema and mapping tests
├── CHANGELOG.md
├── LICENSE
└── requirements.txt
Python 3.10+ is required.
git clone https://github.com/RichieGarafola/multi-file-consolidator.git
cd multi-file-consolidator
python -m venv .venv
# Windows
.venv\Scripts\activate
# macOS / Linux
source .venv/bin/activate
pip install -r requirements.txtstreamlit run app/main.pyThe app opens at http://localhost:8501 by default.
Steps:
- Open the sidebar and toggle Use sample data (enabled by default, loads four quarterly contract files)
- To use your own files, disable sample data and upload 2–10 CSV files
- Adjust the Fuzzy match threshold slider if the auto-mapping misses or incorrectly maps columns
- Review the Schema Map tab to see which columns are present across all files
- Review the Column Mapping tab to verify the auto-detected column assignments
- Use the Consolidated Data tab to preview and filter the merged output
- Export from the Export tab
Sidebar controls:
| Control | Description |
|---|---|
| Use sample data | Loads the four quarterly CSV files automatically |
| Upload CSV files | Upload 2–10 CSV files when sample data is off |
| Fuzzy match threshold | Slider 0.50–1.00 (default 0.70); lower values accept more distant matches |
| Columns to Keep | Multiselect of canonical columns; deselect any to exclude from output |
| Remove duplicates | Toggle deduplication on/off |
| Dedup key column | Column to deduplicate on, or "(full row)" for exact-row deduplication |
Tabs:
| Tab | Contents |
|---|---|
| Schema Map | Coverage matrix (YES/—) and column coverage % bar chart |
| Column Mapping | Auto-detected mapping table; dropped columns listed below |
| Consolidated Data | Full merged table with source filter; pie chart of rows by source file |
| Quality Report | Null count and null % per column; bar chart of columns with nulls |
| Export | Download consolidated CSV, column mapping CSV, dropped columns CSV |
Programmatic usage:
import pandas as pd
from src import (
SIMILARITY_THRESHOLD, build_full_mapping, mapping_to_rename_dicts,
consolidate, consolidation_summary, dropped_columns_report,
)
file_schemas = {
"q1.csv": pd.read_csv("data/q1_contracts.csv", dtype=str).columns.tolist(),
"q2.csv": pd.read_csv("data/q2_contracts.csv", dtype=str).columns.tolist(),
}
full_mapping = build_full_mapping(file_schemas, threshold=SIMILARITY_THRESHOLD)
rename_dicts = mapping_to_rename_dicts(full_mapping)
canonical_cols = list(full_mapping.keys())
named_dfs = [(name, pd.read_csv(f"data/{name}", dtype=str)) for name in file_schemas]
result = consolidate(named_dfs, rename_dicts, canonical_cols, dedup_col="contract_id")
summary = consolidation_summary(result)
print(f"Files: {summary['total_files']}, Rows: {summary['total_rows']}, Deduped: {summary['rows_deduped']}")
result.consolidated.to_csv("consolidated.csv", index=False)
dropped_columns_report(result).to_csv("dropped_columns.csv", index=False)Four quarterly contract files with the same underlying domain but intentionally inconsistent column naming to exercise the fuzzy matching path.
| File | Rows | Notes |
|---|---|---|
q1_contracts.csv |
~60 | Defines canonical schema (9 columns) |
q2_contracts.csv |
~60 | Some columns renamed — demonstrates fuzzy-match path |
q3_contracts.csv |
~50 | Additional naming variations |
q4_contracts.csv |
~45 | Additional naming variations |
Canonical schema (q1_contracts.csv): contract_id, vendor_name, award_amount, contract_type, status, project_manager, agency, start_date, end_date
Regenerate with: python scripts/generate_sample_data.py
pytest tests/ -v70 tests across 9 classes in 3 files; all passing under 2 seconds.
| File | Class | Tests | Scope |
|---|---|---|---|
test_schema.py |
TestInferSchema |
4 | Dict shape, name stored, column list, row count |
test_schema.py |
TestCoverageMatrix |
6 | All columns in index, all files in columns, True/False values, empty input |
test_schema.py |
TestColumnCoveragePct |
5 | 100% for shared columns, 50% for single-file columns, sorted descending |
test_schema.py |
TestSuggestMapping |
7 | Exact match, case-insensitive exact, fuzzy match, no-match → None, one-to-one, empty inputs |
test_schema.py |
TestBuildFullMapping |
4 | Returns dict, canonical from first file, exact match, empty input |
test_schema.py |
TestMappingToRenameDicts |
4 | Rename created, no rename for same name, None not included |
test_consolidator.py |
TestApplyRenames |
4 | Renames correctly, no mutation, unknown columns ignored |
test_consolidator.py |
TestStackDataframes |
6 | _source column added, stacks multiple files, missing canonical filled with None, extra columns tracked |
test_consolidator.py |
TestDeduplicate |
4 | Key-based dedup, full-row dedup, no duplicates unchanged, reset index |
test_consolidator.py |
TestConsolidate |
8 | Full pipeline, source counts, dedup reduces rows, no-dedup keeps all, rename applied, cols_kept, empty list |
test_reporter.py |
TestConsolidationSummary |
5 | Required keys, file count, row count, column count |
test_reporter.py |
TestSourceBreakdown |
4 | DataFrame shape, all sources present, row counts correct |
test_reporter.py |
TestDroppedColumnsReport |
4 | Empty when no drops, reports dropped column, column names present |
test_reporter.py |
TestNullSummary |
5 | No nulls → all zero, detects nulls, empty result → empty report |
- First file defines the schema — canonical columns come entirely from the first uploaded file. A column absent from the first file will not appear in the output.
- Fuzzy matching can make incorrect calls — at lower threshold values, short or similarly-named columns may be mapped to the wrong canonical column. Always review the Column Mapping tab before exporting.
- Each target column is used at most once — if two canonical columns both score highly against the same target column, the better match wins.
- No manual mapping override — column assignments are automatic. If an auto-mapping is wrong, rename the column in the source file or adjust the threshold.
- All values treated as strings — files are read with
dtype=str. Numeric type inference is left to the consumer. - Deduplication keeps first occurrence — when deduplicating by key column, the row from the first file that contributed that key is retained.
- Minimum 2 files required — the app stops if fewer than 2 files are provided.
- Manual mapping override in the UI when auto-detection is wrong
- "Latest wins" deduplication mode — keep the row from the most recently modified file
- Pre-merge value standardization — normalize date formats, strip currency symbols before stacking
- Support for Excel files (
.xlsx) alongside CSV - User-defined canonical schema from a template file
- Column merging — combine two source columns into one canonical column
| Overview | Core Feature | Results |
|---|---|---|
 |
 |
 |
| Application overview with schema coverage matrix | Column mapping and fuzzy match results | Consolidated data output and quality report |
| Document | Description |
|---|---|
| Architecture and Design Decisions | Module reference, algorithm design, and key technical decisions |
| Engineering Decisions | Six annotated architecture decisions with alternatives and trade-off rationale |
| Test Inventory and Coverage | Test class breakdown, coverage details, and CI configuration |
| Data Dictionary and API Reference | Input/output schemas, ConsolidationResult fields, and public API reference |
MIT License — see LICENSE for details.