Language/Python
selenium으로 크롤링하고 Slack에 메시지 전송
건담아빠
2024. 11. 24. 16:57
시에서 운영하는 캠핑장들은 빈자리가 생겨서 알림을 받을수 없는것 같다, 수시로 새로고침해서 예약을 하다 보니 겁나 귀찮다.
캠핑은 좋고 예약은 힘들고 알림이라도 받아서 하는게 어떨까? 그냥 크롤링으로 알림 받는걸 만들어 놓자.
하지만 사용하진 않는다, 배치형태로 돌려야 할 것 같은데 서버를 어떤식으로 할지도 고민해봐야 할듯해서 일단은 대충 만들어 놓고 나중에 진짜로 사용할때 좀더 고급스럽게 커스터 마이징해서 사용하도록 하자.
참고로 python에서 slack에 메시지 보내는 방법은 이전 포스팅에 작성되어 있다.
1. 요구사항
- `금`,`토` (주말) 및 `공휴일`에 자리가 있다면 데이터를 추출
- 빈자리가 있을때 알림이 오게하는게 1차 목표임으로 데이터를 가공하지는 않는다.
- 추출한 데이터를 json형태로 Slack에 전송하자
2. 사용기술
- Python
- Selenium
- Slack
3. 대상사이트
4. 소스코드
4.1. samping.py
datas = api.get_data()에서 A, B사이트에서 남은 사이트 갯수를 가져오고 filtered_data1은 '금', '토', '일', filtered_data2는 include_days = [10, 30] 에서 추가된 공휴일을 가지고 빈 사이트가 있는지 확인 할 수 있다.
from apps.module import api
from apps.module import slack
import json
def main():
datas = api.get_data()
# print("datas : ", datas)
include_days = [10, 30]
# 첫 번째 조건: STATUS가 'ING'이고 week이 '금', '토', '일'이며 'A' 또는 'B'가 0보다 큰 경우
filtered_data1 = [item for item in datas if item.get('STATUS') == 'ING' and item.get('week') in ['금', '토', '일'] and (int(item.get('A', 0)) > 0 or int(item.get('B', 0)) > 0)]
# 두 번째 조건: STATUS가 'ING'이고 day가 10, 20, 22이며 'A' 또는 'B'가 0보다 큰 경우
filtered_data2 = [item for item in datas if item.get('STATUS') == 'ING' and item.get('day') in include_days and (int(item.get('A', 0)) > 0 or int(item.get('B', 0)) > 0)]
if (filtered_data1 is not None and len(filtered_data1) > 0) or (filtered_data2 is not None and len(filtered_data2) > 0):
print("있다.")
print("filtered_data1 : ", filtered_data1)
print("filtered_data2 : ", filtered_data2)
# Slack 메시지 전송
if filtered_data1:
filtered_data1_json = json.dumps(filtered_data1, ensure_ascii=False, indent=2)
slack.send_message(filtered_data1_json)
if filtered_data2:
filtered_data2_json = json.dumps(filtered_data2, ensure_ascii=False, indent=2)
slack.send_message(filtered_data2_json)
if __name__ == '__main__':
main()
4.2. api.py
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import time
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
def get_week_days():
import calendar
from datetime import datetime
# 현재 날짜와 시간 가져오기
now = datetime.now()
# 현재 년도와 월 가져오기
year = now.year
month = now.month
month = now.month
# 이번 달의 첫 날의 요일과 날짜 수 계산
first_day_of_month = datetime(year, month, 1)
first_weekday = first_day_of_month.weekday()
num_days_in_month = calendar.monthrange(year, month)[1]
# 요일 이름 설정 (0: 월요일, 1: 화요일, ..., 6: 일요일)
weekdays = ['월', '화', '수', '목', '금', '토', '일']
week_days = [{'day': 0, 'week': '일'}]
# 시작 요일 이전의 공백 출력
for i in range(first_weekday):
week_days.append({'day': 0, 'week': weekdays[i]})
# 날짜와 요일 출력
for day in range(1, num_days_in_month + 1):
week_days.append({'day': day, 'week': weekdays[(first_weekday + day - 1) % 7]})
# 줄 바꿈 조건 설정
if (first_weekday + day) % 7 == 0:
print()
return week_days
def get_data():
week_days = get_week_days()
# 크롬 옵션 설정
chrome_options = Options()
chrome_options.add_argument("--headless") # headless 모드 설정
# 크롬 드라이버 실행
# driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))
driver = webdriver.Chrome(service=Service())
url = "https://reserve.gmuc.co.kr/mobile/camp/campReservation.do?menu=d&menuFlag=C"
driver.get(url)
# # time.sleep(5)
tr_elements = driver.find_elements(By.CSS_SELECTOR, '.trBefore')
date_count = 0
for tr_element in tr_elements:
tds = tr_element.find_elements(By.TAG_NAME, 'td')
for td in tds:
if len(week_days) > date_count:
# td 요소 안에 있는 모든 하위 요소들 찾기
child_elements = td.find_elements(By.XPATH, ".//*")
# 하위 요소들이 존재하는지 확인
if len(child_elements) > 0:
# td 요소 안에 class가 'done'인 요소 찾기
done_elements = td.find_elements(By.CLASS_NAME, 'done')
if len(done_elements) > 0:
week_days[date_count]['STATUS'] = 'DONE'
else:
# week_days[date_count]['test'] = td.text
circle1 = td.find_element(By.CSS_SELECTOR, '.circle1 > a')
circle2 = td.find_element(By.CSS_SELECTOR, '.circle2 > a')
week_days[date_count]['STATUS'] = 'ING'
week_days[date_count]['A'] = circle1.text
week_days[date_count]['B'] = circle2.text
else:
week_days[date_count]['STATUS'] = 'XXX'
date_count += 1
# WebDriver 종료
driver.quit()
return week_days
if __name__ == '__main__':
datas = get_data()
include_days = [10, 30]
# 첫 번째 조건: STATUS가 'ING'이고 week이 '금', '토', '일'이며 'A' 또는 'B'가 0보다 큰 경우
filtered_data1 = [item for item in datas if item.get('STATUS') == 'ING' and item.get('week') in ['금', '토', '일'] and (int(item.get('A', 0)) > 0 or int(item.get('B', 0)) > 0)]
# 두 번째 조건: STATUS가 'ING'이고 day가 10, 20, 22이며 'A' 또는 'B'가 0보다 큰 경우
filtered_data2 = [item for item in datas if item.get('STATUS') == 'ING' and item.get('day') in include_days and (int(item.get('A', 0)) > 0 or int(item.get('B', 0)) > 0)]
if (filtered_data1 is not None and len(filtered_data1) > 0) or (filtered_data2 is not None and len(filtered_data2) > 0):
print("있다.")
print("filtered_data1 : ", filtered_data1)
print("filtered_data2 : ", filtered_data2)
4.3. slack.py
import slack_sdk
def send_message(message: str):
bot_user_oauth_token = 'Bot User OAuth Token'
client = slack_sdk.WebClient(token=bot_user_oauth_token)
# print("message : ", message)
# 일반 메시지 전송
# slack_msg = f'테스트 메시지 전송'
response = client.chat_postMessage(
channel="private-캠핑-알림",
text=message
)
# 맨션으로 전송
user_id = "dchkang83"
slack_msg = f'<@{user_id}> ' + message
response = client.chat_postMessage(
channel="private-캠핑-알림",
text=slack_msg
)
if __name__ == '__main__':
send_message()
5. 결과
나중에 필요하게 되면 좀더 다듬어서 사용하자!
수고 짝짝짝!