# -*- coding: utf-8 -*-
"""
부동산 실거래가 / 개별공시지가 수집 모듈
=========================================
국토교통부 공공데이터(data.go.kr) OpenAPI 를 호출해서
- 아파트 매매 / 전월세
- 연립·다세대 매매 / 전월세
- 단독·다가구 매매 / 전월세
- 오피스텔 매매 / 전월세
- 개별공시지가
를 받아 pandas DataFrame 으로 돌려줍니다.

인증키는 환경변수 DATA_GO_KR_KEY 또는 같은 폴더의 .env 파일에서 읽습니다.
(data.go.kr 마이페이지 > 인증키의 **일반 인증키(Decoding)** 값을 사용하세요)
"""

from __future__ import annotations

import json
import os
import re as _re
import time
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path

import requests
import pandas as pd

# ---------------------------------------------------------------------------
# 인증키 로딩
# ---------------------------------------------------------------------------

def _load_env_file() -> None:
    """같은 폴더의 .env 파일을 (있으면) 읽어 os.environ 에 넣는다. (의존성 없이 간단 파싱)"""
    env_path = Path(__file__).with_name(".env")
    if not env_path.exists():
        return
    for line in env_path.read_text(encoding="utf-8-sig").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, val = line.partition("=")
        os.environ.setdefault(key.strip(), val.strip().strip('"').strip("'"))


# 개인 비밀키 저장소(JSON). 환경변수 SECRET_JSON 으로 경로 변경 가능.
SECRET_JSON_DEFAULT = Path.home() / ".secret" / "ENVIRONMENT_VARIABLES.json"

# 이 JSON 안의 키 이름 -> 이 스크립트가 쓰는 이름
_SECRET_ALIASES = {
    "DATA_GO_KR_KEY": ["DATA_GO_KR_KEY", "DATA_GO_API_KEY"],
    "VWORLD_KEY": ["VWORLD_KEY", "VW_API_KEY", "VWORLD_API_KEY"],
}


def _load_secret_json() -> None:
    """~/.secret/ENVIRONMENT_VARIABLES.json 에서 '필요한 키만' 골라 os.environ 에 채운다.
    (다른 민감정보는 건드리지 않음)"""
    path = Path(os.environ.get("SECRET_JSON", str(SECRET_JSON_DEFAULT)))
    if not path.exists():
        return
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except Exception:
        return

    # 중첩 구조를 훑어 leaf 문자열만 이름->값 으로 평탄화 (필요한 이름만 관심)
    wanted = {name for names in _SECRET_ALIASES.values() for name in names}
    flat: dict[str, str] = {}

    def walk(o):
        if isinstance(o, dict):
            for k, v in o.items():
                if isinstance(v, (dict, list)):
                    walk(v)
                elif isinstance(v, str) and k in wanted:
                    flat.setdefault(k, v)
        elif isinstance(o, list):
            for v in o:
                walk(v)

    walk(data)

    for target, sources in _SECRET_ALIASES.items():
        if os.environ.get(target):
            continue
        for s in sources:
            if flat.get(s):
                os.environ[target] = flat[s]
                break


def _hydrate() -> None:
    """.env 파일 + 개인 비밀키 JSON 을 모두 반영."""
    _load_env_file()
    _load_secret_json()


def get_service_key() -> str:
    _hydrate()
    key = os.environ.get("DATA_GO_KR_KEY", "").strip()
    if not key:
        raise RuntimeError(
            "인증키가 없습니다. 환경변수 DATA_GO_KR_KEY 를 설정하거나 "
            "이 폴더에 .env 파일을 만들고 DATA_GO_KR_KEY=... 를 넣으세요.\n"
            "(data.go.kr 마이페이지의 '일반 인증키(Decoding)' 값)"
        )
    return key


def has_vworld_key() -> bool:
    _hydrate()
    return bool(os.environ.get("VWORLD_KEY", "").strip())


def get_vworld_key() -> str:
    _hydrate()
    key = os.environ.get("VWORLD_KEY", "").strip()
    if not key:
        raise RuntimeError(
            "VWorld 인증키가 없습니다. 주소→PNU 변환을 쓰려면 .env 에 VWORLD_KEY=... 를 넣으세요.\n"
            "(https://www.vworld.kr 에서 무료 발급, '오픈API 인증키')"
        )
    return key


# ---------------------------------------------------------------------------
# 실거래가 API 정의
# ---------------------------------------------------------------------------
# 모두 apis.data.go.kr/1613000/... 아래에 있음. 응답은 XML.

REALESTATE_BASE = "https://apis.data.go.kr/1613000"

# 표시이름 -> (서비스경로, 오퍼레이션명)
TRADE_SERVICES = {
    "아파트_매매":       ("RTMSDataSvcAptTradeDev", "getRTMSDataSvcAptTradeDev"),
    "아파트_전월세":     ("RTMSDataSvcAptRent",     "getRTMSDataSvcAptRent"),
    "연립다세대_매매":   ("RTMSDataSvcRHTrade",     "getRTMSDataSvcRHTrade"),
    "연립다세대_전월세": ("RTMSDataSvcRHRent",      "getRTMSDataSvcRHRent"),
    "단독다가구_매매":   ("RTMSDataSvcSHTrade",     "getRTMSDataSvcSHTrade"),
    "단독다가구_전월세": ("RTMSDataSvcSHRent",      "getRTMSDataSvcSHRent"),
    "오피스텔_매매":     ("RTMSDataSvcOffiTrade",   "getRTMSDataSvcOffiTrade"),
    "오피스텔_전월세":   ("RTMSDataSvcOffiRent",    "getRTMSDataSvcOffiRent"),
}


@dataclass
class ApiError(Exception):
    code: str
    message: str

    def __str__(self) -> str:
        return f"[{self.code}] {self.message}"


def _xml_items_to_rows(xml_text: str) -> tuple[list[dict], str, int]:
    """실거래가 XML 응답을 파싱해서 (행 리스트, resultCode, totalCount) 반환."""
    root = ET.fromstring(xml_text)

    result_code = (root.findtext(".//resultCode") or "").strip()
    result_msg = (root.findtext(".//resultMsg") or "").strip()
    if result_code and result_code not in ("00", "000"):
        raise ApiError(result_code, result_msg or "알 수 없는 오류")

    total = int((root.findtext(".//totalCount") or "0").strip() or 0)

    rows: list[dict] = []
    for item in root.findall(".//items/item"):
        row = {child.tag: (child.text.strip() if child.text else "") for child in item}
        rows.append(row)
    return rows, result_code, total


def fetch_trade(
    service_name: str,
    lawd_cd: str,
    deal_ymd: str,
    *,
    num_of_rows: int = 1000,
    max_pages: int = 100,
    sleep_sec: float = 0.2,
    session: requests.Session | None = None,
) -> pd.DataFrame:
    """
    실거래가 한 종류를, 한 지역(시군구) x 한 계약월 만큼 전부 받아 DataFrame 으로.

    service_name : TRADE_SERVICES 의 키 (예: '아파트_매매')
    lawd_cd      : 법정동코드 앞 5자리 (시군구 코드, 예: 서울 강남구 '11680')
    deal_ymd     : 계약연월 'YYYYMM' (예: '202406')
    """
    if service_name not in TRADE_SERVICES:
        raise ValueError(f"알 수 없는 서비스: {service_name}. 가능: {list(TRADE_SERVICES)}")

    path, operation = TRADE_SERVICES[service_name]
    url = f"{REALESTATE_BASE}/{path}/{operation}"
    key = get_service_key()
    sess = session or requests.Session()

    all_rows: list[dict] = []
    page = 1
    while page <= max_pages:
        params = {
            "serviceKey": key,
            "LAWD_CD": lawd_cd,
            "DEAL_YMD": deal_ymd,
            "pageNo": page,
            "numOfRows": num_of_rows,
        }
        resp = sess.get(url, params=params, timeout=30)

        # data.go.kr 게이트웨이: 401=키 미인식, 403=이 API 활용신청/승인 안 됨
        if resp.status_code == 401:
            raise ApiError("401", "인증키를 인식하지 못함 (DATA_GO_KR_KEY 확인)")
        if resp.status_code == 403:
            raise ApiError("403", f"이 API 활용신청/승인이 안 됨 → data.go.kr 에서 '{service_name}' 활용신청 필요")
        resp.raise_for_status()

        # 인증키 오류 등은 XML 이 아닌 형태로 오기도 함 -> 방어적 처리
        text = resp.text
        if not text.lstrip().startswith("<"):
            raise ApiError("NON_XML", text[:300])

        rows, _code, total = _xml_items_to_rows(text)
        all_rows.extend(rows)

        if page * num_of_rows >= total or not rows:
            break
        page += 1
        time.sleep(sleep_sec)

    df = pd.DataFrame(all_rows)
    if not df.empty:
        df.insert(0, "조회_계약월", deal_ymd)
        df.insert(0, "조회_지역코드", lawd_cd)
        df.insert(0, "자료구분", service_name)
    return df


def fetch_trades(
    service_names: list[str],
    lawd_cds: list[str],
    deal_ymds: list[str],
    **kwargs,
) -> pd.DataFrame:
    """여러 종류 x 여러 지역 x 여러 월을 한꺼번에 받아 하나의 DataFrame 으로 합침."""
    session = requests.Session()
    frames: list[pd.DataFrame] = []
    for name in service_names:
        for lawd in lawd_cds:
            for ymd in deal_ymds:
                try:
                    df = fetch_trade(name, lawd, ymd, session=session, **kwargs)
                    print(f"  {name} | 지역 {lawd} | {ymd} -> {len(df)}건")
                    frames.append(df)
                except ApiError as e:
                    print(f"  [오류] {name} | {lawd} | {ymd}: {e}")
                except requests.RequestException as e:
                    print(f"  [통신오류] {name} | {lawd} | {ymd}: {e}")
    if not frames:
        return pd.DataFrame()
    return pd.concat(frames, ignore_index=True)


# ---------------------------------------------------------------------------
# 주소 -> PNU 변환 (VWorld)
# ---------------------------------------------------------------------------
# PNU(19) = 법정동코드(10) + 대장구분(1: 토지=1/임야=2) + 본번(4) + 부번(4)

VWORLD_GEOCODE_URL = "https://api.vworld.kr/req/address"
VWORLD_DATA_URL = "https://api.vworld.kr/req/data"


def _parse_jibun(text: str) -> tuple[bool, int, int] | None:
    """주소 문자열 끝의 지번을 파싱 -> (임야여부, 본번, 부번). 실패 시 None."""
    # '산123-4', '산 123', '123-45', '123번지' 등에서 맨 끝 지번을 추출
    m = _re.search(r"(산)?\s*(\d+)(?:-(\d+))?\s*(?:번지)?\s*$", text.strip())
    if not m:
        return None
    is_san = bool(m.group(1))
    bonbun = int(m.group(2))
    bubun = int(m.group(3)) if m.group(3) else 0
    return is_san, bonbun, bubun


def _build_pnu(lawd10: str, is_san: bool, bonbun: int, bubun: int) -> str:
    daejang = "2" if is_san else "1"
    return f"{lawd10}{daejang}{bonbun:04d}{bubun:04d}"


def geocode(address: str, addr_type: str = "PARCEL", *, session: requests.Session | None = None) -> dict | None:
    """VWorld 지오코더로 주소 -> {x(경도), y(위도), lawd10(법정동코드), refined(정제주소)}."""
    sess = session or requests.Session()
    params = {
        "service": "address",
        "request": "getcoord",
        "version": "2.0",
        "crs": "EPSG:4326",
        "address": address,
        "type": addr_type,          # PARCEL(지번) / ROAD(도로명)
        "refine": "true",
        "format": "json",
        "key": get_vworld_key(),
    }
    r = sess.get(VWORLD_GEOCODE_URL, params=params, timeout=30)
    r.raise_for_status()
    data = r.json().get("response", {})
    if data.get("status") != "OK":
        return None
    point = data.get("result", {}).get("point", {})
    structure = data.get("refined", {}).get("structure", {})
    return {
        "x": point.get("x"),
        "y": point.get("y"),
        "lawd10": structure.get("level4LC", ""),
        "refined": data.get("refined", {}).get("text", ""),
    }


def parcel_pnu_by_point(x: str, y: str, *, session: requests.Session | None = None) -> str | None:
    """좌표(경도 x, 위도 y)가 속한 필지의 PNU 를 지적도에서 조회."""
    sess = session or requests.Session()
    params = {
        "service": "data",
        "request": "GetFeature",
        "data": "LP_PA_CBND_BUBUN",     # 연속지적도(부번 필지)
        "geomFilter": f"POINT({x} {y})",
        "crs": "EPSG:4326",
        "columns": "pnu,jibun",
        "geometry": "false",
        "size": "10",
        "format": "json",
        "key": get_vworld_key(),
    }
    r = sess.get(VWORLD_DATA_URL, params=params, timeout=30)
    r.raise_for_status()
    resp = r.json().get("response", {})
    if resp.get("status") != "OK":
        return None
    feats = resp.get("result", {}).get("featureCollection", {}).get("features", [])
    if not feats:
        return None
    return feats[0].get("properties", {}).get("pnu")


def address_to_pnu(address: str, *, session: requests.Session | None = None) -> dict:
    """
    지번/도로명 주소 -> PNU. 반환: {'address','pnu','method','refined','x','y'}.
    실패 시 pnu=None.
      1) 지오코딩(지번, 실패 시 도로명)
      2) 지적도에서 정확 PNU 조회
      3) 실패하면 법정동코드+지번 파싱으로 PNU 조립(폴백)
    """
    sess = session or requests.Session()
    g = geocode(address, "PARCEL", session=sess) or geocode(address, "ROAD", session=sess)
    if not g or not g.get("x"):
        return {"address": address, "pnu": None, "method": "geocode_fail",
                "refined": "", "x": None, "y": None}

    pnu = parcel_pnu_by_point(g["x"], g["y"], session=sess)
    method = "parcel"

    if not pnu and g.get("lawd10"):
        parsed = _parse_jibun(g.get("refined") or address)
        if parsed:
            pnu = _build_pnu(g["lawd10"], *parsed)
            method = "constructed"

    return {"address": address, "pnu": pnu, "method": method if pnu else "no_pnu",
            "refined": g.get("refined", ""), "x": g.get("x"), "y": g.get("y")}


def addresses_to_pnu(addresses: list[str], *, sleep_sec: float = 0.15) -> pd.DataFrame:
    """여러 주소 -> PNU 표. (PNU 조회 결과 확인용)"""
    sess = requests.Session()
    rows = []
    for addr in addresses:
        try:
            info = address_to_pnu(addr, session=sess)
        except requests.RequestException as e:
            info = {"address": addr, "pnu": None, "method": f"error:{e}", "refined": "", "x": None, "y": None}
        print(f"  {addr}  ->  {info['pnu']}  ({info['method']})")
        rows.append(info)
        time.sleep(sleep_sec)
    return pd.DataFrame(rows)


# ---------------------------------------------------------------------------
# 개별공시지가 API
# ---------------------------------------------------------------------------
# 국토교통부_개별공시지가 정보서비스 (NSDI, 1611000)
# 필지고유번호(PNU, 19자리)와 기준연도로 조회.

LANDPRICE_URL = (
    "https://apis.data.go.kr/1611000/nsdi/IndvdLandPriceService/attr/getIndvdLandPriceAttr"
)


def fetch_land_price(
    pnu: str,
    stdr_year: str | int,
    *,
    num_of_rows: int = 100,
    session: requests.Session | None = None,
) -> pd.DataFrame:
    """
    개별공시지가 조회.

    pnu       : 필지고유번호 19자리 (법정동코드10 + 대장구분1 + 본번4 + 부번4)
    stdr_year : 기준연도 (예: 2024). 연도별 공시지가가 여러 행으로 올 수 있음.
    """
    key = get_service_key()
    sess = session or requests.Session()
    params = {
        "serviceKey": key,
        "pnu": pnu,
        "stdrYear": str(stdr_year),
        "format": "xml",
        "numOfRows": num_of_rows,
        "pageNo": 1,
    }
    resp = sess.get(LANDPRICE_URL, params=params, timeout=30)
    resp.raise_for_status()

    text = resp.text
    if not text.lstrip().startswith("<"):
        raise ApiError("NON_XML", text[:300])

    root = ET.fromstring(text)
    # 이 서비스는 result / resultCode 구조가 조금 다름 -> field 여러개 방어적으로
    code = (root.findtext(".//resultCode") or root.findtext(".//RESULT_CODE") or "").strip()
    msg = (root.findtext(".//resultMsg") or root.findtext(".//RESULT_MSG") or "").strip()
    if code and code not in ("00", "000", "NORMAL_SERVICE", "0"):
        raise ApiError(code, msg or "공시지가 조회 오류")

    rows: list[dict] = []
    # 항목 태그가 <field> 로 오는 경우가 많음
    for item in root.findall(".//field"):
        rows.append({child.tag: (child.text.strip() if child.text else "") for child in item})
    if not rows:  # 혹시 items/item 구조면
        for item in root.findall(".//items/item"):
            rows.append({child.tag: (child.text.strip() if child.text else "") for child in item})

    df = pd.DataFrame(rows)
    if not df.empty:
        df.insert(0, "조회_기준연도", str(stdr_year))
        df.insert(0, "조회_PNU", pnu)
    return df


def fetch_land_prices(
    pnus: list[str],
    stdr_year: str | int,
    *,
    sleep_sec: float = 0.2,
) -> pd.DataFrame:
    """여러 필지의 개별공시지가를 한 번에."""
    session = requests.Session()
    frames: list[pd.DataFrame] = []
    for pnu in pnus:
        try:
            df = fetch_land_price(pnu, stdr_year, session=session)
            print(f"  PNU {pnu} | {stdr_year} -> {len(df)}건")
            frames.append(df)
        except ApiError as e:
            print(f"  [오류] PNU {pnu}: {e}")
        except requests.RequestException as e:
            print(f"  [통신오류] PNU {pnu}: {e}")
        time.sleep(sleep_sec)
    if not frames:
        return pd.DataFrame()
    return pd.concat(frames, ignore_index=True)


# ---------------------------------------------------------------------------
# 저장 유틸
# ---------------------------------------------------------------------------

def save(df: pd.DataFrame, path: str | Path) -> None:
    """확장자에 따라 xlsx / csv 저장. (csv 는 엑셀 한글깨짐 방지용 utf-8-sig)"""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    if df.empty:
        print(f"  (저장 안 함: 데이터 0건) {path}")
        return
    if path.suffix.lower() in (".xlsx", ".xls"):
        df.to_excel(path, index=False)
    else:
        df.to_csv(path, index=False, encoding="utf-8-sig")
    print(f"  저장 완료: {path}  ({len(df)}건)")


def month_range(start_ym: str, end_ym: str) -> list[str]:
    """'202401', '202406' -> ['202401', ..., '202406']  (계약월 리스트 생성 도우미)"""
    start = pd.Period(start_ym[:4] + "-" + start_ym[4:6], freq="M")
    end = pd.Period(end_ym[:4] + "-" + end_ym[4:6], freq="M")
    out = []
    p = start
    while p <= end:
        out.append(f"{p.year}{p.month:02d}")
        p += 1
    return out