Celery는 async/비동기적으로 특정한 작업을 돌리기 위해 자주 사용한다. 특히, django와는 찰떡궁합이라고 알려져있다. 하지만 이 celery는 설정이 어렵다면 어렵고, 쉽다면 쉬운편이다.
먼저 Celery를 “쓰려면” 어떤 것들이 필요한지 체크해보자.
$ pip install celery
$ brew install rabbitmq
.bashrc
나 .zshrc
의 마지막 줄에 아래 코드를 추가해준다.export PATH=$PATH:/usr/local/sbin
이로서 rabbitmq-server라는 명령어로 rabbitmq를 실행할 수 있다.
아래와 같이 코드를 작성해 보자. (celery_parser.py)
# celery_parser.py
from celery import Celery
# Celery Setup
app = Celery()
app.conf.timezone = 'Asia/Seoul'
@app.on_after_configure.connect
def periodic_parser(sender, **kwargs):
sender.add_periodic_task(5.0, hello(), name='hello?')
@app.task
def hello():
print('hello!')
위 코드를 작성한 후, 쉘을 두 창을 켠 후 각각 아래 코드를 입력해 준다. (celery_parser.py와 같은 폴더에서)
celery worker -A celery_parser --loglevel=info
<이용법>
celery worker -A 파이썬파일이름 --loglevel=info
위 코드는 Celery의 get_url함수, 즉 app의 Task함수가 실제로 구동될 worker
이며, 아래 코드는 periodic_parser함수 안에서 정의된 sender.add_periodic_task에 의해 첫번째 인자로 전달된 5.0초, 두번째 인자로 전달된 hello 함수를 실행하게 하는 Celery의 beat
이다.
celery beat -A celery_parser --loglevel=info
<이용법>
celery beat -A 파이썬파일이름 --loglevel=info
텔레그램 봇을 Python에서 이용하는 좋은 가이드가 있다.
위 링크를 참고해서 pip로 python-telegram-bot
을 설치하고, 새 봇을 만든 후 token과 id값을 받아오자.
우선 파이썬 파일을 수정하기 전에 target.json
라는 환경변수용 json파일을 아래와 같이 만들어 주자.
{
"BOT_TOKEN":"위에서 받은 숫자9자리:영문+숫자+특수문자 긴것",
"URL":"변화유무를 체크할 URL",
"CHAT_ID":"위에서 받은 id"
}
이제 celery_parser.py
에서 target.json
파일을 불러온 후, 변수로 등록해주고, telegram bot 객체를 만들어 준 후 sendMessage를 이용해 보자.
# celery_parser.py
from celery import Celery
import requests
import json
import os
import datetime
import telegram
# Celery Setup
app = Celery()
app.conf.timezone = 'Asia/Seoul'
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# Parsing/Telegram Environ loads
with open(os.path.join(BASE_DIR, "target.json")) as f:
env = json.loads(f.read())
BOT_TOKEN = env['BOT_TOKEN']
URL = env['URL']
CHAT_ID = env['CHAT_ID']
bot = telegram.Bot(token=BOT_TOKEN)
bot.sendMessage(chat_id=CHAT_ID, text='Started!')
@app.on_after_configure.connect
def periodic_parser(sender, **kwargs):
sender.add_periodic_task(5.0, get_url.s(URL), name='send working time')
@app.task
def get_url(url):
req = requests.get(url)
f = open('temp/req.txt', 'w+')
previous_html = f.read()
new_html = req.text
bot.sendMessage(chat_id=CHAT_ID, text='Working/{}'.format(datetime.datetime.now()))
if previous_html == new_html:
bot.sendMessage(chat_id=CHAT_ID, text='working...')
else:
#TODO: BOT NOTICE
bot.sendMessage(chat_id=CHAT_ID, text='{} 에 변경이 있습니다.'.format(url))
위 코드는 URL에 접속해 html로 저장 후 5초 후 다음 접속 시 사이트에 변동사항이 있으면 변동이 있다는 Telegram 알림을 보내준다.