새 소스 추가 절차¶
체크리스트¶
- [ ] 1. 대상 사이트 분석 (HTML 구조, 데이터 위치, 갱신 주기)
- [ ] 2. 스크래퍼 클래스 구현
- [ ] 3. 파싱 단위 테스트 작성
- [ ] 4. 실제 수집 테스트 (로컬)
- [ ] 5. 파이프라인에 등록
- [ ] 6. Data Catalog 문서 업데이트
- [ ] 7. 빌드 & 배포
1. 대상 사이트 분석¶
수집 전에 확인할 사항:
| 항목 | 확인 내용 |
|---|---|
| URL | 모바일/PC 중 어느 버전이 파싱하기 쉬운가 |
| 렌더링 | SSR(서버 렌더링)인가, CSR(클라이언트 렌더링)인가 |
| 선택자 | 트렌드 데이터를 특정할 CSS 선택자 또는 API 엔드포인트 |
| 갱신 주기 | 데이터가 얼마나 자주 바뀌는가 |
| 제약 | rate limit, 로그인 필요 여부, 새벽 제한 등 |
| robots.txt | 스크래핑 허용 여부 |
2. 스크래퍼 클래스 구현¶
scraper/ 아래에 새 파일 생성:
# scraper/naver.py
import requests
from bs4 import BeautifulSoup
from trend_collector.scraper.base import (
ScrapedKeyword, ScrapedResult, TrendScraper,
)
class NaverTrendScraper(TrendScraper):
@property
def source_name(self) -> str:
return "NAVER"
@property
def base_url(self) -> str:
return "https://..."
def scrape(self) -> ScrapedResult:
response = requests.get(self.base_url, headers=..., timeout=10)
response.raise_for_status()
raw_html = response.text
keywords = self.parse(raw_html)
return ScrapedResult(
source_name=self.source_name,
collected_at=datetime.now(timezone.utc),
raw_html=raw_html,
keywords=keywords,
)
def parse(self, raw_html: str) -> list[ScrapedKeyword]:
soup = BeautifulSoup(raw_html, "html.parser")
# 사이트별 파싱 로직
...
3. 파싱 단위 테스트¶
# tests/test_naver_scraper.py
def test_parse_extracts_keywords():
html = """<실제 HTML 샘플>"""
scraper = NaverTrendScraper()
keywords = scraper.parse(html)
assert len(keywords) == 10
assert keywords[0].rank == 1
4. 실제 수집 테스트¶
PYTHONPATH=src python -c "
from trend_collector.scraper.naver import NaverTrendScraper
result = NaverTrendScraper().scrape()
for kw in result.keywords:
print(f'{kw.rank}위: {kw.keyword} [{kw.status}]')
"
5. 파이프라인에 등록¶
main.py의 build_pipeline():
from trend_collector.scraper.naver import NaverTrendScraper
def build_pipeline() -> TrendCollectPipeline:
scrapers = [
DaumTrendScraper(),
NaverTrendScraper(), # 추가
]
...
6. Data Catalog 업데이트¶
docs/data-catalog/sources.md에 새 소스 명세 추가.