179 lines
7.4 KiB
Python
179 lines
7.4 KiB
Python
"""collect 단계 - HTML / siteLinks 에서 tracking pixels / SNS links / additional domains 추출.
|
|
모두 정규식·도메인 매칭 기반 deterministic 추출 (LLM 미경유)."""
|
|
import re
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
# ── tracking pixels ──────────────────────────────────────────────────────────
|
|
|
|
# 픽셀별 시그니처: 이름 → 정규식 리스트. installed 판정은 OR (하나라도 매치되면 설치된 것).
|
|
# ID 캡처는 모든 패턴 시도 후 첫 non-empty 그룹 사용 (signature-only 패턴과 ID-capture 패턴 혼재 OK).
|
|
_TRACKING_PIXEL_PATTERNS: dict[str, list[re.Pattern]] = {
|
|
"Google Analytics": [
|
|
re.compile(r"googletagmanager\.com/gtag/js\?id=(G-[A-Z0-9]+)", re.IGNORECASE),
|
|
re.compile(r"google-analytics\.com/analytics\.js", re.IGNORECASE),
|
|
re.compile(r"\bgtag\(\s*['\"]config['\"]\s*,\s*['\"](G-[A-Z0-9]+|UA-\d+-\d+)['\"]", re.IGNORECASE),
|
|
re.compile(r"\b(UA-\d+-\d+)\b"),
|
|
],
|
|
"Google Tag Manager": [
|
|
re.compile(r"googletagmanager\.com/gtm\.js\?id=(GTM-[A-Z0-9]+)", re.IGNORECASE),
|
|
re.compile(r"\b(GTM-[A-Z0-9]+)\b"),
|
|
],
|
|
"Facebook Pixel": [
|
|
re.compile(r"connect\.facebook\.net/[^/]+/fbevents\.js", re.IGNORECASE),
|
|
re.compile(r"fbq\(\s*['\"]init['\"]\s*,\s*['\"](\d{10,20})['\"]", re.IGNORECASE),
|
|
re.compile(r"facebook\.com/tr/?\?id=(\d{10,20})", re.IGNORECASE), # noscript image pixel fallback
|
|
re.compile(r"_fbq\.push\(\s*\[\s*['\"]init['\"]\s*,\s*['\"](\d{10,20})['\"]", re.IGNORECASE), # legacy
|
|
],
|
|
"Naver Analytics": [
|
|
re.compile(r"wcs\.naver\.net/wcslog\.js", re.IGNORECASE),
|
|
re.compile(r"\bwcs_add\b", re.IGNORECASE),
|
|
re.compile(r"\bwcs\.inflow\(", re.IGNORECASE),
|
|
re.compile(r"wcs_add\s*\[\s*['\"]wa['\"]\s*\]\s*=\s*['\"]([a-zA-Z0-9_]+)['\"]", re.IGNORECASE), # wa ID
|
|
],
|
|
"Kakao Pixel": [
|
|
re.compile(r"t1\.daumcdn\.net/kas/static/kp\.js", re.IGNORECASE),
|
|
re.compile(r"kakaoPixel\s*\(\s*['\"]?(\d+)['\"]?", re.IGNORECASE),
|
|
],
|
|
}
|
|
|
|
|
|
def extract_tracking_pixels(html: str) -> list[dict]:
|
|
"""HTML 에서 트래킹 픽셀 설치 여부 + ID 추출. 환각 0 — 정규식 매치만 신뢰.
|
|
모든 패턴 시도 → 하나라도 매치되면 installed. ID 는 캡처된 첫 non-empty 그룹 사용."""
|
|
if not html:
|
|
return []
|
|
pixels: list[dict] = []
|
|
for name, patterns in _TRACKING_PIXEL_PATTERNS.items():
|
|
any_match = False
|
|
captured_id: str | None = None
|
|
for pat in patterns:
|
|
m = pat.search(html)
|
|
if not m:
|
|
continue
|
|
any_match = True
|
|
if m.groups() and m.group(1) and not captured_id:
|
|
captured_id = m.group(1)
|
|
if any_match:
|
|
pixels.append({
|
|
"name": name,
|
|
"installed": True,
|
|
"details": captured_id,
|
|
})
|
|
return pixels
|
|
|
|
|
|
# ── main CTA ─────────────────────────────────────────────────────────────────
|
|
|
|
_CTA_KEYWORDS = re.compile(
|
|
r"(상담|예약|문의|신청|등록|진료|Book\s*Now|Consult|Reservation|Contact|Apply)",
|
|
re.IGNORECASE,
|
|
)
|
|
_BUTTON_TAG = re.compile(r"<button\b[^>]*>(.*?)</button>", re.IGNORECASE | re.DOTALL)
|
|
_ANCHOR_TAG = re.compile(r"<a\b[^>]*>(.*?)</a>", re.IGNORECASE | re.DOTALL)
|
|
|
|
|
|
def _clean_text(inner: str) -> str:
|
|
text = re.sub(r"<[^>]+>", "", inner)
|
|
return re.sub(r"\s+", " ", text).strip()
|
|
|
|
|
|
def extract_main_cta(html: str) -> str:
|
|
"""HTML 에서 primary CTA 텍스트 추출. 우선순위: <button> 첫 매치 → <a> 첫 매치.
|
|
CTA 키워드 매칭 + 길이 1~20자 (버튼 라벨 추정)."""
|
|
if not html:
|
|
return ""
|
|
for pat in (_BUTTON_TAG, _ANCHOR_TAG):
|
|
for m in pat.finditer(html):
|
|
text = _clean_text(m.group(1))
|
|
if text and 1 <= len(text) <= 20 and _CTA_KEYWORDS.search(text):
|
|
return text
|
|
return ""
|
|
|
|
|
|
# ── SNS links ────────────────────────────────────────────────────────────────
|
|
|
|
# SNS 도메인 → 표준 platform 이름. siteLinks 필터링 + DOM 위치 판정에 공유.
|
|
_SNS_DOMAINS: dict[str, str] = {
|
|
"facebook.com": "Facebook",
|
|
"instagram.com": "Instagram",
|
|
"youtube.com": "YouTube",
|
|
"youtu.be": "YouTube",
|
|
"tiktok.com": "TikTok",
|
|
"pf.kakao.com": "KakaoTalk",
|
|
"open.kakao.com": "KakaoTalk",
|
|
"blog.naver.com": "Naver Blog",
|
|
"cafe.naver.com": "Naver Cafe",
|
|
"x.com": "X",
|
|
"twitter.com": "X",
|
|
}
|
|
|
|
_FOOTER_BLOCK = re.compile(r"<footer\b[^>]*>(.*?)</footer>", re.IGNORECASE | re.DOTALL)
|
|
_HEADER_BLOCK = re.compile(r"<header\b[^>]*>(.*?)</header>", re.IGNORECASE | re.DOTALL)
|
|
|
|
|
|
def _platform_for(url: str) -> str | None:
|
|
try:
|
|
host = urlparse(url).netloc.lower().lstrip("www.")
|
|
except Exception:
|
|
return None
|
|
for domain, name in _SNS_DOMAINS.items():
|
|
if host == domain or host.endswith("." + domain):
|
|
return name
|
|
return None
|
|
|
|
|
|
def _location_for(url: str, html: str) -> str:
|
|
"""url 이 HTML 의 <header>/<footer> 안에 들어있는지로 위치 판정. 둘 다 아니면 'body'."""
|
|
if not html:
|
|
return ""
|
|
needle = re.escape(url)
|
|
footer = _FOOTER_BLOCK.search(html)
|
|
if footer and re.search(needle, footer.group(1)):
|
|
return "footer"
|
|
header = _HEADER_BLOCK.search(html)
|
|
if header and re.search(needle, header.group(1)):
|
|
return "header"
|
|
return "body"
|
|
|
|
|
|
def extract_sns_links(site_links: list[str], html: str = "") -> list[dict]:
|
|
"""siteLinks 에서 SNS 도메인 매칭. location 은 HTML 의 footer/header tag 블록으로 판정.
|
|
중복 platform 은 첫 URL 만 유지."""
|
|
seen: dict[str, dict] = {}
|
|
for url in site_links or []:
|
|
platform = _platform_for(url)
|
|
if not platform or platform in seen:
|
|
continue
|
|
seen[platform] = {
|
|
"platform": platform,
|
|
"url": url,
|
|
"location": _location_for(url, html),
|
|
}
|
|
return list(seen.values())
|
|
|
|
|
|
# ── additional domains ───────────────────────────────────────────────────────
|
|
|
|
def extract_additional_domains(site_links: list[str], primary_host: str) -> list[dict]:
|
|
"""siteLinks 의 host 중 primary 도메인이 아닌 외부 도메인 모으기. SNS·맵 등 utility 도메인 제외.
|
|
purpose 분류는 LLM 필요 — 여기선 빈 문자열."""
|
|
skip_domains = set(_SNS_DOMAINS) | {
|
|
"map.kakao.com", "map.naver.com", "maps.google.com", "goo.gl",
|
|
}
|
|
primary = (primary_host or "").lower().lstrip("www.")
|
|
seen: dict[str, dict] = {}
|
|
for url in site_links or []:
|
|
if not url or not url.startswith("http"):
|
|
continue
|
|
try:
|
|
host = urlparse(url).netloc.lower().lstrip("www.")
|
|
except Exception:
|
|
continue
|
|
if not host or host == primary or host in seen:
|
|
continue
|
|
if any(host == d or host.endswith("." + d) for d in skip_domains):
|
|
continue
|
|
seen[host] = {"domain": host, "purpose": ""}
|
|
return list(seen.values())
|