o2o-site-AEO/solution/backend/services/grounding/channels.py

"""채널 URL 후보를 믿어도 되는지 판정하는 자리.

Perplexity 가 돌려준 URL 목록에서 **상세 페이지가 아닌 것과 후기 블로그를 걸러낸다.**
외부 호출이 없는 순수 함수라 실호출 없이 테스트할 수 있다.

★ 왜 호출 코드에서 떼어냈나
  "왜 이 URL 이 빠졌나 / 왜 이 쓰레기가 남았나" 는 필터 규칙 문제지 HTTP 문제가 아니다.
  규칙만 따로 읽고 따로 고칠 수 있어야 한다(services/llm/__init__.py 의 겹 설명 참조).

★ 조용히 버리지 않는다. 걸러낸 것은 전부 (url, 사유) 로 돌려주고, 호출측이 로그와
  운영 화면에 싣는다 — 필터가 과해서 진짜 채널을 버렸는지 사람이 판단할 수 있어야 한다.
"""
import json
from dataclasses import dataclass
from urllib.parse import urlparse

from common.enums import LinkChannel
from common.logger import LOG

# 검토할 오탐 후보가 과도하게 늘지 않도록 제한한다.
MAX_LINKS = 6
MAX_LINKS_PER_CHANNEL = 3


# URL 호스트 → 채널 코드. 접미사 매칭이라 서브도메인(place.naver.com 등)도 잡힌다.
_HOST_CHANNEL = (
    ("yanolja.com", LinkChannel.YANOLJA),
    ("goodchoice.kr", LinkChannel.GOODCHOICE),
    ("yeogi.com", LinkChannel.GOODCHOICE),
    ("place.naver.com", LinkChannel.NAVER_PLACE),
    ("naver.me", LinkChannel.NAVER_PLACE),
    ("m.place.naver.com", LinkChannel.NAVER_PLACE),
    # 사람이 실제로 공유하는 주소가 이 호스트다(map.naver.com/p/entry/place/...).
    # 블로그·카페는 위 _NAVER_BLOG_HOSTS 가 먼저 걸러내므로 여기 넣어도 후기가 섞이지 않는다.
    ("map.naver.com", LinkChannel.NAVER_PLACE),
    ("instagram.com", LinkChannel.INSTAGRAM),
)

# 네이버 도메인 중 블로그·카페는 플레이스가 아니다 — 채널 판정을 분리한다.
_NAVER_BLOG_HOSTS = (
    # ★ 안내·도움말 페이지. 가게 채널이 아니다 — 실측으로 '도플로' 수집에
    #   pages.map.naver.com/useful-tips 가 유일한 네이버 링크로 등록돼 사진 0장이 됐다.
    "pages.map.naver.com", "help.naver.com", "guide.naver.com",
    "blog.naver.com", "cafe.naver.com", "m.blog.naver.com", "m.cafe.naver.com",
    "post.naver.com", "m.post.naver.com", "in.naver.com",
)


# ── 상세 페이지 판정 ──────────────────────────────────────────────────────
# ★ 실측(2026-08-27 '핑크비치펜션', 발견 15건)에서 확인된 쓰레기 유형:
#     https://nol.yanolja.com/                     루트 — 이 가게와 무관
#     https://www.goodchoice.kr/                   루트
#     https://nol.yanolja.com/sub-home/pension     카테고리(업종 목록)
#     https://nol.yanolja.com/programmatic/...     SEO 랜딩
#     https://blog.naver.com/...                   후기 4건
#   이런 URL 을 확정 대기 목록에 남기면 사람이 15건을 일일이 봐야 한다.

# 상세 페이지일 수 없는 경로 첫 조각. 숫자 ID 가 뒤에 붙어도 상세가 아니다(SEO 랜딩·목록).
_NON_DETAIL_PREFIXES = frozenset({
    "sub-home", "sub_home", "programmatic", "search", "event", "promotion",
    "category", "theme", "curation", "magazine", "notice", "help", "login", "signup",
})

# 상세 URL 이 숫자 ID 를 갖지 않는 예외 호스트(단축 링크·지도).
_ID_OPTIONAL_HOSTS = ("naver.me", "map.naver.com", "instagram.com")

# 상세 판정에 숫자 ID 를 요구하는 채널. OTA 상세는 항상 숫자 ID 를 갖는다.
_ID_REQUIRED_CHANNELS = (LinkChannel.YANOLJA, LinkChannel.GOODCHOICE, LinkChannel.NAVER_PLACE)

# 탈락 사유 코드 — 운영자가 "왜 빠졌나"를 보는 값이라 문자열을 고정한다.
REASON_NON_HTTP = "non_http"
REASON_DUPLICATE = "duplicate"
REASON_BLOG = "blog"
REASON_ROOT = "root"
REASON_CATEGORY = "category"
REASON_OVERFLOW = "overflow"        # 상한 초과로 잘림 — 버린 게 아니라 순위가 밀린 것



@dataclass(frozen=True)
class DiscoveredLink:
    """발견된 채널 URL 1건. **사실이 아니라 '여기를 보라'는 포인터다.**"""

    url: str
    channel: LinkChannel
    title: str | None = None


def classify_url(url: str) -> LinkChannel:
    """URL → 채널 코드. 모르는 도메인은 ETC 로 떨어뜨린다(버리지 않는다)."""
    try:
        host = (urlparse(url).hostname or "").lower()
    except (ValueError, AttributeError):
        return LinkChannel.ETC
    if not host:
        return LinkChannel.ETC

    # 블로그·카페는 naver.com 접미사에 걸리기 전에 먼저 걸러낸다.
    if any(host == h or host.endswith("." + h) for h in _NAVER_BLOG_HOSTS):
        return LinkChannel.BLOG

    for suffix, channel in _HOST_CHANNEL:
        if host == suffix or host.endswith("." + suffix):
            return channel
    return LinkChannel.ETC


def _path_segments(url: str) -> list[str]:
    """URL 경로를 조각으로 나눈다. 빈 조각은 버린다."""
    try:
        path = urlparse(url).path or ""
    except (ValueError, AttributeError):
        return []
    return [seg for seg in path.split("/") if seg]


def _detail_reject_reason(url: str, channel: LinkChannel) -> str | None:
    """이 URL 이 '이 가게의 상세 페이지' 인가. 아니면 탈락 사유를, 맞으면 None 을 돌려준다.

    ★ 판정을 못 하겠으면 통과시킨다(사람이 확정 단계에서 본다). 여기서 과하게 버리면
      진짜 채널을 잃는데, 그건 필터가 없는 것보다 나쁘다.
    """
    segments = _path_segments(url)

    # 루트('/' 또는 경로 없음)는 업소와 무관한 서비스 첫 화면이다.
    if not segments:
        return REASON_ROOT

    # SEO 랜딩·목록 경로는 숫자 ID 가 뒤에 붙어도 상세가 아니다.
    if segments[0].lower() in _NON_DETAIL_PREFIXES:
        return REASON_CATEGORY

    try:
        host = (urlparse(url).hostname or "").lower()
    except (ValueError, AttributeError):
        host = ""
    # 단축 링크(naver.me)·지도·인스타 프로필은 숫자 ID 를 갖지 않는다 — 경로만 있으면 통과.
    if any(host == h or host.endswith("." + h) for h in _ID_OPTIONAL_HOSTS):
        return None

    # OTA 상세 URL 은 항상 숫자 ID 를 갖는다. 없으면 목록·카테고리다.
    if channel in _ID_REQUIRED_CHANNELS:
        if not any(seg.isdigit() for seg in segments):
            return REASON_CATEGORY

    return None


def filter_links(links: list[DiscoveredLink], include_blogs: bool) -> tuple[list[DiscoveredLink], list[tuple[str, str]]]:
    """발견된 URL 을 '이 가게의 상세 페이지' 인 것만 남긴다.

    ★ 전부 걸러지면 빈 목록을 돌려준다. 억지로 하나 남기지 않는다 —
      '못 찾음'은 정상 결과이고, 사장님 직접 입력으로 폴백한다.
    """
    kept: list[DiscoveredLink] = []
    dropped: list[tuple[str, str]] = []
    per_channel: dict[LinkChannel, int] = {}
    for link in links:
        # 블로그·카페는 채널이 아니라 후기다. 크롤링해도 이 가게의 공식 정보가 아니다.
        if link.channel is LinkChannel.BLOG and not include_blogs:
            dropped.append((link.url, REASON_BLOG))
            continue
        reason = _detail_reject_reason(link.url, link.channel)
        if reason:
            dropped.append((link.url, reason))
            continue
        # ★ 상한 — 같은 채널에서 서로 다른 업소가 우수수 딸려오는 것을 막는다.
        #   links 는 '모델이 고른 것 → search_results' 순으로 들어오므로 앞쪽이 더 믿을 만하다.
        used = per_channel.get(link.channel, 0)
        if used >= MAX_LINKS_PER_CHANNEL or len(kept) >= MAX_LINKS:
            dropped.append((link.url, REASON_OVERFLOW))
            continue
        per_channel[link.channel] = used + 1
        kept.append(link)
    return kept, dropped


def _parse_content_links(content: str) -> list[dict]:
    """구조화 출력(JSON) 본문에서 links 배열을 꺼낸다. 깨져 있으면 빈 목록."""
    if not content:
        return []
    try:
        doc = json.loads(content)
    except (json.JSONDecodeError, TypeError):
        LOG.w("[perplexity] 구조화 출력 파싱 실패 — search_results 만 사용한다")
        return []
    links = doc.get("links")
    return links if isinstance(links, list) else []


def collect_links(payload: dict) -> tuple[list[DiscoveredLink], list[tuple[str, str]]]:
    """모델이 고른 links + search_results(실제 검색 출처)를 합쳐 URL 목록을 만든다.

    search_results 를 함께 쓰는 이유: 모델 답변은 환각이 섞이지만 search_results 는
    실제로 조회된 URL 이라 더 믿을 만하다. 어차피 다음 단계(동일 업소 검증)가 걸러낸다.

    반환: (URL 목록, [(버린 URL, 사유), ...]). 중복·비 http 도 사유로 남긴다."""
    seen: set[str] = set()
    out: list[DiscoveredLink] = []
    dropped: list[tuple[str, str]] = []

    def _add(url, title):
        if not isinstance(url, str):
            return
        url = url.strip()
        if not url:
            return
        if url in seen:
            dropped.append((url, REASON_DUPLICATE))
            return
        if not url.lower().startswith(("http://", "https://")):
            dropped.append((url, REASON_NON_HTTP))
            return
        seen.add(url)
        out.append(DiscoveredLink(url=url, channel=classify_url(url), title=(title or None)))

    choices = payload.get("choices") or []
    content = ""
    if choices and isinstance(choices[0], dict):
        content = ((choices[0].get("message") or {}).get("content")) or ""
    for row in _parse_content_links(content):
        if isinstance(row, dict):
            _add(row.get("url"), row.get("title"))

    for row in payload.get("search_results") or []:
        if isinstance(row, dict):
            _add(row.get("url"), row.get("title"))

    return out, dropped


def search_count(payload: dict) -> int:
    """이번 호출의 검색 횟수. usage 에 없으면 search_results 개수로 대체한다."""
    usage = payload.get("usage") or {}
    for key in ("num_search_queries", "search_queries", "num_searches"):
        value = usage.get(key)
        if isinstance(value, int):
            return value
    results = payload.get("search_results")
    return len(results) if isinstance(results, list) else 0