여러 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 파일만 대상입니다. 하위 폴더를 재귀적으로 뒤지지 않으며 텍스트 메모나 엑셀 통합문서는 합치지 않습니다. 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 파일 경로 ↗ (새 탭)