"""카카오 로컬 API 클라이언트 — 동일 업소 검증 + 지역 정보. 이 서비스에서 카카오 로컬이 하는 일은 두 가지다. 1. **동일 업소 검증** (가장 중요) Perplexity 가 찾아온 채널 URL 이 정말 그 가게 것인지 확인하는 유일한 근거다. 이 단계가 없으면 동명 업소 정보가 섞이고, 남의 가게 체크인 시간이 우리 사이트로 나간다. ★ 애매하면 자동 판정하지 않고 사람에게 넘긴다(pick_match 참고). 2. **지역 정보** (주변 맛집·시설) 좌표 → 행정구역 코드로 바꾸고, 그 코드 단위로 주변 정보를 모은다. ── 비용 (2026-08 기준) ─────────────────────────────────────────────── 무료 쿼터: 키워드 검색 일 10만 · 좌표 변환 일 10만 · 전체 월 300만 초과 단가: 키워드/카테고리 검색 **2원** · 좌표 변환 **0.5원** → ★ 키워드/카테고리 검색이 좌표 변환보다 **4배 비싸다** ★ 무료 쿼터는 개발자 계정의 '첫 번째 활성 앱' 에만 붙는다. dev/stage/prod 앱을 따로 파면 하나만 무료다. 그래서 호출 정책이 이렇다. - 키워드 검색(search_keyword) : 사업장 등록·재검증 때만. 비싸다 - 좌표 변환(coord_to_region) : 싸다. 사업장당 1회면 충분(좌표는 안 바뀐다) - 카테고리 검색(search_category): 비싸다. ★ **행정구역 코드 단위로 캐싱**해야 한다. 같은 지역에 사이트가 50개 생겨도 조회는 1회여야 한다. 캐싱 자체는 local 모듈(local.local_contents, 캐시 키 = region_code)이 책임진다 — 이 클라이언트는 캐시를 두지 않는다. 호출 전에 캐시를 먼저 보라는 뜻이다. 호출 횟수는 전부 LOG.i 로 남긴다(비용 추적). _CALL_COUNTS 로 프로세스 누적도 볼 수 있다. """ 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" # 초과 단가(원). 로그에 함께 남겨 어떤 호출이 비싼지 바로 보이게 한다. _UNIT_COST_KRW = {"keyword": 2.0, "category": 2.0, "coord2region": 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 미설정 — 이 어댑터만 비활성이다. 서버 부팅을 막지 않는다(외부 계약에 부팅이 묶이면 안 된다). 호출측이 잡아 ErrorType.LOCAL_NOT_CONFIGURED 로 응답한다.""" class KakaoRequestFailed(RuntimeError): """카카오 로컬 호출 실패(네트워크·타임아웃·5xx·인증오류). ★ 실패했다고 빈 값을 내보내면 안 된다 — 호출측은 직전 값을 유지하고 내부 알림만 낸다.""" class MatchOutcome(str, Enum): """동일 업소 판정 결과. ※ 이 서비스 안에서만 쓰는 판정 결과라 모듈 지역 enum 으로 둔다. 라우터 응답으로 내보낼 일이 생기면 common/enums.py 로 올려야 한다.""" MATCHED = "matched" # 이 가게가 맞다고 확정 AMBIGUOUS = "ambiguous" # ★ 동명 업소 등 — 사람이 골라야 한다 NO_CANDIDATE = "no_candidate" # 카카오에서 후보를 못 찾음 @dataclass(frozen=True) class KakaoPlace: """카카오 로컬이 돌려준 장소 1건. ★ 카카오 응답의 x 는 경도(longitude), y 는 위도(latitude) 다. 뒤집으면 엉뚱한 지역이 된다.""" 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] 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), place_url=(doc.get("place_url") or None), ) @dataclass(frozen=True) class RegionCode: """행정구역 코드. ★ 지역 정보 캐시의 키다(같은 지역 사이트 50개여도 조회 1회).""" 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: """동일 업소 판정 결과 + **판정 근거**. 근거를 같이 들고 다니는 이유: ambiguous 로 떨어졌을 때 사람이 무엇을 보고 골라야 하는지 알아야 하고, matched 로 확정됐을 때도 나중에 '왜 이 가게로 붙었나' 를 추적해야 한다.""" 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: """★ 검색 결과에서 '이 가게가 맞다' 를 판정한다. 이 서비스에서 가장 비싼 실수가 나는 지점이다. 판정 순서 1. 후보 0건 → NO_CANDIDATE 2. 전화번호가 주어졌고 정확히 1건 일치 → MATCHED (가장 강한 근거) 전화번호 일치가 2건 이상 → 그 부분집합으로 좁혀 상호명 판정을 이어간다 3. 정규화 상호명이 정확히 1건 일치 → MATCHED 4. 정규화 상호명이 2건 이상 일치 → AMBIGUOUS (동명 업소) 5. 정확히 일치하는 상호명이 없음 → AMBIGUOUS (부분일치만으로는 확정하지 않는다) ★ 애매하면 반드시 AMBIGUOUS 로 떨어뜨린다. 억지로 하나 고르면 남의 가게 정보가 섞이고, 그건 사이트가 발행된 뒤에야 드러난다(그때는 이미 예약 클레임이 난 뒤다).""" 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 호출기. 키가 없으면 생성은 되지만 호출 시 KakaoNotConfigured 를 던진다 — 부팅이 외부 계약에 묶이지 않게 하기 위함이다(설정이 비면 이 어댑터만 비활성).""" def __init__(self, api_key: Optional[str] = None, transport=None, timeout: float = 10.0): # api_key 를 명시하지 않으면 설정에서 읽는다. 테스트는 transport 를 주입한다. 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. 호출 1건마다 비용을 로그로 남긴다.""" 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]: """상호명으로 장소를 찾는다. **동일 업소 검증의 입력**이다. ★ 비싸다(초과 시 건당 2원). 사업장 등록·재검증 때만 부른다. x/y 를 주면 그 좌표 근처를 우선한다(x=경도, y=위도). 지역을 아는 경우 후보가 훨씬 깨끗해진다. 반환된 목록은 그대로 pick_match 에 넘긴다 — 여기서 하나를 고르지 않는다.""" 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: """좌표를 행정구역 코드로 바꾼다. ★ 이 코드가 지역 정보 캐시의 키다. 싸다(초과 시 건당 0.5원, 키워드 검색의 1/4). 좌표는 안 바뀌므로 사업장당 1회면 충분하다. 행정동(H)을 우선 반환하고, 없으면 첫 문서를 쓴다.""" 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) # ---- 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]: """좌표 반경 안의 카테고리 장소(주변 맛집·카페·관광지 등)를 찾는다. ★ 비싸다(초과 시 건당 2원 — 좌표 변환의 4배). **반드시 행정구역 코드 단위로 캐싱해서 부른다.** 같은 지역에 사이트가 50개 생겨도 이 호출은 1회여야 한다. 캐싱은 이 클라이언트가 하지 않는다 — local 모듈이 local.local_contents 에 `region_code + content_type` 키로 저장하고, 캐시 미스일 때만 여기를 부른다. `region_code` 인자는 그 캐시 키를 호출 지점에서 명시하게 하려고 받는다(로그에도 남는다). 호출측이 region_code 를 못 주면 캐시 없이 부르고 있다는 뜻이라 경고를 남긴다. region_x = 경도(longitude), region_y = 위도(latitude). 그 지역의 중심 좌표를 넘긴다.""" 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: """상호명(+전화번호)으로 검색해서 동일 업소를 판정한다. 키워드 검색 1회만 쓴다. 결과가 MATCHED 가 아니면 사람이 골라야 한다 — 호출측은 AMBIGUOUS → ErrorType.PLACE_VERIFY_AMBIGUOUS, NO_CANDIDATE → ErrorType.PLACE_VERIFY_NO_CANDIDATE 로 응답한다. ★ name 에는 상호명만 넣는다 — 판정이 정확일치를 보기 때문이다. 지역이 섞인 검색어는 search_query 로 넘긴다(naver.verify_place 와 같은 규약). """ 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()