384b9aa931
- ChromeDriver 경로를 환경 변수로 설정하여 유연한 드라이버 관리 가능 - CI 및 로컬 환경에서 로그 디렉토리를 자동 생성하도록 수정 - requirements.txt에서 selenium 버전을 명시적으로 지정하여 호환성 강화
484 lines
17 KiB
Python
484 lines
17 KiB
Python
"""
|
|
동행복권(dhlottery.co.kr) 로또 6/45 자동 구매 스크립트
|
|
- 매주 금요일 자동 실행 (cron/스케줄러 연동)
|
|
- 5게임 자동 구매
|
|
- Selenium 기반 웹 자동화
|
|
|
|
사전 준비:
|
|
pip install selenium
|
|
Chrome 브라우저 + ChromeDriver 설치 필요
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import shutil
|
|
import logging
|
|
import tempfile
|
|
import urllib.request
|
|
from datetime import datetime
|
|
|
|
# 로그 디렉토리는 CI/로컬 어디서든 보장
|
|
os.makedirs("logs", exist_ok=True)
|
|
|
|
# .env 파일 자동 로드
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
except ImportError:
|
|
pass
|
|
from selenium import webdriver
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.chrome.service import Service
|
|
from selenium.webdriver.chrome.options import Options
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.webdriver.common.keys import Keys
|
|
|
|
# ─────────────────────────────────────────────
|
|
# 설정
|
|
# ─────────────────────────────────────────────
|
|
CONFIG = {
|
|
"USER_ID": os.environ.get("LOTTO_USER_ID", ""),
|
|
"USER_PW": os.environ.get("LOTTO_USER_PW", ""),
|
|
"BUY_COUNT": 1, # 구매 게임 수 (1~10)
|
|
"HEADLESS": True, # True: 브라우저 창 숨김, False: 브라우저 창 표시
|
|
"LOG_FILE": "logs/lotto_log.json",
|
|
"TELEGRAM_BOT_TOKEN": os.environ.get("TELEGRAM_BOT_TOKEN", ""),
|
|
"TELEGRAM_CHAT_ID": os.environ.get("TELEGRAM_CHAT_ID", ""),
|
|
}
|
|
|
|
# 로깅 설정
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
handlers=[
|
|
logging.StreamHandler(),
|
|
logging.FileHandler("logs/lotto_auto.log", encoding="utf-8"),
|
|
],
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def create_driver():
|
|
"""Chrome WebDriver 생성"""
|
|
options = Options()
|
|
# CI 환경에서 setup-chrome이 설치한 Chrome 바이너리 경로 지정
|
|
chrome_bin = os.environ.get("CHROME_BIN", "")
|
|
if chrome_bin:
|
|
options.binary_location = chrome_bin
|
|
if CONFIG["HEADLESS"]:
|
|
options.add_argument("--headless=new")
|
|
options.add_argument("--no-sandbox")
|
|
options.add_argument("--disable-dev-shm-usage")
|
|
options.add_argument("--disable-gpu")
|
|
options.add_argument("--disable-software-rasterizer")
|
|
# 고정 포트(9222)는 CI/병렬 실행 시 충돌 가능성이 커서 pipe 모드 사용
|
|
options.add_argument("--remote-debugging-pipe")
|
|
# 고정 user-data-dir은 lock 충돌(session not created) 유발 가능 -> 매 실행 고유 경로 사용
|
|
chrome_user_data_dir = os.environ.get("CHROME_USER_DATA_DIR")
|
|
auto_created_profile = False
|
|
if not chrome_user_data_dir:
|
|
chrome_user_data_dir = tempfile.mkdtemp(prefix="chrome-user-data-")
|
|
auto_created_profile = True
|
|
options.add_argument(f"--user-data-dir={chrome_user_data_dir}")
|
|
options.add_argument("--window-size=1920,1080")
|
|
options.add_argument("--no-first-run")
|
|
options.add_argument("--no-default-browser-check")
|
|
options.add_argument("--disable-extensions")
|
|
# 자동화 탐지 우회
|
|
options.add_argument("--disable-blink-features=AutomationControlled")
|
|
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
|
options.add_experimental_option("useAutomationExtension", False)
|
|
|
|
chromedriver_path = os.environ.get("CHROMEDRIVER_PATH", "")
|
|
service = Service(executable_path=chromedriver_path) if chromedriver_path else None
|
|
|
|
driver = webdriver.Chrome(service=service, options=options)
|
|
# 종료 시 정리할 수 있도록 임시 프로필 경로 저장
|
|
driver._chrome_user_data_dir = chrome_user_data_dir
|
|
driver._chrome_user_data_dir_auto_created = auto_created_profile
|
|
driver.execute_cdp_cmd(
|
|
"Page.addScriptToEvaluateOnNewDocument",
|
|
{"source": "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"},
|
|
)
|
|
driver.set_page_load_timeout(30) # 페이지 로드 최대 30초
|
|
driver.set_script_timeout(30) # JS 실행 최대 30초
|
|
driver.implicitly_wait(5)
|
|
return driver
|
|
|
|
|
|
def login(driver):
|
|
"""동행복권 사이트 로그인"""
|
|
from selenium.common.exceptions import TimeoutException
|
|
logger.info("로그인 시도 중...")
|
|
try:
|
|
driver.get("https://www.dhlottery.co.kr/login")
|
|
except TimeoutException:
|
|
logger.warning("로그인 페이지 로드 타임아웃 — 로드된 부분으로 계속 진행")
|
|
time.sleep(2)
|
|
|
|
# 아이디/비밀번호 입력
|
|
wait = WebDriverWait(driver, 10)
|
|
id_input = wait.until(EC.element_to_be_clickable((By.ID, "inpUserId")))
|
|
id_input.click()
|
|
id_input.send_keys(CONFIG["USER_ID"])
|
|
|
|
pw_input = wait.until(EC.element_to_be_clickable((By.ID, "inpUserPswdEncn")))
|
|
pw_input.click()
|
|
pw_input.send_keys(CONFIG["USER_PW"])
|
|
|
|
# 로그인 버튼 클릭
|
|
login_btn = wait.until(EC.element_to_be_clickable((By.ID, "btnLogin")))
|
|
login_btn.click()
|
|
|
|
# /login이 URL에서 완전히 사라질 때까지 대기 (최대 15초)
|
|
try:
|
|
WebDriverWait(driver, 15).until(lambda d: "/login" not in d.current_url)
|
|
except:
|
|
pass
|
|
|
|
logger.info(f"로그인 후 URL: {driver.current_url}")
|
|
|
|
# 로그인 성공 확인 (URL이 /login에서 벗어나면 성공)
|
|
if "/login" in driver.current_url:
|
|
logger.error("로그인 실패! 아이디/비밀번호를 확인하세요.")
|
|
return False
|
|
|
|
logger.info("로그인 성공!")
|
|
return True
|
|
|
|
|
|
def check_balance(driver):
|
|
"""예치금 잔액 확인"""
|
|
driver.get("https://www.dhlottery.co.kr/mypage/home")
|
|
time.sleep(2)
|
|
try:
|
|
balance_elem = driver.find_element(By.CSS_SELECTOR, "#totalAmt")
|
|
balance_text = balance_elem.text.replace(",", "").replace("원", "").strip()
|
|
balance = int(balance_text) if balance_text.isdigit() else 0
|
|
logger.info(f"현재 예치금 잔액: {balance:,}원")
|
|
return balance
|
|
except Exception as e:
|
|
logger.warning(f"잔액 확인 실패: {e}")
|
|
return -1
|
|
|
|
|
|
def buy_lotto(driver, game_count=5):
|
|
"""
|
|
로또 6/45 자동 구매
|
|
game_count: 구매할 게임 수 (1~10)
|
|
"""
|
|
logger.info(f"로또 {game_count}게임 자동 구매 시작...")
|
|
|
|
# 로또 구매 페이지 이동 (타임아웃 시 로드된 부분으로 계속 진행)
|
|
from selenium.common.exceptions import TimeoutException
|
|
try:
|
|
driver.get("https://ol.dhlottery.co.kr/olotto/game/game645.do")
|
|
except TimeoutException:
|
|
logger.warning("페이지 로드 타임아웃 — 로드된 부분으로 계속 진행")
|
|
time.sleep(2)
|
|
|
|
# iframe 전환 (로또 구매 페이지는 iframe 내부에 있음)
|
|
try:
|
|
wait = WebDriverWait(driver, 10)
|
|
|
|
def close_layer_alert_if_present(timeout_s: float = 1.5) -> bool:
|
|
"""
|
|
<div class="layer-alert" id="popupLayerAlert"> 가 떠 있으면
|
|
내부 '확인' 버튼을 클릭(= closepopupLayerAlert())해서 닫는다.
|
|
"""
|
|
from selenium.common.exceptions import TimeoutException, WebDriverException
|
|
|
|
try:
|
|
layer = WebDriverWait(driver, timeout_s).until(
|
|
EC.presence_of_element_located((By.ID, "popupLayerAlert"))
|
|
)
|
|
except TimeoutException:
|
|
return False
|
|
|
|
try:
|
|
if not layer.is_displayed():
|
|
return False
|
|
except WebDriverException:
|
|
# stale 등 예외면 "없음"으로 간주
|
|
return False
|
|
|
|
# 우선 버튼 클릭 시도 (DOM: #popupLayerAlert .btns input[value='확인'])
|
|
try:
|
|
btn = driver.find_element(
|
|
By.CSS_SELECTOR,
|
|
"#popupLayerAlert .btns input[type='button'][value='확인']",
|
|
)
|
|
if btn.is_displayed() and btn.is_enabled():
|
|
btn.click()
|
|
time.sleep(0.3)
|
|
return True
|
|
except Exception:
|
|
pass
|
|
|
|
# fallback: 사이트의 close 함수 직접 호출
|
|
try:
|
|
driver.execute_script(
|
|
"if (typeof closepopupLayerAlert === 'function') { closepopupLayerAlert(); }"
|
|
)
|
|
time.sleep(0.3)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
# 모든 레이어 팝업을 JS로 직접 숨기기 (z-index 오버레이 완전 제거)
|
|
driver.execute_script("""
|
|
var popups = document.querySelectorAll(
|
|
'#popupLayerAlert, #popupLayerConfirm, .layer-alert, .layer-popup, .pop_wrap'
|
|
);
|
|
popups.forEach(function(el) { el.style.display = 'none'; });
|
|
""")
|
|
time.sleep(0.5)
|
|
logger.info("팝업 레이어 숨김 처리 완료")
|
|
|
|
# "자동" 탭 선택 — onclick 함수(selectWayTab)를 직접 호출
|
|
driver.execute_script("selectWayTab(1);")
|
|
time.sleep(1)
|
|
time.sleep(1)
|
|
|
|
# 게임 수 설정 (드롭다운에서 선택)
|
|
from selenium.webdriver.support.ui import Select
|
|
try:
|
|
select_elem = driver.find_element(By.ID, "amoundApply")
|
|
select = Select(select_elem)
|
|
select.select_by_value(str(game_count))
|
|
time.sleep(0.5)
|
|
except Exception:
|
|
# 드롭다운이 없는 경우 자동번호 버튼을 game_count번 클릭
|
|
for i in range(game_count):
|
|
auto_select_btn = driver.find_element(By.ID, "btnSelectNum")
|
|
auto_select_btn.click()
|
|
time.sleep(0.5)
|
|
|
|
# "자동번호 선택" 버튼 클릭
|
|
try:
|
|
select_num_btn = driver.find_element(By.ID, "btnSelectNum")
|
|
select_num_btn.click()
|
|
time.sleep(1)
|
|
except:
|
|
pass
|
|
|
|
# "구매하기" 버튼 클릭
|
|
# 구매 직전에 레이어 알림이 뜨는 경우 먼저 닫기
|
|
if close_layer_alert_if_present():
|
|
logger.info("구매 전 popupLayerAlert 감지 — 닫기 처리 완료")
|
|
|
|
buy_btn = wait.until(
|
|
EC.element_to_be_clickable((By.ID, "btnBuy"))
|
|
)
|
|
buy_btn.click()
|
|
time.sleep(2)
|
|
|
|
# 구매 확인 팝업 "확인" 클릭
|
|
confirm_btn = wait.until(
|
|
EC.element_to_be_clickable((By.CSS_SELECTOR, "#popupLayerConfirm input[value='확인']"))
|
|
)
|
|
confirm_btn.click()
|
|
time.sleep(3)
|
|
|
|
logger.info("구매 요청 완료! 구매 결과를 확인합니다...")
|
|
|
|
# 구매 결과 확인
|
|
result = extract_purchase_result(driver)
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"구매 중 오류 발생: {e}")
|
|
return None
|
|
|
|
|
|
def extract_purchase_result(driver):
|
|
"""구매 결과(번호) 추출"""
|
|
result = {
|
|
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"games": [],
|
|
"status": "unknown",
|
|
}
|
|
|
|
try:
|
|
time.sleep(2)
|
|
|
|
# 회차 / 추첨일 추출
|
|
try:
|
|
result["round"] = driver.find_element(By.ID, "buyRound").text
|
|
result["draw_date"] = driver.find_element(By.ID, "drawDate").text
|
|
except:
|
|
pass
|
|
|
|
# 번호 추출: #reportRow li div.nums span
|
|
game_rows = driver.find_elements(By.CSS_SELECTOR, "#reportRow li")
|
|
for row in game_rows:
|
|
nums = row.find_elements(By.CSS_SELECTOR, "div.nums span")
|
|
numbers = [int(n.text) for n in nums if n.text.strip().isdigit()]
|
|
if numbers:
|
|
result["games"].append(numbers)
|
|
|
|
if result["games"]:
|
|
result["status"] = "success"
|
|
logger.info(f"구매 성공! {len(result['games'])}게임")
|
|
for i, game in enumerate(result["games"], 1):
|
|
logger.info(f" 게임 {i}: {game}")
|
|
else:
|
|
result["status"] = "completed_but_numbers_unknown"
|
|
logger.info("번호 추출 실패. 마이페이지에서 확인하세요.")
|
|
|
|
except Exception as e:
|
|
result["status"] = "error"
|
|
result["error"] = str(e)
|
|
logger.warning(f"결과 추출 중 오류: {e}")
|
|
|
|
return result
|
|
|
|
|
|
def send_telegram(result):
|
|
"""Telegram Bot으로 구매 결과 알림 전송"""
|
|
token = CONFIG["TELEGRAM_BOT_TOKEN"]
|
|
chat_id = CONFIG["TELEGRAM_CHAT_ID"]
|
|
if not token or not chat_id:
|
|
return
|
|
|
|
status = result.get("status", "unknown")
|
|
games = result.get("games", [])
|
|
round_no = result.get("round", "")
|
|
draw_date = result.get("draw_date", "")
|
|
|
|
if status == "success" and games:
|
|
games_text = "\n".join(
|
|
f" 게임 {i}: {' '.join(str(n) for n in nums)}"
|
|
for i, nums in enumerate(games, 1)
|
|
)
|
|
text = (
|
|
f"🎟 *로또 구매 완료\!*\n"
|
|
f"회차: {round_no} 추첨일: {draw_date}\n"
|
|
f"{games_text}"
|
|
)
|
|
elif status == "failed":
|
|
text = f"⚠️ *로또 구매 실패*\n{result.get('error', '')}"
|
|
else:
|
|
text = f"❌ *로또 구매 오류*\n{result.get('error', status)}"
|
|
|
|
try:
|
|
payload = json.dumps({
|
|
"chat_id": int(chat_id),
|
|
"text": text,
|
|
"parse_mode": "MarkdownV2",
|
|
}).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
f"https://api.telegram.org/bot{token}/sendMessage",
|
|
data=payload,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
urllib.request.urlopen(req, timeout=10)
|
|
logger.info("Telegram 알림 전송 완료")
|
|
except Exception as e:
|
|
logger.warning(f"Telegram 알림 전송 실패: {e}")
|
|
|
|
|
|
def save_log(result):
|
|
"""구매 기록 저장"""
|
|
log_file = CONFIG["LOG_FILE"]
|
|
history = []
|
|
|
|
if os.path.exists(log_file):
|
|
try:
|
|
with open(log_file, "r", encoding="utf-8") as f:
|
|
history = json.load(f)
|
|
except:
|
|
history = []
|
|
|
|
history.append(result)
|
|
|
|
with open(log_file, "w", encoding="utf-8") as f:
|
|
json.dump(history, f, ensure_ascii=False, indent=2)
|
|
|
|
logger.info(f"구매 기록이 {log_file}에 저장되었습니다.")
|
|
|
|
|
|
def main():
|
|
"""메인 실행 함수"""
|
|
logger.info("=" * 50)
|
|
logger.info("로또 6/45 자동 구매 스크립트 시작")
|
|
logger.info(f"실행 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
logger.info("=" * 50)
|
|
|
|
# 아이디/비밀번호 확인
|
|
if CONFIG["USER_ID"] == "여기에_아이디_입력" or CONFIG["USER_PW"] == "여기에_비밀번호_입력":
|
|
logger.error("아이디/비밀번호가 설정되지 않았습니다!")
|
|
logger.error("환경변수(LOTTO_USER_ID, LOTTO_USER_PW)를 설정하거나 스크립트 내 CONFIG를 수정하세요.")
|
|
sys.exit(1)
|
|
|
|
driver = None
|
|
try:
|
|
driver = create_driver()
|
|
|
|
# 1. 로그인
|
|
if not login(driver):
|
|
logger.error("로그인 실패로 종료합니다.")
|
|
sys.exit(1)
|
|
|
|
# 2. 잔액 확인
|
|
balance = check_balance(driver)
|
|
cost = CONFIG["BUY_COUNT"] * 1000
|
|
if balance >= 0 and balance < cost:
|
|
logger.error(f"예치금 부족! 현재: {balance:,}원, 필요: {cost:,}원")
|
|
logger.error("동행복권 사이트에서 예치금을 충전해주세요.")
|
|
send_telegram({
|
|
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"status": "failed",
|
|
"error": f"예치금 부족! 현재: {balance:,}원, 필요: {cost:,}원",
|
|
})
|
|
sys.exit(1)
|
|
|
|
# 3. 로또 구매
|
|
result = buy_lotto(driver, CONFIG["BUY_COUNT"])
|
|
|
|
# 4. 결과 저장 및 Slack 알림
|
|
if result:
|
|
save_log(result)
|
|
send_telegram(result)
|
|
else:
|
|
failed = {
|
|
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"status": "failed",
|
|
"error": "구매 결과를 확인할 수 없습니다.",
|
|
}
|
|
save_log(failed)
|
|
send_telegram(failed)
|
|
|
|
logger.info("스크립트 실행 완료!")
|
|
|
|
except Exception as e:
|
|
logger.error(f"예상치 못한 오류: {e}")
|
|
error_result = {
|
|
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"status": "error",
|
|
"error": str(e),
|
|
}
|
|
save_log(error_result)
|
|
send_telegram(error_result)
|
|
|
|
finally:
|
|
if driver:
|
|
driver.quit()
|
|
logger.info("브라우저 종료")
|
|
# 자동 생성한 Chrome profile 디렉토리 정리
|
|
try:
|
|
if getattr(driver, "_chrome_user_data_dir_auto_created", False):
|
|
profile_dir = getattr(driver, "_chrome_user_data_dir", "")
|
|
if profile_dir and os.path.isdir(profile_dir):
|
|
shutil.rmtree(profile_dir, ignore_errors=True)
|
|
logger.info(f"임시 Chrome 프로필 정리: {profile_dir}")
|
|
except Exception as e:
|
|
logger.warning(f"임시 Chrome 프로필 정리 실패: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|