16. Vue3 + D3.js 코로나 감염 지도 python 3으로 공공데이터 가져오기

 

앞서 기상청의 공공데이터를 가져온 것을 활용하여 보건복지부 코로나 감염 데이터를 가져와 보기로 한다.

 

공공데이터 활용 전 유의 사항

공공데이터를 활용하기 전에는 항상 목록에서 제공하는 가이드라인 문서를 숙지해야 한다.

데이터를 가져오는 소스코드는 재활용할 수 있으나, 

공공데이터 제공처의 옵션이 따로 있기 때문에 서비스 url 이나 파라미터만 변경해서 사용하면 안된다.

 

보건복지부의 코로나 감염 정보는 전날 정보로 00시 기준으로 데이터가 변경되지 않는다. 

오전이 되어야 새로운 정보를 가져올 수 있다. 

 

이 예제에서는 오전 10시 30분에 가져올 수 있게 하였다.

 

데이터를 가져오는 경우에도 crontab에 등록하여 사용할 수 있지만, 

이번에는 schedule 패키지를 사용하여 지정한 시간에 데이터를 가져왔다.

 

방법은 개인이 선호하는 방법으로 가져오면 될 것이다. 

 

python3 으로 공공데이터(보건복지부) 가져오기

 

구현한 예제 코드는 다음과 같다.

 

# Covid-19

from urllib.request import urlopen
import requests
from urllib.request import Request, urlopen
from urllib.parse import urlencode, quote_plus
import xml.dom.minidom

from xml.etree import ElementTree as etree

import time
import datetime

import mysql.connector

class Covidinfo:  
    def __init__(self):
        self.url =  'http://openapi.data.go.kr/openapi/service/rest/Covid19/getCovid19SidoInfStateJson'           
        self.key = '서비스키'
        self.rtype = 'xml'
    
    def getData(self, rt):
        try:
            url = self.url
            params ={'serviceKey' : self.key,
                    'pageNo' : '1', 
                    'numOfRows' : '20',
                    'startCreateDt' : rt,     
                    'endCreateDt' : rt }
            
            response = requests.get(url, params=params)

            #print(response.content)

            sxml = xml.dom.minidom.parseString(response.content)
            pretty_xml_as_string = sxml.toprettyxml()

            #print(pretty_xml_as_string)

            tree = etree.fromstring(pretty_xml_as_string)
            result = tree.findall(".//item")

            #print(len(result))
            
            ins1 = [x.findtext(".//stdDay") for x in result]
            ins2 = [x.findtext(".//gubun") for x in result]
            ins3 = [x.findtext(".//deathCnt") for x in result]
            ins7 = [x.findtext(".//defCnt") for x in result]
            ins4 = [x.findtext(".//incDec") for x in result]
            ins5 = [x.findtext(".//localOccCnt") for x in result]
            ins6 = [x.findtext(".//overFlowCnt") for x in result]
            

            try:
                conn =  mysql.connector.connect(
                host="아이피",
                port="13306",
                user="root",
                password="암호",
                database='디비명'
                )
                with conn.cursor() as cur:
                    for i in range(0, len(result)):

                        sql = "insert into covid_reals \
                            (stdDay, gubun, deathCnt, defCnt, \
                            incDec, localOccCnt, overFlowCnt ) \
                            values(%s, %s, %s, %s, %s, %s, %s);"
                        val = (rt , ins2[i], ins3[i], ins4[i], ins5[i], ins6[i], ins7[i])
                        cur.execute(sql, val)

                    if (len(result)>0):
                        sql = "update getstate set curtime = now() where tbname = 'covid_reals';"        
                        cur.execute(sql)

                    conn.commit()

            except Exception as e:
                print(e)
                raise Exception("insert exception")
                conn.rollback()
            finally:
                conn.close()
                print("Data ", len(result), "DB Insert")

        except requests.exceptions.Timeout as errd:
            print("Timeout Error : ", errd)
            
        except requests.exceptions.ConnectionError as errc:
            print("Error Connecting : ", errc)
            
        except requests.exceptions.HTTPError as errb:
            print("Http Error : ", errb)

        # Any Error except upper exception
        except requests.exceptions.RequestException as erra:
            print("AnyException : ", erra)   

def getNowTime():
    d = datetime.datetime.now()
    
    nday = d.strftime("%Y%m%d")
    
    return nday

if __name__ =='__main__':
    
    oinfo = Covidinfo()   

    rt = getNowTime()
    print(rt, " : Data Get Start")
    oinfo.getData(rt)

 

이 또한 최적화하지 않았기에 사용자가 최적화하면 될 것 같다.

 

데이터베이스에 확인을 하니 제대로 Insert가 되었다.

 

공공데이터 가져오는 코드는  schedule 패키지는 아래와 같이 등록하고 사용하였다.

schedule.every().day.at("10:30").do(chkCovid)

 

다음 기회가 된다면 schedule 패키지 사용 코드에 대해 소개하겠다.

 

개인적으로 crontab에 등록하여 사용하면 편리하지만, crontab을 사용하지 않는다면 schedule도 좋은 대안인 것 같다.

 

도구의 선택은 각자 알아서 할 일이지만...