o2o-site-AEO/solution/backend/services/external/naver_place_lookup.py

"""상호·주소 → 네이버 플레이스 id.

★ 왜 필요한가
  채널 URL 발견은 Perplexity 가 맡는데, 네이버 플레이스만은 잘 못 찾는다.
  실측(2026-08-27 '도플로'·'버터브루'): 발견 URL 이 전부 야놀자·인스타였고, 필터를
  map.naver.com 까지 넓힌 뒤에도 네이버 쪽은 `pages.map.naver.com/useful-tips` 같은
  안내 페이지가 걸렸다. 검색 언어모델에 맡기기엔 결과가 불안정하고 검색 요금도 든다.

  그런데 우리는 이미 **이 가게가 누구인지 알고 있다**(동일 업소 검증을 통과한 상호·주소).
  그러면 추측할 이유가 없다 — 통합검색 결과에서 상호가 일치하는 place id 를 직접 고른다.

★ 우회하지 않는다. 공개 검색 결과 페이지를 한 번 받아 id 를 읽을 뿐이고,
  막히면 그대로 빈 값을 돌려준다(호출측이 다른 경로로 간다).
"""
import re
from html import unescape
from typing import Optional

import httpx

from common.logger import LOG

SEARCH_URL = "https://m.search.naver.com/search.naver"
REQUEST_TIMEOUT = 20

HEADERS = {
    "User-Agent": ("Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) "
                   "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"),
    "Accept-Language": "ko-KR,ko;q=0.9",
}

_PLACE_ID = re.compile(r"(?:place\.naver\.com/[a-z]+/|/place/|\"placeId\"\s*:\s*\"?)(\d{8,12})")

# id 주변에서 상호를 찾을 때 훑는 범위.
#
# ★ `"name":"…"` JSON 필드를 뽑아 비교하면 안 된다. 그 자리에 실제로 들어 있는 것은
#   리뷰 키워드("주차하기 편해요")나 블로거 닉네임이고, 상호는 다른 형태로 박혀 있다.
#   그래서 **정규화한 원문 조각에 상호가 들어 있는지**로 판정한다 — 마크업 모양이 바뀌어도 버틴다.
_CONTEXT_BEFORE = 600
_CONTEXT_AFTER = 300



def _normalize(text: str) -> str:
    """상호 비교용 정규화. 네이버는 '스테이,머뭄'처럼 구두점을 넣어 표기한다.

    ★ `&` 도 지운다. 검색 결과 원문에는 `&` 로 실려 오기 때문에, 엔티티를 풀어도
      `&` 가 남으면 지역검색이 준 상호(`누에베 풀빌라&리조트`)와 원문 조각의 표기가
      어긋난다. 실측(2026-08-28): 이 한 글자 때문에 place id 를 못 찾아 사장님이
      네이버 지도 주소를 손으로 붙여넣어야 했다 — 10건 중 1건.
    """
    return re.sub(r"[\s,·.\-_'\"()&]", "", (text or "")).lower()


def _region_hint(address: Optional[str]) -> str:
    """주소에서 검색을 좁힐 지역 토막. 시/군/구까지만 쓴다.

    ★ 첫 토막만 쓰면 안 된다 — 그건 광역시·도('경기도')라 오히려 넓어진다.
      실측(2026-09-03 '버터브루'): '버터브루 경기도' 로 찾으면 **다른 동네 동명 업소**의
      id 가 잡히고, 그 id 로 검증하면 남의 가게가 이 사이트의 기준 정보가 된다.
      '버터브루 성남시 중원구' 로 좁히면 정확히 잡힌다.
    ★ 도로명·번지는 넣지 않는다. 검색 결과가 그 주소를 언급한 블로그로 채워져 id 가 사라진다.
    """
    tokens = (address or "").split()
    picked = [t for t in tokens if t.endswith(("시", "군", "구"))]
    return " ".join(picked[:2])


async def _fetch_search_html(query: str) -> Optional[str]:
    """통합검색 결과 페이지 원문. 실패는 None — 호출측이 조용히 폴백한다."""
    try:
        async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT, follow_redirects=True) as client:
            res = await client.get(SEARCH_URL, params={"query": query, "where": "m"}, headers=HEADERS)
        if res.status_code != 200:
            LOG.w(f"[naver_lookup] 검색 실패 HTTP {res.status_code} — query={query}")
            return None
        return res.text
    except httpx.HTTPError as ex:
        LOG.w(f"[naver_lookup] 검색 실패: {ex}")
        return None


def _match_in_html(html: str, name: str) -> Optional[str]:
    """검색 결과 원문에서 이 상호에 해당하는 place id 를 고른다.

    ★ `"name":"…"` 를 뽑아 비교하지 않는다. 그 자리에 실제로 들어 있는 것은 리뷰 키워드
      ("주차하기 편해요")나 블로거 닉네임이라 상호가 아니다. 그래서 **id 주변 원문을
      정규화해 상호가 들어 있는지**로 판정한다 — 마크업이 바뀌어도 버틴다.
    """
    target = _normalize(name)
    if not target:
        return None
    for match in _PLACE_ID.finditer(html):
        # ★ 엔티티를 먼저 푼다. 원문에는 상호가 `누에베 풀빌라&리조트` 처럼 인코딩돼 있어,
        #   그대로 정규화하면 `amp` 라는 없는 글자가 상호 한가운데 남는다.
        #   조각(≈900자)에만 적용한다 — 1.3MB 원문 전체를 후보마다 푸는 것은 낭비다.
        window = _normalize(
            unescape(html[max(0, match.start() - _CONTEXT_BEFORE): match.start() + _CONTEXT_AFTER])
        )
        if target in window:
            return match.group(1)
    return None


async def find_place_ids(query: str, names: list[str]) -> dict[str, str]:
    """후보 상호들에 대해 {상호: place_id} 를 채운다. 못 찾은 상호는 빠진다.

    ★ 왜 후보 목록에 id 를 실어야 하나: 사장님이 후보를 고르는 순간 네이버 플레이스 id 가
      확정되면, 나중에 수집 단계에서 상호를 다시 맞춰 볼 필요가 없다. 이름 맞추기는
      동명 업소·지점명 표기 차이에서 틀리고, 틀리면 남의 가게를 긁는다.

    검색은 **한 번만** 한다 — 후보 5건에 5번 요청하면 네이버가 막는다(429).
    """
    html = await _fetch_search_html(query)
    if not html:
        return {}
    found: dict[str, str] = {}
    for name in names:
        place_id = _match_in_html(html, name)
        if place_id:
            found[name] = place_id
    return found


async def find_place_id(name: str, address: Optional[str] = None) -> Optional[str]:
    """상호(+주소)로 네이버 플레이스 id 를 찾는다. 확신이 없으면 None.

    ★ 이름이 일치하는 후보만 받는다. '비슷한 것 중 첫 번째'를 고르면 남의 가게를
      이 가게의 공식 채널로 등록하게 된다 — 이 제품에서 가장 비싼 실수다.
    """
    query = " ".join(x for x in (name, _region_hint(address)) if x)
    html = await _fetch_search_html(query)
    if not html:
        return None
    place_id = _match_in_html(html, name)
    if place_id:
        LOG.i(f"[naver_lookup] '{name}' → place {place_id}")
        return place_id
    LOG.w(f"[naver_lookup] '{name}' 상호가 일치하는 후보를 찾지 못했다 — 자동 등록하지 않는다")
    return None


def place_url(place_id: str) -> str:
    """수집 어댑터가 그대로 처리할 수 있는 정규 주소."""
    return f"https://m.place.naver.com/place/{place_id}/home"