1. 법정동 코드 전처리
import pandas as pd
1-1. 법정동코드 엑셀 파일 불러오기
#법정동코드 전처리
info_b = pd.read_excel('C:/주~~소/jscode20240801/KIKcd_B.20240801.xlsx', dtype = { '법정동코드' : object })
1-2. 코드 분리
법정동코드를 앞의 5자리와 뒤의 5자리로 나누어 법정동시군구코드, 법정동읍면동코드를 생성한다.
info_b['법정동시군구코드'] = info_b['법정동코드'].str[:5]
info_b['법정동읍면동코드'] = info_b['법정동코드'].str[5:]
1-3. 불필요한 행 제거
법정동읍면동코드가 '00000'인 행을 제거한다.
(대부분 시군구 레벨의 정보일 가능성이 크기 때문이다.)
info_b = info_b[info_b['법정동읍면동코드'] != '00000']
1-4. 필요한 열만 선택
필요한 열만 선택하여 데이터프레임을 구성한다.
info_b = info_b[['법정동코드', '시도명', '시군구명', '읍면동명', '동리명', '법정동시군구코드', '법정동읍면동코드']]
1-5. 인덱스 초기화
인덱스를 초기화한다.
drop=True: 기존 인덱스 제거
info_b = info_b.reset_index(drop = True)
1-6. 공백 만들기
데이터프레임에서 결측값이 있는 경우, 공백 문자열(" ")로 대체합니다.
info_b = info_b.where(pd.notnull(info_b), " ")
1-7. 앞 뒤 공백 제거(str.strip())
시도명, 시군구명, 읍면동명, 동리명 열의 문자열에서 앞뒤 공백을 제거한다.
info_b['시도명'] = info_b['시도명'].str.strip()
info_b['시군구명'] = info_b['시군구명'].str.strip()
info_b['읍면동명'] = info_b['읍면동명'].str.strip()
info_b['동리명'] = info_b['동리명'].str.strip()
1-8. 주소 생성
각 행의 시도명, 시군구명, 읍면동명, 동리명을 조합하여 주소 열을 생성합니다.
info_b['주소'] = info_b['시도명']+" "+ \
info_b['시군구명']+" "+ \
info_b['읍면동명']+" "+ \
info_b['동리명']
1-9. 중복 공백 제거 및 최종 정리
두 개 이상의 공백이 있는 경우 한 개의 공백으로 대체한다.
주소 열의 앞뒤 공백을 제거하여 최종적으로 주소를 정리한다.
info_b['주소'] = info_b['주소'].str.replace(' ', ' ')
info_b['주소'] = info_b['주소'].str.strip()
info_b.head(2)
2. 행정동코드 전처리
2-1. 행정동코드 전처리
행정동코드 열을 문자열(object)로 읽어들였음.
info_h = pd.read_excel('C:/주~소/jscode20240801/KIKcd_H.20240801.xlsx',
dtype = {'행정동코드' : object})
2-2. 행정동시군구코드, 행정동읍면동코드
행정동코드를 앞의 5자리와 뒤의 5자리로 나누어 행정동시군구코드와 행정동읍면동코드를 생성했음.
info_h['행정동시군구코드'] = info_h['행정동코드'].str[:5]
info_h['행정동읍면동코드'] = info_h['행정동코드'].str[5:]
2-3. 행정동 읍면동 자릿수 00000 제거
행정동읍면동코드가 '00000'인 행을 제거했음
info_h = info_h[info_h['행정동읍면동코드'] != '00000']
2-4. 필요한 열만 선택
필요한 열만 선택하여 데이터프레임을 구성했음.
info_h = info_h[['행정동코드', '시도명', '시군구명', '읍면동명', '행정동시군구코드', '행정동읍면동코드']]
2-5. 인덱스 초기화
drop=True: 기존 인덱스를 제거하겠다는 의미
info_h = info_h.reset_index(drop = True)
2-6. 결측값 처리, 주소데이터 생성
데이터프레임에서 결측값이 있는 경우, 공백 문자열(" ")로 대체했음.
info_h = info_h.where(pd.notnull(info_h), ' ')
2-7. 공백제거
시도명, 시군구명, 읍면동명 열의 문자열에서 앞뒤 공백을 제거했음.
info_h['시도명'] = info_h['시도명'].str.strip()
info_h['시군구명'] = info_h['시군구명'].str.strip()
info_h['읍면동명'] = info_h['읍면동명'].str.strip()
2-8. 주소 생성
각 행의 시도명, 시군구명, 읍면동명을 조합하여 주소 열을 생성했음.
info_h['주소'] = info_h['시도명'] + ' ' +\
info_h['시군구명'] + ' ' +\
info_h['읍면동명']
2-9. 중복 공백 제거 및 최종 정리
두 개 이상의 공백이 있는 경우 한 개의 공백으로 대체하고,
주소 열의 앞뒤 공백을 제거하여 최종적으로 주소를 정리했음.
info_h['주소'] = info_h['주소'].str.replace(' ', ' ')
info_h['주소'] = info_h['주소'].str.strip()
2-10. 열 이름 변경
infoh.columns = ['행정동코드','행정동시도명', '행정동시군구명', '행정동읍면동명', '행정동시군구코드', '행정동읍면동코드', '행정동_주소']
3. 데이터 합치기
3-1. 법정동코드 및 행정동코드 데이터 불러오기
info_m = pd.read_excel('C:/주~소/jscode20240801/KIKmix.20240801.xlsx',
dtype = {'법정동코드' : object, '행정동코드' : object})
3-2. 법정동 시군구코드와 읍면동코드 분리
법정동코드를 앞의 5자리와 뒤의 5자리로 나누어 법정동시군구코드와 법정동읍면동코드를 생성했음.
info_m['법정동시군구코드'] = info_m['법정동코드'].str[:5]
info_m['법정동읍면동코드'] = info_m['법정동코드'].str[5:]
3-3. 불필요한 행 제거
법정동읍면동코드가 '00000'인 행을 제거했음.
이는 시군구 레벨의 정보일 가능성이 큼.
info_m = info_m[info_m['법정동읍면동코드'] != '00000']
3-4. 중복된 행 제거
info_m = info_m[['행정동코드', '법정동코드']].drop_duplicates()
3-5.데이터 병합
info_m 데이터프레임을 법정동코드를 기준으로 info_b 데이터프레임과 병합했음.
(how='left'를 사용하여 info_m의 모든 데이터를 유지함.)
이어서 info_m_total 데이터프레임을 행정동코드를 기준으로 info_h 데이터프레임과 병합했음.
info_m_total = pd.merge(info_m, info_b, on = "법정동코드", how = 'left')
info_m_total = pd.merge(info_m_total, info_h, on = "행정동코드", how = 'left')
3-6. 데이터 확인(데이터 전치)
info_m_total 데이터프레임의 상위 3개 행을 확인했으며,
이를 전치(transpose)하여 열이 행으로, 행이 열로 변환된 형태로 출력했음.
info_m_total.head(3).T
3-7. 데이터 저장
info_b, info_h, info_m, info_m_total 데이터프레임을 각각 지정된 경로에 CSV 파일로 저장했음. index=False는 인덱스 정보를 파일에 저장하지 않겠다는 의미임.
info_b.to_csv('C://0.data/info_b.csv', index=False)/0.data/info_h.csv', index=False)
info_h.to_csv('C:/
info_m.to_csv('C:/U0.data/info_m.csv', index=False)/0.data/info_m_total.csv', index=False)
info_m_total.to_csv('C:/
4. API 세팅
4-2. 라이브러리 호출
os, json, time: 기본적인 시스템 조작, JSON 데이터 처리, 시간 제어를 위한 라이브러리
xmltodict: XML 데이터를 Python의 딕셔너리로 변환하기 위한 라이브러리
pandas: 데이터 처리와 분석을 위한 라이브러리
requests, BeautifulSoup: HTTP 요청과 HTML/XML 데이터 파싱을 위한 라이브러리
urllib.parse, urllib.request: URL 인코딩 및 HTTP 요청을 위한 모듈
requests.adapters.HTTPAdapter, requests.packages.urllib3.util.retry.Retry: HTTP 요청에 대한 재시도를 처리하기 위한 모듈
import os
import json
import time
import xmltodict
import pandas as pd
import requests
from bs4 import BeautifulSoup as bs
from urllib.parse import urlencode, quote_plus
from urllib.request import urlopen
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
4-2. 파일 경로 설정
파일을 저장하거나 불러올 경로를 path 변수에 저장했음.
path = 'C:/~/real_info/'
4-3. CSV 파일 불러오기
(참고) 상위 디렉토리로 불러오기 필요 시 '..'
df = pd.read_csv('C:/~/real_info/0.data/info_b.csv')
4-4. 법정동코드의 고유 값 추출
df 데이터프레임의 법정동시군구코드 열에서 고유한 값을 추출하여 LAWD_CD_list 변수에 저장했음.
이 리스트는 이후 API 요청 시 사용할 시군구코드 목록임.
생성한 법정동코드 고유 값 지정
LAWD_CD_list = df['법정동시군구코드'].unique()
4-5. API 키 세팅
api_key = '인코딩키 입력
API 요청에 사용할 API 키를 api_key 변수에 저장했음.
이 부분에서는 실제 API 키를 입력해야 함.
4-6. API 응답 컬럼명 설정 (영문/한글)
API 응답 데이터에서 받을 열의 영문명을 column_nm_en 리스트에 저장했음.
이 리스트는 API 응답 데이터를 처리할 때 열 이름으로 사용할 예정임.
column_nm_en = ['aptDong', 'aptNm', 'aptSeq', 'bonbun', 'bubun', 'buildYear', 'buyerGbn', 'cdealDay', 'cdealType', 'dealAmount', 'dealDay', 'dealMonth', 'dealYear', 'dealingGbn', 'estateAgentSggNm', 'excluUseAr', 'floor', 'jibun', 'landCd', 'landLeaseholdGbn', 'rgstDate', 'roadNm', 'roadNmBonbun', 'roadNmBubun', 'roadNmCd', 'roadNmSeq', 'roadNmSggCd', 'roadNmbCd', 'sggCd', 'slerGbn', 'umdCd', 'umdNm']
API 응답 데이터를 한글로 표현할 열 이름을 column_nm_ko 리스트에 저장했음.
이는 데이터 처리 시 사용자에게 더 직관적으로 제공하기 위해 사용될 수 있음.
column_nm_ko = ['아파트', '단지명', '단지', '법정동본번코드', '법정동부번코드', '건축년도', '매수자', '해제사유발생일', '해제여부', '거래금액', '계약일', '계약월', '계약년도', '거래유형', '중개사소재지', '전용면적', '층', '지번', '법정동지번코드', '토지임대부', '등기일자', '도로명', '도로명건물본번', '도로명건물부번', '도로명코드', '도로명일련번호', '도로명시군구코드', '도로명지상지하', '법정동시군구코드', '매도자', '법정동읍면동코드', '법정동']
4-7. 월 리스트 생성
202101부터 202102까지의 월(년월) 값을 포함하는 리스트를 생성했음.
이 리스트는 이후 API 요청 시 특정 월의 데이터를 가져오기 위해 사용될 수 있음.
월 리스트를 사용한 함수 예제
mo = [m for m in range(202101, 202103)]
4-8. 월 리스트 확장 (주석 처리)
mo.extend([m for m in range(202201, 202213)])
mo.extend([m for m in range(202301, 202313)])
mo.extend([m for m in range(202401, 202409)])
주석으로 처리된 이 코드들은 특정 년도와 월에 해당하는 값을 mo 리스트에 추가하는 역할을 함.
각 주석은 해당 범위의 월 값을 리스트에 확장하기 위해 사용될 수 있음.
202201부터 202212까지, 202301부터 202312까지, 202401부터 202408까지의
월 값을 리스트에 추가하는 코드임.
5. 데이터 전처리
5-1. 함수 정의 및 초기화
apt_extract_df라는 함수를 정의함.
이 함수는 ymd(년월) 값을 입력받아 데이터를 수집하고 처리함.
merge_df는 최종적으로 수집된 데이터를 저장할 데이터프레임으로,
함수 실행 시 처음부터 빈 데이터프레임으로 초기화함.
def apt_extract_df(ymd):
merge_df = pd.DataFrame() # 루프 외부에서 초기화
5-2. 세션 설정 및 재시도 정책 구성
HTTP 요청을 보낼 때 사용할 requests 세션을 설정함.
session = requests.Session()
재시도 및 타임아웃 설정
Retry 객체를 통해 재시도 정책을 구성함.
요청이 실패할 경우 최대 5번까지 재시도하며,
각 재시도 사이의 대기 시간은 점차 증가함 (backoff_factor에 따라 2배씩 증가).
-
서버 오류(500, 502, 503, 504) 발생 시 재시도함.
-
HTTPAdapter를 통해 설정된 재시도 정책을 세션에 적용함.
retry = Retry(
total=5, # 총 5번 재시도
backoff_factor=1, # 재시도 사이의 대기 시간 (2^n 형태로 증가)
status_forcelist=[500, 502, 503, 504], # 서버 오류에 대한 재시도
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
5-3. LAWD_CD 리스트를 순회하며 데이터 수집
LAWD_CD_list에 저장된 각 시군구 코드를 순회하며 데이터를 수집함.
for i in range(len(LAWD_CD_list)):
# URL
url = f'http://apis.data.go.kr/1613000/RTMSDataSvcAptTradeDev/getRTMSDataSvcAptTradeDev?serviceKey={api_key}'
요청 변수 파라미터 설정
API 요청을 위한 기본 URL을 설정하고,
여기에 필요한 요청 변수(pageNo, numOfRows, LAWD_CD, DEAL_YMD)를 추가하여 완전한 요청 URL을 생성함.
-
pageNo: 요청 페이지 번호 (1페이지부터 시작)
-
numOfRows: 한 번의 요청으로 가져올 최대 데이터 개수 (9999개)
-
LAWD_CD: 현재 순회 중인 시군구 코드
-
DEAL_YMD: 함수에 전달된 년월 값
params = '&' + urlencode({ quote_plus('pageNo'): 1, quote_plus('numOfRows'): '9999', quote_plus('LAWD_CD'): LAWD_CD_list[i], quote_plus('DEAL_YMD'): ymd # `ymd` 매개변수 사용으로 수정됨 })
데이터 요청 및 응답 처리
세션을 통해 API 요청을 수행하고, 응답을 받아옴.
이때 요청에 대한 응답 시간이 10초를 넘으면 타임아웃이 발생함.
try:
# 데이터 수집 요청
res = session.get(url + params, timeout=10) # 타임아웃 10초 설정
raise_for_status 메서드를 사용하여 HTTP 오류(4xx, 5xx)가 발생하면 예외를 발생시킴.
res.raise_for_status() # HTTP 오류 발생 시 예외 처리성공적으로 응답을 받으면,
XML 형태의 응답 데이터를 BeautifulSoup을 사용해 파싱하고,
item 태그에 해당하는 모든 데이터를 찾음.
# 응답 성공 시 XML 파싱
soup = bs(res.text, 'xml')
items = soup.find_all('item')
5-5. 개별 항목 파싱 및 데이터프레임 생성
total 데이터프레임을 초기화하여 각 item 태그의 데이터를 저장할 준비를 함.
total = pd.DataFrame()각 item을 순회하면서,
column_nm_en 리스트에 저장된 영문 열 이름에 해당하는 데이터를 추출하여 df_raw 리스트에 추가함.
for k in range(len(items)):
df_raw = []
for j in column_nm_en:
try:
items_data = items[k].find(j).text
df_raw.append(items_data)데이터가 없는 경우 예외를 처리하여 빈 문자열("")을 추가함
except:
df_raw.append("")
df_raw 리스트를 데이터프레임으로 변환하고,
column_nm_ko 리스트를 열 이름으로 사용함.
df = pd.DataFrame([df_raw], columns=column_nm_ko)
생성된 df 데이터프레임을 total 데이터프레임에 누적하여 추가함.
total = pd.concat([total, df], ignore_index=True)5-6. 누적 데이터 저장 및 예외 처리
각 LAWD_CD에 대한 데이터를 merge_df에 누적하여 저장함.
merge_df = pd.concat([merge_df, total], ignore_index=True데이터 수집이 완료되면
현재 시군구 코드(LAWD_CD_list[i])와 누적된 데이터의 행 개수를 출력함.
print(f"법정동코드 LAWD_CD: {LAWD_CD_list[i]} | 데이터 수집 건수: {merge_df.shape}")
데이터 요청 간에 1초씩 대기 시간을 둠 (time.sleep(1)).
time.sleep(1)
except requests.exceptions.RequestException as e:
# 요청 실패 시 오류 메시지 출력
print(f"실패한 법정동코드 LAWD_CD: {LAWD_CD_list[i]} | 오류 내용: {str(e)}")
time.sleep(1)
5-7. 데이터 저장
모든 LAWD_CD가 처리된 후 누적된 DataFrame 저장
수집된 데이터를 저장할 디렉토리 경로를 설정함.
만약 해당 디렉토리가 존재하지 않으면 새로 생성함.
파일 이름은 apttrade{ymd}.csv 형식으로 설정되며, ymd는 함수에 전달된 년월 값임.
최종적으로 누적된 merge_df 데이터프레임을 CSV 파일로 저장함. 저장된 파일 경로를 출력함.
directory = path+'0.data/apt_trade'
if not os.path.exists(directory):
os.makedirs(directory)
file_name = f'apt_trade_{ymd}.csv'
file_path = os.path.join(directory, file_name) # 파일 경로 수정됨
merge_df.to_csv(file_path, index=False)
print(f"파일 저장 경로: {file_path}")6. 반복문 돌리기
for m in mo:
print(f"작업월 : {m}")
apt_extract_df(m)
