.env를 crontab에서 인식하지 못하는 문제 수정
This commit is contained in:
178
app/selenium_manager.py
Normal file
178
app/selenium_manager.py
Normal file
@ -0,0 +1,178 @@
|
||||
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
|
||||
Reference in New Issue
Block a user