[Python] 날씨를 알려주는 나만의 SlackBot - with Github Actions

JaehyoJJAng2023년 08월 19일

개발 환경 설정

requirements.txt

slack_sdk
django-environ
httpx
selectolax
playwright

Slack API 등록

a. https://api.slack.com/apps?new_classic_app=1 사이트 접속

b. 앱 등록하기

c. 슬랙 Bot 등록하기

d. Bot에 권한 등록하기

e. 채널에 weather_bot 초대하기

s

날씨 정보 추출하기

네이버 날씨

채팅에 답변하는 슬랙봇

a. 코드 작성

# post_message.py
from slack_sdk.rtm_v2 import RTMClient

class PostMessage:
    def __init__(self,token: str) -> None:
        self._token = token
        
    def start(self) -> None:
        rtm : RTMClient = RTMClient(token=self._token)
        
        @rtm.on('message')
        def handle(client: RTMClient, event: dict) -> None:
            if 'Hello' in event['text']:
                channel_id = event['channel']
                thread_ts = event['ts']
                user = event['user'] # This is not username but user ID (the format is either U*** or W***)

                client.web_client.chat_postMessage(
                    channel=channel_id,
                    text=f"Hi <@{user}>!",
                    thread_ts=thread_ts
                )
        rtm.start()

# ========================================
# get_token.py
import environ
def get_token(env_file: str='.env') -> environ.Env:
    env : environ.Env = environ.Env(DEBUG=(bool,False))
    env.read_env(env_file=env_file)
    return env

b. 실행

$ python3 app.py

httpx로 Request 보내기

import httpx

url : str = f'https://search.naver.com/search.naver?sm=tab_hty.top&where=nexearch&query={rep.quote_plus("강남 날씨")}&oquery={rep.quote_plus("강남 날씨")}&tqi=i7fWVlp0JXVssuwTYNGssssss8s-483683'

resp : httpx.Response = httpx.get(url=url)

selectolax 로 Response 추출하기

응답 데이터 (response) 를 파싱하여 필요한 데이터만 추출해보기

from selectolax.parser import HTMLParser
import httpx

url : str = f'https://search.naver.com/search.naver?sm=tab_hty.top&where=nexearch&query={rep.quote_plus("강남 날씨")}&oquery={rep.quote_plus("강남 날씨")}&tqi=i7fWVlp0JXVssuwTYNGssssss8s-483683'
resp : httpx.Response = httpx.get(url=url)

html = HTMLParser(resp.text)

# 지역명 추출
area = html.css_first('h2.title').text()

# 현재 기온 추출 - .text(deep=False) : 해당 태그 하위에 존재하는 속성 추적 x
today_temperature : float = float(html.css_first('div._today div.temperature_text > strong').text(deep=False))

# 최저,최고 기온 추출 - html.css : 태그를 리스트로 반환
high_temperature : int = int(html.css('span.temperature_inner > span.highest')[0].text(deep=False))
low_temperature  : int = int(html.css('span.temperature_inner > span.lowest')[0].text(deep=False))

채널에 날씨 정보 보내기

from slack_sdk.rtm_v2 import RTMClient
from typing import Dict,Union
import urllib.parse as rep

"""
1. Slack 채팅의 마지막 두 글자가 "날씨"로 끝나는지 확인
2. httpx 모듈을 사용하여 네이버에서 Slack 채팅글 검색
3. 요청에 대한 응답에서 필요한 데이터 추출
4. 추출한 데이터를 슬랙에 표시
"""

class PostMessage:
    def __init__(self,token: str,WeatherParser) -> None:
        self._token = token
        self.weather : WeatherParser = WeatherParser()
        
    def start(self) -> None:
        rtm : RTMClient = RTMClient(token=self._token)
        
        @rtm.on('message')
        def handle(client: RTMClient, event: dict) -> None:
            keyword : str = event['text']
            if keyword.endswith("날씨"):                
                # Get Response
                response = self.weather.get_response(keyword=keyword)
                print(response)
                
                # Get weather data
                weather_dict : Dict[str,Union[float,int,str]] = self.weather.parsing(response=response)

                # 메시지 보내는 부분
                channel_id = event['channel'] # 채널 ID
                thread_ts = event['ts'] # 댓글 다는 부분

                client.web_client.chat_postMessage(
                    channel=channel_id,
                    text=f"지역명: {weather_dict['area']}\n현재기온: {weather_dict['today_temperature']}\n최고기온: {weather_dict['high_temperature']}\n최저기온: {weather_dict['low_temperature']}\n기상상태: {weather_dict['blind']}",
                    thread_ts=thread_ts
                )
        rtm.start()

날씨 정보 이미지 파일로 저장

날씨 정보를 이미지 파일로 저장하여 슬랙 채널에 전송
playwright 사용

a. playwright 설치

$ pip install playwright
$ playwright install

b. 코드 작성하기

from selectolax.parser import HTMLParser
from typing import Dict,Union
from playwright.sync_api import sync_playwright
import urllib.parse as rep
import httpx
import re
import os

class WeatherParser:        
    def get_weather_info(self,keyword: str) -> Dict[str,Union[float,int,str]]:
        # 데이터 저장할 딕셔너리 변수 선언
        weather_dict : Dict[str,Union[float,int,str]] = dict()
        
        # Set URL
        url : str = f'https://search.naver.com/search.naver?sm=tab_hty.top&where=nexearch&query={rep.quote_plus(keyword)}&oquery={rep.quote_plus(keyword)}&tqi=i7fWVlp0JXVssuwTYNGssssss8s-483683'
        
        # Get Response
        response : httpx.Response = httpx.get(url=url)
        
        # 파서
        html : HTMLParser = HTMLParser(response.text)
        
        # 지역명        
        area = html.css_first('h2.title').text()
        
        # 현재 기온 추출
        today_temperature : float = float(re.sub('[^0-9.]','',html.css_first('div._today div.temperature_text > strong').text()))
        
        # 최저 , 최고 기온
        high_temperature : int = int(re.sub('[^0-9]','',html.css_first('span.temperature_inner > span.highest').text()))
        low_temperature : int = int(re.sub('[^0-9]','',html.css_first('span.temperature_inner > span.lowest').text()))
        
        # 기상 상태
        blind : str = html.css_first('div._today i.wt_icon > span').text()
        
        # 데이터 저장
        weather_dict['area'] = area
        weather_dict['today_temperature'] = today_temperature
        weather_dict['high_temperature'] = high_temperature
        weather_dict['low_temperature'] = low_temperature
        weather_dict['blind'] = blind
        
        print(weather_dict)
        
        return weather_dict

    def get_screenshot(self,keyword: str) -> None:
        # 폴더 지정
        dir : str = 'screenshot/'        
        if not os.path.exists(dir):
            os.mkdir(dir)

        with sync_playwright() as playwright:
            url : str = f'https://search.naver.com/search.naver?sm=tab_hty.top&where=nexearch&query={rep.quote_plus(keyword)}&oquery={rep.quote_plus(keyword)}&tqi=i7fWVlp0JXVssuwTYNGssssss8s-483683'        
            
            # 브라우저 객체
            browser = playwright.chromium.launch(channel='chrome')
                        
            # 브라우저 오픈
            page = browser.new_page(viewport={'width': 1980,'height': 2000})
            
            # 페이지 이동
            page.goto(url=url)
            
            # 스크린샷
            weather_info = page.locator('div.content_wrap')
            weather_info.screenshot(path=os.path.join(dir,'info.png'))
            
            # 브라우저 종료
            browser.close()

# ======================================================================
# post_message.py
class PostMessage:
    def __init__(self,token: str,WeatherParser) -> None:
        self._token = token
        self.weather : WeatherParser = WeatherParser()
        
    def start(self) -> None:
        rtm : RTMClient = RTMClient(token=self._token)
        web_client = WebClient(token=self._token)
        
        @rtm.on('message')
        def handle(client: RTMClient, event: dict) -> None:
            keyword : str = event['text']
            if keyword.endswith("날씨"):
                # Get weather data
                weather_dict : Dict[str,Union[float,int,str]] = self.weather.get_weather_info(keyword=keyword)

                # 메시지 보내는 부분
                channel_id = event['channel'] # 채널 ID
                thread_ts = event['ts'] # 댓글 다는 부분

                client.web_client.chat_postMessage(
                    channel=channel_id,
                    text=f"지역명: {weather_dict['area']}\n현재기온: {weather_dict['today_temperature']}\n최고기온: {weather_dict['high_temperature']}\n최저기온: {weather_dict['low_temperature']}\n기상상태: {weather_dict['blind']}",
                    thread_ts=thread_ts
                )
                
                # 날씨 정보 스크린샷 저장
                self.weather.get_screenshot(keyword=keyword)
                
                # 파일 전송하기
                web_client.files_upload_v2(
                    channel=channel_id,
                    file=os.path.join('screenshot','info.png'),
                    title="날씨 정보",
                    thread_ts = thread_ts
                )
        rtm.start()

c. 실행 결과

d. 파일을 슬랙에 전송하기 위해 권한 추가로 부여하기

작업 자동화

서버 환경에서 스크립트를 특정 시간에 자동으로 실행하도록 크론잡 또는 Github Actions를 이용한 작업 자동화 도입

코드를 아래와 같이 수정

# post_message.py
from slack_sdk.rtm_v2 import RTMClient
from slack_sdk import WebClient
from typing import Dict,Union
import urllib.parse as rep
import os

"""
1. Slack 채팅의 마지막 두 글자가 "날씨"로 끝나는지 확인
2. httpx 모듈을 사용하여 네이버에서 Slack 채팅글 검색
3. 요청에 대한 응답에서 필요한 데이터 추출
4. 추출한 데이터를 슬랙에 표시
"""

class PostMessage:
    def __init__(self,token: str,WeatherParser) -> None:
        self._token = token
        self.weather : WeatherParser = WeatherParser()
        
    def start(self) -> None:
        rtm : RTMClient = RTMClient(token=self._token)
        web_client = WebClient(token=self._token)

        # Get weather data
        weather_dict : Dict[str,Union[float,int,str]] = self.weather.get_weather_info(keyword='병점 날씨')

        # 메시지 보내는 부분
        channel_id = 'C05GV24JJ0P' # 채널 ID

        rtm.web_client.chat_postMessage(
            channel=channel_id,
            blocks=[
                {'type': 'divider'},
                {
                    'type': 'section',
                    'text': {
                        'type': 'plain_text',
                        'text': f"{weather_dict['area']}"
                    }
                },
                {'type': 'divider'},
                {
                    'type': 'section',
                    'text': {
                        'type': 'plain_text',
                        'text': f"현재기온: {weather_dict['today_temperature']}\n최고기온: {weather_dict['high_temperature']}\n최저기온: {weather_dict['low_temperature']}\n기상상태: {weather_dict['blind']}"
                    }
                }
            ],
        )
        
        # 날씨 정보 스크린샷 저장
        self.weather.get_screenshot(keyword='병점 날씨')
        
        # 파일 전송하기
        web_client.files_upload_v2(
            channel=channel_id,
            file=os.path.join('screenshot','info.png'),
            title="날씨 정보",
        )

Crontab

리눅스에서 매일 아침 7시에 스크립트를 실행할 수 있도록 크론잡 설정
위 소스 코드가 모두 서버 환경에 존재하여야 함
date 커맨드를 사용하여 한국 시간대를 사용하고 있는지 확인
소스코드 경로 및 가상환경 경로를 숙지

a. CronTab - Job 등록하기

$ crontab -e

...(생략)...
* * * * * /home/docker/Slack-Python/python-venv/bin/python /home/docker/Slack-Python/Inflearn-Slack-Bot/main.py

b. 서버에 크롬 설치

# 1. 패키지 인덱스 업데이트
$ sudo apt-get update

# 2. wget 설치
$ sudo apt-get install -y wget

# 3. wget 을 사용하여 크롬 패키지 다운로드
$ wget https://dl.google.com/linux/direct/google-chrome-stable_current_amd64.deb

# 4. 다운로드한 크롬 패키지 설치하기
$ sudo dpkg -i google-chrome-stable_current_amd64.deb

Github Actions

a. 워크플로우 작성하기

name: "날씨 슬랙 알림 자동화"

on:
  workflow_dispatch:
  schedule:
    - cron: '0 22 * * *' # 매일 아침 7시에 Job 실행 (깃헙 서버는 UTC 기준이므로 한국 시간 7시에 실행 되지는 않음)

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - name: "1. 소스코드 우분투에 복사"
        uses: actions/checkout@v2

      - name: "2. Ubuntu에 파이썬 개발환경 구축"
        uses: actions/setup-python@v2
        with:
          python-version: "3.10"

      - name: "3. Install dependencies"
        run: |
          if [[ -f requirements/requirements.txt ]]
          then
            pip install -r requirements/requirements.txt
          else
            echo "requirements.txt does not exist"
            exit 1
          fi

      - name: "4. main.py 스크립트 실행"
        env:
          SLACK_BOT_TOKEN: {% raw %} ${{ secrets.SLACK_BOT_TOKEN }} {% endraw %}
        run: |
          python main.py

b. SLACK_BOT_TOKEN 환경변수를 Github Token에서 관리하기

실행 결과

python