[Python] 마스크 재고 API와 텔레그램으로 나만의 마스크 재고 알리미를 만들어보자!

[Python] 공공api를 활용하여 내 주변 공적 마스크 판매처와 마스크 재고를 지도에 시각화해보자!

오늘은 어제 지도 시각화를 하면서 사용했던 마스크 데이터를 제공하는 공공 API와 텔레그램을 활용하여

나만의 텔레그램 마스크 재고 알리미를 만들어보고자 합니다.

 

2020년 9월 30일 업데이트

공적마스크 판매 중단으로 인하여 7월 8일 부로 API 지원이 종료 되었습니다.

공공데이터 포털

 

작업 환경

맥북프로 2017 13인치 - macOS Catalina

jupyter notebook / visual studio Code

 

먼저 좌표기준으로 m 미터 이내의 데이터를 받아오기위해서 어제 api를 사용해서 만들었던 함수를 사용해

텔레그램에서 보낼 메세지를 만들도록 바꾸어보았습니다.

SOMJANG/Korea_Public_Mask_Store_Visualization

 

위도 (lat), 경도 (lng), 반경 m (dist) 미터 안에있는 공적마스크 판매처의 정보를 받아오는 함수

def getNearMaskStoreInfoByGeo(lat, lng, dist):
    url = "https://8oi9s0nnth.apigw.ntruss.com/corona19-masks/v1/storesByGeo/json?lat=" + str(lat) + "&lng=" + str(lng) + "&m=" + str(dist)
    
    req = requests.get(url)
    
    json_data = req.json()
    store_data = json_data['stores']
    
    addrs = []
    codes = []
    latitudes = []
    longitudes = []
    names = []
    types = []
    created_ats = []
    remain_stats = []
    stock_ats = []
    
    for i in tqdm(range(len(store_data))):
        addrs.append(store_data[i]['addr'])
        codes.append(store_data[i]['code'])
        latitudes.append(store_data[i]['lat'])
        longitudes.append(store_data[i]['lng'])
        names.append(store_data[i]['name'])
        types.append(store_data[i]['type'])
        try:
            created_ats.append(store_data[i]['created_at'])
        except:
            created_ats.append("no_data")
        try:
            remain_stats.append(store_data[i]['remain_stat'])
        except:
            remain_stats.append("no_data")
        try:
            stock_ats.append(store_data[i]['stock_at'])
        except:
            stock_ats.append("no_data")
    
    mask_store_info_df = pd.DataFrame({"addr":addrs, "code":codes, "latitude":latitudes, "longitude":longitudes, "name":names, "type":types, "created_at":created_ats, "remain_stat":remain_stats, "stock_at":stock_ats})
    
    return mask_store_info_df

 

받아온 데이터에서 판매가 중단된 곳, 재고가 소진된 곳, 정보가 없는 곳을 제거해주는 함수

def getNoneEmptyStockStore(mask_store_info_by_geo):
    data_df = mask_store_info_by_geo.loc[:, ['name', 'addr', 'remain_stat', 'stock_at', 'created_at']]
    data_df_nan = data_df.dropna()
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'break']
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'empty']
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'no_data']
    new_index = []
    for i in range(len(data_df_nan['name'])):
        new_index.append(i)
    data_df_nan.index = new_index
    
    return data_df_nan

 

정제한 데이터를 바탕으로 보낼 메세지를 만들어주는 함수

마스크 재고가 남아있는 공적판매처가 없는 경우

- 주변 1km 반경 내의 공적판매처의 마스크 재고가 모두 소진되었습니다.

마스그 재고가 남아있는 공적판매처가 있는 경우

- "판매처명 : " + name + "%0A" + "주소 : " + addr + "%0A" + "재고상태 : " + remain_stat_kor[remain_stat] + "%0A" + "재고 갱신 시간 : " + stock_at + "%0A" + "%0A"

def makeMaskStockMessage(mask_stock_info_df):
    
    remain_stat_kor = {'plenty':'100개 이상', 'some':'30개 이상', 'few':'2개이상 30개 미만'}
    
    text = ""
    if len(mask_stock_info_df) == 0:
        text = "주변 1km 반경 내의 공적판매처의 마스크 재고가 모두 소진되었습니다"
    else:
        name_list = list(mask_stock_info_df['name'])
        addr_list = list(mask_stock_info_df['addr'])
        remain_list = list(mask_stock_info_df['remain_stat'])
        stock_at_list = list(mask_stock_info_df['stock_at'])
        
        for i in range(len(mask_stock_info_df['name'])):
            name = name_list[i]
            addr = addr_list[i]
            remain_stat = remain_list[i]
            stock_at = stock_at_list[i]
            
            text = "판매처명 : " + name + "%0A" + "주소 : " + addr + "%0A" + "재고상태 : " + remain_stat_kor[remain_stat] + "%0A" + "재고 갱신 시간 : " + stock_at + "%0A"
            
    return text

위도 (lat), 경도 (lng), 반경 m 미터 (m)의 데이터를 입력 받아 텔레그램이 보낼 메세지를 만들어줄 함수

def makeSendMessage(lat, lng, m):
    mask_store_info_by_geo = getNearMaskStoreInfoByGeo(lat, lng, m)
    mask_stock_info_df = getNoneEmptyStockStore(mask_store_info_by_geo)
    sendMessage = makeMaskStockMessage(mask_stock_info_df)
    return sendMessage

lat = 37.513489
lng = 126.941986
m = 1000

makeSendMessage(lat, lng, m)

노량진역 반경 1km로 테스트를 해보면 이상없이 잘 만들어지는 것을 알 수 있습니다.

1km 로 설정한 이유는 걸어서 다녀올 수 있는 위치의 판매처를 검색하고 싶어서 였습니다.

더 넓은 반경을 원하는 분은 최대 5km까지 설정이 가능하니 m값을 1,000부터 5,000사이의 값으로 잘 설정하면 됩니다.

 

이제 이 함수를 활용하여 텔레그램봇을 만들어보도록 하겠습니다.

나중에 활용하기위해서 .py 파일로 다운로드 해놓습니다.

 

텔레그램 설치하기

먼저 앱스토어에서 텔레그램을 설치합니다.

 

botfather 검색하기

텔레그램 봇 만들기

Start 버튼을 클릭합니다.

/start

텔레그램 봇 만들기를 시작합니다.

/newbot

위의 값을 입력하여 새로운 텔레그램 봇을 만듭니다.

public_mask_notifier_bot

채팅창에 텔레그램 봇의 이름을 입력합니다.

somjang_bot

채팅창에 username을 입력합니다.

 

그럼 답으로 오는 결과 값에서

 

Use this token to access the HTTP API:
token 값 (토큰 값)
Keep your token secure and store it safely, it can be used by anyone to control your bot.

여기서 토큰 값을 기억해줍니다.

 

잘 만들어 졌는지 검색해보기

API에서 받아온 데이터 텔레그램으로 메세지 보내기

이제 텔레그램 라이브러리를 활용하여 아까 위에서 API를 통해 만들어두었던 함수에서 데이터를 받아 텔레그램으로 보내보겠습니다.

먼저 telegram 라이브러리를 설치합니다.

 

telegram / python-telegram-bot 설치하기

주피터 노트북일 경우

!pip3 install telegram

!pip3 install python-telegram-bot

터미널 / 명령프롬프트일 경우

pip3 install telegram

pip3 install python-telegram-bot

설치가 완료되었으면 텔레그램 봇에 새로운 메세지를 보내고 사용자 아이디를 추출합니다.

import telegram

bot = telegram.Bot(token="token 값")

recent_message = []

for i in bot.getUpdates():
    recent_message = i.message
    
message_user_id = recent_message.chat['id']
print(message_user_id)

이를 통해 사용자 id를 확인할 수 있습니다.

 

이 사용자 id와 아까 만들어 두었던 makeSendMessage함수를 가지고

bot.sendMessage 메소드를 활용하여 텔레그램 봇이 메세지를 보내도록 해보았습니다.

import telegram

bot = telegram.Bot(token="1000662221:AAHBAaP1zJWEO88Qf90qSSjGadykxkTIYGI")

lat = 37.513489
lng = 126.941986
m = 1000

bot.sendMessage(chat_id=message_user_id, text=makeSendMessage(lat, lng, m))

노량진역 기준 반경 1km 안에있는 공적판매처 중 재고가 남아있는 약국만 찾아 메세지로 보내줍니다.

여기 까지하고 저는 집에서 가장 가깝다고 생각되는 스마일 약국으로 마스크를 구매하러 다녀왔습니다.

일요일 마스크 5부제 마스크 구매 후기!

일요일이라 마스크 구매에 큰 어려움없이 구매할 수 있었습니다.

하지만 여기까지만 해서는 직접 본인이 메세지를 보내야하는 단점이있고

어느 판매처가 최신의 정보인지를 직접 파악해야하는 문제점이 있었습니다.

 

메세지를 보낼때 재고 갱신 시간을 기준으로 최신의 정보 순으로 정렬하여 메세지를 만들도록하였습니다.

 

재고 입고 시간을 기준으로 최신 정보 순으로 정렬하여 제공받기

def getNoneEmptyStockStore(mask_store_info_by_geo):
    data_df = mask_store_info_by_geo.loc[:, ['name', 'addr', 'remain_stat', 'stock_at', 'created_at']]
    data_df_nan = data_df.dropna()
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'break']
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'empty']
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'no_data']

    data_df_nan = data_df_nan.sort_values(by=['stock_at'], axis=0, ascending=False)

    new_index = []
    for i in range(len(data_df_nan['name'])):
        new_index.append(i)
    data_df_nan.index = new_index
    
    return data_df_nan

기존의 함수에서 DataFrame을 정렬하는 코드를 한줄 추가해주었습니다.

def makeMaskStockMessage(mask_stock_info_df):
    
    remain_stat_kor = {'plenty':'100개 이상', 'some':'30개 이상', 'few':'2개이상 30개 미만'}
    
    text = ""
    if len(mask_stock_info_df) == 0:
        text = "주변 1km 반경 내의 공적판매처의 마스크 재고가 모두 소진되었습니다"
    else:
        name_list = list(mask_stock_info_df['name'])
        addr_list = list(mask_stock_info_df['addr'])
        remain_list = list(mask_stock_info_df['remain_stat'])
        stock_at_list = list(mask_stock_info_df['stock_at'])
        created_at_list = list(mask_stock_info_df['created_at'])
        
        for i in range(len(mask_stock_info_df['name'])):
            name = name_list[i]
            addr = addr_list[i]
            remain_stat = remain_list[i]
            stock_at = stock_at_list[i]
            created_at = created_at_list[i]
            
            text = text + "판매처명 : " + name + "\n" + "주소 : " + addr + "\n" + "재고상태 : " + remain_stat_kor[remain_stat] + "\n" + "재고 입고 시간 : " + stock_at + "\n" + "데이터 갱신 시간 : " + created_at + "\n" +"\n" 
            
    return text

그리고 출력부분에서 데이터 갱신 시간을 추가해주었습니다.

이제 내가 따로 실행을 시켜주지 않아도 자동으로 정해진 시간마다 공적판매처마다 마스크 재고를 받기위해

apScheduler를 활용하였습니다.

 

정해진 시간마다 내용을 보내도록 설정하기

import telegram
from apscheduler.schedulers.blocking import BlockingScheduler

bot = telegram.Bot(token="token값")

def sendStockStateMessage():
    lat = 37.513489
    lng = 126.941986
    m = 1000
    
    bot.sendMessage(chat_id=message_user_id, text=makeSendMessage(lat, lng, m))
    count = count + 1
    
sched = BlockingScheduler({'apscheduler.timezone':'UTC'})
sched.add_job(sendStockStateMessage, 'interval', seconds=5)
sched.start()

테스트를 위해서 5초에 한번씩 내용을 보내도록 설정해보았습니다.

이상없이 잘 보내지는 것을 확인했습니다.

이 역시 .py 파일로 저장하여 정리하여줍니다.

 

geoInfoFromAPIwithGeo.py

#!/usr/bin/env python
# coding: utf-8

import requests
import json
from tqdm import tqdm
import pandas as pd


def getNearMaskStoreInfoByGeo(lat, lng, dist):
    url ="https://8oi9s0nnth.apigw.ntruss.com/corona19-masks/v1/storesByGeo/json?lat=" + str(lat) + "&lng=" + str(lng) + "&m=" + str(dist)
    
    req = requests.get(url)
    
    json_data = req.json()
    store_data = json_data['stores']
    
    addrs = []
    codes = []
    latitudes = []
    longitudes = []
    names = []
    types = []
    created_ats = []
    remain_stats = []
    stock_ats = []
    
    for i in tqdm(range(len(store_data))):
        addrs.append(store_data[i]['addr'])
        codes.append(store_data[i]['code'])
        latitudes.append(store_data[i]['lat'])
        longitudes.append(store_data[i]['lng'])
        names.append(store_data[i]['name'])
        types.append(store_data[i]['type'])
        try:
            created_ats.append(store_data[i]['created_at'])
        except:
            created_ats.append("no_data")
        try:
            remain_stats.append(store_data[i]['remain_stat'])
        except:
            remain_stats.append("no_data")
        try:
            stock_ats.append(store_data[i]['stock_at'])
        except:
            stock_ats.append("no_data")
    
    mask_store_info_df = pd.DataFrame({"addr":addrs, "code":codes, "latitude":latitudes, "longitude":longitudes, "name":names, "type":types, "created_at":created_ats, "remain_stat":remain_stats, "stock_at":stock_ats})
    
    return mask_store_info_df


def getNoneEmptyStockStore(mask_store_info_by_geo):
    data_df = mask_store_info_by_geo.loc[:, ['name', 'addr', 'remain_stat', 'stock_at', 'created_at']]
    data_df_nan = data_df.dropna()
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'break']
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'empty']
    data_df_nan = data_df_nan[data_df_nan['remain_stat'] != 'no_data']

    data_df_nan = data_df_nan.sort_values(by=['stock_at'], axis=0, ascending=False)

    new_index = []
    for i in range(len(data_df_nan['name'])):
        new_index.append(i)
    data_df_nan.index = new_index
    
    return data_df_nan


def makeMaskStockMessage(mask_stock_info_df):
    
    remain_stat_kor = {'plenty':'100개 이상', 'some':'30개 이상', 'few':'2개이상 30개 미만'}
    
    text = ""
    if len(mask_stock_info_df) == 0:
        text = "주변 1km 반경 내의 공적판매처의 마스크 재고가 모두 소진되었습니다"
    else:
        name_list = list(mask_stock_info_df['name'])
        addr_list = list(mask_stock_info_df['addr'])
        remain_list = list(mask_stock_info_df['remain_stat'])
        stock_at_list = list(mask_stock_info_df['stock_at'])
        created_at_list = list(mask_stock_info_df['created_at'])
        
        for i in range(len(mask_stock_info_df['name'])):
            name = name_list[i]
            addr = addr_list[i]
            remain_stat = remain_list[i]
            stock_at = stock_at_list[i]
            created_at = created_at_list[i]
            
            text = text + "판매처명 : " + name + "\n" + "주소 : " + addr + "\n" + "재고상태 : " + remain_stat_kor[remain_stat] + "\n" + "재고 입고 시간 : " + stock_at + "\n" + "데이터 갱신 시간 : " + created_at + "\n" +"\n" 
            
    return text
            



def makeSendMessage(lat, lng, m):
    mask_store_info_by_geo = getNearMaskStoreInfoByGeo(lat, lng, m)
    mask_stock_info_df = getNoneEmptyStockStore(mask_store_info_by_geo)
    sendMessage = makeMaskStockMessage(mask_stock_info_df)
    return sendMessage

sendMessageAtTelegram.py

#!/usr/bin/env python
# coding: utf-8

from getInfoFromAPIwithGeo import makeSendMessage
import telegram
from apscheduler.schedulers.blocking import BlockingScheduler   

def sendStockStateMessage():
    lat = 37.513489
    lng = 126.941986
    m = 1000

    bot = telegram.Bot(token="token 값")

    recent_message = []

    get_id_flag = False

    for i in bot.getUpdates():
        if get_id_flag == False:
            recent_message = i.message
        else:
            break
        
    message_user_id = recent_message.chat['id']
    print("user_id : ", message_user_id)
    
    bot.sendMessage(chat_id=message_user_id, text=makeSendMessage(lat, lng, m))

print("Start")
sendStockStateMessage()
sched = BlockingScheduler({'apscheduler.timezone':'UTC'})
sched.add_job(sendStockStateMessage, 'interval', seconds=600)
sched.start()

이상없이 잘 작동되는 것을 확인했습니다.

이제 이 파일을 안쓰는 노트북으로 옮겨 실행시켜놓으려합니다.

 

먼저 git에 업로드 하였습니다.

git add .
git commit -m "code upload"
git push origin

git clone https://github.com/SOMJANG/Korea_Public_Mask_Stock_Notifier.git

위의 명령어로 코드를 다운로드 받았습니다.

pip install telegram

pip install python-telegram-bot

pip install requests

pip install apscheduler

필요한 라이브러리를 설치 후 

cd Downlods/Korea_Public_Mask_Stock_Notifier

python sendMessageAtTelegram.py

실행시켜주었습니다.

오늘은 여기까지!

 

지금은 안쓰는 컴퓨터에 실행을 시켜두었지만

시간과 자원이된다면 EC2서버에 올려서 실행해보고 싶습니다.

 

읽어주셔서 감사합니다!

 

SOMJANG/Korea_Public_Mask_Stock_Notifier