안녕하세요? 이번 글은 농림축산식품 공공데이터 포털 제공의 가축질병발생정보 오픈 API 사용법을 정리해 보겠습니다.
바로가기: http://data.mafra.go.kr/opendata/data/indexOpenDataDetail.do?data_id=20151204000000000563
먼저 필요한 라이브러리를 가져옵니다. requests는 웹페이지에 요청을 보내고 해당 페이지의 내용을 가져오기 위해 필요한 라이브러리입니다. pandas는 데이터를, geopandas는 지리공간 데이터를 다루기 위한 라이브러리입니다. BeautifulSoup는 웹페이지의 HTML 및 XML 문서를 파싱하고 조작하기 위한 라이브러입니다. 웹 스크래핑(Web scaping)은 웹페이지에서 데이터를 추출하고 수집하는 단계이고 파싱(parsing)은 수집된 데이터를 분석하여 이해하고 처리하는 단계입니다.
import requests
import pandas as pd
import geopandas as gpd
from bs4 import BeautifulSoup가축질병발생정보를 호출하는 call_api() 함수를 정의합니다. 이 함수는 시작위치와 종료위치를 입력 받아 이 인덱스 범위에 해당하는 가축질병발생정보를 API를 통해 호출합니다. "발급받은 API_KEY"는 농림축산식품 공공데이터 포털에서 회원가입 후 신청하시면 됩니다. API_KEY는 API를 호출할 때 인증을 위해 사용됩니다.
API 호출을 위한 URL(url 변수)로 HTTP GET 요청을 보내고 서버로부터 응답을 받아옵니다. 서버 응답의 텍스트 데이터는 xml_data 변수에 저장합니다. 이 데이터는 XML 형식으로 가축질병발생정보를 담고 있습니다. 다음으로 BeautifulSoup 라이브러리를 사용하여 xml_data를 XML 형식으로 파싱하고, 파싱된 XML 데이터(soup 변수)를 리턴합니다. 이렇게 하면 함수 호출자는 파싱된 데이터를 활용할 수 있습니다.
# 가축질병발생정보 호출
def call_api(start_idx, end_idx):
api_key = "발급받은 API_KEY"
base_url = "http://211.237.50.150:7080/openapi/"
api_url = "Grid_20151204000000000316_1"
url = (
f"{base_url}{api_key}/xml/{api_url}/{start_idx}/{end_idx}?"
)
response = requests.get(url)
xml_data = response.text
soup = BeautifulSoup(xml_data, 'xml')
return soup
BeautifulSoup로 파싱된 XML 데이터에서 전체 데이터 개수를 추출하는 함수를 정의합니다. BeautifulSoup의 find 메서드를 사용하여 XML 데이터 내에서 totalCnt 요소의 내용을 문자열로 추출하고 이를 정수로 변환하는 과정입니다. 이렇게 추출된 전체 데이터 개수는 이후 데이터 호출 및 처리 작업에서 사용될 수 있습니다.
# 전체 데이터 개수 추출
def extract_total_count(soup):
total_count = int(soup.find("totalCnt").text)
return total_countBeautifulSoup로 파싱된 XML 데이터에서 가축질병발생정보를 추출하여 데이터프레임을 생성하는 함수 create_dataframe를 정의합니다. 데이터를 저장할 리스트 data_list를 생성하고, soup에서 row 요소(각각의 가축질병발생정보)를 딕셔너리 data로 추출하여 data_list에 반복 추가하는 방식입니다. 추출된 데이터 딕셔너리는 데이터프레임으로 변환 및 반환합니다.
# 가축질병발생정보 데이터프레임 생성
def create_dataframe(soup):
data_list = []
for item in soup.find_all("row"):
data = {
"ROW_NUM": item.find("ROW_NUM").text,
"ICTSD_OCCRRNC_NO": item.find("ICTSD_OCCRRNC_NO").text,
"LKNTS_NM": item.find("LKNTS_NM").text,
"FARM_NM": item.find("FARM_NM").text,
"FARM_LOCPLC_LEGALDONG_CODE": item.find("FARM_LOCPLC_LEGALDONG_CODE").text,
"FARM_LOCPLC": item.find("FARM_LOCPLC").text,
"OCCRRNC_DE": item.find("OCCRRNC_DE").text,
"LVSTCKSPC_CODE": item.find("LVSTCKSPC_CODE").text,
"LVSTCKSPC_NM": item.find("LVSTCKSPC_NM").text,
"OCCRRNC_LVSTCKCNT": item.find("OCCRRNC_LVSTCKCNT").text,
"DGNSS_ENGN_CODE": item.find("DGNSS_ENGN_CODE").text,
"DGNSS_ENGN_NM": item.find("DGNSS_ENGN_NM").text,
"CESSATION_DE": item.find("CESSATION_DE").text,
}
data_list.append(data)
df = pd.DataFrame(data_list)
return df
이제 API를 통해 가축질병발생정보의 전체 데이터 개수를 출력하고 1부터 1000까지 범위에서 데이터프레임을 변환해 보겠습니다.
start_idx = 1 # 시작지점
end_idx = 1000 # 종료지점
#가축질병발생정보 호출
soup = call_api(start_idx, end_idx)
# 전체 데이터 개수
total_count = extract_total_count(soup)
print(f"Total Count: {total_count}")
# 가축질병발생정보 데이터프레임 생성
df = create_dataframe(soup)
df.head()전체 데이터 개수는 44,282건입니다. 데이터요청은 한번에 최대 1000건을 넘을 수 없습니다. 따라서 전체 데이터를 가져오려면 반복문으로 처리해야 합니다.
Total Count: 44282
이번에는 전체 데이터를 데이터프레임으로 변환해 보겠습니다. 전체 데이터를 여러 번에 나누어 API로 가져오고 각 호출 결과를 데이터프레임으로 변환한 후, 이들을 하나로 합쳐서 CSV 파일로 저장하는 작업입니다. %%time를 추가하여 코드 실행 시간을 측정해 보겠습니다.
%%time
# 가축질병발생정보 전체 데이터를 데이터프레임으로 변환
all_data_frames = []
for idx in range(1, total_count + 1, end_idx):
df = create_dataframe(call_api(idx, idx + end_idx - 1))
all_data_frames.append(df)
df = pd.concat(all_data_frames, ignore_index=True)
df.to_csv('livestock_disease.csv', index=False, encoding='cp949')
df.head()실제 시간으로 측정한 실행 시간(Wall time)은 49.8초입니다. 이제 전체 데이터가 livestock_disease.csv라는 파일로 저장되었습니다.
CPU times: total: 22.7 s
Wall time: 49.8 s
