CSVファイルを結合し、元のファイル名を保持する

012つのファイルを5行に結合する

ファイルを結合した後に数値がおかしく見えると、どの元ファイルから来た値なのか追跡しにくくなります。この例では元の4列を維持し、最後に source_file を追加します。各結果行から元のファイル名を直接確認できるようにすることが目的です。まず、入力ファイルのデータ行数である2行と3行を足し、期待される合計が5行であることを書き留めます。

1. 展開したフォルダ内の inputs/sales_01.csv と inputs/sales_02.csv を探します。
2. 両方のファイルをテキストエディタで開き、1行目が date,item,quantity,unit_price になっていることを確認します。
3. example.py があるフォルダでターミナルを開き、以下のコマンドを実行します。
4. outputs/merged.csv を開き、データ行が5行あり、source_file 列が追加されていることを確認します。

python example.py

入力フォルダと出力フォルダを分けることで、以前作成した merged.csv が次回の実行時に入力ファイルとして取り込まれるのを防げます。コードは example.py を基準に inputs も探します。ZIPを展開したときのフォルダ構成をそのまま維持していれば、パスを変更する必要はありません。

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행」（2行）、「sales_02.csv: 3행」（3行）、「완료: 2개 파일 → 5행」（完了: 2ファイル → 5行）と表示されます。1行目のヘッダーはデータ行として数えません。노트（ノート）が2回出てきてもエラーではありません。元の2行はどちらも保持され、合計計算や重複削除は行っていません。

05結果を信頼する前に4つ確認する

1. 入力データ行数の合計5が、結果のデータ行数と一致することを確認します。
2. source_file が sales_01.csv の行が2行、sales_02.csv の行が3行あることを確認します。
3. カンマを含む item 名が1つのセルに残り、二重引用符も消えていないことを確認します。
4. 元の2ファイルをもう一度開き、内容が変更されていないことを確認します。また、outputs がすでに存在するため、スクリプトを再実行すると停止することも確認します。

実務では、行数の確認と合計値の確認は別の作業です。金額の合計が偶然一致していても、1行が欠け、別の1行が重複している可能性があります。この段階では計算を行わないため、まず元ファイル別の行数と元のテキストを比較してください。

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

再実行する前に、最初の outputs フォルダを別名に変更して保存します。inputs 内に同じヘッダーとデータ1行を持つ sales_03.csv を追加します。期待されるデータ行数は6です。新しい行の source_file が sales_03.csv になっていることを確認し、入力追加が追跡列にどう反映されるかを見てください。

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

07不完全な結果を受け入れず、エラーを確認する

エラーの行番号はCSVから読み取った物理行を指します。引用符内に改行がある複雑なCSVでは、スプレッドシート上のデータ行番号と一致しない場合があります。まずメッセージに表示されたファイル名を使って原因箇所を絞り込んでください。

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

このスクリプトは列構成、空の値、CSV構文を確認します。quantity が数値か、date が有効な日付か、amount が業務ルールに合っているかは判断しません。単純な結合後に計算が必要な場合は、数値変換や許容範囲の確認を別の検証ステップとして追加してください。

すべての行をメモリに集め、検証後に保存するため、この例は小規模な業務ファイルでの練習に適しています。非常に大きなCSVでは、一時出力と段階的な処理を使う設計が必要です。保存中のディスクエラーで不完全な結果が残る可能性があるため、完了メッセージが出ていない実行を成功とみなさないでください。同時に入力ファイルが変更される場合やネットワークパスの障害は、この検証範囲には含まれません。

実行・検証の記録

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

• 2つのCSVファイルを5行に結合し、source_file の件数を確認
• カンマと二重引用符を含むセルを保持
• 出力作成前に、不正なヘッダー、不足または余分なフィールド、空の値、不正な引用符を拒否
• ヘッダーだけのCSVとCSVファイルが存在しない場合の処理を確認
• 元ファイルのSHA-256ハッシュが変わっていないことと、既存出力が保持されることを確認

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

参考資料

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

• Python公式ドキュメント · csv と DictReader ↗ (New tab)
• Python公式ドキュメント · pathlib のファイルパス ↗ (New tab)