ssulbox/backend/app/place_service.py

# -*- coding: utf-8 -*-
"""네이버 지도 업장 검색 (키 불필요, Playwright).

`map.naver.com/p/search/{query}` 의 검색 리스트 iframe(`pcmap.place.naver.com/place/list`)
안에 있는 `__APOLLO_STATE__` 를 파싱해, 동명 업장들을 주소로 구분한 후보 목록을 만든다.
각 후보는 `place_url`(map.naver.com/p/entry/place/{id})을 가지므로, 사용자가 하나를 고르면
그 URL 을 그대로 생성 파이프라인(generator/naver.py)에 넘겨 정확히 그 가게를 크롤링한다.

castad 의 NvMapPwScraper 와 같은 목적이지만, NAVER 검색 API 키 없이 지도만으로 후보+place_id 를
동시에 얻는다(castad 는 후보=NAVER API, place_id 해석=Playwright 2단계).
"""
import asyncio
import json
import re
import sys
from urllib.parse import quote

from playwright.async_api import async_playwright

DESKTOP_UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
)


def _extract_apollo_state(html: str) -> dict | None:
    """HTML 안의 `__APOLLO_STATE__ = { ... };` 객체를 중괄호 균형으로 안전하게 추출."""
    i = html.find("__APOLLO_STATE__")
    if i < 0:
        return None
    j = html.find("{", i)
    if j < 0:
        return None
    depth = 0
    in_str = False
    esc = False
    for k in range(j, len(html)):
        c = html[k]
        if in_str:
            if esc:
                esc = False
            elif c == "\\":
                esc = True
            elif c == '"':
                in_str = False
        else:
            if c == '"':
                in_str = True
            elif c == "{":
                depth += 1
            elif c == "}":
                depth -= 1
                if depth == 0:
                    try:
                        return json.loads(html[j : k + 1])
                    except Exception:
                        return None
    return None


def _candidates_from_state(state: dict, limit: int) -> list[dict]:
    out: list[dict] = []
    for key, obj in state.items():
        if not key.startswith("PlaceListBusinessesItem:"):
            continue
        if not isinstance(obj, dict):
            continue
        name = obj.get("name")
        pid = obj.get("id")
        if not (name and pid and str(pid).isdigit()):
            continue
        out.append(
            {
                "title": name,
                "category": obj.get("category") or "",
                # 표시용 주소는 '서울 성동구 금호동3가' 같은 commonAddress 를 우선(동명 구분에 최적)
                "address": obj.get("commonAddress") or obj.get("fullAddress") or "",
                "roadAddress": obj.get("fullAddress") or obj.get("roadAddress") or "",
                "place_url": f"https://map.naver.com/p/entry/place/{pid}",
            }
        )
        if len(out) >= limit:
            break
    return out


async def _search(query: str, limit: int) -> list[dict]:
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=["--disable-blink-features=AutomationControlled", "--no-sandbox"],
        )
        try:
            ctx = await browser.new_context(
                user_agent=DESKTOP_UA, locale="ko-KR", timezone_id="Asia/Seoul",
                viewport={"width": 1280, "height": 800},
                extra_http_headers={"Accept-Language": "ko-KR,ko;q=0.9"},
            )
            page = await ctx.new_page()
            await page.goto(
                f"https://map.naver.com/p/search/{quote(query)}",
                wait_until="domcontentloaded", timeout=40000,
            )
            # 검색 리스트 iframe 이 뜰 때까지 대기(최대 ~12초)
            frame = None
            for _ in range(24):
                for f in page.frames:
                    if "pcmap.place.naver.com/place/list" in f.url:
                        frame = f
                        break
                if frame:
                    break
                # 단일 결과면 곧바로 place 상세로 리다이렉트됨 → 후보 1개로 처리
                if "/place/" in page.url and "/search/" not in page.url:
                    return [{"title": query, "category": "", "address": "",
                             "roadAddress": "", "place_url": page.url}]
                await page.wait_for_timeout(500)
            if not frame:
                return []
            # apollo state 가 채워질 시간을 조금 더 준다
            html = await frame.content()
            state = _extract_apollo_state(html)
            for _ in range(6):
                if state and any(k.startswith("PlaceListBusinessesItem:") for k in state):
                    break
                await page.wait_for_timeout(600)
                html = await frame.content()
                state = _extract_apollo_state(html)
            if not state:
                return []
            return _candidates_from_state(state, limit)
        finally:
            await browser.close()


def _search_blocking(query: str, limit: int) -> list[dict]:
    """별도 스레드에서 자체 이벤트 루프로 Playwright 실행.

    Windows 의 웹서버 루프(Selector)는 서브프로세스를 못 띄워 Playwright 가 즉시 실패한다.
    → 이 함수는 워커 스레드에서 Proactor 루프(윈도우) 를 새로 만들어 그 위에서 돌린다.
    """
    if sys.platform == "win32":
        loop = asyncio.ProactorEventLoop()
    else:
        loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(_search(query, limit))
    finally:
        loop.close()


async def search_places(query: str, limit: int = 8, timeout: float = 35.0) -> list[dict]:
    """업장명으로 네이버 지도 후보 목록을 반환. 실패/타임아웃이면 빈 리스트."""
    query = (query or "").strip()
    if len(query) < 2:
        return []
    try:
        return await asyncio.wait_for(asyncio.to_thread(_search_blocking, query, limit), timeout=timeout)
    except Exception:
        return []