o2o-site-AEO/solution/backend/services/collector/yanolja_adapter.py

"""야놀자(NOL) 국내숙소 어댑터 — Playwright 로 상세페이지를 렌더링해 객실·사진을 수집한다.

★ nol.yanolja.com 은 Next.js CSR 페이지라 httpx(정적 HTML)로는 못 읽는다 — 그래서
  static_html_adapter 가 아니라 이 어댑터가 Playwright 로 실제 브라우저 렌더링을 거친다.
  이건 봇 탐지 우회가 아니라 JS 렌더링이 필요한 페이지를 읽는 통상적인 방법이다 —
  캡차 우회·IP 회전·지문 위장 같은 건 하지 않는다(registry.py 의 영구 금지 원칙 그대로 유지).
  차단(403·챌린지 등)을 만나면 그대로 실패로 돌려주고 재시도·우회하지 않는다.

수집 원칙
  - 페이지에 보이는 값만 옮긴다. 가격은 수집하지 않는다(불안정하고 예약 시점에 따라 바뀐다).
  - 이미지는 원본 URL 그대로만 남긴다(origin_url) — 재게시 여부는 발행 게이트가 판단한다.
  - 객실(unit)별 값은 scope="unit" 로 담는다. 숙소소개·시설/서비스·이용안내·예약공지는
    전부 크롤링하지만 fact 로 만들지 않는다 — 숙소소개(intro)·객실소개(room_intro)는
    allow_llm=True 필드라 그대로 넣으면 절대규칙 7을 어기고, 나머지 셋은 스키마의
    특정 필드와 1:1로 안 맞는다. 넷 다 RawSource.text 로 보존한다. 시설/서비스·이용안내·
    예약공지는 확정 링크의 payload.links[].stayGuide로도 전달해 원문을 표시한다.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import Optional

from playwright.async_api import Page, TimeoutError as PWTimeoutError, async_playwright

from common.enums import LinkChannel, PlaceCategory
from common.logger import LOG
from services.collector.base import CollectedFact, CollectedMedia, RawSource

SEARCH_URL = "https://nol.yanolja.com/"
DETAIL_URL_RE = re.compile(r"nol\.yanolja\.com/stay/domestic/\d+", re.IGNORECASE)

# 검색결과 카드는 <a data-card-type="basic" ... aria-label="{업체명} 상품 상세 보기" href="...">
RESULT_CARD_SELECTOR = 'a[data-card-type="basic"]'

SECTION_IDS = {
    "rooms": "PLACE_SECTION",
    "overview": "OVERVIEW_SECTION",
    "service": "SERVICE_SECTION",
    "policy": "POLICY_SECTION",
    "reservation": "RESERVATION_SECTION",
}

# 원문을 보존할 섹션들. 생성 근거와 확정된 NOL 안내 표시가 같은 수집 원문을 쓴다.
_TEXT_ONLY_SECTIONS = (
    ("overview", "숙소 소개"),
    ("service", "시설/서비스"),
    ("policy", "이용 안내"),
    ("reservation", "예약 공지"),
)

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


@dataclass
class _RoomInfo:
    name: str
    description: Optional[str] = None
    capacity: Optional[str] = None
    bed: Optional[str] = None
    images: list[str] = field(default_factory=list)


def _extract_stay_id(url: str) -> Optional[str]:
    m = re.search(r"/stay/domestic/(\d+)", url)
    return m.group(1) if m else None


def _is_real_photo_url(src: str) -> bool:
    return bool(src) and "static/images" not in src


def _parse_capacity(capacity: Optional[str]) -> tuple[Optional[str], Optional[str]]:
    if not capacity:
        return None, None
    std_m = re.search(r"기준\s*(\d+)인", capacity)
    max_m = re.search(r"최대\s*(\d+)인", capacity)
    return (std_m.group(1) if std_m else None, max_m.group(1) if max_m else None)


def _parse_rooms_from_section_text(section_text: str) -> list[_RoomInfo]:
    """PLACE_SECTION.innerText 는 "N / M"(사진 장수) 로 객실 카드 수만큼 반복된다.
    가격·예약 버튼 이하는 무시한다. 사이트 구조가 바뀌면 이 정규식도 손봐야 한다."""
    rooms: list[_RoomInfo] = []
    chunks = re.split(r"\n?\d+\s*\n/\n\d+\n", section_text)
    for chunk in chunks[1:]:
        lines = [l.strip() for l in chunk.split("\n") if l.strip()]
        if not lines:
            continue
        name = lines[0]
        capacity_m = re.search(r"기준\s*\d+인\s*/\s*최대\s*\d+인", chunk)
        bed_m = re.search(r"(킹|퀸|더블|싱글|트윈)\s*침대\s*\d+개", chunk)
        desc_lines = []
        for l in lines[1:]:
            if l.startswith("기준") or "체크인" in l or l == "숙박":
                break
            desc_lines.append(l.strip("()"))
        rooms.append(
            _RoomInfo(
                name=name,
                description=" ".join(desc_lines) if desc_lines else None,
                capacity=capacity_m.group(0) if capacity_m else None,
                bed=bed_m.group(0) if bed_m else None,
            )
        )
    return rooms


async def _scroll_through_page(page: Page) -> None:
    prev_height = -1
    for _ in range(30):
        await page.mouse.wheel(0, 1200)
        await page.wait_for_timeout(250)
        cur_height = await page.evaluate("document.body.scrollHeight")
        if cur_height == prev_height:
            break
        prev_height = cur_height
    await page.evaluate("window.scrollTo(0, 0)")


async def _get_section_text(page: Page, element_id: str) -> str:
    try:
        await page.evaluate(
            "(id) => { const el = document.getElementById(id); if (el) el.scrollIntoView({block:'center'}); }",
            element_id,
        )
        await page.wait_for_timeout(400)
        return await page.evaluate(
            "(id) => { const el = document.getElementById(id); return el ? el.innerText : ''; }",
            element_id,
        ) or ""
    except Exception:
        return ""


async def _get_room_images_by_name(page: Page, element_id: str) -> list[tuple[str, list[str]]]:
    """카드 구조가 "사진 캐러셀 → <h2>객실명</h2> → 설명" 순이라, h2 를 만나기 전까지
    쌓인 이미지가 그 h2 의 몫이다."""
    try:
        await page.evaluate(
            "(id) => { const el = document.getElementById(id); if (el) el.scrollIntoView({block:'center'}); }",
            element_id,
        )
        await page.wait_for_timeout(400)
        raw = await page.evaluate(
            """(id) => {
                const el = document.getElementById(id);
                if (!el) return [];
                const nodes = el.querySelectorAll('img, h2');
                const result = [];
                let buf = [];
                for (const n of nodes) {
                    if (n.tagName === 'IMG') {
                        if (n.src) buf.push(n.src);
                    } else if (n.tagName === 'H2') {
                        result.push({name: n.textContent.trim(), images: [...new Set(buf)]});
                        buf = [];
                    }
                }
                return result;
            }""",
            element_id,
        ) or []
    except Exception:
        return []
    return [(item["name"], [u for u in item["images"] if _is_real_photo_url(u)]) for item in raw]


async def _get_section_images(page: Page, element_id: str) -> list[str]:
    try:
        await page.evaluate(
            "(id) => { const el = document.getElementById(id); if (el) el.scrollIntoView({block:'center'}); }",
            element_id,
        )
        await page.wait_for_timeout(400)
        srcs = await page.evaluate(
            """(id) => {
                const el = document.getElementById(id);
                if (!el) return [];
                return [...new Set(Array.from(el.querySelectorAll('img')).map(i => i.src).filter(Boolean))];
            }""",
            element_id,
        ) or []
    except Exception:
        return []
    return [u for u in srcs if _is_real_photo_url(u)]


async def _get_stay_name(page: Page) -> Optional[str]:
    try:
        h1 = await page.query_selector("h1")
        return (await h1.inner_text()).strip() if h1 else None
    except Exception:
        return None


async def _new_page():
    """호출마다 브라우저를 새로 띄우고 닫는다 — 다른 어댑터처럼 상태를 들고 있지 않는다."""
    pw = await async_playwright().start()
    browser = await pw.chromium.launch(headless=True)
    context = await browser.new_context(locale="ko-KR", user_agent=_UA)
    page = await context.new_page()
    return pw, browser, page


async def search_by_address(address: str, timeout_ms: int = 15000) -> Optional[tuple[str, str]]:
    """주소로 검색해 첫 검색결과의 (상세 URL, 업체명)을 돌려준다. 못 찾으면 None.

    discover 단계(주소만 아는 상태에서 링크를 찾는 쪽)가 쓴다. 카드를 클릭하지 않고
    href 를 직접 읽어 이동한다 — 클릭 시 새 탭이 뜨거나 배너에 가로채이는 문제를 피한다.
    """
    pw, browser, page = await _new_page()
    try:
        await page.goto(SEARCH_URL, wait_until="domcontentloaded")
        search_box = page.get_by_role("combobox", name="검색어 입력")
        await search_box.click()
        await search_box.fill(address)
        await search_box.press("Enter")

        try:
            first_card = page.locator(RESULT_CARD_SELECTOR).first
            await first_card.wait_for(state="visible", timeout=timeout_ms)
        except PWTimeoutError:
            return None

        name = await first_card.get_attribute("aria-label") or await first_card.inner_text()
        detail_url = await first_card.get_attribute("href")
        if not detail_url:
            return None

        await page.goto(detail_url, wait_until="domcontentloaded")
        return page.url, name
    except Exception as ex:
        LOG.w(f"[yanolja] 주소 검색 실패(계속): {type(ex).__name__}: {ex}")
        return None
    finally:
        await browser.close()
        await pw.stop()


class YanoljaAdapter:
    """야놀자(NOL) 국내숙소 상세페이지 → 객실 fact·사진."""

    id = "yanolja"

    def can_handle(self, url: str) -> bool:
        return bool(DETAIL_URL_RE.search((url or "").lower()))

    async def fetch(self, url: str, category: Optional[PlaceCategory] = None) -> RawSource:
        pw, browser, page = await _new_page()
        try:
            await page.goto(url, wait_until="domcontentloaded", timeout=60000)
            try:
                await page.wait_for_selector("h1", timeout=15000)
            except PWTimeoutError:
                # ★ 우회하지 않는다 — 차단·비정상 응답이면 그대로 실패로 돌린다.
                pass

            name = await _get_stay_name(page)
            if not name:
                return RawSource.failure(url, self.id, "상세페이지를 읽지 못했다(차단되었거나 존재하지 않는 페이지)", LinkChannel.YANOLJA)

            await _scroll_through_page(page)

            rooms_text = await _get_section_text(page, SECTION_IDS["rooms"])
            rooms = _parse_rooms_from_section_text(rooms_text)

            room_images = await _get_room_images_by_name(page, SECTION_IDS["rooms"])
            if len(room_images) == len(rooms):
                for room, (_, images) in zip(rooms, room_images):
                    room.images = images
            else:
                images_by_name = dict(room_images)
                for room in rooms:
                    room.images = images_by_name.get(room.name, [])

            gallery = await _get_section_images(page, SECTION_IDS["overview"])

            # 숙소소개·시설서비스·이용안내·예약공지 — fact 로 만들 스키마 필드가 없어
            # RawSource.text 로만 싣는다(생성 근거). intro 계열은 allow_llm=True 라
            # fact 로 만들면 안 된다(절대규칙 7, 2026-08-31 사고: TourAPI 가 원문을 그대로
            # intro 로 밀어넣어 LLM 소개문을 영영 못 보이게 만들었다).
            section_texts: list[str] = []
            for section_key, label in _TEXT_ONLY_SECTIONS:
                body = await _get_section_text(page, SECTION_IDS[section_key])
                if body.strip():
                    section_texts.append(f"[{label}]\n{body.strip()}")
            page_text = "\n\n".join(section_texts)[:20000]
        except Exception as ex:
            return RawSource.failure(url, self.id, f"{type(ex).__name__}: {ex}", LinkChannel.YANOLJA)
        finally:
            await browser.close()
            await pw.stop()

        facts: list[CollectedFact] = []
        media: list[CollectedMedia] = []

        for src in gallery:
            media.append(CollectedMedia(origin_url=src, label="갤러리"))

        for room in rooms:
            std_cap, max_cap = _parse_capacity(room.capacity)
            facts.append(CollectedFact(key="room_type", value=room.name, scope="unit", unit_name=room.name))
            if std_cap:
                facts.append(CollectedFact(key="standard_capacity", value=std_cap, scope="unit", unit_name=room.name))
            if max_cap:
                facts.append(CollectedFact(key="max_capacity", value=max_cap, scope="unit", unit_name=room.name))
            if room.bed:
                facts.append(CollectedFact(key="bed_type", value=room.bed, scope="unit", unit_name=room.name))
            # ★ room_intro 도 allow_llm=True 라 수집하지 않는다(위 intro 와 같은 이유).
            for src in room.images:
                media.append(CollectedMedia(origin_url=src, label="객실 사진", unit_name=room.name))

        if not facts and not media:
            return RawSource.failure(url, self.id, "객실·소개 정보를 찾지 못했다", LinkChannel.YANOLJA)

        LOG.i(f"[yanolja] {name} — 객실 {len(rooms)}개 · fact {len(facts)}건 · 사진 {len(media)}장")
        return RawSource(
            url=url,
            adapter_id=self.id,
            channel=LinkChannel.YANOLJA,
            text=page_text,
            facts=facts,
            media=media,
        )