Mina Choi
11d30bb3d1
여러 줄 주석이 설명보다 경위(예전·실측·지적)를 적고 있어 읽는 사람이 결론을 찾기 어려웠다. - ts·tsx·js·mjs·css·py 478개: 여러 줄 주석은 첫 문장 한 줄로, 과거형·날짜 문장은 삭제 - 주석 위치는 TypeScript 파서·파이썬 tokenize/ast 로 찾는다 — 문자열 안의 # · /* 는 건드리지 않는다 - eslint·ts·noqa·type: ignore 같은 지시 주석은 그대로 둔다 파이썬 275개 정리 전후 AST 동일, TS 298개 주석 뺀 토큰 동일(빈 JSX 주석 10곳만 차이). site·frontend·admin tsc, site vitest 105 passed Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
367 lines
14 KiB
Python
367 lines
14 KiB
Python
"""카카오 로컬 API 클라이언트 — 동일 업소 검증 + 지역 정보."""
|
|
|
|
import re
|
|
from collections import Counter
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
from common.logger import LOG
|
|
from config.server_configs import external_api_config
|
|
|
|
_BASE_URL = "https://dapi.kakao.com"
|
|
_KEYWORD_URL = f"{_BASE_URL}/v2/local/search/keyword.json"
|
|
_COORD2REGION_URL = f"{_BASE_URL}/v2/local/geo/coord2regioncode.json"
|
|
_CATEGORY_URL = f"{_BASE_URL}/v2/local/search/category.json"
|
|
_ADDRESS_URL = f"{_BASE_URL}/v2/local/search/address.json"
|
|
|
|
# 초과 단가(원).
|
|
_UNIT_COST_KRW = {"keyword": 2.0, "category": 2.0, "coord2region": 0.5, "address": 0.5}
|
|
|
|
# 프로세스 누적 호출 수 — 생성 1건당 검색 횟수를 세는 근거.
|
|
_CALL_COUNTS: Counter = Counter()
|
|
|
|
# 카테고리 그룹 코드(카카오 정의).
|
|
CATEGORY_RESTAURANT = "FD6" # 음식점
|
|
CATEGORY_CAFE = "CE7" # 카페
|
|
CATEGORY_ATTRACTION = "AT4" # 관광명소
|
|
CATEGORY_ACCOMMODATION = "AD5" # 숙박
|
|
CATEGORY_CONVENIENCE = "CS2" # 편의점
|
|
CATEGORY_PARKING = "PK6" # 주차장
|
|
|
|
|
|
class KakaoNotConfigured(RuntimeError):
|
|
"""KAKAO_REST_API_KEY 미설정 — 이 어댑터만 비활성이다."""
|
|
|
|
|
|
class KakaoRequestFailed(RuntimeError):
|
|
"""카카오 로컬 호출 실패(네트워크·타임아웃·5xx·인증오류)."""
|
|
|
|
|
|
class MatchOutcome(str, Enum):
|
|
"""동일 업소 판정 결과."""
|
|
|
|
MATCHED = "matched" # 이 가게가 맞다고 확정
|
|
AMBIGUOUS = "ambiguous" # 동명 업소 등 — 사람이 골라야 한다
|
|
NO_CANDIDATE = "no_candidate" # 카카오에서 후보를 못 찾음
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class KakaoPlace:
|
|
"""카카오 로컬이 돌려준 장소 1건."""
|
|
|
|
kakao_place_id: str # id — 동일 업소 판정의 유일 키
|
|
name: str # place_name
|
|
road_address: Optional[str] # road_address_name
|
|
address: Optional[str] # address_name (지번)
|
|
phone: Optional[str]
|
|
latitude: Optional[float] # y
|
|
longitude: Optional[float] # x
|
|
category_name: Optional[str]
|
|
# 업종 자동 판별의 입력.
|
|
category_group_code: Optional[str]
|
|
place_url: Optional[str]
|
|
|
|
@classmethod
|
|
def from_document(cls, doc: dict) -> "KakaoPlace":
|
|
return cls(
|
|
kakao_place_id=str(doc.get("id") or ""),
|
|
name=doc.get("place_name") or "",
|
|
road_address=(doc.get("road_address_name") or None),
|
|
address=(doc.get("address_name") or None),
|
|
phone=(doc.get("phone") or None),
|
|
latitude=_to_float(doc.get("y")), # y = 위도
|
|
longitude=_to_float(doc.get("x")), # x = 경도
|
|
category_name=(doc.get("category_name") or None),
|
|
category_group_code=(doc.get("category_group_code") or None),
|
|
place_url=(doc.get("place_url") or None),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RegionCode:
|
|
"""행정구역 코드."""
|
|
|
|
code: str # 행정동/법정동 코드
|
|
region_1depth_name: str # 시·도
|
|
region_2depth_name: str # 시·군·구
|
|
region_3depth_name: str # 읍·면·동
|
|
region_type: Optional[str] = None # "H"=행정동, "B"=법정동
|
|
|
|
@property
|
|
def full_name(self) -> str:
|
|
parts = [self.region_1depth_name, self.region_2depth_name, self.region_3depth_name]
|
|
return " ".join(p for p in parts if p)
|
|
|
|
@classmethod
|
|
def from_document(cls, doc: dict) -> "RegionCode":
|
|
return cls(
|
|
code=str(doc.get("code") or ""),
|
|
region_1depth_name=doc.get("region_1depth_name") or "",
|
|
region_2depth_name=doc.get("region_2depth_name") or "",
|
|
region_3depth_name=doc.get("region_3depth_name") or "",
|
|
region_type=doc.get("region_type"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class MatchResult:
|
|
"""동일 업소 판정 결과 + **판정 근거**."""
|
|
|
|
outcome: MatchOutcome
|
|
place: Optional[KakaoPlace] = None # MATCHED 일 때만 채워진다
|
|
candidates: list[KakaoPlace] = field(default_factory=list) # 사람이 고를 후보 전체
|
|
reason: str = "" # 근거 코드(기계 판독용)
|
|
detail: str = "" # 근거 설명(사람 판독용)
|
|
|
|
@property
|
|
def is_matched(self) -> bool:
|
|
return self.outcome == MatchOutcome.MATCHED
|
|
|
|
|
|
# 문자열 정규화
|
|
_WS_RE = re.compile(r"\s+")
|
|
_DIGIT_RE = re.compile(r"\D")
|
|
|
|
|
|
def normalize_name(name: str) -> str:
|
|
"""상호명 비교용 정규화 — 공백 제거 + 소문자화."""
|
|
return _WS_RE.sub("", (name or "")).lower()
|
|
|
|
|
|
def normalize_phone(phone: str) -> str:
|
|
"""전화번호 비교용 정규화 — 숫자만 남긴다('033-672-0000' → '0336720000')."""
|
|
return _DIGIT_RE.sub("", phone or "")
|
|
|
|
|
|
def _to_float(value) -> Optional[float]:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
# 동일 업소 판정
|
|
def pick_match(name: str, candidates: list[KakaoPlace], phone: Optional[str] = None) -> MatchResult:
|
|
"""검색 결과에서 '이 가게가 맞다' 를 판정한다."""
|
|
if not candidates:
|
|
return MatchResult(
|
|
outcome=MatchOutcome.NO_CANDIDATE,
|
|
candidates=[],
|
|
reason="no_candidate",
|
|
detail=f"'{name}' 으로 카카오 로컬에서 후보를 찾지 못했습니다.",
|
|
)
|
|
|
|
pool = list(candidates)
|
|
phone_note = ""
|
|
|
|
# 2) 전화번호 — 동명 업소를 가르는 가장 강한 근거.
|
|
normalized_phone = normalize_phone(phone) if phone else ""
|
|
if normalized_phone:
|
|
phone_hits = [c for c in pool if normalize_phone(c.phone or "") == normalized_phone]
|
|
if len(phone_hits) == 1:
|
|
return MatchResult(
|
|
outcome=MatchOutcome.MATCHED,
|
|
place=phone_hits[0],
|
|
candidates=pool,
|
|
reason="phone_exact",
|
|
detail=f"전화번호({phone})가 정확히 1건과 일치합니다: {phone_hits[0].name}",
|
|
)
|
|
if len(phone_hits) > 1:
|
|
# 전화번호까지 같은 후보가 여럿 — 지점 등록 등.
|
|
pool = phone_hits
|
|
phone_note = f" (전화번호 일치 {len(phone_hits)}건으로 좁힘)"
|
|
|
|
# 3~5) 상호명 정확 일치
|
|
target = normalize_name(name)
|
|
name_hits = [c for c in pool if normalize_name(c.name) == target]
|
|
|
|
if len(name_hits) == 1:
|
|
return MatchResult(
|
|
outcome=MatchOutcome.MATCHED,
|
|
place=name_hits[0],
|
|
candidates=pool,
|
|
reason="name_exact",
|
|
detail=f"상호명이 정확히 1건과 일치합니다{phone_note}: {name_hits[0].name}",
|
|
)
|
|
|
|
if len(name_hits) > 1:
|
|
return MatchResult(
|
|
outcome=MatchOutcome.AMBIGUOUS,
|
|
candidates=name_hits,
|
|
reason="name_duplicate",
|
|
detail=(
|
|
f"'{name}' 과 상호명이 같은 업소가 {len(name_hits)}건입니다{phone_note}. "
|
|
f"주소로 구분해야 합니다: "
|
|
+ " / ".join(f"{c.name}({c.road_address or c.address or '주소없음'})" for c in name_hits)
|
|
),
|
|
)
|
|
|
|
return MatchResult(
|
|
outcome=MatchOutcome.AMBIGUOUS,
|
|
candidates=pool,
|
|
reason="name_no_exact",
|
|
detail=(
|
|
f"'{name}' 과 정확히 일치하는 상호명이 없습니다{phone_note}. 후보 {len(pool)}건 중 사람이 골라야 합니다: "
|
|
+ " / ".join(f"{c.name}({c.road_address or c.address or '주소없음'})" for c in pool[:5])
|
|
),
|
|
)
|
|
|
|
|
|
# 클라이언트
|
|
class KakaoLocalClient:
|
|
"""카카오 로컬 API 호출기."""
|
|
|
|
def __init__(self, api_key: Optional[str] = None, transport=None, timeout: float = 10.0):
|
|
# api_key 를 명시하지 않으면 설정에서 읽는다.
|
|
self._api_key = external_api_config.kakao_rest_api_key if api_key is None else api_key
|
|
self._transport = transport
|
|
self._timeout = timeout
|
|
self._client: Optional[httpx.AsyncClient] = None
|
|
|
|
# 내부
|
|
@property
|
|
def enabled(self) -> bool:
|
|
"""키가 설정돼 있는지."""
|
|
return bool(self._api_key)
|
|
|
|
def _headers(self) -> dict:
|
|
if not self._api_key:
|
|
raise KakaoNotConfigured(
|
|
"KAKAO_REST_API_KEY 가 설정되지 않았습니다 — .env 또는 config 의 [ExternalApiConfig] 를 확인하세요"
|
|
)
|
|
return {"Authorization": f"KakaoAK {self._api_key}"}
|
|
|
|
def _get_client(self) -> httpx.AsyncClient:
|
|
if self._client is None or self._client.is_closed:
|
|
kwargs = {"timeout": httpx.Timeout(self._timeout, connect=5.0)}
|
|
if self._transport is not None:
|
|
kwargs["transport"] = self._transport
|
|
self._client = httpx.AsyncClient(**kwargs)
|
|
return self._client
|
|
|
|
async def aclose(self):
|
|
if self._client is not None and not self._client.is_closed:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
async def _get(self, url: str, params: dict, kind: str) -> dict:
|
|
"""공통 GET."""
|
|
headers = self._headers() # 키 없으면 여기서 KakaoNotConfigured
|
|
_CALL_COUNTS[kind] += 1
|
|
LOG.i(
|
|
f"[kakao] {kind} 호출 (누적 {_CALL_COUNTS[kind]}회, "
|
|
f"초과 시 건당 {_UNIT_COST_KRW.get(kind, 0):g}원) params={ {k: v for k, v in params.items() if k != 'query'} }"
|
|
)
|
|
try:
|
|
resp = await self._get_client().get(url, params=params, headers=headers)
|
|
except httpx.TimeoutException as ex:
|
|
raise KakaoRequestFailed(f"카카오 로컬 {kind} 타임아웃: {ex}") from ex
|
|
except httpx.HTTPError as ex:
|
|
raise KakaoRequestFailed(f"카카오 로컬 {kind} 요청 실패: {type(ex).__name__}: {ex}") from ex
|
|
|
|
if resp.status_code == 401:
|
|
raise KakaoNotConfigured(f"카카오 로컬 인증 실패(401) — REST API 키를 확인하세요: {resp.text[:200]}")
|
|
if resp.status_code != 200:
|
|
raise KakaoRequestFailed(f"카카오 로컬 {kind} 응답 오류 status={resp.status_code} body={resp.text[:200]}")
|
|
|
|
try:
|
|
return resp.json()
|
|
except ValueError as ex:
|
|
raise KakaoRequestFailed(f"카카오 로컬 {kind} 응답 파싱 실패: {ex}") from ex
|
|
|
|
# 1) 상호명 → 주소·좌표·전화
|
|
async def search_keyword(
|
|
self, name: str, x: Optional[float] = None, y: Optional[float] = None, size: int = 15
|
|
) -> list[KakaoPlace]:
|
|
"""상호명으로 장소를 찾는다."""
|
|
params: dict = {"query": name, "size": max(1, min(size, 15))}
|
|
if x is not None and y is not None:
|
|
params["x"] = str(x) # 경도
|
|
params["y"] = str(y) # 위도
|
|
data = await self._get(_KEYWORD_URL, params, "keyword")
|
|
return [KakaoPlace.from_document(d) for d in (data.get("documents") or [])]
|
|
|
|
# 2) 좌표 → 행정구역 코드
|
|
async def coord_to_region(self, lat: float, lon: float) -> RegionCode:
|
|
"""좌표를 행정구역 코드로 바꾼다."""
|
|
data = await self._get(_COORD2REGION_URL, {"x": str(lon), "y": str(lat)}, "coord2region")
|
|
docs = data.get("documents") or []
|
|
if not docs:
|
|
raise KakaoRequestFailed(f"좌표 → 행정구역 변환 결과 없음 (lat={lat}, lon={lon})")
|
|
# 행정동(H) 우선 — 생활권 기준이라 주변 정보와 더 잘 맞는다.
|
|
picked = next((d for d in docs if d.get("region_type") == "H"), docs[0])
|
|
return RegionCode.from_document(picked)
|
|
|
|
# 2-1) 주소 → 좌표
|
|
async def geocode_address(self, address: str) -> Optional[tuple[float, float]]:
|
|
"""도로명·지번 주소 → (위도, 경도)."""
|
|
query = (address or "").strip()
|
|
if not query:
|
|
return None
|
|
data = await self._get(_ADDRESS_URL, {"query": query, "size": "1"}, "address")
|
|
docs = data.get("documents") or []
|
|
if not docs:
|
|
LOG.i(f"[kakao] 주소 → 좌표 결과 없음: {query[:60]}")
|
|
return None
|
|
lat, lon = _to_float(docs[0].get("y")), _to_float(docs[0].get("x")) # y=위도 · x=경도
|
|
if lat is None or lon is None:
|
|
return None
|
|
return lat, lon
|
|
|
|
# 3) 주변 맛집·시설
|
|
async def search_category(
|
|
self,
|
|
region_x: float,
|
|
region_y: float,
|
|
category_group_code: str,
|
|
radius: int = 2000,
|
|
size: int = 15,
|
|
region_code: Optional[str] = None,
|
|
) -> list[KakaoPlace]:
|
|
"""좌표 반경 안의 카테고리 장소(주변 맛집·카페·관광지 등)를 찾는다."""
|
|
if not region_code:
|
|
LOG.w(
|
|
"[kakao] search_category 를 region_code 없이 호출했습니다 — "
|
|
"행정구역 코드 단위 캐시를 거치지 않으면 같은 지역을 사이트 수만큼 반복 조회합니다"
|
|
)
|
|
params = {
|
|
"category_group_code": category_group_code,
|
|
"x": str(region_x), # 경도
|
|
"y": str(region_y), # 위도
|
|
"radius": str(max(0, min(radius, 20000))), # 카카오 상한 20km
|
|
"size": str(max(1, min(size, 15))),
|
|
"sort": "distance",
|
|
}
|
|
data = await self._get(_CATEGORY_URL, params, "category")
|
|
docs = [KakaoPlace.from_document(d) for d in (data.get("documents") or [])]
|
|
LOG.i(f"[kakao] category={category_group_code} region_code={region_code or '미지정'} → {len(docs)}건")
|
|
return docs
|
|
|
|
# 조합: 상호명 하나로 동일 업소까지
|
|
async def verify_place(
|
|
self,
|
|
name: str,
|
|
phone: Optional[str] = None,
|
|
*,
|
|
search_query: Optional[str] = None,
|
|
) -> MatchResult:
|
|
"""상호명(+전화번호)으로 검색해서 동일 업소를 판정한다."""
|
|
candidates = await self.search_keyword(search_query or name)
|
|
result = pick_match(name, candidates, phone)
|
|
detail = f"'{name}'" if not search_query or search_query == name else f"'{name}' (검색어 '{search_query}')"
|
|
LOG.i(f"[kakao] 동일 업소 판정 {detail} → {result.outcome.value} ({result.reason})")
|
|
return result
|
|
|
|
|
|
def call_counts() -> dict:
|
|
"""프로세스 누적 호출 수 — 생성 1건당 검색 횟수를 세는 근거(비용 리포트용)."""
|
|
return dict(_CALL_COUNTS)
|
|
|
|
|
|
def reset_call_counts():
|
|
"""테스트/리포트 구간 분리용."""
|
|
_CALL_COUNTS.clear()
|