헤더를 유지하면서 큰 CSV를 작은 파일로 나누기
CSV를 데이터 레코드 수에 따라 나누고, 각 출력 파일에 헤더를 넣으면서 원본은 그대로 보존합니다. 7개 레코드로 구성된 합성 예제를 이용해 분할 경계를 확인하고, 모든 레코드가 원래 순서대로 유지되는지 검증합니다.
파일 크기로 비교 대상을 추리고 SHA-256 해시값을 비교한 뒤, 원본을 변경하지 않고 검토용 CSV를 작성합니다. 7개 파일로 구성된 합성 예제를 이용해 보고서에 포함될 경로와 제외될 경로를 확인합니다.
이런 분께 맞아요중복 가능성이 있는 파일을 찾되, 검색과 삭제 판단을 분리해서 진행하려는 사람을 위한 안내입니다.
이 작업은 먼저 파일을 바이트 크기에 따라 묶습니다. 크기가 다르면 바이트 내용이 완전히 같을 수 없으므로, 경로가 여러 개 있는 크기 그룹만 해시를 계산하면 됩니다. 각 크기 그룹 안에서는 SHA-256 해시값이 일치하는 파일을 중복 후보로 분류합니다. 파일명 일치는 필수 조건도 충분조건도 아닙니다.
중복 후보 그룹은 삭제 지시가 아닙니다. 내용이 같아도 서로 다른 폴더에서 다른 용도로 사용될 수 있으며, 해시값이 일치한다고 동일성이 절대적으로 증명되는 것도 아닙니다. 보고서에는 일치하는 경로를 모두 기록하며, 남길 파일을 자동으로 선택하지 않습니다.
아래 데이터는 이 글을 위해 작성한 합성 데이터입니다. 코드를 create_duplicate_demo.py로 저장하고 작업 폴더에서 실행하세요. 바이너리 모드로 기록하므로 내용이 정확하게 유지됩니다. 줄바꿈은 추가되지 않으며, 빈 파일은 0바이트입니다.
from pathlib import Path
DEMO = Path("outputs") / "duplicate_demo"
FILES = {
"archive/a_saved.txt": b"abc",
"empty_a.txt": b"",
"empty_b.txt": b"",
"notes/a.txt": b"abc",
"notes/a_copy.txt": b"abc",
"notes/other.txt": b"xyz",
"unique.txt": b"solo",
}
DEMO.parent.mkdir(parents=True, exist_ok=True)
DEMO.mkdir() # Refuse to reuse an existing destination.
for relative_path, content in FILES.items():
target = DEMO / relative_path
target.parent.mkdir(parents=True, exist_ok=True)
with target.open("xb") as stream:
stream.write(content)
python create_duplicate_demo.py| 상대 경로 | 정확한 텍스트 내용 | 바이트 수 |
|---|---|---|
| archive/a_saved.txt | abc | 3 |
| empty_a.txt | 비어 있음 | 0 |
| empty_b.txt | 비어 있음 | 0 |
| notes/a.txt | abc | 3 |
| notes/a_copy.txt | abc | 3 |
| notes/other.txt | xyz | 3 |
| unique.txt | solo | 4 |
7개 파일의 내용은 총 16바이트입니다. 크기가 3바이트인 파일은 4개지만, notes/other.txt의 내용은 abc가 들어 있는 3개 파일과 다릅니다. 크기가 같다는 조건만으로는 충분하지 않다는 점을 확인하도록 구성했습니다.
python find_duplicate_files.py스크립트는 경로를 해석한 결과 보고서 대상 폴더가 원본 폴더 안에 있으면 거부합니다. 검색 과정에 자체 출력물이 포함되는 것을 막기 위한 조치입니다. 상대 경로의 기준은 여전히 터미널의 작업 디렉터리입니다.
검색 과정에서 발견한 심볼릭 링크는 건너뛰고, 일반 파일이 아닌 항목도 제외합니다. 디렉터리 탐색이나 파일 읽기 중 오류가 발생하면 실행을 중단합니다. 해시 계산 전후의 메타데이터 비교로 일부 동시 변경을 감지하지만, 파일을 잠근 스냅샷을 만드는 것은 아닙니다.
import csv
import hashlib
import os
import stat
from pathlib import Path
SOURCE = Path("outputs") / "duplicate_demo"
OUTPUT_DIR = Path("outputs") / "duplicate_review"
CHUNK_BYTES = 1024 * 1024
def signature(info: os.stat_result) -> tuple[int, ...]:
return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns)
def raise_walk_error(error: OSError) -> None:
raise error
def file_digest(path: Path, expected: os.stat_result) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
before = os.fstat(stream.fileno())
if not stat.S_ISREG(before.st_mode) or signature(before) != signature(expected):
raise RuntimeError(f"File changed before hashing: {path}")
while chunk := stream.read(CHUNK_BYTES):
digest.update(chunk)
after = os.fstat(stream.fileno())
if signature(after) != signature(expected) or signature(path.lstat()) != signature(expected):
raise RuntimeError(f"File changed during hashing: {path}")
return digest.hexdigest()
def main() -> None:
root = SOURCE.resolve(strict=True)
if not root.is_dir():
raise ValueError("SOURCE must be a directory.")
if OUTPUT_DIR.resolve().is_relative_to(root):
raise ValueError("The report folder must be outside SOURCE.")
OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR.mkdir() # Stop if the destination already exists.
by_size = {}
scanned = 0
for directory, directories, filenames in os.walk(
root, followlinks=False, onerror=raise_walk_error
):
base = Path(directory)
directories[:] = sorted(
name for name in directories if not (base / name).is_symlink()
)
for name in sorted(filenames):
path = base / name
info = path.lstat()
if not stat.S_ISREG(info.st_mode):
continue
by_size.setdefault(info.st_size, []).append((path, info))
scanned += 1
groups = []
hashed = 0
for size, members in sorted(by_size.items()):
if len(members) < 2:
continue
by_hash = {}
for path, initial in members:
digest = file_digest(path, initial)
by_hash.setdefault(digest, []).append(path)
hashed += 1
for digest, paths in sorted(by_hash.items()):
if len(paths) > 1:
ordered = sorted(paths, key=lambda path: path.as_posix())
groups.append((size, digest, ordered))
report = OUTPUT_DIR / "duplicates.csv"
with report.open("x", encoding="utf-8", newline="") as stream:
writer = csv.writer(stream)
writer.writerow([
"group_id", "size_bytes", "sha256", "relative_path", "review_status"
])
for number, (size, digest, paths) in enumerate(groups, start=1):
for path in paths:
writer.writerow([
f"G{number:03d}", size, digest,
path.relative_to(root).as_posix(), "UNREVIEWED"
])
matched = sum(len(paths) for _, _, paths in groups)
print(f"Scanned {scanned} regular file paths; hashed {hashed}.")
print(f"Duplicate candidates: {len(groups)} groups, {matched} paths.")
print(f"Review CSV: {report.as_posix()}")
if __name__ == "__main__":
main()
예상 보고서는 헤더 외에 데이터 행이 5개입니다. 그룹은 크기, 해시값 순으로 정렬되고, 그룹 안의 경로도 정렬됩니다. 여기서는 해시값을 계산하지 않았으므로 아래 표에서 sha256 열은 생략했습니다. 실제로 실행하면 스크립트가 해당 열을 채웁니다.
| group_id | size_bytes | relative_path | review_status |
|---|---|---|---|
| G001 | 0 | empty_a.txt | UNREVIEWED |
| G001 | 0 | empty_b.txt | UNREVIEWED |
| G002 | 3 | archive/a_saved.txt | UNREVIEWED |
| G002 | 3 | notes/a.txt | UNREVIEWED |
| G002 | 3 | notes/a_copy.txt | UNREVIEWED |
해시 계산 대상은 빈 파일 2개와 3바이트 파일 4개를 합친 6개입니다. 4바이트 파일은 크기가 같은 다른 파일이 없어 해시 계산 단계에서 제외됩니다. 아래 콘솔 출력은 수동으로 도출한 예상값이며 실제 실행 로그가 아닙니다.
Scanned 7 regular file paths; hashed 6.
Duplicate candidates: 2 groups, 5 paths.
Review CSV: outputs/duplicate_review/duplicates.csv빈 파일은 의도적으로 만든 자리표시자일 수 있습니다. 내용이 같다고 불필요한 것은 아니며, 삭제해도 파일 내용에 해당하는 바이트가 줄어드는 것은 아닙니다.
이 항목들은 독자가 직접 수행할 확인 사항이며, 이 글을 작성하면서 실행한 시험이 아닙니다. 검색을 완료했지만 그룹이 0개인 보고서와 예외가 발생해 중단된 검색은 구분해야 합니다.
| 증상 | 확인할 사항 |
|---|---|
| FileNotFoundError | 준비용 스크립트를 먼저 실행하고, 터미널의 작업 디렉터리를 기준으로 SOURCE를 확인하세요. |
| FileExistsError | 이전 대상 폴더를 검토하고 outputs 아래의 새 폴더를 선택하세요. 이미 존재하는 빈 대상 폴더도 사용할 수 없습니다. |
| PermissionError 또는 다른 읽기 오류 | 접근 문제를 해결하거나 검색 범위를 명시적으로 줄이세요. 중단된 검색을 완료된 것으로 취급하지 마세요. |
| 해시 계산 전이나 도중 파일 변경 감지 | 원본을 수정하는 응용 프로그램을 중지한 뒤, 새 출력 대상 폴더를 지정해 다시 검색하세요. |
| 예상과 다른 바이트 크기 | 텍스트 편집기가 줄바꿈을 추가했거나 인코딩을 바꿨을 수 있습니다. 바이너리로 기록하는 준비용 스크립트로 예제를 다시 만드세요. |
| 실패 후 빈 폴더나 일부만 작성된 보고서가 남음 | 중단된 실행은 대상 폴더를 실행 전 상태로 되돌리지 않습니다. 해당 결과를 완료된 보고서와 분리해 두세요. |
해시 계산에서는 내용을 일정 크기씩 읽지만, 파일 메타데이터와 경로, 그룹 정보는 메모리에 저장합니다. 따라서 파일 수와 관계없이 일정한 메모리만 사용하는 목록 작성 방식은 아닙니다. 시각적 유사성이나 문서의 의미가 아니라 전체 바이트를 비교하므로, 겉보기에는 같은 문서도 해시값이 다를 수 있습니다.
보고서는 독립적인 물리적 사본 수가 아니라 경로 수를 셉니다. 하드 링크는 내용을 별도로 저장하지 않으면서 여러 경로가 보고되게 할 수 있으므로, 보고된 크기만큼 디스크 공간이 절약된다고 보장할 수 없습니다. Windows 정션, 마운트 지점, 악의적인 동시 변경에는 현재의 심볼릭 링크 검사 외에 추가 처리가 필요합니다.
2026-09-20 · 예제 수동 검토 · 대상: Python 3.12 · 표준 라이브러리: csv, hashlib, os, pathlib, stat · 실행하지 않음
설명과 예제는 직접 작성했습니다. 관련 동작과 개념은 아래 공식 자료에서 확인할 수 있습니다.