178 lines
6.2 KiB
Python
178 lines
6.2 KiB
Python
import logging
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from contextlib import contextmanager
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.options import Options
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.common.exceptions import StaleElementReferenceException, TimeoutException
|
|
|
|
# 로깅 설정
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class SeleniumManager:
|
|
"""Selenium WebDriver 관리 클래스"""
|
|
|
|
WEATHER_SELECTORS = {
|
|
'tab_button': (By.XPATH, '//*[@id="digital-forecast"]/div[1]/div[3]/div[1]/div/div/a[2]'),
|
|
'list_button': (By.XPATH, '//*[@id="digital-forecast"]/div[1]/div[3]/ul/div[1]/a[2]'),
|
|
'target_element': (By.XPATH, '/html/body/div[1]/main/div[2]/div[1]'),
|
|
}
|
|
|
|
def __init__(self, headless=True, window_size=(1802, 1467), timeout=10):
|
|
"""
|
|
Args:
|
|
headless: 헤드리스 모드 여부
|
|
window_size: 브라우저 윈도우 크기
|
|
timeout: WebDriverWait 기본 타임아웃 (초)
|
|
"""
|
|
self.headless = headless
|
|
self.window_size = window_size
|
|
self.timeout = timeout
|
|
self.temp_dir = None
|
|
self.driver = None
|
|
|
|
def _setup_chrome_options(self):
|
|
"""Chrome 옵션 설정"""
|
|
options = Options()
|
|
|
|
if self.headless:
|
|
options.add_argument('--headless')
|
|
|
|
options.add_argument(f'--window-size={self.window_size[0]},{self.window_size[1]}')
|
|
options.add_argument('--no-sandbox')
|
|
options.add_argument('--disable-dev-shm-usage')
|
|
options.add_argument('--disable-gpu')
|
|
|
|
# 임시 사용자 데이터 디렉토리 생성 (중복 실행 문제 방지)
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
options.add_argument(f'--user-data-dir={self.temp_dir}')
|
|
|
|
return options
|
|
|
|
def start(self):
|
|
"""WebDriver 시작"""
|
|
try:
|
|
options = self._setup_chrome_options()
|
|
self.driver = webdriver.Chrome(options=options)
|
|
logger.info("Selenium WebDriver 시작됨")
|
|
except Exception as e:
|
|
logger.error(f"WebDriver 시작 실패: {e}")
|
|
self.cleanup()
|
|
raise
|
|
|
|
def cleanup(self):
|
|
"""WebDriver 종료 및 임시 파일 정리"""
|
|
if self.driver:
|
|
try:
|
|
self.driver.quit()
|
|
logger.info("WebDriver 종료됨")
|
|
except Exception as e:
|
|
logger.warning(f"WebDriver 종료 중 오류: {e}")
|
|
|
|
if self.temp_dir and os.path.exists(self.temp_dir):
|
|
try:
|
|
shutil.rmtree(self.temp_dir)
|
|
logger.info(f"임시 디렉토리 삭제됨: {self.temp_dir}")
|
|
except Exception as e:
|
|
logger.warning(f"임시 디렉토리 삭제 실패: {e}")
|
|
|
|
@contextmanager
|
|
def managed_driver(self):
|
|
"""Context manager를 통한 자동 정리"""
|
|
try:
|
|
self.start()
|
|
yield self.driver
|
|
finally:
|
|
self.cleanup()
|
|
|
|
def wait(self):
|
|
"""WebDriverWait 인스턴스 반환"""
|
|
return WebDriverWait(self.driver, self.timeout)
|
|
|
|
def click_with_retry(self, selector, max_retries=5, sleep_time=1):
|
|
"""
|
|
재시도 로직을 포함한 요소 클릭
|
|
|
|
Args:
|
|
selector: (By, xpath) 튜플
|
|
max_retries: 최대 재시도 횟수
|
|
sleep_time: 재시도 사이의 대기 시간 (초)
|
|
|
|
Returns:
|
|
성공 여부
|
|
"""
|
|
for attempt in range(max_retries):
|
|
try:
|
|
wait = self.wait()
|
|
element = wait.until(EC.presence_of_element_located(selector))
|
|
wait.until(EC.element_to_be_clickable(selector))
|
|
element.click()
|
|
logger.info(f"요소 클릭 성공: {selector}")
|
|
return True
|
|
except StaleElementReferenceException:
|
|
logger.warning(f"시도 {attempt + 1}: StaleElementReferenceException 발생, 재시도 중...")
|
|
if attempt < max_retries - 1:
|
|
import time
|
|
time.sleep(sleep_time)
|
|
except TimeoutException:
|
|
logger.warning(f"시도 {attempt + 1}: TimeoutException 발생, 재시도 중...")
|
|
if attempt < max_retries - 1:
|
|
import time
|
|
time.sleep(sleep_time)
|
|
except Exception as e:
|
|
logger.error(f"시도 {attempt + 1}: 예상치 못한 오류 {type(e).__name__}: {e}")
|
|
if attempt < max_retries - 1:
|
|
import time
|
|
time.sleep(sleep_time)
|
|
|
|
logger.error(f"최대 재시도 횟수 초과: {selector}")
|
|
return False
|
|
|
|
def get_element(self, selector):
|
|
"""
|
|
요소 선택 및 반환
|
|
|
|
Args:
|
|
selector: (By, xpath) 튜플
|
|
|
|
Returns:
|
|
WebElement 또는 None
|
|
"""
|
|
try:
|
|
wait = self.wait()
|
|
element = wait.until(EC.presence_of_element_located(selector))
|
|
logger.info(f"요소 획득 성공: {selector}")
|
|
return element
|
|
except TimeoutException:
|
|
logger.error(f"요소 대기 시간 초과: {selector}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"요소 획득 실패 {type(e).__name__}: {e}")
|
|
return None
|
|
|
|
def take_element_screenshot(self, selector, output_path):
|
|
"""
|
|
요소 스크린샷 저장
|
|
|
|
Args:
|
|
selector: (By, xpath) 튜플
|
|
output_path: 저장 경로
|
|
|
|
Returns:
|
|
성공 여부
|
|
"""
|
|
try:
|
|
element = self.get_element(selector)
|
|
if element is None:
|
|
return False
|
|
|
|
element.screenshot(output_path)
|
|
logger.info(f"스크린샷 저장 성공: {output_path}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"스크린샷 저장 실패 {type(e).__name__}: {e}")
|
|
return False |