Aislo/B01_Dashboard/B01_Dashboard_Map.py

"""회사 주소 지도 — VWorld 주소검색·배경지도 타일 (2026-09-06 사용자 지시).

이미 쓰던 VWorld 키를 그대로 쓴다. 키가 화면으로 새지 않게 타일도 서버가 받아 넘긴다.
"""

from __future__ import annotations

import asyncio
import json
import urllib.parse
import urllib.request
from typing import Any

from B04_PreProcess.B04_PreProcess_Engine_VWorld import VWORLD_API_KEY

_GEOCODE_URL = "https://api.vworld.kr/req/address"
_TILE_URL = "http://api.vworld.kr/req/wmts/1.0.0/{key}/Base/{z}/{y}/{x}.png"


def _fetch(url: str, timeout: int = 5) -> bytes:
    request = urllib.request.Request(url, headers={"User-Agent": "Aislo/1.0"})
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return response.read()


def _geocode_sync(address: str) -> dict[str, Any] | None:
    """도로명으로 먼저 찾고, 없으면 지번으로 다시 찾는다."""
    for address_type in ("ROAD", "PARCEL"):
        query = urllib.parse.urlencode(
            {
                "service": "address",
                "request": "getcoord",
                "version": "2.0",
                "crs": "epsg:4326",
                "address": address,
                "refine": "true",
                "simple": "false",
                "format": "json",
                "type": address_type,
                "key": VWORLD_API_KEY,
            }
        )
        try:
            body = json.loads(_fetch(f"{_GEOCODE_URL}?{query}").decode("utf-8"))
        except Exception:
            continue
        point = (body.get("response") or {}).get("result", {}).get("point")
        if point:
            return {"lon": float(point["x"]), "lat": float(point["y"])}
    return None


async def geocode_address(address: str) -> dict[str, Any] | None:
    return await asyncio.to_thread(_geocode_sync, address.strip())


async def fetch_base_tile(z: int, x: int, y: int) -> bytes:
    url = _TILE_URL.format(key=VWORLD_API_KEY, z=z, y=y, x=x)
    return await asyncio.to_thread(_fetch, url)


_SEARCH_URL = "https://api.vworld.kr/req/search"


def _search_sync(query: str, size: int) -> list[dict[str, Any]]:
    """도로명으로 먼저 찾고, 비면 지번으로 다시 찾는다 — 후보를 목록으로 돌려준다."""
    for category in ("road", "parcel"):
        params = urllib.parse.urlencode(
            {
                "service": "search",
                "request": "search",
                "version": "2.0",
                "crs": "EPSG:4326",
                "size": str(size),
                "page": "1",
                "query": query,
                "type": "address",
                "category": category,
                "format": "json",
                "errorformat": "json",
                "key": VWORLD_API_KEY,
            }
        )
        try:
            body = json.loads(_fetch(f"{_SEARCH_URL}?{params}", timeout=8).decode("utf-8"))
        except Exception:
            continue
        result = (body.get("response") or {}).get("result") or {}
        items = result.get("items") or []
        found: list[dict[str, Any]] = []
        for item in items:
            address = item.get("address") or {}
            point = item.get("point") or {}
            try:
                lon = float(point["x"])
                lat = float(point["y"])
            except (KeyError, TypeError, ValueError):
                continue
            found.append(
                {
                    "zipcode": address.get("zipcode") or "",
                    "road": address.get("road") or "",
                    "parcel": address.get("parcel") or "",
                    "building": address.get("bldnm") or "",
                    "category": category,
                    "lat": lat,
                    "lon": lon,
                }
            )
        if found:
            return found
    return []


async def search_address(query: str, size: int = 10) -> list[dict[str, Any]]:
    return await asyncio.to_thread(_search_sync, query.strip(), size)


def _address_items(lon: float, lat: float) -> dict[str, dict[str, Any]]:
    """좌표 → VWorld getAddress 결과(종류 road · parcel → 한 칸) · 실패 = 빈 dict."""
    query = urllib.parse.urlencode(
        {
            "service": "address",
            "request": "getAddress",
            "version": "2.0",
            "crs": "epsg:4326",
            "point": f"{lon},{lat}",
            "type": "both",
            "format": "json",
            "key": VWORLD_API_KEY,
        }
    )
    try:
        body = json.loads(_fetch(f"{_GEOCODE_URL}?{query}").decode("utf-8"))
    except Exception:
        return {}
    items = (body.get("response") or {}).get("result") or []
    return {item.get("type"): item for item in items if isinstance(item, dict)}


def _reverse_sync(lon: float, lat: float) -> str | None:
    """좌표 → 주소 한 줄(도로명 먼저 · 없으면 지번)."""
    items = _address_items(lon, lat)
    return next(
        (items[k]["text"] for k in ("road", "parcel") if items.get(k, {}).get("text")), None
    )


def _place_sync(lon: float, lat: float) -> dict[str, str | None]:
    """좌표 → {address, region}(591-1 기준점) — 산은 지번 먼저 · region = 「시군구 읍면」."""
    items = _address_items(lon, lat)
    item = next((items[k] for k in ("parcel", "road") if items.get(k, {}).get("text")), None)
    if not item:
        return {"address": None, "region": None}
    part = item.get("structure") or {}
    town = "level4L" if item.get("type") == "parcel" else "level4A"  # 도로명은 4L = 길 이름
    region = " ".join(v for k in ("level2", "level3", town) if (v := part.get(k)))
    return {"address": item["text"], "region": region or None}


async def reverse_geocode(lon: float, lat: float) -> str | None:
    return await asyncio.to_thread(_reverse_sync, lon, lat)


async def reverse_place(lon: float, lat: float) -> dict[str, str | None]:
    return await asyncio.to_thread(_place_sync, lon, lat)