"""site_import HTML 清洗与抽取纯函数(spec §3.1)。 全部无状态纯函数,crawler 编排层消费。禁第三方解析库(项目零新依赖) ——stdlib regex 实现,鲁棒性以 tests/test_site_import_extract.py 的 fixture 覆盖为准(正则处理 HTML 的边界不求完美解析)。 """ from __future__ import annotations import html as _html import re from urllib.parse import urljoin, urlsplit, urlunsplit MAX_TEXT_LEN = 8192 THIN_THRESHOLD = 200 # —— clean_html —— # 容器级块(含内容整体删除):backref \1 保证开闭标签同名(大小写不敏感) _RE_BLOCK = re.compile( r"<(script|style|nav|footer|header)\b[^>]*>.*?", re.IGNORECASE | re.DOTALL, ) # EOF 未闭合块(>500KB 截断的 fetch 可能切在 script/style 中间)——配对清理后兜底 _RE_BLOCK_UNCLOSED = re.compile( r"<(script|style|nav|footer|header)\b[^>]*>.*$", re.IGNORECASE | re.DOTALL, ) _RE_COMMENT = re.compile(r"", re.DOTALL) _RE_TAG = re.compile(r"<[^>]+>") _RE_WS = re.compile(r"\s+") # —— extract_color:只看内联 style 属性 + 外部 css_text,不扫全文(避免误收 URL 等) —— # \sstyle 防 data-style 等复合属性名误命中(与 \ssrc/\shref 同款守卫) _RE_STYLE_ATTR = re.compile(r"""\sstyle\s*=\s*["']([^"']*)["']""", re.IGNORECASE) # 6 位优先、3 位展开;负向前瞻防截取更长 hex(如 #aabbccdd)的前缀 _RE_HEX = re.compile(r"#([0-9a-fA-F]{6}|[0-9a-fA-F]{3})(?![0-9a-fA-F])") # —— extract_logo_url —— # \ssrc 防 data-src / srcset 误命中 _RE_IMG_SRC = re.compile( r"""]*?\ssrc\s*=\s*["']([^"']+)["']""", re.IGNORECASE) _RE_URL_DATE = re.compile(r"/(\d{4})/(\d{1,2})/") # —— extract_contacts —— _RE_WA = re.compile(r"wa\.me/\+?(\d+)", re.IGNORECASE) _RE_MAILTO = re.compile( r"mailto:([A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,})", re.IGNORECASE) # —— discover_links —— _RE_A_HREF = re.compile( r"""]*?\shref\s*=\s*["']([^"']+)["']""", re.IGNORECASE) _RE_PAGE_N = re.compile(r"/page/\d+/?$", re.IGNORECASE) _LINK_KEYWORDS = ( "about", "company", "product", "solution", "contact", "nosotros", "empresa", "productos", "contacto", "关于", "产品", "联系", ) # —— find_residue:模板残留清单(命中返回描述串,供 LLM 摘要与人工复核) —— _RESIDUE_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(r"1?[-.]?555[-.]?\d{3}[-.]?\d{4}"), "占位电话号码(555 号段)"), (re.compile(r"Zemez", re.IGNORECASE), "模板厂牌残留(Zemez)"), (re.compile(r"Add Your Heading", re.IGNORECASE), "占位标题(Add Your Heading Text Here)"), (re.compile(r"\(800\)\s*123-0045"), "占位电话号码((800) 123-0045)"), (re.compile(r"Glasgow D04 89GR"), "占位地址(Glasgow D04 89GR)"), (re.compile(r"Lorem ipsum", re.IGNORECASE), "占位文本(Lorem ipsum)"), ] def clean_html(html_text: str) -> str: """去 script/style/nav/footer/header/注释,剩余标签转文本,空白折叠,>8192 截断。 先删配对块、再兜底删 EOF 未闭合块——顺序不可倒(未闭合模式会吞掉 配对块之后的全部文本)。 """ t = _RE_BLOCK.sub(" ", html_text) t = _RE_BLOCK_UNCLOSED.sub(" ", t) t = _RE_COMMENT.sub(" ", t) t = _RE_TAG.sub(" ", t) t = _html.unescape(t) t = _RE_WS.sub(" ", t).strip() return t[:MAX_TEXT_LEN] def _collect_hex(text: str, counts: dict[str, int]) -> None: """把 text 里的 #hex 计入 counts(3 位先展开,键归一小写 #rrggbb)。""" for m in _RE_HEX.finditer(text): h = m.group(1) if len(h) == 3: # #abc → #aabbcc h = "".join(c * 2 for c in h) key = "#" + h.lower() counts[key] = counts.get(key, 0) + 1 def _is_neutral(hex6: str) -> bool: """白/黑/灰系判定(hex6 形如 #rrggbb)——中性色不作品牌主色。""" r, g, b = (int(hex6[i:i + 2], 16) for i in (1, 3, 5)) if r > 240 and g > 240 and b > 240: return True # 白系 if r < 15 and g < 15 and b < 15: return True # 黑系 if abs(r - g) < 16 and abs(g - b) < 16 and abs(r - b) < 16: return True # 灰系(低饱和) return False def extract_color(html_text: str, css_text: str | None) -> str | None: """统计内联 style 与 css_text 的 #hex,排除白黑灰后取最高频(平票首现)。 返回小写 #rrggbb;全被排除或无 hex → None。 """ counts: dict[str, int] = {} for m in _RE_STYLE_ATTR.finditer(html_text): _collect_hex(m.group(1), counts) if css_text: _collect_hex(css_text, counts) eligible = {k: n for k, n in counts.items() if not _is_neutral(k)} if not eligible: return None # dict 保插入序 + max 平票取先遍历者 → 首现胜出 return max(eligible.items(), key=lambda kv: kv[1])[0] def extract_logo_url(html_text: str) -> str | None: """ 含 "logo"(不区分大小写)的 URL。 多个时按 URL 里 /YYYY/MM/ 日期取最新(WordPress uploads 路径形态); 有日期者优先参与比较,全无日期保持原序取首;无 → None。 """ dated: list[tuple[tuple[int, int], str]] = [] undated: list[str] = [] for m in _RE_IMG_SRC.finditer(html_text): src = m.group(1).strip() if "logo" not in src.lower(): continue dm = _RE_URL_DATE.search(src) if dm: dated.append(((int(dm.group(1)), int(dm.group(2))), src)) else: undated.append(src) if dated: return max(dated, key=lambda kv: kv[0])[1] if undated: return undated[0] return None def extract_contacts(html_text: str) -> dict[str, str | None]: """wa.me/(可带 +)→ "+"+纯数字归一 E164;mailto: → email。 无则 {"whatsapp": None, "email": None}。 """ wa = _RE_WA.search(html_text) mail = _RE_MAILTO.search(html_text) return { "whatsapp": "+" + wa.group(1) if wa else None, "email": mail.group(1) if mail else None, } def find_residue(title_text: str, html_text: str) -> list[str]: """在标题 + 页面里找模板残留,返回命中的残留描述串(无命中为空表)。""" hay = f"{title_text}\n{html_text}" return [desc for pat, desc in _RESIDUE_PATTERNS if pat.search(hay)] def discover_links(base_url: str, html_text: str) -> list[str]: """同源链接发现:所有 → urljoin 绝对化 → 同源 http(s) 过滤 → 去锚点去查询 → 去重保序 → 关键词命中(about/product/关于/…与 /page/N) 稳定排前。返回全部同源链接(非关键词页不剔除,仅排后),抓取广度交给调用方。 """ base = urlsplit(base_url) origin = (base.scheme.lower(), base.netloc.lower()) links: list[str] = [] seen: set[str] = set() for m in _RE_A_HREF.finditer(html_text): url = urljoin(base_url, m.group(1).strip()) parts = urlsplit(url) if parts.scheme.lower() not in ("http", "https"): continue # mailto: / javascript: / #锚 等剔除 if (parts.scheme.lower(), parts.netloc.lower()) != origin: continue # 跨源剔除 clean = urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")) if clean not in seen: seen.add(clean) links.append(clean) def _priority(u: str) -> int: low = u.lower() if _RE_PAGE_N.search(low) or any(k in low for k in _LINK_KEYWORDS): return 0 return 1 return sorted(links, key=_priority) # sorted 稳定 → 组内保持首现序 def is_thin(text: str) -> bool: """清洗后文本 <200 字符视为薄页(模板空壳信号)。""" return len(text.strip()) < THIN_THRESHOLD