o2o-site-AEO/solution/backend/scripts/import_yanolja_test_place.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
야놀자 크롤링 결과를 테스트 place 하나에 적재하는 1회성 스크립트.
=================================================================

★ 운영 파이프라인이 아니다 — services/collector/registry.py 에 등록하지 않고,
  worker 잡큐에도 연결하지 않는다. 개발자가 수동으로 실행하는 CLI 전용.
  (registry.py·static_html_adapter.py 의 야놀자 관련 구조적 차단은 그대로 유효하다 —
   이 스크립트는 그 결론을 뒤집지 않는다. 테스트 목적 한정, 실제 배포 전 별도 협의 필요.)

동작
  1. place_id 로 places 테이블에서 업장을 조회한다. 주소는 따로 입력받지 않고
     그 업장의 road_address(없으면 address)를 그대로 검색어로 쓴다.
  2. verified_at 이 비어 있으면 중단한다(운영 run_collect 와 같은 가드 —
     동일 업소 검증 전에는 수집하지 않는다).
  3. scripts/yanolja_search_and_crawl.py 로 그 주소를 검색·크롤링한다.
  4. 결과를 CollectedFact/CollectedMedia 로 매핑해서, 기존 collect_service.py 의
     ensure_units/store_facts/store_media 를 그대로 호출한다 — FactService 를 그대로
     통과하므로 새 값은 UNVERIFIED/PENDING_OWNER/PENDING_REVIEW 로만 만들어진다.
     이 상태를 이 스크립트가 직접 VERIFIED/APPROVED 로 바꾸는 일은 없다.

사용법 (solution/backend 에서, 가상환경 안에서)
  python scripts/import_yanolja_test_place.py --place-id <UUID>              (dry-run: 매핑만 출력)
  python scripts/import_yanolja_test_place.py --place-id <UUID> --commit     (실제 DB 적재)

  PGSSLMODE=disable DB_PASSWORD=... 를 dev-env-quirks 메모대로 주입해야 한다(APP_ENV=local 기본).
"""
from __future__ import annotations

import argparse
import asyncio
import os
import re
import sys
import uuid

# scripts/ 는 solution/backend 바로 아래이므로, backend 자체(부모 디렉터리)를
# sys.path 에 넣어야 common/services 등을 import 할 수 있다.
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault("APP_ENV", "local")

from playwright.sync_api import sync_playwright  # noqa: E402
from sqlalchemy import select  # noqa: E402

from scripts.yanolja_search_and_crawl import StayData, search_and_crawl  # noqa: E402

from common.database.db_session_manager import DB_SESSION_MNG  # noqa: E402
from common.database.model.models import places  # noqa: E402
from common.enums import DBWRType, ErrorType  # noqa: E402
from services.collector.base import CollectedFact, CollectedMedia, RawSource  # noqa: E402
from services.collect_service import ensure_units, store_facts, store_media  # noqa: E402


def _parse_capacity(capacity: str | None) -> tuple[str | None, str | None]:
    """'기준 2인 / 최대 4인' → ('2', '4')."""
    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 stay_data_to_source(data: StayData) -> RawSource:
    """StayData → RawSource(facts/media). lodging.json 의 scope=unit 필드 중
    크롤러가 실제로 채울 수 있는 것만 매핑한다(room_type/standard_capacity/max_capacity 는
    required=true 라 반드시 채운다. weekday_price 등 요금은 크롤러가 안 모으므로 비워둔다).
    """
    facts: list[CollectedFact] = []
    media: list[CollectedMedia] = []

    if data.overview.strip():
        facts.append(CollectedFact(key="intro", value=data.overview.strip(), scope="place"))
    for url in data.photos:
        media.append(CollectedMedia(origin_url=url, label="갤러리"))

    for room in data.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))
        if room.description:
            facts.append(CollectedFact(key="room_intro", value=room.description, scope="unit", unit_name=room.name))
        for url in room.images:
            media.append(CollectedMedia(origin_url=url, label="객실 사진", unit_name=room.name))

    return RawSource(url=data.url, adapter_id="yanolja_test_import", facts=facts, media=media)


def _print_preview(source: RawSource) -> None:
    print(f"\n[미리보기] {source.url}")
    place_facts = [f for f in source.facts if f.scope == "place"]
    print(f"  place-scope fact {len(place_facts)}건: {[f.key for f in place_facts]}")
    for unit_name in source.unit_names():
        unit_facts = [f for f in source.facts if f.unit_name == unit_name]
        unit_media = [m for m in source.media if m.unit_name == unit_name]
        print(f"  [{unit_name}] fact {len(unit_facts)}건, 사진 {len(unit_media)}장")
        for f in unit_facts:
            print(f"      {f.key} = {f.value}")
    place_media = [m for m in source.media if m.unit_name is None]
    print(f"  place-scope 사진 {len(place_media)}장")


async def _load_place(place_id: str):
    stmt = select(places).where(places.place_id == uuid.UUID(place_id), places.deleted == False)  # noqa: E712
    err, row = await DB_SESSION_MNG.execute_lambda(
        places.DBType(),
        DBWRType.DB_READ.value,
        lambda s: DB_SESSION_MNG.execute(s, stmt),
    )
    if err != ErrorType.SUCCESS or not row:
        return None
    return row[0]


def _crawl_sync(address: str, headful: bool) -> StayData:
    """동기 Playwright API는 asyncio 루프가 도는 스레드에서 못 쓴다 — 별도 스레드에서 실행한다."""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=not headful)
        context = browser.new_context(
            locale="ko-KR",
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                "(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36"
            ),
        )
        page = context.new_page()
        try:
            return search_and_crawl(page, address)
        finally:
            browser.close()


async def run(place_id: str, commit: bool, headful: bool) -> None:
    place = await _load_place(place_id)
    if place is None:
        raise SystemExit(f"place를 찾을 수 없습니다: {place_id}")
    if place.verified_at is None:
        raise SystemExit(
            "동일 업소 검증(verify) 전에는 수집하지 않습니다 — 운영 collect_service.run_collect 와 같은 가드."
        )

    address = place.road_address or place.address
    if not address:
        raise SystemExit("이 place에는 주소가 없습니다.")

    print(f"[대상] {place.name} ({place_id}) — 검색 주소: {address}")

    data = await asyncio.to_thread(_crawl_sync, address, headful)

    print(f"[크롤링 완료] {data.name} — 객실 {len(data.rooms)}개, 갤러리 사진 {len(data.photos)}장")
    source = stay_data_to_source(data)
    _print_preview(source)

    if not commit:
        print("\n(dry-run) DB에는 쓰지 않았습니다. 실제로 저장하려면 --commit 을 주세요.")
        await DB_SESSION_MNG.dispose_all()
        return

    from common.models.gmodel import UserInfo

    actor = UserInfo(user_id=str(place.owner_user_id), id="yanolja_test_import", role=1)

    unit_map = await ensure_units(place_id, [source])
    facts_stat = await store_facts(actor, place_id, [source], unit_map)
    media_stat = await store_media(place_id, [source], unit_map)

    print(f"\n[적재 완료] unit {len(unit_map)}개, fact={facts_stat}, media={media_stat}")
    await DB_SESSION_MNG.dispose_all()


def main() -> None:
    parser = argparse.ArgumentParser(description="야놀자 크롤링 → 테스트 place 객실 데이터 임포트(1회성)")
    parser.add_argument("--place-id", required=True, help="대상 place_id(UUID)")
    parser.add_argument("--commit", action="store_true", help="실제로 DB에 적재(기본은 dry-run)")
    parser.add_argument("--headful", action="store_true", help="브라우저 창을 보이게 실행(디버깅용)")
    args = parser.parse_args()
    asyncio.run(run(args.place_id, args.commit, args.headful))


if __name__ == "__main__":
    main()