환경 : 라즈베리파이(라즈비안 써서 CronJob으로 매시간 돌리지만 소스는 OS안가림 셀레니움 드라이버 패스 제외)
언어 : 파이썬3
라이브러리 : Selenium(Chrome)
DB : sqlite3
DB구성
본격적으로 체크할 상품 정보를 적을 테이블
ITEM_ID ID로 자동상승(AI)를 킨건 좀 오버긴 한데 없으면 그것되로 불편
SITE_TYEP SITE테이블과 연동할 필드
NAME 상품 이름 메일 출력용으로 씀
OPTION1,2,3 URL의 KEY가 될 값, 3은 안쓰지만 일단 넣어둠
DEL 0이 아니면 더 이상 트레킹 안할 려고 추가
ITEM 테이블과 연동해서 URL 작성용으로 만든 테이블
SITE_TYPE ITEM과 연동용으로 ID용으로 씀
URL 사이트 URL {0}~{2}까지 지원해서 URL 생성용
URL_TYPE MS사이트가 KEY가 2개라서 급하게 추가
SITE_MEMO 진짜 메모용... 실 소스에 전혀 안씀
ITEM 테이블 실제로 들어간 값
SITE테이블 실제 값
파이썬 소스
제 개인 정보 나올 수 있는것만 지우고 올립니다
from selenium import webdriver from selenium.webdriver.chrome.options import Options import smtplib, ssl import email.utils from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText import sqlite3 #import logging #logging.basicConfig(level=logging.DEBUG, filename="test.log", format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p') options = Options() options.add_argument('--headless') options.add_argument('--disable-gpu') options.add_argument('--no-sandbox') chrome_prefs = {} options.experimental_options["prefs"] = chrome_prefs chrome_prefs["profile.default_content_settings"] = {"images": 2} chrome_prefs["profile.managed_default_content_settings"] = {"images": 2} driver = webdriver.Chrome(executable_path="/usr/lib/chromium-browser/chromedriver", chrome_options=options) dbPath = "./" dbFileName = "PriceChk.sqlite3" conn = sqlite3.connect(dbPath + dbFileName) cur = conn.cursor() dataList = cur.execute("select B.ITEM_ID, B.NAME, A.SITE_TYPE, A.URL, A.URL_TYPE, B.OPTION1, B.OPTION2, B.OPTION3 from SITE A, ITEM B where B.DEL=0 AND A.SITE_TYPE=B.SITE_TYPE").fetchall() for data in dataList: url = "" if data[4] == 1: url = data[3].format(data[5]) elif data[4] == 2: url = data[3].format(data[5], data[6]) elif data[4] == 3: url = data[3].format(data[5], data[6], data[7]) #print(url) driver.get(url) newPrice = "" if data[2].startswith("MS"): if data[2] == "MSM2": buyText = "Buy " for btn in driver.find_elements_by_xpath("//button[@data-focus-rank='600']"): tempValue = btn.get_attribute("aria-label") if tempValue.startswith(buyText): newPrice = tempValue[len(buyText):] break else: priceValue = driver.find_element_by_xpath("//meta[@itemprop='price']") newPrice = priceValue.get_attribute("content") elif data[2].startswith("AM"): priceValue = driver.find_element_by_xpath("//-nput[@name='displayedPrice']") newPrice = priceValue.get_attribute("value") oldPriceSelect = cur.execute("select ITEM_ID, PRICE, DATE from DATA where ITEM_ID = ? order by DATE desc limit 1", (str(data[0]),)).fetchone() oldPrice = "" if not oldPriceSelect is None: oldPrice = oldPriceSelect[1] if oldPrice != newPrice: cur.execute("insert into DATA ( ITEM_ID, PRICE, DATE ) VALUES (?, ?, datetime('now', 'localtime'))", (data[0], newPrice)) fromaddr = <<보내는 메일 주소>> toaddr = <<받는 메일 주소>> msgBody = "{0}의 가격 변경\n{1}→{2}\n{3}".format(data[1], oldPrice, newPrice, url) msg = MIMEMultipart() msg['From'] = email.utils.formataddr(('가격봇', fromaddr)) msg['To'] = email.utils.formataddr(('너님', toaddr)) msg['Subject'] = "{0}의 가격 변경".format(data[1]) msg.attach(MIMEText(msgBody)) id = fromaddr password = <<메일 계정 패스워드>> context = ssl.create_default_context() with smtplib.SMTP(<<매일SMTP서버>>, <<포트번호>>) as server: server.ehlo() # Can be omitted server.starttls(context=context) server.ehlo() # Can be omitted server.login(id, password) server.sendmail(fromaddr, toaddr, msg.as_string()) conn.commit() conn.close() driver.close() driver.quit() |