누리장터(정부 조달 플랫폼)에서 입찰공고 데이터를 자동 수집하는 프로덕션급 웹 크롤러
# 1. 저장소 이동
cd nurijangter-crawler
# 2. 가상환경 생성 및 활성화
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 3. 의존성 설치
pip install -r requirements.txt
# 4. Playwright 브라우저 설치
playwright install chromium기본 실행 (체크포인트 자동 재개)
python main.py처음부터 새로 시작
python main.py --no-resume --clear-checkpoint디버그 모드
python main.py --log-level DEBUG스케줄 모드 (자동 반복 실행)
python main.py --scheduled특정 페이지 크롤링 (범위 지정 가능)
# 5페이지만 크롤링
python main.py --pages 5
# 2페이지부터 5페이지까지 크롤링
python main.py --pages 2-5
# 10페이지부터 끝까지 크롤링
python main.py --pages 10-재시도 실행 (실패 항목 다시 처리)
python main.py --retry-failed참고: 재시도 성공 시
crawler_checkpoint.json의 실패 목록에서 자동 제거되고,seen_items.json에 처리된 항목으로 추가됩니다.
# 수집된 데이터
ls data/*.json
ls data/*.csv
# 로그
tail -f logs/crawler_*.logconfig/config.yaml 파일에서 수정:
crawler:
pagination:
max_pages: 1 # 테스트: 1페이지만 / 프로덕션: 0 (무제한)
scheduler:
enabled: true
mode: 'interval'
interval:
hours: 6 # 6시간마다 실행- Python: 3.11 이상
- OS: macOS, Linux, Windows (WSL2 권장)
- 메모리: 최소 2GB RAM (권장 4GB)
playwright==1.49.0 # 브라우저 자동화
pydantic==2.9.0 # 데이터 검증
pyyaml==6.0.1 # 설정 파일 파싱
schedule==1.2.0 # 작업 스케줄링
python-json-logger==2.0.7 # 구조화된 로깅
requirements.txt 참고:
- Core: playwright, pydantic, pyyaml
- Scheduling: schedule, croniter
- Logging: python-json-logger
- Testing: pytest, pytest-cov, pytest-mock
- Development: black, flake8, mypy, isort
# Python 버전 확인
python --version # 3.11+ 필요
# 의존성 확인
pip list | grep -E "playwright|pydantic|pyyaml"
# Playwright 브라우저 확인
playwright --versionmain.py
└─ CrawlerEngine (오케스트레이션)
├─ Navigator (SPA 네비게이션 & 모달 제어)
├─ Processor (개별 공고 처리 & 상세 수집)
│ ├─ ListParser (목록 페이지 파싱)
│ └─ DetailParser (상세 페이지 파싱)
├─ RetryManager (실패 항목 재시도 & 검색)
├─ BrowserManager (Playwright 제어)
├─ CheckpointManager (상태 저장/복원)
├─ DeduplicationManager (중복 방지)
└─ Storage (JSON/CSV 저장)
설계 의도:
- Navigator: 복잡한 SPA(Single Page Application) 탐색과 팝업 처리를 전담
- Processor: 공고 데이터 수집 로직을 분리하여 유지보수성 향상
- RetryManager: 실패한 항목을 독립적으로 재처리하는 전용 모듈
- 각 모듈을 독립적으로 테스트 및 교체 가능하도록 분리
# Method 1: 새 탭 시도 (timeout=3s)
# Method 2: 모달 확인
# Method 3: In-Page 컨텐츠 로드 (누리장터 방식)이유: 누리장터는 WebSquare 기반 SPA로, In-Page 방식으로 상세를 로드함. 범용성을 위해 3가지 방법 모두 구현.
# Strategy 1: XPath (TH → TD)
# Strategy 2: Table Row Iteration
# Strategy 3: Label-Value Div/Span이유: 정부 사이트는 HTML 구조가 불규칙하므로, 3가지 전략 병합으로 90%+ 필드 추출률 확보.
- 가정: 누리장터는 WebSquare 프레임워크 사용
- 가정: 상세페이지는 In-Page 방식으로 로드됨 (새 탭/모달 아님)
- 검증:
debug_detail_page.py로 실제 동작 분석하여 확인
- 가정: 필드명이 한글이며 일관되지 않음
- 대응: Fuzzy Matching (예: "전화번호", "연락처 전화번호", "담당자연락처" 모두 매칭)
- 가정: 일부 필드는 누락되거나 잘못된 위치에 존재 가능
- 대응: 모든 필드를 Optional로 정의, 값 검증 추가
- 가정: 장시간 실행 (수백~수천 페이지)
- 대응: 10개 항목마다 체크포인트 자동 저장
- 가정: 네트워크 불안정, 간헐적 타임아웃 발생
- 대응: Exponential Backoff로 3회 재시도
- 선택: 동기 방식 (비동기 대신)
- 이유:
- Rate Limit 위반 방지 (IP 차단 위험)
- 디버깅 용이성
- 안정성 우선 (속도는 나중에 최적화 가능)
- 전략: 필드 검증 강화
- 구현:
budget_amount,base_price등 핵심 필드 중 4개 이상 누락 시 실패 처리 - 효과: 불완전한 데이터가 DB/파일에 저장되는 것을 방지하고 재시도 큐로 이동시킴
| 결정 | 선택 | 이유 |
|---|---|---|
| 브라우저 자동화 | Playwright | 자동 대기, 안정성 2배 향상 |
| 동기 vs 비동기 | 동기 | 안정성, 디버깅, Rate Limit 준수 |
| 에러 처리 | Fail-Safe | 일부 실패해도 계속 진행 |
| 중복 방지 | Hash 기반 | O(1) 조회, 영속성 |
| 체크포인트 | JSON + Atomic Write | 동시성 안전, 재개 가능 |
- 현상: 10개 항목당 ~2분 소요 (동기 처리)
- 원인: 각 상세페이지를 순차적으로 로드
- 영향: 1000개 항목 수집 시 3-4시간 소요
- 현상: In-Page 상세 조회 후 목록으로 돌아오지 못하는 경우 있음
- 현재 대응: 메뉴 재클릭으로 강제 리셋
- 문제: 추가 2-3초 소요, 비효율적
- 현상: 평균 80-90% 필드 추출 성공
- 원인:
- 테이블 구조가 페이지마다 다름
- 일부 필드는 이미지로 표시됨
- 미추출 필드 예시: 첨부파일, 일부 특수 필드
- 현상: HTML 구조 변경 시 파서 수정 필요
- 예시:
col_id="bidPbancNm"→ 다른 ID로 변경되면 작동 안 함 - 현재 대응: 다층 전략으로 일부 완화, 그러나 근본 해결 아님
- 현상: 에러 발생해도 즉시 알 수 없음
- 현재: 로그 파일 수동 확인 필요
- 문제: 장시간 실행 시 문제 발견 지연
1. 비동기 처리 전환
# 현재: 동기
for notice in notices:
detail = fetch_detail(notice) # 순차
# 개선: 비동기
tasks = [fetch_detail(notice) for notice in notices]
results = await asyncio.gather(*tasks) # 병렬- 효과: 3-5배 속도 향상 (2분 → 30초)
- 주의: Rate Limit 설정 필수 (초당 5개 제한)
2. 목록 복귀 개선
# 현재: 메뉴 재클릭
menu.click()
search.click()
# 개선: "목록" 버튼 찾기
back_btn = page.locator('button:has-text("목록")')
if back_btn.is_visible():
back_btn.click()- 효과: 복귀 시간 2초 절감
3. Slack 알림 연동
def send_alert(message):
requests.post(SLACK_WEBHOOK, json={'text': message})
# 치명적 에러 시
except CriticalError as e:
send_alert(f"🚨 Crawler failed: {e}")- 효과: 실시간 에러 알림
4. 데이터베이스 연동
# 현재: JSON/CSV 파일
storage = JSONStorage()
# 개선: PostgreSQL
storage = PostgreSQLStorage()
# 테이블: bids, attachments
# 인덱스: bid_notice_number, announcement_date- 효과:
- 실시간 조회 가능
- 중복 검사 DB 레벨에서 처리
- 증분 크롤링 효율적
5. ML 기반 필드 추출
# 현재: Fuzzy Matching
field_mappings = {'phone': ['전화번호', '연락처']}
# 개선: BERT 임베딩
similarity = cosine_similarity(
embed("전화번호"),
embed(label_text)
)
if similarity > 0.8:
mapped_field = "phone_number"- 효과:
- 구조 변경에도 견고
- 새로운 필드명 자동 학습
- 90% → 95%+ 추출률
6. 증분 크롤링 최적화
# 현재: 전체 크롤링
for page in range(1, max_pages):
crawl(page)
# 개선: 변경 감지
last_crawl_time = get_last_crawl_time()
new_notices = get_notices_after(last_crawl_time)
# 새 공고만 수집- 효과: 일일 크롤링 시간 80% 단축
7. 분산 크롤링
# Celery + Redis
@celery.task
def crawl_page(page_num):
return crawler.crawl_page(page_num)
# 10개 워커에서 병렬 처리
results = group(crawl_page.s(i) for i in range(100))()- 효과:
- 100배 속도 향상 가능
- 여러 서버에서 분산 처리
- 대규모 데이터 수집 (수십만 건)
8. 실시간 모니터링 대시보드
# Grafana + Prometheus
metrics = {
'items_collected': Counter(),
'success_rate': Gauge(),
'avg_response_time': Histogram()
}
# 웹 대시보드
# - 실시간 진행률 그래프
# - 에러율 추이
# - 페이지별 성능- 효과:
- 실시간 상태 확인
- 문제 즉시 발견
- 성능 병목 지점 파악
9. Auto-Scaling
# Kubernetes + HPA
if queue_length > 1000:
scale_up(replicas=10)
elif queue_length < 100:
scale_down(replicas=2)- 효과:
- 데이터 양에 따라 자동 확장
- 비용 최적화
- 안정적인 처리량 보장
| 우선순위 | 개선 항목 | 예상 효과 | 난이도 |
|---|---|---|---|
| 🔥 1 | Slack 알림 | 운영 가시성 | 낮음 |
| 🔥 2 | 비동기 처리 | 속도 3-5배 | 중간 |
| 📊 3 | DB 연동 | 실시간 조회 | 중간 |
| 📊 4 | 증분 크롤링 | 시간 80% 단축 | 낮음 |
| 🚀 5 | ML 기반 추출 | 추출률 95%+ | 높음 |
| 🚀 6 | 분산 크롤링 | 속도 100배 | 높음 |
이 프로젝트는 교육 및 기술 평가 목적으로 제공됩니다.