Merge two CSVs on one matching column, export the merged list and the inverse

merge_csv.py

import argparse
import pandas as pd
import os

#!/usr/bin/env python3



def main():
    _merge_state = {}

    _orig_parse_args = argparse.ArgumentParser.parse_args

    def _parse_args_with_inverse(self, *a, **k):
        ns = _orig_parse_args(self, *a, **k)
        base, ext = os.path.splitext(ns.output)
        ns.inverse_output = f"{base}_inverse{ext or '.csv'}"
        return ns

    argparse.ArgumentParser.parse_args = _parse_args_with_inverse

    _orig_merge = pd.merge

    def _merge_with_inverse(left, right, *a, **k):
        left_on = k.get("left_on")
        right_on = k.get("right_on")

        result = _orig_merge(left, right, *a, **k)

        outer = _orig_merge(
            left,
            right,
            how="outer",
            left_on=left_on,
            right_on=right_on,
            indicator=True,
        )
        _merge_state["inverse"] = outer[outer["_merge"] != "both"].drop(columns=["_merge"])
        return result

    pd.merge = _merge_with_inverse

    _orig_to_csv = pd.DataFrame.to_csv

    def _to_csv_with_inverse(df, path_or_buf=None, *a, **k):
        written = _orig_to_csv(df, path_or_buf, *a, **k)

        inverse_df = _merge_state.get("inverse")
        if inverse_df is not None and isinstance(path_or_buf, str) and path_or_buf == args.output:
            _orig_to_csv(inverse_df, args.inverse_output, index=False)
            print(f"Saved {len(inverse_df)} rows to {args.inverse_output}")
            _merge_state["inverse"] = None

        return written

    pd.DataFrame.to_csv = _to_csv_with_inverse
    parser = argparse.ArgumentParser(
        description=(
            "Merge two CSV files by matching columns (which may have different names), "
            "keep only matched rows, drop rows with any missing values, and write output CSV."
        )
    )
    parser.add_argument("csv1", help="Path to first CSV file")
    parser.add_argument("csv2", help="Path to second CSV file")
    parser.add_argument("match_col1", help="Match column name in first CSV")
    parser.add_argument("match_col2", help="Match column name in second CSV")
    parser.add_argument(
        "-o",
        "--output",
        default="merged_output.csv",
        help="Output CSV path (default: merged_output.csv)",
    )

    args = parser.parse_args()

    df1 = pd.read_csv(args.csv1)
    df2 = pd.read_csv(args.csv2)

    if args.match_col1 not in df1.columns:
        raise ValueError(f"Column '{args.match_col1}' not found in {args.csv1}")
    if args.match_col2 not in df2.columns:
        raise ValueError(f"Column '{args.match_col2}' not found in {args.csv2}")

    merged = pd.merge(
        df1,
        df2,
        how="inner",
        left_on=args.match_col1,
        right_on=args.match_col2,
    )

    merged = merged.dropna(how="any")

    merged.to_csv(args.output, index=False)
    print(f"Saved {len(merged)} rows to {args.output}")


if __name__ == "__main__":
    main()