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

"""TourAPI 콘텐츠 조회 — 검증된 상호·좌표로 contentId 를 직접 해석한다."""
import math
import re
from typing import Optional
from urllib.parse import unquote, urlencode

import httpx

from common.enums import PlaceCategory
from common.logger import LOG
from config.server_configs import external_api_config
# 출처 주소 규칙은 어댑터 한 곳에서만 정의한다.
from services.collector.tour_api_adapter import SOURCE_URL

BASE_URL = "https://apis.data.go.kr/B551011/KorService2"
REQUEST_TIMEOUT = 20

# 우리 업종 → TourAPI contentTypeId.
CATEGORY_TO_CONTENT_TYPE = {
    PlaceCategory.LODGING: "32",
    PlaceCategory.CAFE: "39",
    PlaceCategory.RESTAURANT: "39",
}

# 같은 업소로 볼 좌표 거리 상한.
MAX_DISTANCE_M = 500

# 상호를 짧혀가며 다시 물어보는 최대 횟수.
MAX_QUERY_ATTEMPTS = 4


def normalize(text: str) -> str:
    """상호 대조용 정규화."""
    return re.sub(r"[\s,·.\-_'\"()&]", "", (text or "")).lower()


def _name_matches(query_name: str, candidate: str) -> bool:
    """정규화 후 한쪽이 다른 쪽을 포함하면 같은 업소로 본다."""
    a, b = normalize(query_name), normalize(candidate)
    if not a or not b:
        return False
    if min(len(a), len(b)) <= 2:
        return a == b
    return a in b or b in a


def _distance_m(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
    """두 좌표의 거리(m)."""
    r = 6_371_000
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dp, dl = math.radians(lat2 - lat1), math.radians(lng2 - lng1)
    h = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
    return 2 * r * math.asin(math.sqrt(h))


def is_configured() -> bool:
    return bool((external_api_config.tour_api_key or "").strip())


def content_url(content_id: str) -> str:
    """이 레코드를 가리키는 출처 주소(인증키 없음)."""
    return SOURCE_URL.format(content_id=content_id)


async def _search(client: httpx.AsyncClient, key: str, **params) -> list[dict]:
    query = urlencode(
        {"serviceKey": unquote(key), "MobileOS": "ETC", "MobileApp": "o2o-web4ai",
         "_type": "json", "numOfRows": "20", "pageNo": "1", **params},
        safe="",
    )
    res = await client.get(f"{BASE_URL}/searchKeyword2?{query}")
    if res.status_code != 200:
        raise RuntimeError(f"searchKeyword2 HTTP {res.status_code}")
    try:
        payload = res.json()
    except ValueError:
        raise RuntimeError(f"searchKeyword2 응답이 JSON 이 아니다: {res.text[:160]}")

    header = payload.get("response", {}).get("header", {})
    code = str(header.get("resultCode") or "")
    if code not in ("0000", "00"):
        raise RuntimeError(f"searchKeyword2 실패 [{code}] {header.get('resultMsg')}")

    items = (payload.get("response", {}).get("body", {}).get("items") or {}).get("item")
    if isinstance(items, dict):
        return [items]
    return items or []


def _query_candidates(name: str) -> list[str]:
    """검색어 후보 — 전체 상호부터 시작해 **뒤 토큰을 하나씩 떼며** 짧게 만든다."""
    tokens = [t for t in re.split(r"\s+", (name or "").strip()) if t]
    if not tokens:
        return []
    out: list[str] = []
    for end in range(len(tokens), 0, -1):
        q = " ".join(tokens[:end])
        if len(normalize(q)) >= 2 and q not in out:
            out.append(q)
        if len(out) >= MAX_QUERY_ATTEMPTS:
            break
    return out


_TYPE_LABEL = {"32": "숙박", "39": "음식점·카페", "12": "관광지", "14": "문화시설", "28": "레포츠", "38": "쇼핑"}


async def _warn_if_other_content_type(key: str, name: str, expected: str) -> None:
    """같은 상호가 **다른 콘텐츠 타입**으로 등록돼 있으면 로그로 알린다."""
    try:
        async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
            for query in _query_candidates(name):
                found = await _search(client, key, keyword=query)
                for item in found:
                    other = str(item.get("contenttypeid") or "")
                    if other and other != expected and _name_matches(name, str(item.get("title") or "")):
                        LOG.w(
                            f"[tour_lookup] ★ 업종 불일치 — '{name}' 은 TourAPI 에 "
                            f"{_TYPE_LABEL.get(other, other)}({other}) 로 등록돼 있는데 "
                            f"이 사업장은 {_TYPE_LABEL.get(expected, expected)}({expected}) 로 조회했다. "
                            f"사업장 업종 등록을 확인하세요."
                        )
                        return
                if found:
                    return
    except (httpx.HTTPError, RuntimeError):
        return          # 진단용이라 실패해도 조용히 넘어간다


async def find_content_id(
    name: str,
    category: PlaceCategory,
    *,
    latitude: Optional[float] = None,
    longitude: Optional[float] = None,
) -> Optional[tuple[str, str]]:
    """(contentId, contentTypeId) 또는 None."""
    key = (external_api_config.tour_api_key or "").strip()
    if not key:
        return None
    content_type = CATEGORY_TO_CONTENT_TYPE.get(category)
    if not content_type:
        return None

    items: list[dict] = []
    try:
        async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
            for query in _query_candidates(name):
                items = await _search(client, key, keyword=query, contentTypeId=content_type)
                if items:
                    if query != name:
                        LOG.i(f"[tour_lookup] '{name}' 0건 → '{query}' 로 축약해 {len(items)}건")
                    break
    except httpx.HTTPError as ex:
        LOG.w(f"[tour_lookup] 조회 실패(계속): {type(ex).__name__}: {ex}")
        return None
    except RuntimeError as ex:
        LOG.w(f"[tour_lookup] 조회 실패(계속): {ex}")
        return None

    best: Optional[tuple[float, dict]] = None
    for item in items:
        title = str(item.get("title") or "")
        if not _name_matches(name, title):
            continue

        distance = 0.0
        if latitude is not None and longitude is not None:
            try:
                distance = _distance_m(latitude, longitude, float(item["mapy"]), float(item["mapx"]))
            except (KeyError, TypeError, ValueError):
                # 좌표가 없는 항목은 거리로 거를 수 없다 — 상호만 맞으면 후순위 후보로 둔다.
                distance = float(MAX_DISTANCE_M)
            if distance > MAX_DISTANCE_M:
                LOG.i(f"[tour_lookup] '{title}' 상호는 맞지만 {distance:.0f}m 떨어져 있어 제외")
                continue
        if best is None or distance < best[0]:
            best = (distance, item)

    if best is None:
        # 타입을 안 걸고 한 번 더 본다 — 업종을 잘못 등록하면 여기서만 알 수 있다.
        await _warn_if_other_content_type(key, name, content_type)
        LOG.i(f"[tour_lookup] '{name}' 일치하는 TourAPI 콘텐츠 없음 (후보 {len(items)}건)")
        return None

    item = best[1]
    LOG.i(
        f"[tour_lookup] '{name}' → contentId={item['contentid']} "
        f"({item.get('title')}, {best[0]:.0f}m)"
    )
    return str(item["contentid"]), str(item.get("contenttypeid") or content_type)