723 lines
35 KiB
Python
723 lines
35 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
네이버 플레이스 크롤링 레이어 (AI브리핑 · 사진) — animation 3종(조선/삼국지/그리스로마) 공유본
|
|
====================================================================================
|
|
animation 계열이 공유하는 '링크 1개' 네이버 크롤러. ★ 이 파일 하나로 자립 동작.
|
|
· fetch_briefing 가게이름 + 업종타입(cat) + AI 브리핑(불릿)
|
|
· fetch_and_select 공식 사진 크롤 + Gemini 비전 선별(★인물 항상 제외)
|
|
공유 헬퍼(MOBILE_UA / STEALTH / extract_place)는 한 번만 정의.
|
|
※ 클립(fetch_clips)·지도(fetch_map)는 narration/kakao 폴더 삭제와 함께 제거됨(animation 미사용).
|
|
|
|
[정리한 것]
|
|
- 방문자 후기(fetch_reviews) 제거: AI요약(브리핑)만 나레이션 근거로 쓰므로 불필요.
|
|
- fetch_briefing 은 항상 3-tuple (store, bullets, cat) 반환. cat 안 쓰는 형식은 그냥 무시.
|
|
- 사진 선별은 항상 '인물 사진 제외'(초상권). 끄고 싶으면 fetch_and_select(..., exclude_person=False).
|
|
|
|
[봇 차단 회피] 자동화 흔적 줄인 스텔스 컨텍스트 + 모바일 UA 로 접근.
|
|
설치: pip install playwright google-genai pillow httpx ; playwright install chromium
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import shutil
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urlparse
|
|
|
|
import httpx
|
|
from google.genai import types
|
|
from PIL import Image
|
|
from playwright.async_api import async_playwright
|
|
|
|
|
|
# ============================== 공유 헬퍼 (UA · 스텔스 · place id) ==============================
|
|
|
|
MOBILE_UA = ("Mozilla/5.0 (iPhone; CPU iPhone OS 16_5 like Mac OS X) "
|
|
"AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Mobile/15E148 Safari/604.1")
|
|
# 자동화 탐지 우회용 init 스크립트
|
|
STEALTH = ("Object.defineProperty(navigator,'webdriver',{get:()=>undefined});"
|
|
"Object.defineProperty(navigator,'languages',{get:()=>['ko-KR','ko']});"
|
|
"window.chrome={runtime:{}};")
|
|
|
|
VISION_MODEL = "gemini-2.5-flash"
|
|
|
|
|
|
def extract_place(final_url: str):
|
|
"""리다이렉트된 최종 URL → (업종타입, place id). 예: ('restaurant', '1234567')."""
|
|
m = re.search(r"place\.naver\.com/([a-z]+)/(\d+)", final_url)
|
|
if m:
|
|
return m.group(1), m.group(2)
|
|
m2 = re.search(r"(\d{7,})", final_url)
|
|
return "restaurant", (m2.group(1) if m2 else None)
|
|
|
|
|
|
def looks_like_naver_place(s: str) -> bool:
|
|
return bool(s) and ("naver.me/" in s or "place.naver.com" in s or "map.naver.com" in s)
|
|
|
|
|
|
def _clean_store(raw: str) -> str:
|
|
"""og:title/title 값에서 실제 가게명만 정리. 예: '골목냉면 : 네이버\\x1c' → '골목냉면'."""
|
|
if not raw:
|
|
return ""
|
|
s = re.sub(r"[\x00-\x1f]", "", raw) # 제어문자 제거
|
|
s = re.split(r"\s*[:\-|]\s*네이버", s)[0] # ' : 네이버', ' - 네이버 지도' 접미사 제거
|
|
s = re.sub(r"\s+", " ", s).strip()
|
|
return s.splitlines()[0].strip() if s else ""
|
|
|
|
|
|
async def _store_name(page) -> str:
|
|
"""가게명 추출. og:title 메타(가장 안정적) 우선 → title → 화면 요소.
|
|
섹션 헤딩('AI 브리핑' 등)이나 '네이버' 포함 값은 배제한다.
|
|
(네이버가 span.GHAhO 같은 난독 클래스를 자주 바꿔 화면 셀렉터만으론 불안정하므로 메타 우선.)"""
|
|
async def _meta():
|
|
return await page.locator("meta[property='og:title']").first.get_attribute("content", timeout=1500)
|
|
for getter in (_meta, page.title):
|
|
try:
|
|
name = _clean_store(await getter())
|
|
if name and name != "AI 브리핑" and "네이버" not in name:
|
|
return name
|
|
except Exception:
|
|
pass
|
|
for sel in ("span.GHAhO", "#_title span", "h2"): # 최후 폴백(섹션명 배제)
|
|
try:
|
|
t = (await page.locator(sel).first.inner_text(timeout=1200)).strip().splitlines()[0]
|
|
if t and t != "AI 브리핑" and not t.startswith("AI "):
|
|
return t
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
async def _new_page(p, headful=False, viewport=None, device_scale_factor=None):
|
|
"""공통 브라우저 컨텍스트(스텔스 + 모바일 UA) → (browser, ctx, page)."""
|
|
browser = await p.chromium.launch(
|
|
headless=not headful,
|
|
args=["--disable-blink-features=AutomationControlled", "--no-sandbox"])
|
|
kwargs = dict(user_agent=MOBILE_UA, locale="ko-KR",
|
|
viewport=viewport or {"width": 414, "height": 896},
|
|
extra_http_headers={"Accept-Language": "ko-KR,ko;q=0.9"})
|
|
if device_scale_factor:
|
|
kwargs["device_scale_factor"] = device_scale_factor
|
|
ctx = await browser.new_context(**kwargs)
|
|
await ctx.add_init_script(STEALTH)
|
|
page = await ctx.new_page()
|
|
return browser, ctx, page
|
|
|
|
|
|
# ============================== 브리핑 (이름 · 업종타입 · AI요약) ==============================
|
|
|
|
# 홈에서 'AI 브리핑' 불릿 문장만 뽑는 브라우저측 스크립트.
|
|
# [핵심] 브리핑 본문 문장은 출처번호(<a>1</a>…)를 '자식'으로 품은 노드 안에 있어서
|
|
# leaf(자식없는) 추출로는 절대 안 잡힌다(라벨/번호만 leaf로 남는다).
|
|
# → 인트로 + 본문을 모두 품은 컨테이너의 innerText 를 통째로 떠서 줄 단위로 판다.
|
|
# → 본문 불릿과 '출처 리뷰 인용문'을 가르는 구조 신호:
|
|
# 브리핑 불릿 문장 = 뒤에 출처번호만 따라옴.
|
|
# 리뷰 인용문 = 문장 뒤 몇 줄 안에 '날짜(26.05.13.)'가 붙는다.
|
|
# → 문장 뒤 3줄 안에 날짜가 나오면 거기서부터 리뷰 → 그 직전까지가 브리핑.
|
|
# 이 방식은 라벨(객실/시설/위치/서비스…)이 뭐로 바뀌든 안 흔들린다.
|
|
_EXTRACT_JS = r"""() => {
|
|
const hasHeader = [...document.querySelectorAll('body *')]
|
|
.some(e => e.children.length === 0 && /^AI\s*브리핑/.test((e.innerText||'').trim()));
|
|
|
|
// 인트로 + 본문을 함께 품은 '가장 안쪽' 컨테이너
|
|
const INTRO = /(리뷰[를에]?\s*(바탕|기반|활용))|요약(했|해|합|하면)|정리한 (정보|내용)|다음과 같습니다/;
|
|
const cands = [...document.querySelectorAll('body *')]
|
|
.filter(e => { const t = e.innerText || '';
|
|
return INTRO.test(t) && t.length > 40 && /[가-힣]{6,}/.test(t); });
|
|
cands.sort((a, b) => (a.innerText||'').length - (b.innerText||'').length);
|
|
if (!cands.length) return {found: hasHeader, bullets: []};
|
|
const block = cands[0].innerText || '';
|
|
|
|
const lines = block.split('\n').map(s => s.trim()).filter(Boolean);
|
|
const isNum = s => /^\d+$/.test(s);
|
|
const isDate = s => /^\d{2}\.\d{2}\.\d{2}\.?$/.test(s) || /^\d{4}\s*년/.test(s);
|
|
const isSent = s => s.length >= 12 && /[가-힣]/.test(s) && !isNum(s) && !INTRO.test(s);
|
|
|
|
// 인트로 다음부터, 리뷰 인용(문장 뒤 3줄 내 날짜) 시작 전까지의 문장 = 브리핑 불릿
|
|
let start = lines.findIndex(t => INTRO.test(t));
|
|
start = start < 0 ? 0 : start + 1;
|
|
const bullets = [];
|
|
for (let i = start; i < lines.length; i++) {
|
|
if (!isSent(lines[i])) continue;
|
|
if (lines.slice(i + 1, i + 4).some(isDate)) break; // 리뷰 인용 시작 → 종료
|
|
bullets.push(lines[i].replace(/^[가-힣A-Za-z ]{1,12}:\s*/, '')); // '객실 정보: ' 라벨 제거
|
|
}
|
|
return {found: hasHeader || bullets.length > 0, bullets};
|
|
}"""
|
|
|
|
|
|
async def _fetch(url, headful=False):
|
|
async with async_playwright() as p:
|
|
browser, ctx, page = await _new_page(p, headful=headful)
|
|
|
|
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
|
|
await page.wait_for_timeout(2500)
|
|
cat, pid = extract_place(page.url)
|
|
if not pid:
|
|
await browser.close()
|
|
raise RuntimeError("place id 를 못 찾음. URL 확인 또는 headful=True 로 화면 확인.")
|
|
|
|
# 항상 표준 홈 URL 로 이동(리다이렉트 직후엔 섹션이 덜 렌더됨)
|
|
await page.goto(f"https://m.place.naver.com/{cat}/{pid}/home",
|
|
wait_until="domcontentloaded", timeout=30000)
|
|
await page.wait_for_timeout(2000)
|
|
|
|
store = await _store_name(page)
|
|
|
|
# [핵심] AI 브리핑은 페이지 하단에서 lazy-load 된다.
|
|
# 헤더('AI 브리핑')가 보여도 본문(불릿)은 그 섹션이 화면에 들어와야 네트워크로 채워진다.
|
|
# → 예전 코드는 헤더가 8초 안에 보이면 스크롤을 '안 하고' 바로 추출해서,
|
|
# 본문이 아직 안 그려진 순간엔 빈 결과(found=False)가 나왔다(간헐 실패의 원인).
|
|
# → 이제는 브리핑 섹션을 화면에 띄우고, 본문 불릿이 실제로 나올 때까지 스크롤하며 폴링한다.
|
|
res = {"found": False, "bullets": []}
|
|
for attempt in range(14): # 최대 ~14회(≈11초) 폴링
|
|
# 브리핑 헤더를 뷰에 올려 본문 로딩을 유발(있으면). 없으면 아래로 조금씩 스크롤.
|
|
try:
|
|
loc = page.get_by_text("AI 브리핑", exact=False).first
|
|
await loc.scroll_into_view_if_needed(timeout=1500)
|
|
except Exception:
|
|
await page.mouse.wheel(0, 1600)
|
|
await page.wait_for_timeout(700)
|
|
res = await page.evaluate(_EXTRACT_JS)
|
|
if res.get("bullets"):
|
|
# 불릿이 처음 잡혀도 본문이 아직 다 안 채워졌을 수 있다(2/3만 렌더된 순간).
|
|
# 한 번 더 기다렸다 재추출해서 더 많이 잡히면 그걸 택한 뒤 종료.
|
|
await page.wait_for_timeout(900)
|
|
res2 = await page.evaluate(_EXTRACT_JS)
|
|
if len(res2.get("bullets", [])) > len(res.get("bullets", [])):
|
|
res = res2
|
|
break
|
|
|
|
await browser.close()
|
|
|
|
if not res.get("found"):
|
|
return store, None, cat
|
|
seen, bullets = set(), []
|
|
for t in res.get("bullets", []):
|
|
t = re.sub(r"\s+", " ", t).strip()
|
|
if t and t not in seen:
|
|
seen.add(t); bullets.append(t)
|
|
return store, bullets, cat
|
|
|
|
|
|
def fetch_briefing(url, headful=False):
|
|
"""동기 진입점. (가게이름, [불릿...], 업종타입) 반환.
|
|
업종타입 = 링크의 플레이스 타입(restaurant/accommodation/hairshop…). 브리핑 없으면 불릿이 None.
|
|
※ animation 처럼 업종이 필요 없는 형식은 3번째 값을 그냥 무시하면 된다."""
|
|
return asyncio.run(_fetch(url, headful=headful))
|
|
|
|
|
|
# ============================== 리뷰 크롤링 (브리핑 없을 때 폴백) ==============================
|
|
|
|
# 방문자 리뷰 '본문' 셀렉터. 네이버가 난독 클래스(pui__…)를 자주 바꾸므로 우선순위대로 시도해
|
|
# '충분히(3개+) 잡히는 첫 셀렉터'를 쓴다. ★ 최다매칭이 아님 — 작성자/메타 링크(a.pui__GStJHb 등)가
|
|
# 더 많이 잡혀서 그걸 고르면 후기가 아니라 '작성자 이름·리뷰수'가 수집되는 문제가 있었다.
|
|
# 리뷰 본문 전용 클래스(.pui__vn15t2)를 최우선으로 둔다.
|
|
_REVIEW_JS = r"""() => {
|
|
const SELS = ['.pui__vn15t2', 'a.pui__xtsQN-', 'div.pui__vn15t2', 'span.pui__gfuUIT'];
|
|
const clean = s => (s || '').replace(/\s*더보기\s*$/,'').replace(/\s+/g,' ').trim();
|
|
for (const sel of SELS) {
|
|
let els;
|
|
try { els = [...document.querySelectorAll(sel)]; } catch (e) { continue; }
|
|
const texts = els.map(e => clean(e.innerText))
|
|
.filter(t => t.length >= 8 && /[가-힣]/.test(t));
|
|
if (texts.length >= 3) return texts;
|
|
}
|
|
return [];
|
|
}"""
|
|
|
|
|
|
async def _fetch_reviews(url, max_reviews=30, headful=False):
|
|
"""방문자 리뷰 탭에서 '더보기'를 모두 펼쳐 리뷰 본문을 수집. (가게이름, [리뷰문장…]) 반환."""
|
|
async with async_playwright() as p:
|
|
browser, ctx, page = await _new_page(p, headful=headful)
|
|
|
|
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
|
|
await page.wait_for_timeout(2500)
|
|
cat, pid = extract_place(page.url)
|
|
if not pid:
|
|
await browser.close()
|
|
raise RuntimeError("place id 를 못 찾음.")
|
|
store = await _store_name(page)
|
|
|
|
# 방문자 리뷰 탭으로 이동(구조에 따라 /review/visitor 또는 /review)
|
|
reviews, seen = [], set()
|
|
for tab in ("review/visitor", "review"):
|
|
try:
|
|
resp = await page.goto(f"https://m.place.naver.com/{cat}/{pid}/{tab}",
|
|
wait_until="domcontentloaded", timeout=25000)
|
|
except Exception:
|
|
continue
|
|
if resp and resp.status >= 400:
|
|
continue
|
|
await page.wait_for_timeout(2200)
|
|
|
|
# 스크롤로 리뷰를 더 로드하면서, 매 라운드 '더보기'를 눌러 본문을 전부 펼친 뒤 수집.
|
|
stagnant = 0
|
|
for _ in range(16):
|
|
# ★ 리뷰 '본문(.pui__vn15t2)' 안의 더보기만 클릭한다.
|
|
# 페이지 전체의 '더보기'를 무차별 클릭하면 사진/방문정보 더보기가 눌려
|
|
# 리뷰 목록이 통째로 사라지는(0개 수집) 문제가 있었다 → 본문 내부로 한정.
|
|
try:
|
|
bodies = page.locator(".pui__vn15t2")
|
|
bn = await bodies.count()
|
|
for i in range(bn):
|
|
mb = bodies.nth(i).locator(
|
|
"xpath=.//*[normalize-space(text())='더보기']")
|
|
if await mb.count():
|
|
try:
|
|
await mb.first.click(timeout=400)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
await page.wait_for_timeout(400)
|
|
|
|
for t in await page.evaluate(_REVIEW_JS):
|
|
if t not in seen:
|
|
seen.add(t); reviews.append(t)
|
|
|
|
if len(reviews) >= max_reviews:
|
|
break
|
|
before = len(reviews)
|
|
# 무한스크롤 트리거: 문서 맨 아래로 점프 + 휠 + 대기(리뷰 추가 로드 유발)
|
|
try:
|
|
await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
|
|
except Exception:
|
|
pass
|
|
await page.mouse.wheel(0, 3200)
|
|
await page.wait_for_timeout(1100)
|
|
stagnant = stagnant + 1 if len(reviews) == before else 0
|
|
if stagnant >= 4: # 4라운드 연속 새 리뷰 없음 → 끝까지 봤다고 판단
|
|
break
|
|
|
|
if reviews: # 이 탭에서 뭔가 건졌으면 다른 탭은 시도 안 함
|
|
break
|
|
|
|
await browser.close()
|
|
return store, reviews[:max_reviews]
|
|
|
|
|
|
def fetch_reviews(url, max_reviews=30, headful=False):
|
|
"""동기 진입점. 방문자 리뷰 본문을 '더보기'까지 펼쳐 최대 max_reviews 개 수집. (가게이름, [리뷰…])."""
|
|
return asyncio.run(_fetch_reviews(url, max_reviews=max_reviews, headful=headful))
|
|
|
|
|
|
_REVIEW_SUMMARY_SCHEMA = {
|
|
"type": "object",
|
|
"properties": {
|
|
"bullets": {"type": "array", "items": {"type": "string"},
|
|
"description": "가게 홍보 소재로 쓸 핵심 특징 불릿(각 한 문장)"},
|
|
},
|
|
"required": ["bullets"],
|
|
}
|
|
|
|
|
|
def summarize_reviews(client, store, reviews, model="gemini-2.5-flash", n=4):
|
|
"""방문자 리뷰들을 Gemini 로 요약해 '브리핑 대체' 불릿 리스트로 반환(실패 시 빈 리스트).
|
|
AI 브리핑이 없는 가게에서, 브리핑과 같은 형식(짧은 특징 문장들)을 만들어 나레이션 근거로 쓴다."""
|
|
reviews = [r for r in reviews if r][:40]
|
|
if not reviews:
|
|
return []
|
|
joined = "\n".join(f"- {r}" for r in reviews)
|
|
sname = store or "이 가게"
|
|
prompt = (
|
|
f"다음은 '{sname}'를 방문한 손님들이 남긴 실제 리뷰 모음이다.\n"
|
|
f"이 리뷰들을 종합해서, 가게를 홍보 영상 소재로 쓸 수 있도록 핵심 특징을 "
|
|
f"{n}개의 짧은 불릿으로 요약하라.\n"
|
|
f"- ★ 첫 번째 불릿은 반드시 가게 이름 '{sname}'(으)로 시작해, 이 가게가 어떤 곳/무슨 메뉴인지 "
|
|
f"한 문장으로 소개하라. (예: '{sname}은(는) …가 유명한 …맛집입니다.')\n"
|
|
f"- 나머지 불릿에도 맥락이 필요하면 '{sname}'을(를) 자연스럽게 써도 된다. 절대 'OO식당' 같은 "
|
|
"익명 표현으로 바꾸지 마라 — 반드시 실제 상호를 쓴다.\n"
|
|
"- 각 불릿은 한 문장. 메뉴/맛/분위기/서비스/특징 위주로.\n"
|
|
"- 여러 리뷰에서 반복되는 공통점을 우선. 한두 명만 말한 지엽적 내용은 제외.\n"
|
|
"- 리뷰에 없는 사실을 지어내지 마라(과장·거짓 금지).\n\n"
|
|
f"[리뷰]\n{joined}"
|
|
)
|
|
try:
|
|
resp = client.models.generate_content(
|
|
model=model, contents=prompt,
|
|
config=types.GenerateContentConfig(
|
|
temperature=0.4, response_mime_type="application/json",
|
|
response_schema=_REVIEW_SUMMARY_SCHEMA),
|
|
)
|
|
out, seen = [], set()
|
|
for b in json.loads(resp.text).get("bullets", []):
|
|
b = re.sub(r"\s+", " ", str(b)).strip()
|
|
if b and b not in seen:
|
|
seen.add(b); out.append(b)
|
|
return out
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
# ============================== 사진 (크롤 + Gemini 비전 선별, 인물 제외) ==============================
|
|
|
|
# 공식 대표사진 CDN(업체 등록)만 후보로 사용.
|
|
# 손님 리뷰 사진(pup-review-*)은 저작권/초상권이 '리뷰 작성자 개인'에게 있어
|
|
# 가게 동의만으론 광고 영상에 쓸 수 없으므로 제외한다.
|
|
# 공식 대표사진이 오는 호스트(둘 다 '업체가 직접 등록'한 사진이라 광고 사용 가능):
|
|
# · ldb-phinf : 음식점/일반 업체가 등록한 공식 대표사진
|
|
# · naverbooking-phinf : 네이버 '예약'형 숙소(펜션·게하·글램핑)의 공식 객실 사진
|
|
# → 예약 숙소는 ldb 대신 이쪽으로 온다 (이게 'accommodation 사진 0장'의 원인이었음)
|
|
# 손님 리뷰 사진(pup-review-*)은 초상권 문제로 의도적 제외.
|
|
OFFICIAL_HOST = "ldb-phinf.pstatic.net"
|
|
BOOKING_HOST = "naverbooking-phinf.pstatic.net"
|
|
REVIEW_HOST = "pup-review-phinf.pstatic.net" # (참고: 의도적으로 후보에서 제외)
|
|
CAND_HOSTS = (OFFICIAL_HOST, BOOKING_HOST)
|
|
|
|
|
|
def _underlying(url: str) -> str:
|
|
"""search.pstatic.net/common?...&src=<진짜URL> 프록시면 src 를 풀어서 반환."""
|
|
m = re.search(r"[?&]src=([^&]+)", url)
|
|
if m:
|
|
return unquote(m.group(1))
|
|
return url
|
|
|
|
|
|
async def crawl_and_download(url, dst_dir, max_candidates=14, headful=False):
|
|
"""사진 탭 → 공식 대표사진을 dst_dir 에 내려받고 (가게이름, [경로]) 반환."""
|
|
dst_dir = Path(dst_dir)
|
|
dst_dir.mkdir(parents=True, exist_ok=True)
|
|
seen, seen_tab, store = [], [], ""
|
|
cur = {"label": "업체"} # 현재 보고 있는 사진탭 카테고리(응답에 태깅)
|
|
|
|
async with async_playwright() as p:
|
|
browser, ctx, page = await _new_page(p, headful=headful)
|
|
|
|
def on_resp(resp):
|
|
u = resp.url
|
|
if "phinf.pstatic.net" in u or "search.pstatic.net/common" in u:
|
|
seen.append(u); seen_tab.append(cur["label"])
|
|
page.on("response", on_resp)
|
|
|
|
# 링크 → place id → 사진 탭
|
|
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
|
|
await page.wait_for_timeout(2000)
|
|
cat, pid = extract_place(page.url)
|
|
if not pid:
|
|
await browser.close()
|
|
raise RuntimeError("place id 를 못 찾음.")
|
|
store = await _store_name(page)
|
|
|
|
await page.goto(f"https://m.place.naver.com/{cat}/{pid}/photo",
|
|
wait_until="domcontentloaded", timeout=30000)
|
|
await page.wait_for_timeout(2000)
|
|
|
|
# [핵심] 사진탭 상단 카테고리 탭(업체/내부/음식·음료/외부/풍경...)을 돌며 로드.
|
|
# 기본 '업체' 탭엔 대표사진 몇 장뿐 → 다른 탭을 눌러야 음식·음료(와인 등)·내부가 나온다.
|
|
# (세션이 길면 헤드리스 브라우저가 불안정해지므로 탭 수·스크롤을 적당히 제한)
|
|
try:
|
|
chips = [c.strip() for c in await page.locator("a.sbkBy").all_inner_texts()]
|
|
except Exception:
|
|
chips = []
|
|
targets = [c for c in chips if c and c != "클립"][:6] or [None] # 클립(영상)은 제외
|
|
for label in targets:
|
|
cur["label"] = label or "업체" # 이 탭에서 로드되는 사진을 이 라벨로 태깅
|
|
try:
|
|
if label is not None:
|
|
await page.locator("a.sbkBy", has_text=label).first.click(timeout=3000)
|
|
await page.wait_for_timeout(900)
|
|
for _ in range(5): # 각 탭에서 스크롤하며 lazy 로드 유발
|
|
await page.mouse.wheel(0, 2500)
|
|
await page.wait_for_timeout(300)
|
|
except Exception:
|
|
break # 연결 끊기면 모은 만큼으로 진행
|
|
await page.wait_for_timeout(300)
|
|
|
|
# 후보 = 탭(업체/내부/음식·음료/외부…)을 '고루 섞어' 뽑는다.
|
|
# 앞에서 N장만 자르면 먼저 로드된 업체(food) 탭이 다 차지해 내부/외부가 잘린다
|
|
# → 탭별로 모은 뒤 라운드로빈으로 채워 다양성을 강제한다. 각 탭 안에선 공식 먼저.
|
|
tab_imgs, bases = {}, set() # tab -> [(host, base)…]
|
|
official_n = 0
|
|
for u, tab in zip(seen, seen_tab):
|
|
real = _underlying(u)
|
|
host = urlparse(real).netloc
|
|
if host not in CAND_HOSTS: # 공식 대표사진(ldb-phinf)만 통과
|
|
continue
|
|
if "icon" in real or "profile" in real:
|
|
continue
|
|
base = real.split("?")[0]
|
|
if base in bases:
|
|
continue
|
|
bases.add(base)
|
|
tab_imgs.setdefault(tab, []).append((host, base))
|
|
official_n += 1
|
|
# 메뉴판·동영상 탭은 광고컷으로 약하므로 뒤로 미룬다
|
|
def _tab_rank(t):
|
|
return 1 if (t and ("메뉴" in t or "동영상" in t)) else 0
|
|
order = sorted(tab_imgs.keys(), key=_tab_rank)
|
|
candidates, ptr = [], {t: 0 for t in order}
|
|
while len(candidates) < max_candidates and any(ptr[t] < len(tab_imgs[t]) for t in order):
|
|
for t in order:
|
|
if ptr[t] < len(tab_imgs[t]):
|
|
candidates.append(tab_imgs[t][ptr[t]][1]); ptr[t] += 1
|
|
if len(candidates) >= max_candidates:
|
|
break
|
|
tabsum = ", ".join(f"{t}:{len(v)}" for t, v in tab_imgs.items())
|
|
print(f"■ 사진 후보 {len(candidates)}장 (공식 {official_n}, "
|
|
f"전체 응답 {len(seen)}) / 탭별 [{tabsum}]")
|
|
# [진단] 후보 0인데 응답은 많을 때, 실제로 어떤 CDN 호스트에서 왔는지 확인용.
|
|
# 여기에 CAND_HOSTS 가 아닌 호스트가 잡히면 (리뷰/블로그 아닌 한) 추가하면 된다.
|
|
if official_n == 0 and seen:
|
|
from collections import Counter
|
|
hosts = Counter(urlparse(_underlying(u)).netloc for u in seen)
|
|
print(" [진단] 응답 호스트별: "
|
|
+ ", ".join(f"{h}:{c}" for h, c in hosts.most_common()))
|
|
|
|
# 1차: 브라우저 컨텍스트로 다운로드(쿠키/헤더 유지). 연결 끊기면 남은 건 httpx 폴백.
|
|
paths, got = [], set()
|
|
for i, u in enumerate(candidates, 1):
|
|
try:
|
|
r = await ctx.request.get(u, timeout=15000)
|
|
if not r.ok:
|
|
continue
|
|
data = await r.body()
|
|
if len(data) < 3000: # 너무 작으면 깨진/아이콘
|
|
continue
|
|
fp = dst_dir / f"cand{i:02d}.jpg"
|
|
fp.write_bytes(data)
|
|
paths.append(fp); got.add(i)
|
|
except Exception:
|
|
break # 연결이 죽었을 가능성 → 폴백으로
|
|
try:
|
|
await browser.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# 2차(폴백): 1차에서 못 받은 후보를 httpx 로 직접 다운로드 (공개 CDN, Referer 헤더로 차단 회피)
|
|
missing = [(i, u) for i, u in enumerate(candidates, 1) if i not in got]
|
|
if missing:
|
|
hdr = {"User-Agent": MOBILE_UA, "Referer": "https://m.place.naver.com/"}
|
|
with httpx.Client(headers=hdr, timeout=15, follow_redirects=True) as hc:
|
|
for i, u in missing:
|
|
try:
|
|
rr = hc.get(u)
|
|
if rr.status_code == 200 and len(rr.content) >= 3000:
|
|
fp = dst_dir / f"cand{i:02d}.jpg"
|
|
fp.write_bytes(rr.content); paths.append(fp)
|
|
except Exception:
|
|
continue
|
|
paths.sort()
|
|
print(f"■ 다운로드 {len(paths)}장 → {dst_dir}")
|
|
return store, paths
|
|
|
|
|
|
# 비전 채점 스키마 — has_person 포함(인물 제외 판정용).
|
|
_SCORE_SCHEMA = {
|
|
"type": "object",
|
|
"properties": {
|
|
"scores": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"index": {"type": "integer"},
|
|
"category": {"type": "string",
|
|
"enum": ["food", "interior", "exterior", "menu", "drink", "other"]},
|
|
"subject": {"type": "string"}, # 핵심 피사체 한두 단어(다양성 판단용)
|
|
"quality": {"type": "integer"}, # 1~10
|
|
"good_for_ad": {"type": "boolean"},
|
|
"has_person": {"type": "boolean"}, # 사람(전신/일부/뒷모습 포함)이 보이면 true
|
|
},
|
|
"required": ["index", "category", "subject", "quality", "good_for_ad", "has_person"],
|
|
},
|
|
}
|
|
},
|
|
"required": ["scores"],
|
|
}
|
|
|
|
|
|
def vision_score(client, paths, cap=16):
|
|
"""각 사진을 Gemini 비전으로 채점. [{index,category,subject,quality,good_for_ad,has_person}] 반환."""
|
|
paths = paths[:cap]
|
|
parts = [types.Part.from_text(text=(
|
|
"다음 사진들은 한 음식점의 사진이다. 각 사진을 평가하라.\n"
|
|
"- category: food(음식)/drink(음료)/interior(내부)/exterior(외부/간판)/menu(메뉴판)/other\n"
|
|
"- subject: 핵심 피사체를 한두 단어로(예: 구운고기, 랍스터, 스테이크, 와인, 파스타, 수영장야경, 실내홀, 외관).\n"
|
|
" ★ 거의 같은 피사체면 같은 단어로 통일하라(예: 그릴에 구운 고기 사진 여러 장은 전부 '구운고기').\n"
|
|
" 이 값으로 비슷한 사진을 묶어 다양하게 고른다.\n"
|
|
"- quality: 광고 영상에 쓸 만큼 잘 나왔는지 1~10 (선명·밝기·구도·먹음직스러움). 흐리거나 어둡거나 잘리면 낮게.\n"
|
|
"- has_person: 사진 안에 사람이 조금이라도 보이면 true (얼굴·전신·일부·뒷모습·손 등 무엇이든).\n"
|
|
" 사람이 전혀 없는 빈 공간/음식/사물 사진만 false.\n"
|
|
"- good_for_ad: 가게 홍보 컷으로 바로 써도 좋은가(true/false).\n"
|
|
" ★ 다음은 무조건 category='other', good_for_ad=false, quality 1~3 으로:\n"
|
|
" · 로고/브랜드 글자/간판 텍스트만 큰 이미지\n"
|
|
" · 메뉴판·가격표·영수증 등 글자 위주 이미지\n"
|
|
" · 사람이 보이는 사진(초상권 문제 — 사람 없는 컷이 많으니 사람 있으면 제외), 텍스트 그래픽/포스터만 있는 이미지\n"
|
|
" 즉 '음식·음료·매장 공간'이 또렷하게 보이고 사람이 없는 사진만 good_for_ad=true.\n"
|
|
"각 사진 바로 앞에 [사진 N] 표시가 있다. 그 N 을 index 로.")), ]
|
|
for i, p in enumerate(paths):
|
|
parts.append(types.Part.from_text(text=f"[사진 {i}]"))
|
|
parts.append(types.Part.from_bytes(data=Path(p).read_bytes(), mime_type="image/jpeg"))
|
|
resp = client.models.generate_content(
|
|
model=VISION_MODEL, contents=parts,
|
|
config=types.GenerateContentConfig(
|
|
temperature=0.2, response_mime_type="application/json",
|
|
response_schema=_SCORE_SCHEMA),
|
|
)
|
|
return json.loads(resp.text).get("scores", [])
|
|
|
|
|
|
BAD_CATS = {"menu", "other"}
|
|
|
|
|
|
def _dhash(path, size=8):
|
|
"""차이 해시(dHash, 64bit). 시각적으로 비슷한 사진끼리는 해밍거리가 작다.
|
|
(같은 그릴/같은 수영장뷰처럼 '거의 같은 그림'을 잡아내 중복 선택을 막는 용도.)"""
|
|
try:
|
|
img = Image.open(path).convert("L").resize((size + 1, size), Image.LANCZOS)
|
|
px = list(img.getdata())
|
|
bits = 0
|
|
for r in range(size):
|
|
row = px[r * (size + 1):(r + 1) * (size + 1)]
|
|
for c in range(size):
|
|
bits = (bits << 1) | (1 if row[c] < row[c + 1] else 0)
|
|
return bits
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _ham(a, b):
|
|
if a is None or b is None:
|
|
return 999
|
|
return bin(a ^ b).count("1")
|
|
|
|
|
|
def too_similar(path, others, thresh=8):
|
|
"""path 의 차이해시(dHash)가 others(이미지 경로 리스트) 중 어느 것과든 해밍거리 <= thresh 면 True.
|
|
= 시각적으로 '거의 같은 그림'. 사진/클립이 서로 너무 비슷하게 뽑히는 걸 막는 데 쓴다(중복 방지)."""
|
|
h = _dhash(path)
|
|
if h is None:
|
|
return False
|
|
return any(_ham(h, _dhash(o)) <= thresh for o in others)
|
|
|
|
|
|
def _subj(s):
|
|
"""다양성 판단 키: 피사체(subject) 우선, 없으면 종류(category)."""
|
|
return (s.get("subject") or s.get("category") or "other").strip().lower()
|
|
|
|
|
|
def _ngrams(s, k=2):
|
|
s = s.replace(" ", "")
|
|
return {s[i:i + k] for i in range(len(s) - k + 1)} or {s}
|
|
|
|
|
|
def _sim_subj(a, b):
|
|
"""두 피사체가 사실상 같은 종류인가(예: '구운고기' vs '구운고기새우' → True)."""
|
|
if not a or not b:
|
|
return False
|
|
if a in b or b in a:
|
|
return True
|
|
return bool(_ngrams(a) & _ngrams(b)) # 2글자 조각이 하나라도 겹치면 비슷
|
|
|
|
|
|
def pick_best(paths, scores, n=4, per_cat=2, dup_thresh=8, exclude_person=True):
|
|
"""채점 결과로 '종류·피사체 안 겹치게 잘 나온' N장 선택 → 경로 리스트.
|
|
|
|
비전 모델이 피사체를 너무 잘게 쪼개는 경우가 있어(구운고기/구운고기새우/문어새우 …
|
|
전부 사실상 '그릴 음식') 단순 피사체 분산만으론 비슷한 그림이 몰린다. 그래서:
|
|
1차) 종류(category) 우선 분산 + 피사체 비유사 — 최대한 다른 종류·다른 피사체로.
|
|
2차) 같은 종류 2번째 허용하되 '비유사 피사체'만(그릴 다음엔 디저트 같은 식).
|
|
3차) 그래도 모자라면 피사체 유사 허용(동일사진 hash 만 제외).
|
|
4차) 최후: 동일사진만 피해 채움.
|
|
로고/메뉴판/텍스트(other·menu, good_for_ad=false)는 후보에서 제외.
|
|
★ exclude_person=True(기본): 사람이 보이는 사진은 초상권 문제로 끝까지 제외.
|
|
"""
|
|
by_idx = {s["index"]: s for s in scores if 0 <= s.get("index", -1) < len(paths)}
|
|
ranked = sorted(by_idx.values(),
|
|
key=lambda s: (s.get("good_for_ad", False), s.get("quality", 0)),
|
|
reverse=True)
|
|
usable = [s for s in ranked
|
|
if s.get("good_for_ad") and s.get("category") not in BAD_CATS
|
|
and not (exclude_person and s.get("has_person"))] # 인물 제외(초상권)
|
|
|
|
hashes = {paths[s["index"]]: _dhash(paths[s["index"]]) for s in ranked}
|
|
chosen, chosen_h, chosen_subj, cat_cnt = [], [], [], {}
|
|
|
|
def is_dup(p):
|
|
h = hashes.get(p)
|
|
return any(_ham(h, ch) <= dup_thresh for ch in chosen_h)
|
|
|
|
def take(s):
|
|
p = paths[s["index"]]
|
|
chosen.append(p); chosen_h.append(hashes.get(p)); chosen_subj.append(_subj(s))
|
|
c = s.get("category", "other"); cat_cnt[c] = cat_cnt.get(c, 0) + 1
|
|
|
|
def fits(s, new_cat, distinct_subj, respect_cap):
|
|
p = paths[s["index"]]; c = s.get("category", "other")
|
|
if p in chosen or is_dup(p):
|
|
return False
|
|
if respect_cap and cat_cnt.get(c, 0) >= per_cat:
|
|
return False
|
|
if new_cat and cat_cnt.get(c, 0) >= 1:
|
|
return False
|
|
if distinct_subj and any(_sim_subj(_subj(s), cs) for cs in chosen_subj):
|
|
return False
|
|
return True
|
|
|
|
# (카테고리상한, 새카테고리우선, 피사체비유사) 단계적 완화.
|
|
# 3단계: 한 카테고리로 쏠린 가게(전부 food 등)는 상한을 풀되 '피사체 다양성'은 끝까지 지킨다.
|
|
# → 한정식집처럼 다 음식이어도 돌솥밥/불고기/생선구이/전… 서로 다른 걸 고른다.
|
|
for new_cat, distinct, cap in [
|
|
(True, True, True), (False, True, True),
|
|
(False, True, False), (False, False, False)]:
|
|
for s in usable:
|
|
if len(chosen) >= n:
|
|
break
|
|
if fits(s, new_cat, distinct, cap):
|
|
take(s)
|
|
if len(chosen) >= n:
|
|
break
|
|
# 최후: 동일사진만 피해 나머지로 채움 (단, exclude_person 이면 사람 있는 사진은 끝까지 제외)
|
|
if len(chosen) < n:
|
|
for s in ranked:
|
|
if len(chosen) >= n:
|
|
break
|
|
if exclude_person and s.get("has_person"):
|
|
continue
|
|
p = paths[s["index"]]
|
|
if p not in chosen and not is_dup(p):
|
|
chosen.append(p); chosen_h.append(hashes.get(p)); chosen_subj.append(_subj(s))
|
|
if chosen_subj:
|
|
print(f" 선별 피사체: {', '.join(chosen_subj)}")
|
|
return chosen
|
|
|
|
|
|
def fetch_and_select(url, out_dir, n=4, client=None, max_candidates=16,
|
|
headful=False, exclude_person=True):
|
|
"""전체 파이프라인: 크롤링 → (있으면)비전 선별 → out_dir 에 photoNN.jpg 저장.
|
|
|
|
반환 (가게이름, [최종 사진 경로]). client 없으면 비전 없이 앞에서 N장.
|
|
exclude_person=True(기본): 인물 사진 제외(초상권). 셋(animation/narration/kakao) 다 기본 적용.
|
|
"""
|
|
out_dir = Path(out_dir)
|
|
tmp = out_dir / "_cand"
|
|
if tmp.exists():
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
store, cand = asyncio.run(crawl_and_download(url, tmp, max_candidates, headful))
|
|
if not cand:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
return store, []
|
|
|
|
if client is not None and len(cand) > n:
|
|
try:
|
|
scores = vision_score(client, cand)
|
|
chosen = pick_best(cand, scores, n=n, exclude_person=exclude_person)
|
|
print(f"■ 비전 선별: {len(cand)}장 → {len(chosen)}장")
|
|
except Exception as e:
|
|
print(f" (비전 선별 실패 → 앞에서 {n}장: {type(e).__name__})")
|
|
chosen = cand[:n]
|
|
else:
|
|
chosen = cand[:n]
|
|
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
finals = []
|
|
for i, src in enumerate(chosen, 1):
|
|
dst = out_dir / f"photo{i:02d}.jpg"
|
|
shutil.copyfile(src, dst)
|
|
finals.append(dst)
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
return store, finals
|
|
|