"""채널 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