사용한 API는 공공데이터포털 농촌진흥청 국립농업과학원_농업기상 관측지점 상세정보 / 기본 관측데이터 조회 두가지


머신러닝 공부하는중이라 데이터 확보가 중요하다보니 자연스레 api로 데이터 취득하는걸 해보게 됨......

파이썬 독학 들어간지 이제야 반년이라 코드 개더러움 주의

더 쉽게 짤 수 있는 부분이 있거나 불필요한 부분이 있는 등 피드백 할게 있으면 가르쳐주면 고맙겠음



일단 api 호출 필수 조건 네가지 중 세가지((시)도 구분 코드 OR 관측지점코드, 관측시작~종료일, API KEY)는 베이스로 두고

1. (시)도 구분 코드를 사용한 경우

1-2. (시)도 구분 코드로 데이터를 한번에 불러올 때,그 안에 불러올 필요가 없는 관측 지점이 존재하는 경우

2. 관측지점코드(1 개소)를 사용한 경우

에 대응할 수 있도록 코딩해봤음


마지막 네 번째 필수 조건인 page no는 그냥 list 형태인 [1, 2] 로 지정해놨는데, 해당 API의 경우 하루치 데이터를 모두 불러오는 데에 반드시 2페이지가 필요해서 굳이 머리아프게 페이지 수까지 유동적으로 변하게는 하지 않았음

그리고 api key는 난 무조건 decoding 키만 사용할 수 있던데 뭐 구글에서 하는 말 들어보니까 encoding 키를 써야 할 수도 있다 해서 일단은 바꿀 수 있게는 해 뒀음



추가로 넣어놓은 기능들이

3. 수시로 HTTP Error(api 서버 에러)를 띄우면서 끊겨대는 바람에 부득이하게 error 발생 시 대처할 수단을 코드에 넣었음

4. 정상적으로 한 지점의 데이터를 받고 나면 이 지점들의 정보를 리스트로 저장해뒀다가 오류떠서 멈춰버리면 해당 정상 다운로드 완료된 목록을 지점 리스트에서 빼버려서 이미 다운받은걸 다시 다운받는 일이 없게 만들었음

5. 데이터가 없어 api의 'item' 부분이 텅 비어있는 경우가 있는데 이때는 아예 데이터가 없는 경우니 에러로 멈출 바에야 그냥 무시하고 계속 다운받는게 맞다고 봐서 데이터가 없으면 그냥 없는대로 돌아가게 해 뒀음. **다만 가끔 가다 보면 'items' 나 'body' 자체가 없는 경우가 있는데 이건 api 서버에 오류가 생겼을 가능성도 없지 않다 생각해서 배제했음

5. 하다가 어떠한 원인으로 다운로드 진행이 막히면 일단 받아온 지점까지의 데이터만을 저장할건지, 또한 지점 리스트에서 이 오류가 발생한 지점을 빼버릴 것인지를 input을 사용해서 그때그때 선택하게 만들어둠

6. 에러 발생한 지점들은 따로 error_spot 변수에 저장해둬서 나중에 오류 해결되면 spot_code 자리에 error_spot 변수로 바꿔치기만 하면 다시 에러가 발생했던 지점만 다운받을 수 있도록 해 놨음

7. error 뜨는것도 바로바로 정보를 확인할 수 있게 정리해놨음.

8. 마지막으로 이왕 데이터 api 통해서 불러오는거 지점명이 value, 지점코드가 key값인 딕셔너리로 저장해뒀다가 나중에 spot code가 어디 지점인지 한글로 보고 싶으면 바로 볼 수 있게 했음


개인적으론 관측지점코드도 list로 여러개 입력해서 한번에 쫙 받을 수 있게 하고 싶었는데 자꾸 에러떠서 포기함

어째서 이 조건이랑 저 조건이랑 받아오는 데이터 모양새가 다른거임


이건 코딩 실력이 늘어나면 그때 해봐야지









jupyter notebook으로 만듬



코드 텍스트로 작성해둔거

#라이브러리 호출 및 파라미터 정리

import pandas as pd

import requests, xmltodict, json



api_encoding = '---'

api_decoding = '---'


spot_url = 'http://apis.data.go.kr/---'

data_url = 'http://apis.data.go.kr/---'


page_size = '100'

page_no = [1, 2]

obsr_spot_code = ''


df = pd.DataFrame()

date_list, spot_record, error_spot = [],[],[]

spot_dict = {}



#함수 정의

def spot_data(param, del_spot_code, div): #지점 정보 받아오기

    global api_key

    

    params = {'serviceKey' : api_key, 'Page_Size': '20', 'Page_No': '1'}

    params.update(param)

        

    response = requests.get(spot_url, params=params)

    dict_response = xmltodict.parse(response.content)

    json_string = json.dumps(dict_response['response']['body']['items'], ensure_ascii=False)

                

    jsonObj = json.loads(json_string)

    

    if div == 'do_se': # 도 구분 코드를 사용하여 불러온 api 구조와 지점 code를 사용하여 불러온 api의 행렬이 달라서 추가함.

        df_spot = pd.DataFrame(jsonObj['item'])

    else:

        df_spot = pd.DataFrame(jsonObj).transpose()

    

    spot_code = df_spot['Obsr_Spot_Code'].values.tolist()

    spot_name = df_spot['Obsr_Spot_Nm'].values.tolist()

    

    no = 0

    for i in spot_code: #spot_code 와 spot_name dictionary <-- 에러 발생 지점 확인을 위해 spot_code를 키값으로 딕셔너리화

        spot_dict.update({i : spot_name[no]})

        no = no + 1

        

    if div == 'do_se' and del_spot_code != ['']: #없앨 spot code를 입력받았다면 해당 spot code들을 제거함

        spot_code = list(set(spot_code).difference(set(del_spot_code)))

        

    print(f'spot_code: {spot_code}')

    

    return spot_code, spot_dict



def spot_list(do_se_code, obsr_spot_code, del_spot_code): #지점code 지점정보api에서 받아와서 list화

    spot_code, spot_name, spot_dict = [], [], {} #초기화

    

    if obsr_spot_code =='':

        param = {'Do_Se_Code' : do_se_code}

        spot_code, spot_dict = spot_data(param, del_spot_code, 'do_se')

    else:

        param = {'Obsr_Spot_Code' : obsr_spot_code}

        spot_code, spot_dict = spot_data(param, del_spot_code, 'obsr')

        

    return spot_code, spot_dict



def from_to(date_from, date_to): #다운받을 일자 list화

    

    date_list = [] #초기화

    for i in pd.date_range(date_from, date_to).format(formatter=lambda x: x.strftime("%Y-%m-%d")):

        date_list.append(i)

    print(f'date_list: {date_list}')

    

    return date_list



def yes_or_no(question): #에러뜰 시 질문

    while "(y/n) 중 하나를 입력":

        reply = str(input(question+' (y/n): ')).lower().strip()

        if reply[0] == 'y':

            return True

        if reply[0] == 'n':

            return False



def csv_download(spot, date): #csv download

    global date_list

    

    df.to_csv(f"f:/Result/result_{spot}_{date_list[0]} ~ {date}.csv",index=False,header=True, encoding="utf-8-sig")

    print(f'result_{spot}_{date_list[0]} ~ {date}.csv 출력 완료')



def api_load(spot_code, date_list, page_no, spot_dict): #데이터 다운로드

    global df, api_key, spot_record

    

    for spot in spot_code:

        

        for date in date_list:

            

            for page in page_no:

                

                print(f'spot: {spot}, date: {date}, Page_no: {page}')

                headers = {'Content-Type': 'application/xml; charset=utf-8-sig'} #HTTP Error 방지용?

                params = {'serviceKey' : api_decoding, 'Page_Size': '100', 'Page_No': page, 'date' : date, 'Obsr_Spot_Code' : spot} #API requests


                res = requests.get(data_url, headers = headers, params=params)

                dict_res = xmltodict.parse(res.content)

                

                

                try: #KeyError 발생 여부 확인

                    json_string = json.dumps(dict_res['response']['body']['items'], ensure_ascii=False)

                except KeyError: #service error 발생 시 KeyError가 출력됨. spot_code에서 작업된 date는 제거하고 셀 강제종료

                    

                    error_spot.append(spot)

                    print(f'!!!KeyError!!!\n Error Information: {dict_res}')

                    

                    answer = yes_or_no("해당 관측 지점의 관측 데이터를 현재 시점까지만 저장하시겠습니까? y: yes, n: no") # y: 저장. n: 코드 작동 중지.

                    answer_val = yes_or_no("해당 관측 지점의 지점 code를 list에서 제거하시겠습니까? y:yes, n: no") # y: 리스트에서 제거, n: 리스트에서 유지

                    

                    if answer == True:

                        csv_download(spot, date)

                        

                    if answer_val == True:

                        spot_record.append(spot) #<-- 정상 작업된 Spot은 변수에 저장

                        

                    if spot_record != []:

                        spot_code = list(set(spot_code).difference(set(spot_record)))

                    print(f'에러 발생 정보: date - {date}, Page_no - {page}, Obsr_Spot_Code - {spot}, Obsr_spot_Nm - {spot_dict[spot]}, 남은 spot: {spot_code}')

                    

                    return spot_code, error_spot #변경된 spot_code와 에러가 저장된 error_spot 반환

                

                else:

                    jsonObj = json.loads(json_string)

                

                        

                try: #데이터 없어도 계속 돌아가도록 하기 위해 try 사용

                    if page == 1:

                        df1 = pd.DataFrame(jsonObj['item'])

                    elif page == 2:

                        df2 = pd.DataFrame(jsonObj['item'])

                        combined = pd.concat([df1, df2])

                        df = pd.concat([df, combined])

                    else:

                        print('ERROR: pageno list check')

                except:

                    print(f'{date} 데이터 없음.')

                    

    

        if date == date_list[-1] or answer == True:

            csv_download(spot, date)

            spot_record.append(spot) #<-- 정상 작업된 Spot은 변수에 저장

        else:

            print(f' 저장 안됨. date_list 정상적으로 생성되었는지 확인: {date}, {spot}, {spot_dict[spot]}')

            break

        

    return spot_code, error_spot



# '' 안에 다운받을 데이터셋의 정보 입력 (API 기술명세서 참고) <-- [do_se_code, obsr_spot_code 둘 중 하나는 반드시 입력], [date_from, date_to, api_key는 빠짐없이 입력]

#############################################################################################

do_se_code = '8'                 # (시)도 구분 코드 #ex) 8 (--- 코드)

obsr_spot_code = ''   # 관측지점코드 #ex) --- (--- 코드), 코드 하나만 입력할 수 있음

del_spot_code = ['---', '---']    # (도 구분 코드로 다운받을 경우) 다운받을 필요가 없는 관측지점 코드를 list 형태로 입력

date_from = '2022-08-15'        # 관측시작일 ex) 0000-00-00 (부터) <-- 시, 분, 초, 밀리초 입력 불가

date_to = '2022-08-20'          # 관측종료일 ex) 0000-00-00 (까지) <-- 시, 분, 초, 밀리초 입력 불가

api_key = api_decoding          # api_decoding OR api_incoding 선택

#############################################################################################

## llst 형태: ['---', '---', '---', ...]


spot_code, spot_dict = spot_list(do_se_code, obsr_spot_code, del_spot_code)

date_list = from_to(date_from, date_to)



#데이터 다운로드

spot_code, error_spot = api_load(spot_code, date_list, page_no, spot_dict)