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

"""MockAdapter — 네트워크를 타지 않는 가짜 수집기. **Phase 1 에 등록된 유일한 어댑터.**

크롤링 법적 검토가 끝나지 않아(docs/DECISIONS.md 1-1) 실제 수집 어댑터를 등록할 수 없다.
그렇다고 수집 이후 단계(fact 검증 전이 · 사람 확인 큐 · 사이트 빌드)를 못 만들 이유는 없으므로,
같은 계약(SourceAdapter)을 만족하는 가짜 소스를 두고 파이프라인 전체를 여기에 물려 돌린다.

**뱉는 fact key 는 전부 업종 스키마(common/category_schema)에 있는 것만 쓴다.**
스키마에 없는 key 는 fact 기록 단계에서 FACT_INVALID_KEY 로 전부 거부되므로,
그런 목데이터는 파이프라인을 검증하지 못하는 무의미한 데이터가 된다.
그래서 이 파일은 기동 시 자기 목데이터를 스키마와 대조하고, 어긋나면 즉시 예외를 던진다.

URL 규약 (업종을 URL 에서 읽어 결정적으로 동작한다)
    mock://lodging/pension-1
    mock://cafe/cafe-1?channel=naver_place
    https://mock.test/restaurant/r-1
"""
from urllib.parse import parse_qs, urlparse

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

_SCHEMES = ("mock",)
_HOSTS = ("mock.test",)

# URL 에 쓰는 업종 이름 → PlaceCategory. category_schema 의 `name` 과 같은 표기를 쓴다.
_CATEGORY_BY_NAME = {
    "lodging": PlaceCategory.LODGING,
    "cafe": PlaceCategory.CAFE,
    "restaurant": PlaceCategory.RESTAURANT,
    "clinic": PlaceCategory.CLINIC,
}

_CHANNEL_BY_NAME = {c.name.lower(): c for c in LinkChannel}


# ---- 업종별 목데이터 --------------------------------------------------------
# (key, value) — 전부 해당 업종 스키마에 존재하는 key 여야 한다.
# 소개문 계열(intro/room_intro/…)은 넣지 않는다. 그건 수집물이 아니라 generator 가 쓰는 문장이다.
_PLACE_FACTS = {
    PlaceCategory.LODGING: [
        ("check_in_time", "15:00"),
        ("check_out_time", "11:00"),
        ("cancel_policy", "이용 7일 전 100% 환불, 3일 전 50% 환불, 당일 취소 불가"),
        ("cooking_allowed", "true"),
        ("pet_allowed", "false"),
        ("smoking", "false"),
        ("parking", "true"),
        ("parking_capacity", "10"),
        ("wifi", "true"),
        ("bbq_available", "true"),
        ("bbq_fee", "20000"),
        ("breakfast", "false"),
        ("reception_hours", "09:00~21:00"),
    ],
    PlaceCategory.CAFE: [
        ("business_hours", "10:00~22:00"),
        ("closed_days", "매주 화요일"),
        ("break_time", "15:00~16:00"),
        ("last_order", "21:30"),
        ("pet_allowed", "true"),
        ("kids_allowed", "true"),
        ("parking", "true"),
        ("parking_free_hours", "2"),
        ("seat_count", "40"),
        ("wifi", "true"),
        ("power_outlet", "true"),
        ("takeout", "true"),
        ("signature_menu", "흑임자 라떼"),
        ("price_range", "5,000~9,000원"),
    ],
    PlaceCategory.RESTAURANT: [
        ("business_hours", "11:00~21:00"),
        ("closed_days", "매주 월요일"),
        ("break_time", "15:00~17:00"),
        ("last_order", "20:30"),
        ("reservation_required", "true"),
        ("reservation_channel", "전화 예약"),
        ("pet_allowed", "false"),
        ("kids_allowed", "true"),
        ("room_available", "true"),
        ("group_seat_max", "20"),
        ("parking", "true"),
        ("takeout", "true"),
        ("signature_menu", "한우 물회"),
        ("price_range", "15,000~30,000원"),
    ],
    PlaceCategory.CLINIC: [
        ("operating_hours", "평일 10:00~19:00, 토 10:00~14:00"),
        ("closed_days", "일요일·공휴일"),
        ("specialties", "피부과, 성형외과"),
        ("medical_staff", "피부과 전문의 2인, 성형외과 전문의 1인"),
        ("reservation_required", "true"),
        ("reservation_channel", "네이버 예약, 전화"),
        ("consultation_fee", "초진 상담료 없음"),
        ("insurance_covered", "질환 치료는 보험 적용, 미용 시술은 비급여"),
        ("cancel_policy", "예약 1일 전까지 변경 가능, 당일 취소는 재예약 제한"),
        ("night_clinic", "목요일 21:00까지"),
        ("weekend_clinic", "토요일 14:00까지"),
        ("parking", "true"),
    ],
}

# 단위(객실·메뉴·프로그램)별 목데이터 — {단위이름: [(key, value)]}
_UNIT_FACTS = {
    PlaceCategory.LODGING: {
        "A동 스탠다드": [
            ("room_type", "스탠다드"),
            ("standard_capacity", "2"),
            ("max_capacity", "4"),
            ("bed_type", "퀸 1"),
            ("has_kitchen", "true"),
            ("weekday_price", "120000"),
            ("weekend_price", "180000"),
        ],
        "B동 복층": [
            ("room_type", "복층"),
            ("standard_capacity", "4"),
            ("max_capacity", "6"),
            ("bed_type", "퀸 1 + 싱글 2"),
            ("has_kitchen", "true"),
            ("weekday_price", "190000"),
            ("weekend_price", "260000"),
        ],
    },
    PlaceCategory.CAFE: {
        "흑임자 라떼": [("menu_name", "흑임자 라떼"), ("menu_price", "6500")],
        "바스크 치즈케이크": [("menu_name", "바스크 치즈케이크"), ("menu_price", "7000")],
    },
    PlaceCategory.RESTAURANT: {
        "한우 물회": [("menu_name", "한우 물회"), ("menu_price", "18000")],
        "성게 비빔밥": [("menu_name", "성게 비빔밥"), ("menu_price", "22000"), ("menu_min_order", "2")],
    },
    PlaceCategory.CLINIC: {
        "레이저 토닝": [
            ("program_name", "레이저 토닝"),
            ("program_duration", "30"),
            ("price_adult", "100000"),
            ("anesthesia", "마취 없음"),
            ("downtime", "당일 일상생활 가능"),
            ("session_count", "5회 권장"),
        ],
        "보톡스": [
            ("program_name", "보톡스"),
            ("program_duration", "20"),
            ("price_adult", "150000"),
            ("anesthesia", "연고 마취"),
            ("downtime", "1~2일 미세 부기"),
        ],
    },
}

# 사진 — (라벨 후보, 단위 이름). 단위 이름이 None 이면 사업장 공용 사진.
_MEDIA = {
    PlaceCategory.LODGING: [
        ("외관", None),
        ("A동 침실", "A동 스탠다드"),
        ("A동 거실", "A동 스탠다드"),
        ("B동 침실", "B동 복층"),
        ("바비큐장", None),
    ],
    PlaceCategory.CAFE: [
        ("매장 내부", None),
        ("테라스", None),
        ("흑임자 라떼", "흑임자 라떼"),
        ("바스크 치즈케이크", "바스크 치즈케이크"),
    ],
    PlaceCategory.RESTAURANT: [
        ("매장 외관", None),
        ("홀 전경", None),
        ("한우 물회", "한우 물회"),
        ("성게 비빔밥", "성게 비빔밥"),
    ],
    PlaceCategory.CLINIC: [
        ("병원 외관", None),
        ("상담실", None),
        ("레이저 장비", "레이저 토닝"),
        ("시술실", "보톡스"),
    ],
}


def _validate_mock_data():
    """목데이터의 모든 key 가 업종 스키마에 있는지 확인한다. 최초 import 시 1회.

    어긋나면 즉시 예외 — 스키마에 없는 key 로 만든 목데이터는 fact 기록에서 전부 거부되므로,
    '테스트는 도는데 실제로는 하나도 안 들어가는' 상태를 배포 전에 드러낸다."""
    for category, rows in _PLACE_FACTS.items():
        schema = get_schema(category)
        for key, _value in rows:
            spec = schema.get(key)
            if spec is None:
                raise CollectError(f"MockAdapter 목데이터 오류: {schema.name} 스키마에 없는 key={key}")
            if spec.scope != "place":
                raise CollectError(f"MockAdapter 목데이터 오류: {schema.name}.{key} 는 place 스코프가 아니다")

    for category, units in _UNIT_FACTS.items():
        schema = get_schema(category)
        for unit_name, rows in units.items():
            for key, _value in rows:
                spec = schema.get(key)
                if spec is None:
                    raise CollectError(f"MockAdapter 목데이터 오류: {schema.name} 스키마에 없는 key={key} (단위 {unit_name})")
                if spec.scope != "unit":
                    raise CollectError(f"MockAdapter 목데이터 오류: {schema.name}.{key} 는 unit 스코프가 아니다")


_validate_mock_data()


class MockAdapter:
    """가짜 수집기. 같은 URL 이면 항상 같은 결과를 돌려준다(fetched_at 만 다르다).

    업종은 URL 에서 읽는다 — SourceAdapter.fetch(url) 계약을 그대로 지키면서
    업종별 목데이터를 고를 수 있게 하기 위함이다.
    """

    id = "mock"

    def can_handle(self, url: str) -> bool:
        try:
            parsed = urlparse(url or "")
        except ValueError:
            return False
        if parsed.scheme in _SCHEMES:
            return True
        return parsed.scheme in ("http", "https") and parsed.hostname in _HOSTS

    async def fetch(self, url: str) -> RawSource:
        """URL 에서 업종을 읽어 그 업종의 목데이터를 돌려준다.

        업종을 못 읽으면 예외가 아니라 실패 결과(ok=False)로 돌려준다 —
        채널 하나가 이상해도 나머지 수집이 멈추면 안 된다."""
        if not self.can_handle(url):
            return RawSource.failure(url, self.id, f"MockAdapter 가 처리할 수 없는 URL: {url}")

        category = self._category_of(url)
        if category is None:
            return RawSource.failure(url, self.id, f"URL 에서 업종을 읽지 못했다(가능: {sorted(_CATEGORY_BY_NAME)}): {url}")

        facts = [CollectedFact(key=k, value=v, scope="place") for k, v in _PLACE_FACTS[category]]
        for unit_name, rows in _UNIT_FACTS[category].items():
            facts += [CollectedFact(key=k, value=v, scope="unit", unit_name=unit_name) for k, v in rows]

        media = [
            CollectedMedia(
                origin_url=f"https://mock.test/img/{category.name.lower()}/{idx}.jpg",
                label=label,
                unit_name=unit_name,
            )
            for idx, (label, unit_name) in enumerate(_MEDIA[category], start=1)
        ]

        schema = get_schema(category)
        LOG.d(f"[collector:{self.id}] {schema.name} fact {len(facts)}건 · 사진 {len(media)}장 (url={url})")
        return RawSource(
            url=url,
            adapter_id=self.id,
            channel=self._channel_of(url),
            html=f"<html><body><h1>{schema.label} 목데이터</h1></body></html>",
            text=f"{schema.label} 목데이터",
            facts=facts,
            media=media,
        )

    # ---- URL 파싱 ----------------------------------------------------------
    @staticmethod
    def _category_of(url: str):
        """mock://lodging/... 또는 https://mock.test/lodging/... 에서 업종을 읽는다."""
        parsed = urlparse(url)
        # mock:// 는 첫 조각이 hostname 으로, https://mock.test/ 는 path 첫 조각으로 들어온다.
        candidates = [parsed.hostname or ""] + [p for p in parsed.path.split("/") if p]
        for token in candidates:
            category = _CATEGORY_BY_NAME.get(token.lower())
            if category is not None:
                return category
        return None

    @staticmethod
    def _channel_of(url: str) -> LinkChannel:
        """?channel=yanolja 로 채널을 지정할 수 있다. 없으면 ETC."""
        query = parse_qs(urlparse(url).query)
        name = (query.get("channel") or [""])[0].lower()
        return _CHANNEL_BY_NAME.get(name, LinkChannel.ETC)

    # ---- 편의 ---------------------------------------------------------------
    @staticmethod
    def url_for(category, slug: str = "1", channel: str | None = None) -> str:
        """이 어댑터가 처리할 수 있는 mock URL 을 만든다(테스트·시드용)."""
        cat = category if isinstance(category, PlaceCategory) else PlaceCategory(category)
        name = next(n for n, c in _CATEGORY_BY_NAME.items() if c is cat)
        suffix = f"?channel={channel}" if channel else ""
        return f"mock://{name}/{slug}{suffix}"