[파이썬 100강] 53강. os.walk로 대규모 디렉터리 탐색 제어하기

Last updated on 
[파이썬 100강] 53강. os.walk로 대규모 디렉터리 탐색 제어하기

[파이썬 100강] 53강. os.walk로 대규모 디렉터리 탐색 제어하기

파일 자동화가 커지기 시작하면 결국 부딪히는 문제가 있습니다. “어디까지 순회할지”, “어떤 폴더는 건너뛸지”, “파일이 너무 많은데 메모리를 버티는지” 같은 운영 이슈입니다. pathlib.rglob()만으로도 많은 일을 할 수 있지만, 제외 규칙을 정교하게 제어하거나 디렉터리 순회 자체를 통제해야 할 때는 os.walk()가 훨씬 명확한 선택이 됩니다.

이번 강의에서는 서론을 길게 끌지 않고 바로 실전 기준으로 갑니다. 핵심은 단순 순회 문법이 아니라, 탐색 범위 제어 + 예외 내성 + 후속 처리(복사/백업/리포트) 연결입니다. 초보가 가장 많이 하는 실수는 “일단 전부 순회한 뒤 조건으로 걸러내면 되겠지”라고 생각하는 건데, 이 방식은 디렉터리 규모가 커질수록 속도와 안정성에서 큰 손해를 봅니다.

핵심 개념

  • os.walk(top)는 (현재경로, 하위디렉터리목록, 파일목록) 튜플을 반복적으로 반환하는 제너레이터 기반 순회 도구입니다.
  • topdown=True일 때 dirs[:]를 수정하면 하위 탐색 대상 자체를 줄일 수 있어, 제외 규칙(.git, __pycache__, node_modules)을 매우 효율적으로 적용할 수 있습니다.
  • 대규모 파일 시스템 자동화는 “모든 파일을 모은 뒤 처리”보다 “순회하면서 즉시 필터링/집계/기록”하는 스트리밍 방식이 안정적입니다.

os.walk()를 제대로 이해하면 파일 탐색이 단순한 for문이 아니라 “정책 실행 지점”으로 바뀝니다. 예를 들어, 백업 대상에서 임시 폴더를 제외해야 한다면 파일 단위에서 나중에 거르는 것이 아니라 디렉터리 단계에서 아예 내려가지 않게 만드는 편이 훨씬 빠르고 안전합니다. 탐색 깊이를 제한하고 싶으면 현재 경로의 깊이를 계산해 하위 디렉터리 리스트를 비우는 방식으로 구현할 수 있고, 특정 확장자만 다루고 싶다면 파일 목록 단계에서 즉시 필터링하면 됩니다.

또 하나 중요한 포인트는 실패 허용 전략입니다. 실제 운영 환경에서는 권한 문제(PermissionError)나 사라진 파일(FileNotFoundError)을 종종 만나게 됩니다. 이때 전체 작업을 바로 중단할지, 실패를 로그에 기록하고 다음 대상으로 넘어갈지 결정해야 합니다. 이 정책을 초기에 정해 두면 자동화 스크립트가 “한 번 에러 나면 멈추는 데모 코드”에서 “현장에서 돌아가는 운영 코드”로 올라갑니다.

기본 사용

예제 1) 가장 기본 패턴: 확장자별 파일 탐색

>>> import os
>>> root = "project_data"
>>> csv_files = []
>>> for current, dirs, files in os.walk(root):
...     for name in files:
...         if name.endswith(".csv"):
...             csv_files.append(os.path.join(current, name))
>>> isinstance(csv_files, list)
True

해설:

  • current는 현재 순회 중인 디렉터리, dirs는 하위 디렉터리명 목록, files는 파일명 목록입니다.
  • 파일 전체 경로는 os.path.join(current, name)으로 만들고, 확장자 필터는 가능한 한 이 단계에서 바로 적용합니다.

예제 2) 제외 폴더 제어: 순회 비용 줄이기

>>> import os
>>> root = "project_data"
>>> excluded = {".git", "__pycache__", "node_modules"}
>>> visited = []
>>> for current, dirs, files in os.walk(root, topdown=True):
...     dirs[:] = [d for d in dirs if d not in excluded]
...     visited.append(current)
>>> len(visited) >= 0
True

해설:

  • 핵심은 dirs[:] = ... 입니다. 이걸 통해 “내려갈 하위 디렉터리” 자체를 바꿉니다.
  • topdown=True일 때만 이런 방식의 제어가 자연스럽게 동작합니다. 큰 폴더를 미리 제외하면 체감 속도 차이가 큽니다.

예제 3) 실전형 미니 케이스: 용량 집계 리포트 만들기

>>> import os
>>> from collections import defaultdict
>>> root = "logs"
>>> size_by_ext = defaultdict(int)
>>> count_by_ext = defaultdict(int)
>>> for current, dirs, files in os.walk(root):
...     for name in files:
...         path = os.path.join(current, name)
...         ext = os.path.splitext(name)[1].lower() or "(noext)"
...         try:
...             size_by_ext[ext] += os.path.getsize(path)
...             count_by_ext[ext] += 1
...         except OSError:
...             pass
>>> sorted(count_by_ext.items())[:3]
[]

해설:

  • 파일 수/총 용량을 확장자 단위로 집계하면 “어떤 파일이 용량을 잡아먹는지”를 빠르게 파악할 수 있습니다.
  • OSError를 개별 파일 단위에서 처리해 전체 작업 지속성을 확보합니다.

예제 4) 깊이 제한 순회 패턴

>>> import os
>>> root = "dataset"
>>> root_depth = root.rstrip(os.sep).count(os.sep)
>>> max_depth = 2
>>> limited = []
>>> for current, dirs, files in os.walk(root, topdown=True):
...     depth = current.count(os.sep) - root_depth
...     if depth >= max_depth:
...         dirs[:] = []
...     limited.append((current, depth))
>>> all(depth <= max_depth for _, depth in limited)
True

해설:

  • 탐색 깊이를 제한하면 예상치 못한 깊은 하위 구조로 내려가며 시간이 폭증하는 문제를 막을 수 있습니다.
  • 데이터 레이크/로그 아카이브처럼 트리가 깊은 환경에서 특히 유용합니다.

자주 하는 실수

실수 1) 제외 폴더를 파일 단계에서만 걸러 순회 비용 폭증

>>> import os
>>> skipped = {"node_modules", ".git"}
>>> selected = []
>>> for current, dirs, files in os.walk("repo"):
...     if any(part in skipped for part in current.split(os.sep)):
...         continue
...     for f in files:
...         if f.endswith(".py"):
...             selected.append((current, f))

원인:

  • 이미 제외 대상 디렉터리 내부까지 내려간 뒤에 continue를 하므로, 불필요한 순회 비용을 다 치르고 있습니다.
  • “필터링은 했으니 괜찮다”라고 착각하기 쉬운데, 성능 병목은 탐색 단계에서 이미 발생합니다.

해결:

>>> import os
>>> skipped = {"node_modules", ".git"}
>>> selected = []
>>> for current, dirs, files in os.walk("repo", topdown=True):
...     dirs[:] = [d for d in dirs if d not in skipped]
...     for f in files:
...         if f.endswith(".py"):
...             selected.append((current, f))

제외 규칙은 파일이 아니라 디렉터리 진입 전에 적용해야 합니다. 이 한 줄(dirs[:])이 실제 실행 시간을 크게 바꿉니다.

실수 2) 전체 결과를 한 번에 리스트로 모아 메모리 압박

>>> import os
>>> all_files = []
>>> for current, dirs, files in os.walk("huge_data"):
...     for f in files:
...         all_files.append(os.path.join(current, f))
>>> len(all_files)
0

원인:

  • 후속 작업이 단순 집계/전송/복사인데도 모든 경로를 메모리에 쌓았습니다.
  • 파일 수가 수십만~수백만으로 커지면 메모리 사용량이 급격히 증가합니다.

해결:

>>> import os
>>> processed = 0
>>> for current, dirs, files in os.walk("huge_data"):
...     for f in files:
...         path = os.path.join(current, f)
...         # 여기서 즉시 처리: 복사/해시/집계/DB기록 등
...         processed += 1
>>> processed >= 0
True

스트리밍 처리로 전환하면 메모리 사용량을 거의 일정하게 유지할 수 있습니다.

실수 3) 심볼릭 링크 순회 정책을 정하지 않아 의도치 않은 확장 탐색

  • 증상: 탐색 범위가 갑자기 커지거나, 동일 경로를 반복 방문하는 것처럼 보입니다.
  • 원인: followlinks=True 사용 시 링크된 외부 디렉터리까지 내려가며 정책 밖 경로를 탐색했습니다.
  • 해결: 기본값(followlinks=False)을 유지하고, 링크를 따라가야 한다면 허용 루트 검증과 최대 깊이 제한을 함께 둡니다.

실무 패턴

  • 입력 검증 규칙: 시작 경로가 실제 디렉터리인지(os.path.isdir) 먼저 확인하고, 허용 루트 바깥 경로는 즉시 거부합니다.
  • 로그/예외 처리 규칙: 에러는 (path, error_type, message) 형태로 누적하고 마지막에 요약 리포트를 출력합니다. 중간에 조용히 삼키지 않습니다.
  • 재사용 함수/구조화 팁: iter_target_files(root, include_ext, exclude_dirs, max_depth) 같은 제너레이터 함수로 탐색 정책을 고정하면 팀 코드 품질이 안정됩니다.
  • 성능/메모리 체크 포인트: 파일 수가 많은 환경에서는 전체 리스트화 금지, 집계는 카운터/딕셔너리로 즉시 누적, 필요 시 날짜 단위 배치 실행을 적용합니다.

추가로, 운영 자동화에서는 “탐색”과 “작업”을 분리하는 게 좋습니다. 예를 들어 1단계에서 대상 파일 목록을 로그로 남기고, 2단계에서 복사/삭제를 수행하면 장애가 났을 때 재현이 쉬워집니다. 또 dry-run 모드에서 “총 탐색 디렉터리 수, 파일 수, 제외된 디렉터리 수”를 먼저 보여 주면 사람이 검토하기 쉬워지고, 잘못된 경로를 조기에 발견할 수 있습니다. 이런 패턴은 단순해 보여도 실제 사고 예방 효과가 큽니다.

오늘의 결론

한 줄 요약: os.walk()의 진짜 힘은 파일 찾기가 아니라, 순회 정책을 코드로 명확하게 통제하는 데 있습니다.

기억할 것:

  • 큰 폴더 자동화에서 제외 규칙은 dirs[:]로 디렉터리 진입 전에 적용하세요.
  • 모든 경로를 모으지 말고 순회하면서 즉시 처리하는 스트리밍 패턴을 기본으로 삼으세요.
  • 권한/링크/깊이 제한 정책을 먼저 정하면 운영 중 예외와 성능 이슈를 크게 줄일 수 있습니다.

연습문제

  1. workspace_data/를 순회해 .csv와 .json만 대상으로 수집하되, tmp, .git, node_modules 디렉터리는 내려가지 않도록 구현하세요.
  2. 위 탐색 함수에 max_depth 인자를 추가해 깊이 제한 기능을 넣고, 실제로 제한이 적용되는지 (path, depth) 로그를 출력해 확인하세요.
  3. 순회 중 파일 크기를 확장자별로 집계해 “개수 상위 3개 확장자”와 “총 용량 상위 3개 확장자”를 각각 출력하는 리포트를 만드세요.

이전 강의 정답

  1. source_reports/ 하위 .csv만 backup_reports/로 구조 유지 복사
>>> from pathlib import Path
>>> import shutil
>>> def backup_csv_tree(src_root: str, dst_root: str) -> int:
...     src = Path(src_root)
...     dst = Path(dst_root)
...     copied = 0
...     for p in src.rglob("*.csv"):
...         if not p.is_file():
...             continue
...         rel = p.relative_to(src)
...         target = dst / rel
...         target.parent.mkdir(parents=True, exist_ok=True)
...         shutil.copy2(p, target)
...         copied += 1
...     return copied
>>> isinstance(backup_csv_tree("source_reports", "backup_reports"), int)
True
  1. 원본/대상 .csv 개수 비교 검증 함수
>>> from pathlib import Path
>>> def verify_csv_count(src_root: str, dst_root: str) -> tuple[int, int, bool]:
...     src_count = sum(1 for _ in Path(src_root).rglob("*.csv"))
...     dst_count = sum(1 for _ in Path(dst_root).rglob("*.csv"))
...     ok = (src_count == dst_count)
...     if not ok:
...         print(f"[WARN] csv count mismatch: src={src_count}, dst={dst_count}")
...     return src_count, dst_count, ok
>>> isinstance(verify_csv_count("source_reports", "backup_reports"), tuple)
True
  1. 날짜 기반 zip 생성 후 dist/ 이동 + 실패 로그 기록
>>> from datetime import datetime
>>> from pathlib import Path
>>> import shutil
>>> def archive_and_move(src_dir: str, dist_dir: str):
...     errors = []
...     try:
...         date_str = datetime.now().strftime("%Y%m%d")
...         base = Path("archives") / f"archive_{date_str}"
...         base.parent.mkdir(parents=True, exist_ok=True)
...         Path(dist_dir).mkdir(parents=True, exist_ok=True)
...         zip_path = shutil.make_archive(str(base), "zip", root_dir=src_dir)
...         final = shutil.move(zip_path, str(Path(dist_dir) / f"archive_{date_str}.zip"))
...         return final, errors
...     except Exception as e:
...         errors.append((type(e).__name__, str(e)))
...         return None, errors
>>> final_path, failed = archive_and_move("archives", "dist")
>>> isinstance(failed, list)
True

실습 환경/재현 정보

  • 실행 환경: conda env python100 (Python 3.11.14)
  • 가정한 OS: macOS/Linux 공통
  • 사용 모듈: os, os.path, collections, pathlib, shutil
  • 재현 절차:
    1. workspace_data, logs, repo, huge_data 같은 샘플 디렉터리를 생성합니다.
    2. 예제 1~4를 순서대로 실행하며 탐색 결과 수, 제외 동작, 깊이 제한 동작을 확인합니다.
    3. 연습문제를 구현한 뒤 실제 트리에서 파일 수/용량 집계가 예상대로 나오는지 검증합니다.
  • 검증 포인트: 제외 폴더 미진입 여부, 메모리 과다 사용 여부(리스트화 금지), 에러 로그 누락 여부, 결과 리포트 재현 가능성