파이썬 CSV 보고서 자동화 - 여러 파일 합치기와 오류 행 기록 JSON 요약 (파이썬 자동화 3단원)
이 단원에서 배우는 것
사무 자동화에서 가장 흔한 일은 "여러 사람이 보낸 표를 하나로 합쳐 요약하기"다. 엑셀로 복사해 붙이고 피벗 표를 만드는 일을 매주 반복하고 있다면 이 단원의 스크립트로 대신할 수 있다. 7단원에서 CSV와 JSON을 읽고 쓰는 법을, 10단원에서 예외 처리를 배웠다. 이번에는 둘을 합쳐 일부 데이터가 틀려도 끝까지 돌고, 무엇이 틀렸는지 정확히 보고하는 보고서 스크립트를 만든다.
- 폴더 안의 CSV 여러 개를
glob으로 모아 한 번에 집계한다. - 틀린 행은 건너뛰되 파일 이름·줄 번호·이유를
errors.csv에 남긴다. - 엑셀에서 한글이 깨지지 않는 CSV와, 다른 프로그램이 읽을 JSON 요약을 동시에 만든다.
문제 상황
서점은 강남·부산·판교 세 지점이 있다. 지점장은 매일 밤 판매 내역을 20260922_강남.csv 같은 이름으로 본사 공유 폴더에 올린다. 본사 담당자는 다음 날 아침 세 파일을 열어 도서별 판매 순위와 지점별 매출을 정리해 대표에게 보낸다.
그런데 지점 파일이 늘 깨끗하지는 않다.
- 부산 지점은 수량 칸에
두권이라고 한글로 적기도 한다. - 제목 칸을 비워 둔 행이 있다.
- 판교 지점은 반품을
-1로 적는다. 반품은 따로 처리하기로 했으므로 매출에 넣으면 안 된다. - 강남 지점은 가격을
"33,000"처럼 천 단위 콤마를 넣어 적는다. - 판교 지점은 엑셀에서 "CSV UTF-8"로 저장해서 파일 맨 앞에 보이지 않는 표시(BOM)가 붙어 있다.
손으로 할 때는 담당자가 눈으로 보고 알아서 고쳤다. 스크립트는 이런 행을 만나면 멈추거나, 더 나쁘게는 조용히 틀린 합계를 낸다. 목표는 멈추지도 말고 숨기지도 말 것이다. 10단원에서 말한 "배치 한 건의 실패로 전체가 죽지 않게 하되 실패를 숨기지 않는다"를 실제로 구현한다.
그림 · 지점별 CSV 를 합치고 잘못된 행을 따로 모으는 흐름 — 세 지점 파일 11행을 한 행씩 검사해 정상 8행은 도서별·지점별 합계로, 잘못된 3행은 errors.csv 로 보낸다. 멈추지 않되 버린 행을 숨기지 않는다.
완성 스크립트
report.py로 저장한다. 사용법은 python3 report.py 입력폴더 출력폴더다.
"""지점별 판매 CSV 여러 개를 합쳐 도서별·지점별 합계 보고서를 만든다."""
import csv
import json
import sys
from pathlib import Path
REQUIRED = ["date", "title", "qty", "price"]
def to_int(text, label):
try:
return int(text.replace(",", ""))
except ValueError:
raise ValueError(f"{label}이 숫자가 아님: {text!r}") from None
def parse_row(row):
"""정상이면 (qty, amount), 아니면 ValueError 를 던진다."""
missing = [col for col in REQUIRED if not (row.get(col) or "").strip()]
if missing:
raise ValueError("빈 칸: " + ", ".join(missing))
qty = to_int(row["qty"], "수량")
price = to_int(row["price"], "단가")
if qty <= 0:
raise ValueError(f"수량이 0 이하: {qty}")
return qty, qty * price
def collect(input_dir):
by_title = {}
by_branch = {}
errors = []
for path in sorted(input_dir.glob("*.csv")):
branch = path.stem.split("_")[-1]
with open(path, encoding="utf-8-sig", newline="") as f:
for line_no, row in enumerate(csv.DictReader(f), start=2):
try:
qty, amount = parse_row(row)
except ValueError as e:
errors.append({"file": path.name, "line": line_no, "reason": str(e)})
continue
title = row["title"].strip()
item = by_title.setdefault(title, {"qty": 0, "amount": 0})
item["qty"] += qty
item["amount"] += amount
by_branch[branch] = by_branch.get(branch, 0) + amount
return by_title, by_branch, errors
def write_reports(out_dir, by_title, by_branch, errors):
out_dir.mkdir(parents=True, exist_ok=True)
ranking = sorted(by_title.items(), key=lambda kv: kv[1]["amount"], reverse=True)
with open(out_dir / "summary.csv", "w", encoding="utf-8-sig", newline="") as f:
writer = csv.writer(f)
writer.writerow(["순위", "도서", "수량", "매출"])
for rank, (title, item) in enumerate(ranking, start=1):
writer.writerow([rank, title, item["qty"], item["amount"]])
with open(out_dir / "errors.csv", "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["file", "line", "reason"])
writer.writeheader()
writer.writerows(errors)
report = {
"total_amount": sum(by_branch.values()),
"by_branch": by_branch,
"top3": [title for title, _ in ranking[:3]],
"error_count": len(errors),
}
with open(out_dir / "report.json", "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
return report
def main():
if len(sys.argv) != 3:
print("사용법: python3 report.py 입력폴더 출력폴더")
sys.exit(2)
input_dir, out_dir = Path(sys.argv[1]), Path(sys.argv[2])
by_title, by_branch, errors = collect(input_dir)
report = write_reports(out_dir, by_title, by_branch, errors)
print(f"총매출 {report['total_amount']:,}원")
for branch, amount in sorted(by_branch.items()):
print(f" {branch}: {amount:,}원")
print(f"건너뛴 행 {len(errors)}건 -> {out_dir / 'errors.csv'}")
if __name__ == "__main__":
main()한 줄씩 해설
to_int와 parse_row — 한 행의 검사를 한곳에 모은다
parse_row는 CSV 한 행(딕셔너리)을 받아 정상이면 (수량, 금액)을 돌려주고, 이상하면 ValueError를 던진다. 검사 규칙이 이 함수 하나에 모여 있어서 "반품도 받아 주자"는 요청이 오면 여기만 고치면 된다.
먼저 필수 칸이 비었는지 본다. row.get(col) or ""는 열 자체가 없어서 None이 오는 경우까지 빈 문자열로 바꾼다. 거기에 .strip()을 붙여 공백만 들어 있는 칸도 빈 칸으로 친다. 빈 칸이 여러 개면 한 번에 모두 알려 준다.
to_int는 콤마를 지우고 int()로 바꾼다. 실패하면 파이썬의 영어 메시지(invalid literal for int() with base 10) 대신 어느 칸이 무슨 값이라서 틀렸는지 한국어로 다시 던진다. 오류 보고서를 읽는 사람은 지점 직원이지 개발자가 아니다. from None은 원래 예외의 긴 추적 정보를 붙이지 말라는 뜻이다. {text!r}는 값을 따옴표와 함께 보여 줘서 ' 3'처럼 공백이 섞인 값도 눈에 띄게 한다.
금액은 2단원에서 강조한 대로 float이 아니라 정수(원 단위)로 계산한다. 소수가 필요한 통화라면 decimal.Decimal을 쓴다.
그림 · parse_row 가 한 행을 검사하는 순서 — 빈 칸, 수량 숫자, 단가 숫자, 수량 0 이하를 차례로 본다. 실습 데이터의 잘못된 세 행은 각각 다른 단계에서 걸리고, 33,000 처럼 쉼표가 든 단가는 쉼표를 지운 뒤 통과한다.
collect — 파일 여러 개, 행 여러 개를 돈다
for path in sorted(input_dir.glob("*.csv")):
branch = path.stem.split("_")[-1]입력 폴더의 CSV를 이름순으로 모두 읽는다. 지점 이름은 파일 이름 규칙(날짜_지점.csv)에서 꺼낸다. stem이 20260922_강남이니 _로 잘라 마지막 조각을 쓴다. 파일 이름 규칙에 기대는 코드이므로 규칙을 지점과 합의해 두어야 한다.
with open(path, encoding="utf-8-sig", newline="") as f:
for line_no, row in enumerate(csv.DictReader(f), start=2):encoding="utf-8-sig"는 파일 맨 앞의 BOM이 있으면 떼고, 없으면 일반 UTF-8로 읽는다. BOM이 있는 파일과 없는 파일을 같은 코드로 읽을 수 있어서, 엑셀에서 온 CSV를 읽을 때는 이쪽이 안전하다. 줄 번호는 헤더가 1번 줄이므로 2부터 센다. 오류 보고서에 적힌 줄 번호로 지점 직원이 엑셀에서 바로 그 행을 찾을 수 있어야 하기 때문이다.
try:
qty, amount = parse_row(row)
except ValueError as e:
errors.append({"file": path.name, "line": line_no, "reason": str(e)})
continue이 네 줄이 "멈추지도 숨기지도 않는다"의 구현이다. try 범위를 parse_row 호출 한 줄로 좁힌 것에 주의한다. 집계 코드까지 try 안에 넣으면 집계 코드의 버그도 "잘못된 행"으로 분류돼 버린다. 10단원에서 본 대로 예상한 실패만 잡는다.
집계는 딕셔너리 두 개로 한다. by_title.setdefault(title, {"qty": 0, "amount": 0})는 처음 보는 제목이면 0으로 시작하는 딕셔너리를 넣고, 있으면 기존 것을 돌려준다. 그래서 바로 +=를 할 수 있다. 지점별 합계는 by_branch.get(branch, 0) + amount로 같은 일을 한다. 13단원의 Counter나 defaultdict를 써도 된다.
write_reports — 사람용 CSV, 기계용 JSON
매출순 정렬은 sorted(by_title.items(), key=lambda kv: kv[1]["amount"], reverse=True)다. items()가 (제목, {"qty":…, "amount":…}) 쌍을 주므로 kv[1]["amount"]가 정렬 기준이다. lambda는 이름 없는 한 줄 함수다.
CSV는 utf-8-sig로 쓴다. 윈도우 엑셀은 BOM이 없는 UTF-8 CSV를 더블클릭으로 열면 한글을 깨뜨려 보여 준다. BOM을 붙여 쓰면 엑셀이 UTF-8로 알아본다. 반대로 JSON은 BOM 없이 utf-8로 쓴다. 많은 JSON 파서가 BOM을 오류로 취급한다. 용도에 따라 인코딩을 다르게 고른 것이다.
오류 목록은 한 건도 없어도 헤더만 있는 errors.csv를 만든다. 파일이 없으면 "오류가 없었다"인지 "스크립트가 거기까지 못 갔다"인지 구분할 수 없기 때문이다.
report.json에는 다른 프로그램(메신저 알림 봇, 대시보드)이 쓸 요약만 담는다. 합계, 지점별 매출, 상위 3개 제목, 오류 건수다.
표 · report.py 가 쓰는 출력 파일 세 개
| 파일 | 인코딩 | 읽는 쪽 | 내용 |
|---|---|---|---|
summary.csv | utf-8-sig | 사람 (엑셀) | 순위·도서·수량·매출 |
errors.csv | utf-8-sig | 사람 (지점 직원) | file·line·reason |
report.json | utf-8 | 다른 프로그램 | 총매출·지점별 매출·상위 3권·오류 수 |
실행 결과
실습용 sales 폴더에는 위에서 말한 문제를 모두 담은 세 지점 파일이 있다.
$ python3 report.py sales out
총매출 429,000원
강남: 183,000원
부산: 74,000원
판교: 172,000원
건너뛴 행 3건 -> out/errors.csv$ cat out/summary.csv
순위,도서,수량,매출
1,파이썬 개론,7,196000
2,SQL 첫걸음,5,110000
3,네트워크 기초,3,90000
4,리팩터링,1,33000$ cat out/errors.csv
file,line,reason
20260922_부산.csv,3,수량이 숫자가 아님: '두권'
20260922_부산.csv,5,빈 칸: title
20260922_판교.csv,3,수량이 0 이하: -1$ cat out/report.json
{
"total_amount": 429000,
"by_branch": {
"강남": 183000,
"부산": 74000,
"판교": 172000
},
"top3": [
"파이썬 개론",
"SQL 첫걸음",
"네트워크 기초"
],
"error_count": 3
}틀린 세 행은 합계에서 빠졌고, 각각 어느 파일 몇 번째 줄인지 남았다. 강남 지점의 "33,000"은 콤마를 지우고 정상 처리됐고, BOM이 붙은 판교 파일도 문제없이 읽혔다. errors.csv를 지점장에게 그대로 보내면 고쳐서 다시 올려 줄 수 있다.
표 · 잘못된 행이 걸린 검사와 errors.csv 에 남은 실제 이유 (report-errors 실행 결과)
| 검사 | 실습 데이터 | errors.csv 의 reason |
|---|---|---|
| ① 필수 칸 | 부산 지점 제목 없는 행 | 빈 칸: title |
| ② 수량 숫자 | 부산 지점 두권 | 수량이 숫자가 아님: '두권' |
| ③ 단가 숫자 | 강남 지점 "33,000" | 걸리지 않음 (콤마를 지운 33000) |
| ④ 수량 1 이상 | 판교 지점 반품 -1 | 수량이 0 이하: -1 |
실무에서 자주 틀리는 것
1. BOM 때문에 첫 번째 열만 KeyError가 난다
엑셀에서 저장한 CSV를 encoding="utf-8"로 읽으면 이런 일이 생긴다.
import csv
with open("sales/20260922_판교.csv", encoding="utf-8", newline="") as f:
first = next(csv.DictReader(f))
print("utf-8 키 목록:", list(first.keys()))
with open("sales/20260922_판교.csv", encoding="utf-8-sig", newline="") as f:
first = next(csv.DictReader(f))
print("utf-8-sig 키 목록:", list(first.keys()))$ python3 demo_bom.py
utf-8 키 목록: ['\ufeffdate', 'title', 'qty', 'price']
utf-8-sig 키 목록: ['date', 'title', 'qty', 'price']첫 열 이름이 'date'가 아니라 '\ufeffdate'다. 화면에 출력하면 둘이 똑같이 보여서 한참 헤맨다. row["date"]에서 KeyError: 'date'가 나는데 파일을 열어 보면 분명 date라고 적혀 있다면 BOM을 의심한다. 읽을 때는 utf-8-sig가 정답이다.
2. 문제 있는 행을 조용히 버린다
try:
qty = int(row["qty"])
except ValueError:
continue # 어디서 몇 건이 빠졌는지 아무도 모른다이렇게 쓰면 스크립트는 멀쩡히 끝나고 합계만 조금 모자란다. 대표는 그 숫자로 발주를 결정한다. 건너뛸 때는 반드시 무엇을 왜 건너뛰었는지 남기고, 화면에도 건수를 출력한다. 오류가 전체의 일정 비율을 넘으면 보고서 자체를 만들지 않고 실패로 끝내는 규칙을 추가하는 것도 좋다(연습 과제 2).
3. 금액을 float으로 더한다
원 단위 정수는 정수로 더한다. float("28000")으로 바꿔 수천 건을 더하면 2단원에서 본 것처럼 끝자리가 어긋날 수 있고, 보고서에 196000.0 같은 값이 찍혀 엑셀에서 다시 서식을 고쳐야 한다.
4. 출력 파일을 입력 폴더에 쓴다
python3 report.py sales sales처럼 출력 폴더를 입력 폴더와 같게 주면, 다음 실행 때 summary.csv와 errors.csv도 *.csv에 걸려 지점 파일로 읽힌다. 열 이름이 달라 전부 오류 행으로 기록되거나, 운이 나쁘면 형식이 맞아서 합계에 섞인다. 입력과 출력 폴더는 분리하고, 가능하면 입력 파일 이름 규칙(*_*.csv)을 더 좁게 잡는다.
스스로 확인하기
- 판교 지점이 앞으로 반품을 음수로 계속 보내기로 했다. 반품은 매출에서 빼되
errors.csv에는 남기지 않으려면parse_row의 어느 부분을 어떻게 바꾸는가? - 전체 행 중 오류가 10%를 넘으면 보고서를 만들지 않고 종료 코드 1로 끝내고 싶다.
main에 무엇을 추가해야 하는가?collect가 전체 행 수를 알려 주지 않는다는 점에 주의하라. - 같은 파일이 실수로 두 번 올라왔다(
20260922_강남.csv와20260922_강남 (1).csv). 지금 스크립트는 어떻게 동작하며, 어떻게 막겠는가?
정답
if qty <= 0: raise ...를if qty == 0:으로 좁힌다. 음수 수량은 그대로 통과해amount도 음수가 되므로 도서별·지점별 합계에서 자동으로 빠진다. 다만 반품이 판매보다 많은 도서는 수량이 음수로 보고서에 나오니, 순위에서 제외할지 정책을 정해야 한다.collect에서 읽은 행 수를 세어(total += 1을try앞에 두고) 함께 돌려준 뒤,main에서if total and len(errors) / total > 0.1:이면 오류 목록만 출력하고sys.exit(1)한다. 이때write_reports보다 먼저 검사해야 틀린 보고서가 남지 않는다.- 지점 이름이
(1)을 포함한강남 (1)이 되어 네 번째 지점으로 집계되고, 도서별 합계는 강남 판매분이 두 번 더해진다. 오류도 나지 않는다. 지점 이름을 미리 정한 집합({"강남", "부산", "판교"})과 비교해 모르는 지점이면 파일 전체를 오류로 기록하고, 같은 날짜·지점 파일이 두 개 이상이면 중단하도록 검사를 추가한다.
다음 단원에서는 표 대신 로그 파일을 다룬다. 형식이 정해진 CSV와 달리 로그는 한 줄 한 줄이 자유로운 문장이라, 필요한 부분을 뽑아내는 도구로 정규 표현식과 Counter를 쓴다.