不動産実売価格データのクローリングガイド — アパート・オフィステルの市場価格を自動収集

import requests
import xml.etree.ElementTree as ET
import pandas as pd

# API 설정 — 디코딩된 키를 사용하세요
SERVICE_KEY = "발급받은_디코딩_API_키"
BASE_URL = "http://openapi.molit.go.kr/OpenAPI_ToolInstall/service/rest/RTMSOBJSvc/getRTMSDataSvcAptTrade"

def get_apt_trade(lawd_cd: str, deal_ymd: str) -> pd.DataFrame:
    """
    아파트 매매 실거래가를 조회합니다.

    Args:
        lawd_cd: 법정동 앞 5자리 코드 (예: "11680" = 서울 강남구)
        deal_ymd: 계약년월 (예: "202601")

    Returns:
        거래 데이터가 담긴 DataFrame
    """
    params = {
        "serviceKey": SERVICE_KEY,
        "LAWD_CD": lawd_cd,
        "DEAL_YMD": deal_ymd,
        "numOfRows": "9999",
        "pageNo": "1"
    }

    response = requests.get(BASE_URL, params=params, timeout=30)
    response.raise_for_status()

    root = ET.fromstring(response.text)

    # 에러 체크
    result_code = root.find(".//resultCode")
    if result_code is not None and result_code.text != "00":
        result_msg = root.find(".//resultMsg")
        raise Exception(f"API 오류: {result_msg.text if result_msg is not None else 'Unknown'}")

    items = root.findall(".//item")
    if not items:
        return pd.DataFrame()

    data = []
    for item in items:
        row = {}
        for child in item:
            text = child.text.strip() if child.text else ""
            row[child.tag] = text
        data.append(row)

    return pd.DataFrame(data)


# 사용 예시: 서울 강남구(11680) 2026년 1월 거래 조회
df = get_apt_trade("11680", "202601")
print(f"조회된 거래 수: {len(df)}")
if not df.empty:
    # 거래금액 정수 변환 (쉼표와 공백 제거)
    df["거래금액_만원"] = df["거래금액"].str.replace(",", "").str.strip().astype(int)
    print(df[["거래금액_만원", "아파트", "전용면적", "층"]].head(10))

import time

# 서울시 구별 법정동 앞 5자리 코드
SEOUL_CODES = {
    "강남구": "11680", "서초구": "11650", "송파구": "11710",
    "강동구": "11740", "마포구": "11440", "용산구": "11170",
    "성동구": "11200", "광진구": "11215", "중구": "11140",
    "종로구": "11110", "영등포구": "11560", "동작구": "11590",
    "관악구": "11620", "강서구": "11500", "양천구": "11470",
    "구로구": "11530", "금천구": "11545", "노원구": "11350",
    "도봉구": "11320", "강북구": "11305", "성북구": "11290",
    "동대문구": "11230", "중랑구": "11260", "은평구": "11380",
    "서대문구": "11410"
}

def collect_seoul_yearly(year: int) -> pd.DataFrame:
    """서울시 전체 구의 1년치 아파트 매매 데이터를 수집합니다."""
    all_data = []
    months = [f"{year}{m:02d}" for m in range(1, 13)]
    total = len(SEOUL_CODES) * len(months)
    count = 0

    for gu_name, code in SEOUL_CODES.items():
        for month in months:
            count += 1
            try:
                df = get_apt_trade(code, month)
                if not df.empty:
                    df["구"] = gu_name
                    all_data.append(df)
                print(f"[{count}/{total}]  {gu_name} {month}: {len(df)}건")
            except Exception as e:
                print(f"[{count}/{total}]  {gu_name} {month}: {e}")
            time.sleep(0.3)  # API 호출 간격

    if all_data:
        result = pd.concat(all_data, ignore_index=True)
        return result
    return pd.DataFrame()

# 2025년 전체 데이터 수집
result = collect_seoul_yearly(2025)
result.to_csv("서울_아파트_실거래가_2025.csv", index=False, encoding="utf-8-sig")
print(f"\n총 {len(result)}건 수집 완료!")

# 단지 기본 정보
GET https://fin.land.naver.com/complexes/{complexNo}

# 단지 매물 목록
GET https://fin.land.naver.com/complexes/{complexNo}/articles

# 단지 실거래가
GET https://fin.land.naver.com/complexes/{complexNo}/real-estates/trades

# 지역별 단지 목록
GET https://fin.land.naver.com/front-api/v1/complex/marker

import requests
import json
import time

def get_naver_complex(complex_no: str) -> dict | None:
    """
    네이버 부동산에서 아파트 단지 정보를 조회합니다.

    주의: 네이버 내부 API는 수시로 변경될 수 있습니다.
    이 코드가 동작하지 않으면 브라우저 개발자 도구에서
    최신 API 엔드포인트를 확인하세요.
    """
    url = f"https://fin.land.naver.com/complexes/{complex_no}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                       "AppleWebKit/537.36 (KHTML, like Gecko) "
                       "Chrome/120.0.0.0 Safari/537.36",
        "Referer": "https://land.naver.com/",
        "Accept": "application/json"
    }

    try:
        response = requests.get(url, headers=headers, timeout=10)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"HTTP {response.status_code}: {complex_no}")
            return None
    except requests.exceptions.JSONDecodeError:
        # HTML이 반환된 경우 (API 변경 가능성)
        print(f"JSON 파싱 실패 — API 엔드포인트가 변경되었을 수 있습니다")
        return None
    except requests.exceptions.RequestException as e:
        print(f"요청 실패: {e}")
        return None