249 lines
9.3 KiB
Python
249 lines
9.3 KiB
Python
"""HTML 정규식 기반 deterministic 추출 — tracking pixels / SNS / additional domains / main CTA."""
|
|
import re
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
# ── tracking pixels ──────────────────────────────────────────────────────────
|
|
|
|
# 픽셀별 시그니처: 하나라도 매치되면 installed, ID 는 첫 non-empty 그룹.
|
|
_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 추출."""
|
|
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,
|
|
)
|
|
# CTA 카테고리 우선순위 (전화>카톡>예약). 매치 없으면 결과에 포함 X.
|
|
_CTA_CATEGORIES: list[tuple[str, re.Pattern]] = [
|
|
("전화 상담", re.compile(r"전화|\bCall\b|\bPhone\b", re.IGNORECASE)),
|
|
("카카오톡 상담", re.compile(r"카(카오)?톡|Kakao", re.IGNORECASE)),
|
|
("온라인 예약", re.compile(r"예약|Reserv|Book", 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()
|
|
|
|
|
|
_MAIN_CTA_LIMIT = 3
|
|
|
|
|
|
def _classify_cta(text: str) -> str | None:
|
|
"""CTA 텍스트 → 카테고리 (전화/카톡/예약). 매치 없으면 None."""
|
|
for label, pat in _CTA_CATEGORIES:
|
|
if pat.search(text):
|
|
return label
|
|
return None
|
|
|
|
|
|
def extract_main_cta(html: str) -> str:
|
|
"""<button>·<a> 라벨을 카테고리로 분류 후 우선순위 순서로 최대 _MAIN_CTA_LIMIT 개 join."""
|
|
if not html:
|
|
return ""
|
|
found: set[str] = set()
|
|
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):
|
|
cat = _classify_cta(text)
|
|
if cat:
|
|
found.add(cat)
|
|
ordered_labels = [label for label, _ in _CTA_CATEGORIES]
|
|
result = [label for label in ordered_labels if label in found]
|
|
return " + ".join(result[:_MAIN_CTA_LIMIT])
|
|
|
|
|
|
# ── SNS links ────────────────────────────────────────────────────────────────
|
|
|
|
# SNS 도메인 → 표준 platform 이름.
|
|
_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)
|
|
# <a> 태그 안의 href 만 매치 — <link>/<script> 등 리소스 호출 제외 (CDN 노이즈 차단).
|
|
_ANCHOR_HREF_PATTERN = re.compile(
|
|
r"""<a\b[^>]*\bhref\s*=\s*['"](https?://[^'"\s]+)['"]""",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _platform_for(url: str) -> str | None:
|
|
try:
|
|
host = urlparse(url).netloc.lower().removeprefix("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 이 <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_anchor_hrefs(html: str) -> list[str]:
|
|
"""<a href="http..."> 만 추출 — 순서 보존, 중복 제거."""
|
|
if not html:
|
|
return []
|
|
seen: dict[str, None] = {}
|
|
for url in _ANCHOR_HREF_PATTERN.findall(html):
|
|
if url not in seen:
|
|
seen[url] = None
|
|
return list(seen.keys())
|
|
|
|
|
|
def extract_sns_links(html: str) -> list[dict]:
|
|
"""anchor href 에서 SNS 도메인 매칭. 중복 platform 은 첫 URL 만 유지."""
|
|
seen: dict[str, dict] = {}
|
|
for url in _extract_anchor_hrefs(html):
|
|
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 (글로벌/다국어 사이트만) ─────────────────────────────────
|
|
|
|
# 언어 코드 → 한국어 라벨.
|
|
_LANG_LABEL: dict[str, str] = {
|
|
"en": "영어", "eng": "영어",
|
|
"zh": "중국어", "cn": "중국어", "chn": "중국어",
|
|
"ja": "일본어", "jp": "일본어",
|
|
"ko": "한국어", "kor": "한국어", "kr": "한국어",
|
|
"vi": "베트남어", "vn": "베트남어",
|
|
"th": "태국어", "thai": "태국어",
|
|
"ru": "러시아어",
|
|
"es": "스페인어",
|
|
"mn": "몽골어",
|
|
"ar": "아랍어", "arab": "아랍어",
|
|
"id": "인도네시아어",
|
|
"de": "독일어",
|
|
"fr": "프랑스어",
|
|
"pt": "포르투갈어",
|
|
}
|
|
|
|
_LANG_LI_PATTERN = re.compile(
|
|
r'<li\b[^>]*\bdata-lang\s*=\s*[\'"]([^\'"]+)[\'"][^>]*>(.*?)</li>',
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
|
|
|
|
def extract_additional_domains(html: str, primary_host: str) -> list[dict]:
|
|
"""글로벌/다국어 사이트 anchor 수집 — data-lang 마커 + URL 서브도메인 prefix 로 식별."""
|
|
if not html:
|
|
return []
|
|
primary = (primary_host or "").lower().removeprefix("www.")
|
|
seen: dict[str, dict] = {}
|
|
|
|
def add(host: str, lang_kr: str):
|
|
host = host.lower().removeprefix("www.")
|
|
if not host or host == primary or host in seen:
|
|
return
|
|
seen[host] = {"domain": host, "purpose": lang_kr}
|
|
|
|
# 1) <li data-lang="xx"> 안의 anchor
|
|
for m in _LANG_LI_PATTERN.finditer(html):
|
|
lang_kr = _LANG_LABEL.get(m.group(1).lower())
|
|
if not lang_kr:
|
|
continue
|
|
a_m = re.search(r'<a[^>]*href\s*=\s*[\'"](https?://[^\'"\s]+)[\'"]', m.group(2), re.IGNORECASE)
|
|
if a_m:
|
|
add(urlparse(a_m.group(1)).netloc, lang_kr)
|
|
|
|
# 2) URL 서브도메인 prefix 가 언어 코드
|
|
for url in _extract_anchor_hrefs(html):
|
|
host = urlparse(url).netloc.lower().removeprefix("www.")
|
|
parts = host.split(".")
|
|
if len(parts) >= 3:
|
|
lang_kr = _LANG_LABEL.get(parts[0])
|
|
if lang_kr:
|
|
add(host, lang_kr)
|
|
|
|
return list(seen.values())
|
|
|
|
|