CSV 열 수가 맞지 않는 행을 원문 위치와 함께 찾기
CSV를 열었는데 일부 값이 옆 열로 밀렸다면 먼저 각 레코드의 필드 수를 확인하세요. 따옴표 안 쉼표와 줄바꿈은 CSV 읽기에 맡기고, 읽은 필드 수를 머리글의 3개 열과 비교하면 부족한 행과 넘치는 행의 원문 위치를 찾을 수 있습니다. 화면에 보이는 줄 번호와 데이터 레코드 번호는 따로 남깁니다.
이번 실습은 가상 id,qty,memo CSV를 사용합니다. 열 수가 맞는 레코드 2개, 4개 필드인 레코드 1개, 2개 필드인 레코드 1개를 새 JSON 보고서에 기록합니다. 닫히지 않은 따옴표는 별도 파일에서 확인합니다. 원본 값이나 수량을 고치거나 문제 행을 삭제하지 않습니다.
1 열 수 오류와 파싱 오류를 구분합니다
CSV의 한 레코드는 따옴표 안 줄바꿈 때문에 여러 물리 줄에 걸칠 수 있습니다. "메모, 포함"도 쉼표를 포함한 필드 하나입니다. 쉼표 문자만 기준으로 나누면 이 메모를 둘로 잘못 셉니다.
csv.reader가 필드 목록을 반환했다면 그 목록의 길이를 3과 비교합니다. 4개면 초과, 2개면 부족입니다. 닫히지 않은 인용문처럼 reader가 레코드를 반환하지 못한 경우에는 해당 필드 수를 추정하지 않고 parse_error로 구분합니다.
입력은 심볼릭 링크가 아닌 일반 CSV 파일로 한정합니다. 폴더와 FIFO 같은 비일반파일은 열기 전에 거부합니다. UTF-8 CSV이며 UTF-8 BOM도 받습니다. 첫 레코드의 머리글은 소문자 id,qty,memo 순서로 정확히 일치해야 합니다. 앞뒤 공백을 지우거나 열 순서를 자동으로 바꾸지 않습니다. 머리글 뒤의 빈 레코드도 필드 0개인 너비 오류로 기록합니다. 머리글만 있으면 데이터 0개의 완전한 스캔이며, 필요한 최소 거래 건수는 별도 업무 기준으로 확인해야 합니다.
2 물리 6줄에 데이터 레코드 4개가 들어 있습니다
실습 ZIP을 빈 폴더에 풀고 fixtures/mixed.csv를 텍스트 편집기로 확인합니다. 파일은 다음과 같습니다.
id,qty,memo
A,2,"메모, 포함"
B,3,"첫 줄
둘째 줄"
C,4,추가,필드
D,5머리글을 논리 레코드 1로 세면 A부터 D까지는 2~5입니다. B의 메모는 물리 3~4줄에 걸치지만 데이터 레코드는 하나입니다.
| 구분 | 논리 레코드 번호 | 물리 시작 줄 | 물리 끝 줄 | 필드 수 | 열 수 판정 |
|---|---|---|---|---|---|
| 머리글 | 1 | 1 | 1 | 3 | 기준 열 |
| A | 2 | 2 | 2 | 3 | 일치 |
| B | 3 | 3 | 4 | 3 | 일치 |
| C | 4 | 5 | 5 | 4 | 초과 |
| D | 5 | 6 | 6 | 2 | 부족 |
C에서 어느 쉼표가 잘못됐는지, D에 어떤 값이 빠졌는지는 이 판정만으로 알 수 없습니다. 보고서의 위치를 찾아 원본 자료와 대조합니다.
3 필드 수와 줄 위치를 남기는 전체 코드
다음 코드는 ZIP의 code/audit_width.py와 같습니다. Python 3.12 표준 라이브러리만 사용합니다. 추가 패키지나 API 키는 필요하지 않습니다.
Python csv 공식 문서에 따라 기본 reader의 필드는 문자열로 유지합니다. qty에 02나 수량 확인 필요가 있어도 여기서는 숫자 변환이나 수량 검사를 하지 않습니다. 인용 쉼표와 인용 줄바꿈을 처리하도록 newline=''인 문자열 스트림을 사용합니다.
reader.line_num 설명은 읽은 물리 줄 수와 반환한 레코드 수가 다를 수 있다고 안내합니다. 코드에서는 읽기 직전의 끝 줄에 1을 더해 시작 줄을 정하고, 읽은 뒤의 line_num을 끝 줄로 남깁니다.
"""Read a synthetic CSV and report field widths without changing its values."""
import argparse
import csv
import hashlib
import io
import json
import os
import stat
import sys
import tempfile
from pathlib import Path
HEADER = ["id", "qty", "memo"]
MAX_BYTES = 1_000_000
MAX_RECORDS = 10_000
def physical_position(reader, previous_end):
return previous_end + 1, reader.line_num
def inspect(source):
source = Path(source)
if not stat.S_ISREG(source.lstat().st_mode):
raise ValueError("입력은 심볼릭 링크가 아닌 일반 CSV 파일이어야 합니다")
with source.open("rb") as handle:
raw = handle.read(MAX_BYTES + 1)
if len(raw) > MAX_BYTES:
raise ValueError("입력 한도 1,000,000바이트를 초과했습니다")
text = raw.decode("utf-8-sig", errors="strict")
reader = csv.reader(io.StringIO(text, newline=""), strict=True)
old_limit = csv.field_size_limit(MAX_BYTES)
try:
try:
header = next(reader, None)
except csv.Error as error:
raise ValueError("머리글 파싱 실패: " + str(error)) from error
if header != HEADER:
raise ValueError("머리글은 id,qty,memo 순서로 정확히 지정하세요")
start, end = physical_position(reader, 0)
header_position = {"source_record": 1, "physical_start_line": start,
"physical_end_line": end}
records, parse_error = [], None
previous_end = end
while True:
try:
fields = next(reader)
except StopIteration:
break
except csv.Error as error:
start, consumed = physical_position(reader, previous_end)
parse_error = {
"attempted_source_record": len(records) + 2,
"physical_start_line": start,
"consumed_through_line": consumed,
"message": str(error),
}
break
if len(records) >= MAX_RECORDS:
raise ValueError("데이터 레코드 한도 10,000개를 초과했습니다")
start, end = physical_position(reader, previous_end)
count = len(fields)
issue = None
if count < len(HEADER):
issue = "too_few_fields"
elif count > len(HEADER):
issue = "too_many_fields"
records.append({
"source_record": len(records) + 2,
"physical_start_line": start, "physical_end_line": end,
"field_count": count, "expected_field_count": len(HEADER),
"width_issue": issue, "fields": fields,
})
previous_end = end
finally:
csv.field_size_limit(old_limit)
width_errors = sum(row["width_issue"] is not None for row in records)
complete = parse_error is None
status = "parse_error" if parse_error else ("width_errors" if width_errors else "ok")
return {
"schema_version": 1, "source_file": source.name,
"source_bytes": len(raw), "source_sha256": hashlib.sha256(raw).hexdigest(),
"source_changes_written": 0, "header": HEADER,
"header_position": header_position,
"diagnostic_report_complete": True, "scan_complete": complete,
"file_status": status,
"counts": {
"scope": "all_data_records" if complete else "parsed_prefix_only",
"parsed_data_records": len(records),
"matching_width_records": len(records) - width_errors,
"width_error_records": width_errors,
"total_data_records": len(records) if complete else None,
},
"records": records, "parse_error": parse_error,
}
def write_report(source, target):
source, target = Path(source), Path(target)
if source.resolve() == target.resolve():
raise ValueError("원본과 보고서는 서로 다른 경로여야 합니다")
if target.suffix.lower() != ".json":
raise ValueError("보고서는 새 .json 파일로 지정하세요")
if target.exists() or target.is_symlink():
raise ValueError("기존 출력 보존: 새 --report 이름을 지정하세요")
report = inspect(source)
rendered = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
target.parent.mkdir(parents=True, exist_ok=True)
temporary, committed = None, False
try:
with tempfile.NamedTemporaryFile(
mode="w", encoding="utf-8", newline="\n", dir=target.parent,
prefix=".audit-width-", suffix=".tmp", delete=False,
) as handle:
temporary = Path(handle.name)
handle.write(rendered)
handle.flush()
os.fsync(handle.fileno())
try:
os.link(temporary, target)
except FileExistsError as error:
raise ValueError("기존 출력 보존: 보고서 확정 중 경로가 충돌했습니다") from error
committed = True
finally:
if temporary is not None:
try:
temporary.unlink(missing_ok=True)
except OSError:
if not committed:
raise
print("알림: 완성된 임시파일 정리 실패, 출력 보고서는 완전합니다", file=sys.stderr)
return report
def main():
parser = argparse.ArgumentParser(description="CSV 열 수와 원문 줄 위치를 읽기 전용 검산")
parser.add_argument("--input", required=True)
parser.add_argument("--report", required=True)
args = parser.parse_args()
try:
report = write_report(args.input, args.report)
except (ValueError, OSError) as error:
print("실행 실패: " + str(error), file=sys.stderr)
return 2
counts = report["counts"]
print(f"상태 {report['file_status']} / 읽은 레코드 {counts['parsed_data_records']} / "
f"열 수 일치 {counts['matching_width_records']} / 너비 오류 {counts['width_error_records']} / "
f"범위 {counts['scope']} / 원본 변경 0")
return 0 if report["file_status"] == "ok" else 1
if __name__ == "__main__":
raise SystemExit(main())strict=True는 reader가 감지하는 잘못된 CSV 입력에서 예외를 내게 합니다. 모든 CSV 문법 규칙이나 업무 양식의 하자를 검증한다는 보장은 아닙니다. 열 개수, 정확한 머리글과 파일 한도는 코드에서 따로 확인합니다.
파일은 최대 1,000,000바이트, 데이터 레코드는 최대 10,000개입니다. 한도를 넘기면 전체 실행을 중단합니다. 코드의 필드 길이 한도도 입력 바이트 한도에 맞춰 두었으며, 파일 한도가 먼저 적용됩니다. 이는 대용량 처리 성능을 보장하는 기준이 아닙니다.
4 정상 너비 2개와 오류 2개를 대조합니다
터미널을 ZIP을 푼 폴더로 옮깁니다. python --version으로 Python 3.12 환경인지 확인한 뒤 각 명령을 따로 실행합니다. out 폴더는 필요하면 만들어집니다.
python --version
python code/audit_width.py --input fixtures/mixed.csv --report out/mixed.json출력은 다음과 같습니다. 이 파일에는 너비 오류가 있으므로 종료 코드는 1입니다. 보고서는 만들어져야 합니다.
상태 width_errors / 읽은 레코드 4 / 열 수 일치 2 / 너비 오류 2 / 범위 all_data_records / 원본 변경 0out/mixed.json의 scan_complete는 true, file_status는 width_errors입니다. counts는 다음과 같아야 합니다.
{
"scope": "all_data_records",
"parsed_data_records": 4,
"matching_width_records": 2,
"width_error_records": 2,
"total_data_records": 4
}records에서 C는 source_record=4, 물리 5~5줄, too_many_fields입니다. D는 source_record=5, 물리 6~6줄, too_few_fields입니다. fields에는 원래 읽은 문자열 목록을 남기므로 C의 네 값도 그대로 확인할 수 있습니다.
정상 너비 두 레코드만 담은 파일도 실행합니다.
python code/audit_width.py --input fixtures/valid.csv --report out/valid.json이 경우 상태는 ok, 읽은 레코드는 2개, 너비 오류는 0개이고 종료 코드는 0입니다. 여기서 ok는 정해진 CSV 읽기와 열 수 검사를 통과했다는 뜻입니다. ID 중복, 수량의 숫자 형식이나 메모 내용까지 확인한 것은 아닙니다.
5 인용문이 닫히지 않으면 확실히 읽은 부분만 셉니다
fixtures/unclosed.csv는 A 레코드 뒤에서 B의 따옴표를 닫지 않은 파일입니다.
id,qty,memo
A,2,"메모, 포함"
B,3,"닫히지 않은 메모
이어서 적은 줄python code/audit_width.py --input fixtures/unclosed.csv --report out/unclosed.json이 명령은 종료 코드 1과 함께 완전한 진단보고서를 저장합니다. file_status=parse_error, scan_complete=false이며 diagnostic_report_complete=true입니다. CSV 전체를 읽지 못했다는 사실과, 그 상태를 담은 JSON 보고서가 완성됐다는 사실을 구분합니다.
counts.scope는 parsed_prefix_only입니다. 파싱 오류 전에 확실히 읽은 접두부분은 A 하나이므로 parsed_data_records=1, matching_width_records=1, width_error_records=0입니다. 파일 전체의 데이터 건수는 확정하지 못했으므로 total_data_records=null로 남깁니다. 이 숫자를 파일 전체의 정상 1건으로 요약하면 안 됩니다.
오류 위치는 읽으려던 논리 레코드 3, 물리 시작 줄 3, reader가 소비한 마지막 줄 4입니다. consumed_through_line=4는 파싱 시도가 4줄까지 읽었다는 뜻입니다. 정상적으로 닫힌 레코드의 끝 줄은 확인되지 않았으므로 physical_end_line을 오류 위치에 넣지 않습니다. 이후의 레코드 수나 너비도 추정하지 않고 멈춥니다.
6 종료 코드와 보고서 유무를 함께 확인합니다
| 종료 코드 | 상황 | 새 최종 JSON |
|---|---|---|
| 0 | 파싱과 열 수 검사에 문제가 없는 완전한 진단 | 있음 |
| 1 | 너비 오류 또는 데이터 파싱 오류를 담은 완전한 진단 | 있음 |
| 2 | 읽기 실패, UTF-8 해석 실패, 머리글 오류, 한도 초과, 출력 경로 문제 | 없음 |
코드 1은 불완전하게 저장된 파일을 뜻하지 않습니다. 파싱 실패도 필요한 상태와 확인된 접두부분을 담은 완전한 진단보고서입니다. 반면 머리글이 잘못되면 기준 열 자체를 확정하지 않으므로 실행 실패로 처리합니다.
python code/audit_width.py --input fixtures/bad_header.csv --report out/bad_header.json종료 코드 2가 나오고 새 out/bad_header.json은 없어야 합니다. 기존 파일이 있던 경로라면 기존 파일을 보존합니다.
보고서는 원본과 다른 새 .json 경로로 지정합니다. 다음 두 명령은 각각 기존 출력 충돌과 원본 동일 경로를 확인하는 의도한 실패입니다. 첫 명령은 앞에서 만든 out/mixed.json이 있는 상태에서 실행합니다.
python code/audit_width.py --input fixtures/mixed.csv --report out/mixed.json
python code/audit_width.py --input fixtures/mixed.csv --report fixtures/mixed.csv둘 다 종료 코드 2이며, 기존 보고서와 원본 바이트는 바뀌지 않아야 합니다. 다시 보고하려면 out/mixed_second.json처럼 새 이름을 쓰세요.
저장할 때는 JSON을 같은 출력 폴더의 임시파일에 끝까지 쓴 뒤 새 이름으로 확정합니다. Python os.link 안내에 있는 하드 링크를 사용해 확정 시점에 생긴 출력 충돌도 덮어쓰지 않습니다. 이 방식은 하드 링크를 지원하는 파일시스템이 필요합니다. 지원하지 않는 위치에서는 종료 코드 2로 멈추므로, 지원되는 로컬 실습 폴더에서 실행하세요.
7 자동 시험으로 원본과 기존 출력을 지킵니다
다음 명령은 제공된 가상 파일과 시험용 임시폴더만 사용합니다.
python -m unittest discover -s code -p 'test_*.py' -v정상 완료하면 28개 시험의 마지막에 OK가 나옵니다. 인용 쉼표·줄바꿈, 물리 줄 위치, prefix 집계, 머리글 오류, 1 MB와 10,000레코드 경계, 기존 출력과 원본 동일 경로를 확인합니다. 저장과 확정의 실패를 시험 내부에서 만들어 새 최종 파일과 임시파일이 남지 않는지도 확인합니다. 각 CLI 시험은 원본 SHA-256이 실행 전후 같은지 대조합니다.
ZIP의 expected/mixed.json, expected/valid.json, expected/unclosed.json은 제공된 파일의 비교용 결과입니다. 입력을 바꾸면 해시와 레코드 내용도 달라질 수 있습니다. 임의의 새 파일에 같은 숫자가 나와야 한다는 뜻은 아닙니다.
프로세스 강제 종료, 전원 장애나 폴더 권한의 갑작스러운 변경까지 복구하는 도구는 아닙니다. 실행이 강제로 끊겼다면 숨김 임시파일이 남아 있는지 확인하고, 파일 내용을 확인하기 전에는 기존 결과를 지우지 마세요. 출력 이름이 생기는 시점에는 JSON 쓰기가 끝나 있지만, 이 코드가 저장장치의 장애 복구를 보장하지는 않습니다.
8 위치를 찾은 다음에는 원본 근거로 고칩니다
먼저 열 수 보고서에서 논리 레코드와 물리 시작·끝 줄을 찾습니다. 너비 오류는 해당 위치의 원문과 원본 자료를 대조하고, 인용문 오류는 닫힘 위치부터 확인합니다. 이 실습은 자동 복구를 하지 않으므로 수정본을 만들 때에는 새 파일로 저장한 뒤 다시 검사하세요.
열 구조를 확인한 다음 수량을 합치려면 CSV 점포 수량 합계와 저장값 검산을 이어 볼 수 있습니다. 옮긴 표의 ID와 행이 원본에서 빠지지 않았는지는 샘플 표의 ID와 원본 대조에서 별도로 확인합니다.
실습 자료는 코드, 가상 CSV, 예상 JSON, 자동 시험과 README를 담은 reader.zip입니다. 빈 실습 폴더에 풀어 README의 명령부터 실행하세요.
전체 코드·가상 CSV·기대 결과·테스트 ZIP 다운로드
작성자 소개
