feat(site-import): HTML 清洗与抽取纯函数
spec §3.1 Task 3:clean_html(去 script/style/nav/footer/header/注释 + 空白折叠 + 8K 截断)、extract_color(内联 style+CSS 合并计数,白黑灰 排除,平票首现)、extract_logo_url(/YYYY/MM/ 日期新者胜出)、 extract_contacts(wa.me 归一 E164 + mailto)、find_residue(模板残留 清单)、discover_links(同源绝对化去锚去询,关键词稳定排前)、is_thin。 零新依赖(stdlib regex),纯函数无 IO,15 测试钉位。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,187 @@
|
||||
"""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[^>]*>.*?</\1\s*>",
|
||||
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 等) ——
|
||||
_RE_STYLE_ATTR = re.compile(r"""style\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"""<img\b[^>]*?\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"""<a\b[^>]*?\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 截断。"""
|
||||
t = _RE_BLOCK.sub(" ", html_text)
|
||||
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:
|
||||
"""<img src> 含 "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/<digits>(可带 +)→ "+"+纯数字归一 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]:
|
||||
"""同源链接发现:所有 <a href> → 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
|
||||
Reference in New Issue
Block a user