도로명 주소 API를 이용해서 도로명 주소를 구주소 변환 파이썬
by 개발자   2025-10-21 15:40:07   조회수:13
import pandas as pd
import requests
import time
import json
import os

# === 설정 ===
# 중요: 아래는 개발 테스트용 샘플 키입니다. 반드시 실제 발급받은 인증키로 변경하세요!
#      이 키를 그대로 사용하시면 '승인되지 않은 KEY' 에러가 발생합니다.
API_KEY = "devU01TX0FVVEgyMDI1MTAyMTEzNDcwNjExNjM0ODU="
INPUT_FILE = "부천시고객회원_주소.xlsx"
OUTPUT_FILE = "부천시고객회원_주소_지번변환_결과a.csv"

def get_jibun_address(road_addr, retries=3):
    """
    (수정) 주소 검색 API를 호출하여 지번 주소, 우편번호, 시군구, 읍면동을 추출합니다.
    검색 실패 시 검색어 마지막 단어를 제거하며 최대 10회 재시도합니다.
    """
    # 반환할 기본값
    default_result = {'jibunAddr': None, 'zipNo': None, 'sggNm': None, 'emdNm': None}
   
    current_keyword = str(road_addr).strip()

    for attempt in range(10): # 최대 10회 재시도
        if not current_keyword:
            # print("  [검색 중단] 검색어가 비어있습니다.")
            break

        print(f"  [시도 {attempt+1}/10] 검색어: '{current_keyword}'")
       
        # --- 주소 검색 API 호출 (addrLinkApi) ---
        search_url = "https://www.juso.go.kr/addrlink/addrLinkApiJsonp.do"
        search_params = {
            "confmKey": API_KEY, "currentPage": 1, "countPerPage": 1,
            "keyword": current_keyword, "resultType": "json"
        }

        try:
            search_response = requests.get(search_url, params=search_params, timeout=10)
           
            # <<<--- 요청 결과 원본 텍스트를 콘솔에 출력 (요청 사항) --- START
            print(f"  [API 응답 원문]\n{search_response.text}")
            # <<<--- 요청 결과 원본 텍스트를 콘솔에 출력 (요청 사항) --- END

            search_response.raise_for_status()
            search_text = search_response.text.strip()
           
            if not search_text or search_text == "()":
                print(f"  [검색 실패] API로부터 빈 응답 수신. (API 키가 유효한지 확인하세요)")
                return default_result
               
            if search_text.startswith("(") and search_text.endswith(")"):
                search_text = search_text[1:-1]
           
            search_data = json.loads(search_text)

            if search_data['results']['common']['errorCode'] != '0':
                error_message = search_data['results']['common']['errorMessage']
                print(f"  [검색 실패] API 오류: {error_message}")
                # API 자체 오류 시에는 재시도 의미 없으므로 바로 반환
                return default_result
           
            if search_data['results']['common']['totalCount'] != '0':
                # 정상 결과에서 정보 추출
                juso_info = search_data['results']['juso'][0]
                jibun_addr = juso_info.get('jibunAddr')
                zip_no = juso_info.get('zipNo')
                sgg_nm = juso_info.get('sggNm') # 시군구 정보 추출
                emd_nm = juso_info.get('emdNm') # 읍면동 정보 추출

                print(f"  → 변환 성공: {jibun_addr} (우편번호: {zip_no}, 시군구: {sgg_nm}, 읍면동: {emd_nm})")
               
                # 시군구와 읍면동을 각각 별개의 값으로 반환
                return {'jibunAddr': jibun_addr, 'zipNo': zip_no, 'sggNm': sgg_nm, 'emdNm': emd_nm}

            # 검색 결과가 없는 경우, 다음 시도를 위해 검색어 조정
            print(f"  → 결과 없음. 검색어를 조정하여 재시도합니다.")
            keyword_parts = current_keyword.split()
            if len(keyword_parts) > 1:
                current_keyword = " ".join(keyword_parts[:-1])
            else:
                # 단어가 하나만 남았으면 더 이상 줄일 수 없으므로 중단
                break

        except json.JSONDecodeError as e:
            print(f"  [에러] API 응답을 분석할 수 없습니다 (JSON 형식 오류): {e}")
            return default_result
        except requests.exceptions.RequestException as e:
            print(f"  [에러] 네트워크 또는 HTTP 오류: {e}")
            if retries > 0:
                print("  1초 후 재시도합니다...")
                time.sleep(1)
                # 네트워크 오류 시에는 현재 검색어로 재시도
                # 이 부분은 외부 함수의 재귀 호출 대신 간단화된 형태로 유지
            return default_result # 네트워크 오류 시 재시도 후에도 실패하면 기본값 반환
        except Exception as e:
            print(f"  [알 수 없는 에러] '{current_keyword}': {e}")
            return default_result

    print(f"  [최종 검색 실패] '{road_addr}'에 대한 유효한 주소를 찾지 못했습니다.")
    return default_result

# === 메인 실행 로직 ===
try:
    df = pd.read_excel(INPUT_FILE, engine='openpyxl')
except FileNotFoundError:
    print(f"오류: 입력 파일 '{INPUT_FILE}'을 찾을 수 없습니다.")
    exit()
except Exception as e:
    print(f"입력 파일 '{INPUT_FILE}'을 읽는 중 오류가 발생했습니다: {e}")
    exit()

if 'Address' not in df.columns:
    print("오류: 입력 파일의 컬럼 중 'Address'를 찾을 수 없습니다.")
    exit()

print("주소 변환 작업 시작...")
total_addresses = len(df)
SLEEP_TIME = 1

# 최종 출력될 파일의 컬럼 정의 ('시군구', '읍면동'이 별도로 포함됨)
output_columns = list(df.columns) + ['우편번호', 'Address_Jibun', '시군구', '읍면동']

# 결과 파일이 없거나 비어있으면 헤더(컬럼명)를 포함하여 새로 생성
if not os.path.exists(OUTPUT_FILE) or os.path.getsize(OUTPUT_FILE) == 0:
    pd.DataFrame(columns=output_columns).to_csv(OUTPUT_FILE, index=False, mode='w', encoding='utf-8-sig')

# 데이터프레임의 각 행을 순회하며 주소 변환 실행
for i, row in df.iterrows():
    addr = row.get('Address')
   
    current_row_df = pd.DataFrame([row])
   
    if pd.isna(addr) or not str(addr).strip():
        print(f"[{i+1}/{total_addresses}] 빈 주소 건너뛰기")
        details = get_jibun_address("")
    else:
        addr_str = str(addr)
        print(f"[{i+1}/{total_addresses}] 조회 시작: {addr_str}")
        details = get_jibun_address(addr_str)
   
    # 변환된 결과를 새로운 컬럼에 추가
    current_row_df['우편번호'] = details['zipNo']
    current_row_df['Address_Jibun'] = details['jibunAddr']
    # '시군구' 컬럼에 API로부터 받은 'sggNm' 값을 할당
    current_row_df['시군구'] = details['sggNm']
    # '읍면동' 컬럼에 API로부터 받은 'emdNm' 값을 할당
    current_row_df['읍면동'] = details['emdNm']
   
    # 처리된 한 행의 결과를 CSV 파일에 이어서 쓰기 (append 모드)
    # output_columns 순서에 따라 '시군구', '읍면동'이 별도 컬럼으로 저장됨
    current_row_df.reindex(columns=output_columns).to_csv(
        OUTPUT_FILE, index=False, mode='a', header=False, encoding='utf-8-sig'
    )

    time.sleep(SLEEP_TIME)

print(f"\n변환 완료! 결과 저장: {OUTPUT_FILE}")