複数のCSVを結合し、元のファイル名を残す

01二つのファイルを五行にまとめる

複数のファイルを結合した後に数値がおかしいと、どの元ファイルから来たのか探しにくくなります。この例では既存の四列を残し、最後にsource_fileを追加します。結果の各行から元のファイル名をすぐに確認できるようにするのが目的です。まず、二つのファイルのデータ行数2と3を足し、期待値の5を書き留めてください。

1. 展開したフォルダーでinputs/sales_01.csvとinputs/sales_02.csvを確認します。
2. 二つのファイルをテキストエディターで開き、先頭行がdate,item,quantity,unit_priceになっているか確認します。
3. example.pyがあるフォルダーのターミナルで、以下のコマンドを実行します。
4. outputs/merged.csvを開き、五行のデータとsource_file列を確認します。

python example.py

入力と出力のフォルダーを分けているため、再実行しても前のmerged.csvが入力ファイルに混ざりません。コードが入力を探す基準も、example.pyの場所に固定しています。フォルダー構成を保って展開すれば、パスを別途修正する必要はありません。

02結合前に列構成を統一する

列名だけでなく、順序も同じである必要があります。たとえば、あるファイルでquantityの代わりにqtyを使ったり、unit_priceの位置を変えたりすると停止します。意味を推測して自動で合わせる機能を設けていないため、別の部署の書式が混ざったことにすぐ気づけます。

合成データには「메모지, 대형」と、二重引用符を含む「표지 "파랑"」を入れています。CSVを単純にカンマでsplitすると、このような値を誤って分割しやすくなります。引用の規則は標準のcsvモジュールに任せ、読み取った行をそのまま書き戻します。

03実行に使うコード全体

collect_rowsは各CSVのヘッダーと行を確認し、元のファイル名を追加します。すべての検査を通過した後で、mainが新しい出力を作ります。後のファイルの列に問題があっても、先のファイルだけを結合した結果が残らないよう、この順序にしています。

"""inputs의 같은 구조 CSV를 결합합니다. 각 행에 원본 파일명을 붙입니다."""

import csv
import sys
from pathlib import Path

BASE = Path(__file__).resolve().parent
INPUT = BASE / "inputs"
OUTPUT = BASE / "outputs"
FIELDS = ["date", "item", "quantity", "unit_price"]


def collect_rows():
    if INPUT.is_symlink() or not INPUT.is_dir():
        raise ValueError("inputs 폴더가 없거나 링크입니다.")
    if getattr(INPUT, "is_junction", lambda: False)():
        raise ValueError("연결 디렉터리는 처리하지 않습니다.")
    # inputs 바로 아래의 .csv 파일만 읽습니다. 순서를 명시해 결과를 재현합니다.
    paths = sorted(
        (p for p in INPUT.iterdir() if p.suffix.lower() == ".csv"),
        key=lambda p: p.name.casefold(),
    )
    if not paths:
        raise ValueError("inputs에 CSV 파일이 없습니다.")
    merged = []
    counts = []
    for path in paths:
        if path.is_symlink() or not path.is_file():
            raise ValueError(f"일반 CSV 파일이 아닙니다: {path.name}")
        with path.open("r", encoding="utf-8-sig", newline="") as stream:
            reader = csv.DictReader(stream, strict=True)
            if reader.fieldnames != FIELDS:
                raise ValueError(f"{path.name}: 열 이름과 순서는 {FIELDS}여야 합니다.")
            count = 0
            for row in reader:
                # 열이 많으면 None 키, 적으면 None 값이 생깁니다.
                if None in row or any(value is None or not value.strip() for value in row.values()):
                    raise ValueError(f"{path.name}, {reader.line_num}행: 열 수 또는 빈 값을 확인하세요.")
                merged.append({**row, "source_file": path.name})
                count += 1
            counts.append((path.name, count))
    return merged, counts


def main():
    if OUTPUT.exists() or OUTPUT.is_symlink():
        raise FileExistsError("outputs가 이미 있습니다. 기존 결과를 옮긴 뒤 실행하세요.")
    rows, counts = collect_rows()  # 모든 파일을 검증한 다음에 출력합니다.
    OUTPUT.mkdir()
    target = OUTPUT / "merged.csv"
    with target.open("x", encoding="utf-8-sig", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=FIELDS + ["source_file"])
        writer.writeheader()
        writer.writerows(rows)
    for name, count in counts:
        print(f"{name}: {count}행")
    print(f"완료: {len(counts)}개 파일 → {len(rows)}행")
    print(target)
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, csv.Error) as error:
        print(f"중지: {error}", file=sys.stderr)
        raise SystemExit(2)

DictReaderでは、データの列が多すぎるとNoneキー、少なすぎるとNone値が生じることを検査に利用しています。空文字列や空白だけの値も拒否します。newlineに空文字列を指定し、UTF-8 BOMを処理する文字コードを明示して、改行と韓国語を一貫して扱います。

04行数と出典を併せて照合する

ターミナルには順に「sales_01.csv: 2행」「sales_02.csv: 3행」「완료: 2개 파일 → 5행」と表示されます。結果の先頭行であるヘッダーはデータ数に含めません。노트が二行あるのはエラーではありません。異なる元の行を保持した結果であり、合計計算や重複削除は行っていません。

05結果を使う前に四つの点を確認する

1. 各入力の行数の合計5が、結果のデータ行数と一致するか確認します。
2. source_fileがsales_01.csvの行は二つ、sales_02.csvの行は三つあるか確認します。
3. カンマを含む品目名が一つのセルに収まり、二重引用符も残っているか確認します。
4. 元の二つのファイルを開き直して内容が変わっていないか確認し、再実行すると既存のoutputsによって停止するか確認します。

実務では、行数の確認と合計の確認は別々の検査です。金額の合計が偶然一致していても、一行が抜け、別の行が二回入っているかもしれません。この段階ではまだ計算を行わないため、まず出典ごとの件数と元のテキストを照合する方が明確です。

06三つ目のファイルを追加して練習する

再実行するには、最初のoutputsを別名にして保管します。inputsにsales_03.csvを追加し、同じヘッダーとデータ一行を入れてみてください。期待されるデータ数は六つです。新しい行のsource_fileにsales_03.csvが入ることまで確認すると、入力の追加と追跡列の関係が分かります。

対象はinputs直下にある拡張子.csvのファイルだけです。サブフォルダーを再帰的に探したり、テキストメモやExcelブックを結合したりはしません。CSVファイルが一つもなければ、入力の選択が誤っているとして停止します。ヘッダーだけの正常なCSVは0行として数えるため、他のファイルの結合を妨げません。

07部分的な結果ではなくエラーを確認する

エラーの行番号は、CSVを読み取った物理的な行を基準にしています。引用された値の中に改行を含む複雑なCSVでは、スプレッドシートのデータ行番号と異なる場合があります。まずメッセージのファイル名を使って、調べる元ファイルを絞ってください。

08結合とデータ検証の境界

このスクリプトは、列構成、空の値、CSV構文を検査します。quantityが数値か、dateが実在する日付か、金額が業務ルールに合っているかまでは判断しません。結合後に計算が必要なら、数値への変換と許容範囲の確認を、別の検証段階として追加する必要があります。

すべての行をメモリに集め、検査完了後に保存するため、小さな業務ファイルでの練習に適しています。非常に大きなCSVには、一時出力と段階的な処理の設計が必要です。保存中にディスクエラーが起きると部分的な結果が残る場合があるため、完了メッセージがなければ成功と見なさないでください。同時に入力が変更される状況や、ネットワークパスの障害は検証範囲外です。

実行・検証の記録

2026-09-19 · Windows 11 · CPython 3.12.14 · 追加パッケージなし · 配布版の一時コピーで実行

• 2個のCSVを5行に結合し、source_fileごとの件数を確認
• カンマや二重引用符を含むセルを保持
• ヘッダー不正、列の不足・追加、空の値、不正な引用符を出力作成前に拒否
• ヘッダーのみのCSVと、CSVがない場合を処理
• 元ファイルのSHA-256一致と既存の出力の保全を確認

サイト全体の執筆・検証方針

参考資料

説明と例は独自に作成しました。関連する動作や概念は、以下の公式資料で確認できます。

• Python公式ドキュメント · csvとDictReader ↗ (新しいタブ)
• Python公式ドキュメント · pathlibによるファイルパス ↗ (新しいタブ)