o2o-site-AEO/solution/backend/services/grounding/channels.py
Mina Choi 11d30bb3d1 [chore] solution,admin,ontology: 코드 주석을 한 줄로 — 히스토리 주석 삭제
여러 줄 주석이 설명보다 경위(예전·실측·지적)를 적고 있어 읽는 사람이 결론을 찾기 어려웠다.

- ts·tsx·js·mjs·css·py 478개: 여러 줄 주석은 첫 문장 한 줄로, 과거형·날짜 문장은 삭제
- 주석 위치는 TypeScript 파서·파이썬 tokenize/ast 로 찾는다 — 문자열 안의 # · /* 는 건드리지 않는다
- eslint·ts·noqa·type: ignore 같은 지시 주석은 그대로 둔다

파이썬 275개 정리 전후 AST 동일, TS 298개 주석 뺀 토큰 동일(빈 JSX 주석 10곳만 차이).
site·frontend·admin tsc, site vitest 105 passed

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 16:05:19 +09:00

208 lines
7.4 KiB
Python

"""채널 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 호스트 → 채널 코드.
_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/...).
("map.naver.com", LinkChannel.NAVER_PLACE),
("instagram.com", LinkChannel.INSTAGRAM),
)
# 네이버 도메인 중 블로그·카페는 플레이스가 아니다 — 채널 판정을 분리한다.
_NAVER_BLOG_HOSTS = (
# 안내·도움말 페이지.
"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",
)
# ── 상세 페이지 판정 ──────────────────────────────────────────────────────
# 상세 페이지일 수 없는 경로 첫 조각.
_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 를 요구하는 채널.
_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 → 채널 코드."""
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 이 '이 가게의 상세 페이지' 인가."""
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
# 상한 — 같은 채널에서 서로 다른 업소가 우수수 딸려오는 것을 막는다.
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 목록을 만든다."""
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 = 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