나의 피와 땀을 흘려가며 만들었던 크롤러... 여러 에피소드가 많았는데 그래도 이제는 어찌어찌 잘 굴러가는 중이라 벨로그에 정리해본다.
원래는 집 와서 작업 이어서 하려고 했는데 분명 세팅 맞춰뒀건만... 안열려. 몰라몰라 안해!
🗂️ 크롤러 마이크로서비스 아키텍처
GAENCHWIS/ # 프로젝트 루트 디렉토리
├── chrome_extension/ # 크롬 확장 프로그램 관련 코드
├── client/ # 프론트엔드 클라이언트 코드
└── server/ # 백엔드 서버 코드
├── crawler/ # 크롤링 서비스 메인 디렉토리
│ ├── __init__.py
│ ├── base/ # 크롤러 기본 구조 정의
│ │ ├── __init__.py
│ │ └── base_crawler.py # 크롤러 추상 클래스 (템플릿 메서드 패턴)
│ ├── crawlers/ # 사이트별 크롤러 구현
│ │ ├── __init__.py
│ │ ├── jobkorea_crawler.py # 잡코리아 크롤러 구현
│ │ └── saramin_crawler.py # 사람인 크롤러 구현
│ └── common/ # 공통 유틸리티 및 설정
│ ├── __init__.py
│ ├── config.py # 환경 설정 관리 (dataclass 기반)
│ ├── aws_client.py # AWS 서비스 연동 (싱글톤 패턴)
│ ├── enums.py # 상태 및 타입 열거형 정의
│ ├── constants.py # 상수 및 설정값 정의
│ ├── utils.py # 공통 유틸리티 함수
│ └── driver_setup.py # 웹드라이버 설정 (봇 감지 방지 로직)
├── Dockerfile # 도커 컨테이너 설정
├── requirements.txt # Python 패키지 의존성
└── main.py # 애플리케이션 진입점 (API/CLI 인터페이스) 당초에는 사람인 크롤러와 잡코리아 크롤러, 이후에 확장성을 고려하여 base_crawler.py를 만들고, 크롤러 사이트를 추가할 때는 이를 상속받아서 사용할 수 있게 하려고 했다.
그런데? 어쩌다보니... 잡코리아 크롤러를 못 쓰게 됨 ㅋㅎ 크롤링 함부로 하면 안된다는 사실을 이번에 많이 배워갑니다... 그래서 결국 잡코리아는 코드는 있지만 현재 사용은 못하는 중이다. 사람인만 크롤링 하게 되었어요... 머쓱...
🖨️ 크롤러
classDiagram
class BaseCrawler {
<<abstract>>
-output_dir: str
-driver: WebDriver
-logger: Logger
+initialize()
+cleanup()
+natural_scroll()
+wait_random()
+wait_for_element()
+safe_page_navigation()
#crawl()*
}
class SaraminCrawler {
-url: str
-dynamodb: DynamoDB
+crawl()
-_setup_logger()
-_setup_aws_resources()
-_initialize_repositories()
-_generate_hash()
-process_and_save_data()
-_save_company_info()
-_process_tags()
-_save_job_posting()
-_save_job_tags()
-crawl_jobs()
}
class CrawlerExecutor {
-config: Config
-logger: Logger
+execute_crawler()
-_setup_logger()
}
class APIServer {
-app: Flask
-config: Config
-crawler_executor: CrawlerExecutor
+run_crawler()
+healthcheck()
+get_status()
+run()
}
BaseCrawler <|-- SaraminCrawler
CrawlerExecutor --> SaraminCrawler
APIServer --> CrawlerExecutor📥 base_crawler.py
BaseCrawler는 모든 크롤러의 기본 기능을 제공한다.
추상 클래스로 구현했기 때문에 다른 사이트 크롤러를 추가하기 쉽게 만들었다.
자원 관리나 초기화 로직을 포함하고 있고, 봇 감지를 방지하기 위한 기본 메서드들을 제공하는 곳이다.
📨 saramin_crawler.py
SaraminCrawler에서의 데이터 처리 파이프라인
🖥️ 수집 → 파싱 → 정규화 → 저장
중요한 함수들만 몇 가지 설명하려고 한다.
데이터 수집
crawl() 함수
def crawl(self):
# 크롤링 메인 프로세스
try:
if not os.path.exists(self.output_dir):
os.mkdir(self.output_dir)
print(f"출력 디렉토리 생성: {self.output_dir}")
self.driver.get(self.url)
print("URL 접속 시도:", self.url)
# 초기 페이지 로딩 대기
if not self.wait_for_element(By.CSS_SELECTOR, "div.box_item"):
raise TimeoutException("초기 페이지 로딩 실패")
print("페이지 로딩 완료")
# 채용공고 크롤링
saramin_data = self.crawl_jobs()
print(f"크롤링된 데이터 수: {len(saramin_data)}")
if saramin_data:
print("데이터 처리 시작")
try:
# CSV 파일 저장
output_file = os.path.join(self.output_dir, 'saramin_job.csv')
save_to_csv(saramin_data, output_file)
print(f"CSV 파일 저장 완료: {output_file}")
# DynamoDB에 데이터 저장
print("DynamoDB 저장 시작")
self.process_and_save_data(saramin_data)
print("DynamoDB 저장 완료")
except PermissionError as e:
print(f"파일 저장 권한 오류: {str(e)}")
print("DynamoDB 저장 시작")
self.process_and_save_data(saramin_data)
print("DynamoDB 저장 완료")
except Exception as e:
print(f"데이터 처리 중 오류 발생: {str(e)}")
raise크롤링 프로세스의 메인 진입점이다. 출력 디렉토리를 만들고, URL에 접속하고 초기 페이지 로딩 후 크롤링 데이터를 CSV로 저장한다. 이건... DynamoDB 저장하기 전에 데이터를 잘 긁어오는지 확인하려고 만들었던 거라... 지워야 하는데... 심지어 주석도 어디서 에러나는지 몰라서 열심히 그리고 끔찍하게 달아둔 모습니다. 이거 언제 다 정리하냐... 암튼 DynamoDB에 저장하는 프로세스도 여기에서 실행한다.
crawl_jobs()
def crawl_jobs(self) -> List[Dict]:
saramin_list = []
max_pages = 2
for page in range(1, max_pages + 1):
try:
if page > 1:
try:
# 페이지네이션 영역 찾기
pagination = self.wait_for_element(By.CLASS_NAME, "PageBox")
if pagination:
# 다음 페이지 버튼 찾기 (page 속성값으로 찾기)
next_page_button = self.driver.find_element(
By.CSS_SELECTOR,
f"button.BtnType.SizeS[page='{page}']"
)
# 버튼이 보이도록 스크롤
self.driver.execute_script(
"arguments[0].scrollIntoView({behavior: 'smooth', block: 'center'});",
next_page_button
)
self.wait_random(1, 2) # 스크롤 후 자연스러운 대기
# 클릭 전 약간의 지연
self.wait_random(0.5, 1)
# 버튼 클릭
next_page_button.click()
# 페이지 로딩 대기
self.wait_random(2, 3)
# 새 페이지의 컨텐츠가 로드될 때까지 대기
if not self.wait_for_element(By.CLASS_NAME, "box_item"):
self.logger.warning(f"페이지 {page}의 컨텐츠를 찾을 수 없습니다.")
continue
else:
self.logger.warning(f"페이지 {page}의 페이지네이션을 찾을 수 없습니다.")
continue
except Exception as e:
self.logger.error(f"페이지 {page} 이동 중 오류 발생: {str(e)}")
continue
# 채용공고 항목 대기
if not self.wait_for_element(By.CLASS_NAME, "box_item"):
self.logger.warning(f"페이지 {page}에서 채용공고 항목을 찾을 수 없음")
continue
# 자연스러운 스크롤 동작
self.natural_scroll()
# HTML 파싱
soup = BeautifulSoup(self.driver.page_source, 'html.parser')
job_items = soup.select('.box_item')
if not job_items:
print(f"페이지 {page}: 공고 항목을 찾을 수 없습니다. 다음 페이지로 진행합니다.")
continue
self.logger.info(f"페이지 {page} - 발견된 채용공고 수: {len(job_items)}")
for item in job_items:
parsed_item = self._parse_job_item(item)
if parsed_item:
saramin_list.append(parsed_item)
self.logger.info(f"수집된 공고: {parsed_item['회사명']} - {parsed_item['공고제목']}")
self.wait_random(2, 3) # 자연스러운 딜레이
self.wait_random(15, 30) # 페이지 이동 전 딜레이
except TimeoutException:
self.logger.error(f"페이지 {page} 처리 중 시간 초과")
continue
except Exception as e:
self.logger.error(f"페이지 {page} 처리 중 오류 발생: {str(e)}")
continue
self.logger.info(f"총 수집된 공고 수: {len(saramin_list)}")
return saramin_list실제 공고 데이터를 수집하는 기능은 이 함수가 담당하고 있다. 페이지네이션 처리나 자연스러운 스크롤 동작을 구현하고 있고, 여기에서 BeautifulSoup을 통해서 HTML을 파싱한다.
데이터 파싱
_parse_job_item()
개별 채용 공고 항목을 파싱하는 역할을 담당하는 중요한 함수이다.
{
'회사명': company,
'공고제목': title,
'직무분야': sector,
'근무지': location,
'경력/고용형태': career,
'학력': education,
'마감일': deadline,
'등록일': posted,
'공고URL': job_url
}이런 내용들을 파싱하는데,
- _parse_sector(): 직무분야 정보 파싱
- _parse_deadline(): 마감일 문자열 변환
- _extract_rec_idx(): 공고 고유 ID 추출
디테일한 작업이 필요한 경우에는 따로 보조 함수를 분리해서 기능을 분리할 수 있도록 했다.
데이터 정규화
process_and_save_data()
def process_and_save_data(self, saramin_data: List[Dict]) -> None:
print(f"처리할 데이터 개수: {len(saramin_data)}")
# 크롤링한 데이터 처리 및 저장
for job_data in saramin_data:
try:
print(f"현재 처리중인 데이터: {job_data}")
# 고유 ID 생성
company_id = self._generate_hash(job_data['회사명'])
post_id = self._generate_hash(job_data['공고URL'])
print(f"생성된 ID - company_id: {company_id}, post_id: {post_id}")
# 회사 정보 저장
try:
self._save_company_info(company_id, job_data)
except Exception as e:
print(f"회사 정보 저장 실패: {str(e)}")
raise
# 태그 처리 및 저장
try:
tags = self._process_tags(job_data)
except Exception as e:
print(f"태그 처리 실패: {str(e)}")
raise
# 채용 공고 저장
try:
self._save_job_posting(company_id, post_id, job_data)
except Exception as e:
print(f"채용공고 저장 실패: {str(e)}")
raise
# 공고-태그 매핑 저장
try:
self._save_job_tags(post_id, tags)
except Exception as e:
print(f"태그매핑 저장 실패: {str(e)}")
raise
self.logger.info(f"데이터 처리 완료: {job_data['회사명']} - {job_data['공고제목']}")
except Exception as e:
self.logger.error(f"데이터 처리 중 오류 ({job_data['회사명']}): {str(e)}")
print(f"전체 처리 실패: {str(e)}")
continue 수집된 데이터를 정규화하고 저장하는 기능이다.
고유의 ID를 만들고, 회사 정보를 정규화하고, 태그 데이터들을 처리한다. 처리할 게 굉장히 많았다. 채용 공고 정보도 정규화 한 다음 태그와 공고 사이의 매핑도 생성한다.
_process_tags()
각종 태그들을 처리하는 함수이다.
- 직무분야 (skill)
- 경력/고용형태 (position)
- 학력 (education)
- 지역 (location)
- _process_location_tag(): 지역 정보 계층 구조화
- _process_skill_tags(): 직무분야 태그 생성
- _process_career_type_tags(): 경력/고용형태 태그 생성
- _process_education_tag(): 학력 태그 생성
지역 같은 경우 서울이나 인천, 경기 지역 전체 카테고리에 들어갈 경우 level1, 구나 시 단위로 들어갈 경우 level2를 부여하여 지역 필터링을 용이하게 할 수 있도록 구성했다. 아... 좀 더 고칠 거 있는데 머였는지 까먹음. 해도해도 계속 뭐가 생겨요...
데이터 저장
회사 정보 저장
# _save_company_info() 함수
company_data = {
'PK': f"COMPANY#{company_id}",
'SK': f"METADATA#{company_id}",
'company_id': company_id,
'company_name': job_data['회사명'],
'created_at': datetime.now().isoformat(),
'updated_at': datetime.now().isoformat(),
'GSI1PK': "COMPANY#ALL",
'GSI1SK': job_data['회사명']
}채용공고 저장
# _save_job_posting() 함수
job_data_processed = {
'PK': f"COMPANY#{company_id}",
'SK': f"JOB#{post_id}",
'post_id': post_id,
'post_name': job_data['공고제목'],
'company_id': company_id,
'company_name': job_data['회사명'],
'is_closed': deadline_str,
'post_url': job_data['공고URL'],
'rec_idx': rec_idx,
'status': 'active',
...
}태그 매핑 저장
# _save_job_tags() 함수
mapping_data = {
'PK': f"JOB#{post_id}",
'SK': f"TAG#{tag_id}",
'job_tag_id': job_tag_id,
'job_id': post_id,
'tag_id': tag_id,
'created_at': datetime.now().isoformat(),
'GSI1PK': f"TAG#{tag_id}",
'GSI1SK': f"JOB#{post_id}"
}이런 식으로 데이터를 DynamoDB에 저장할 수 있게 하는 함수들이다.
공통 유틸리티 및 설정
주요 컴포넌트만 설명을 좀 달겠음.
환경 설정
config.py
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class Config:
# API 설정
API_KEY: str = os.getenv('API_KEY', 'default_api_key')
# 서버 설정
HOST: str = os.getenv('HOST', '0.0.0.0')
PORT: int = int(os.getenv('PORT', '8000'))
DEBUG: bool = os.getenv('FLASK_ENV') == 'development'
# AWS 설정
AWS_REGION: str = os.getenv('AWS_REGION', 'ap-northeast-2')
AWS_ACCESS_KEY_ID: Optional[str] = os.getenv('AWS_ACCESS_KEY_ID')
AWS_SECRET_ACCESS_KEY: Optional[str] = os.getenv('AWS_SECRET_ACCESS_KEY')
# 크롤러 설정
CHROME_DRIVER_PATH: str = os.getenv('CHROME_DRIVER_PATH', '/usr/local/bin/chromedriver')
HEADLESS: bool = os.getenv('HEADLESS', 'true').lower() == 'true'
# 로깅 설정
LOG_LEVEL: str = os.getenv('LOG_LEVEL', 'INFO')
LOG_FILE: str = os.getenv('LOG_FILE', 'crawler_executor.log')
def __post_init__(self):
"""설정 유효성 검증"""
# AWS 인증 정보가 없을 경우 경고
if not self.AWS_ACCESS_KEY_ID or not self.AWS_SECRET_ACCESS_KEY:
print("Warning: AWS credentials not found in environment variables")
# 환경변수로 AWS 설정
if self.AWS_ACCESS_KEY_ID and self.AWS_SECRET_ACCESS_KEY:
os.environ['AWS_ACCESS_KEY_ID'] = self.AWS_ACCESS_KEY_ID
os.environ['AWS_SECRET_ACCESS_KEY'] = self.AWS_SECRET_ACCESS_KEY
os.environ['AWS_REGION'] = self.AWS_REGION환경 변수를 기반으로 설정을 관리하고 있으며, AWS 인증 정보와 크롤러 설정을 관리한다.
AWS 서비스 연동
aws_client.py
import boto3
from typing import Dict, Any
from .config import Config
class AWSClient:
_instances: Dict[str, Any] = {}
@classmethod
def get_client(cls, service_name: str):
if service_name not in cls._instances:
try:
config = Config()
if service_name == 'dynamodb':
cls._instances[service_name] = boto3.resource(
service_name,
aws_access_key_id=config.AWS_ACCESS_KEY_ID,
aws_secret_access_key=config.AWS_SECRET_ACCESS_KEY,
region_name=config.AWS_REGION
)
else:
cls._instances[service_name] = boto3.client(
service_name,
aws_access_key_id=config.AWS_ACCESS_KEY_ID,
aws_secret_access_key=config.AWS_SECRET_ACCESS_KEY,
region_name=config.AWS_REGION
)
except Exception as e:
raise Exception(f"AWS 클라이언트 생성 실패: {str(e)}")
return cls._instances[service_name] AWS 서비스를 연동하는 부분이다. DynamoDB 리소스나 일반 클라이언트를 생성한다. 원래는... aws_service에 관련된 폴더를 모듈화해서 다른 백엔드 서버에서 이용할 수 있게 하려고 했는데(?) 마이크로서비스에서는(?) 안된다고 함... 그래서 이렇게 나눈거였다. 그냥 DynamoDB 리소스에만 접근할 수 있게 수정할 예정이다. 언젠간... 암튼 여기에서 aws 연동을 관리한다. 한 번 만들어뒨 아주 요긴하게 사용하고 있는 코드이다.
크롬 드라이버 설정
driver_setup.py
def get_chrome_version():
try:
# Chrome 버전 확인 (Docker 환경에서만 동작)
if os.environ.get('DOCKER_CONTAINER'):
version = os.popen('google-chrome --version').read()
version = version.strip('Google Chrome ').strip().split('.')[0]
return version
return None
except Exception as e:
print(f"Chrome 버전 확인 실패: {str(e)}")
return None
def get_chromedriver_url(version):
try:
# 특정 버전의 ChromeDriver URL 가져오기
response = requests.get(f'https://chromedriver.storage.googleapis.com/LATEST_RELEASE_{version}')
if response.status_code == 200:
driver_version = response.text.strip()
# ChromeDriver 다운로드 URL 반환
return f"https://chromedriver.storage.googleapis.com/{driver_version}/chromedriver_linux64.zip"
except Exception as e:
print(f"ChromeDriver URL 가져오기 실패: {str(e)}")
return None
# 웹 드라이버 설정
def setup_driver():
# Chrome 옵션 객체 생성
options = webdriver.ChromeOptions()
# Docker 환경 감지
in_docker = os.environ.get('DOCKER_CONTAINER', False)
# [봇 감지 방지를 위한 핵심 설정]
# 0. 기본 옵션
options.add_argument('--disable-infobars') # 자동화된 테스트 소프트웨어에 의해 제어되고 있다는 알림 배너 비활성화
options.add_argument('--disable-extensions')
options.add_argument('--disable-features=NetworkService')
# 1. 랜덤 User-Agent 설정
options.add_argument(f'user-agent={random.choice(CrawlerConfig.USER_AGENTS)}') # 랜덤 User-Agent로 봇 감지 방지
# 2. 자동화 흔적 제거
options.add_argument('--disable-blink-features=AutomationControlled') # 자동화 감지 플래그 제거
options.add_experimental_option('excludeSwitches', ['enable-logging']) # 자동화 관련 로그 숨김
options.add_experimental_option('useAutomationExtension', False)
# 3. 웹 드라이버 관련 플래그 제거
options.add_argument('--disable-web-security') # 웹 보안 정책 우회ㅑ
options.add_argument('--ignore-certificate-errors') # 인증서 오류 무시
options.add_argument('--ignore-ssl-errors') # SSL 오류 무시
# 4. 실제 브라우저처럼 보이게 하는 설정
options.add_argument('--start-maximized') # 실제 사용자처럼 전체 화면으로 사용
options.add_argument('--lang=ko_KR') # 한국어 설정으로 현지화
# 5. 필수 성능 옵션
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
# Docker 환경 전용 설정
if in_docker:
options.add_argument('--headless')
options.binary_location = '/usr/bin/google-chrome'
# 6. 추가 봇 감지 방지 설정
prefs = {
'profile.default_content_setting_values.notifications': 2,
'credentials_enable_service': False,
'profile.password_manager_enabled': False
}
options.add_experimental_option('prefs', prefs)
# 페이지 로드 전략 변경 (normal: 모든 리소스 로드 대기)
options.page_load_strategy = 'normal'
try:
if in_docker:
# Docker 환경에서는 설치된 Chrome 버전 확인
chrome_version = get_chrome_version()
print(f"Detected Chrome vserion: {chrome_version}")
if not chrome_version:
raise Exception("Chrome 버전을 확인할 수 없습니다.")
service = Service()
else:
# 로컬 환경에서는 자동으로 ChromeDriverManager 설치 및 관리
service = Service(ChromeDriverManager().install())
# WebDriver 인스턴스 생성
driver = webdriver.Chrome(
service=service,
options=options
)
# 7. WebDriver 속성 숨기기
driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
driver.implicitly_wait(30) # 요소 대기 시간 설정
return driver
except Exception as e:
print(f"ChromeDriver 초기화 실패: {str(e)}")
raise이 친구 만드는데 제일 오래 걸렸다.
주요 기능
1. 크롬 버전 확인
2. 크롬 드라이버 URL 획득
3. 웹 드라이버 설정
봇인거 들키면 가차없이 차단당하는 사이트가 있다. (알고싶지 않았어요) 차단 몇 번 당해보고 나서야 이런저런 설정이 필요하다는 걸 알아서 피눈물 흘리며 설정들을 찾아 헤맸다. 잘 알지도 못하고 여기저기서 설정을 긁어오는 바람에 뭐가 봇 감지에 도움이 되고 되지 않는지 몰랐기 때문에 AI 한 번 돌려서 설정을 정리했다. 이런 설정들이 필요하다고 하네요...
나머지는 설정에 관련된 파일이다.
느낀점
확장성을 고려한 설계를 하려고 노력했던 결과물이다. 잘 된건지... 하지만 결국 사람인 크롤러 하나만 사용하게 된 아이러니... 하지만 나중엔 다른 사이트를 더 붙여서 사용할 수 있겠죠?
만드는 데 정말 오래걸리고, 지금도 DB 관련해서 스키마 바뀌면 수정해야 하고, 여전히 수정하고 개선할 점들이 많은 애물단지같은 서비스지만 돌아가는 거 보면 밥 안먹어도 될만큼 뿌듯하다.
나는 갱스터 크롤러... 프로젝트라서 예의없게 여기저기 크롤링을 막 하고 다녔는데 실제로 상업적인 서비스에서는 이러면 안된다. 알고 있습니다...! 일단 우리 서비스에는 필요한 데이터들이기 때문에 열심히 긁긴 했다. 후... 나머지도 잘 마무리하고 프로젝트가 어서 끝났으면 좋겠다. 건강이 안좋아지는 것이 느껴져...!
본 포스팅은 글로벌소프트웨어캠퍼스와 교보DTS가 함께 진행하는 챌린지입니다.
영차영차 😎