diff --git a/app/platform/site_import/extract.py b/app/platform/site_import/extract.py new file mode 100644 index 0000000..080abe5 --- /dev/null +++ b/app/platform/site_import/extract.py @@ -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[^>]*>.*?", + 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"""]*?\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 截断。""" + 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: + """ 含 "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 diff --git a/tests/test_site_import_extract.py b/tests/test_site_import_extract.py new file mode 100644 index 0000000..e08adaf --- /dev/null +++ b/tests/test_site_import_extract.py @@ -0,0 +1,173 @@ +"""site_import HTML 清洗与抽取纯函数（spec §3.1）——fixture 内联，覆盖即标准。""" +from app.platform.site_import.extract import ( + clean_html, + discover_links, + extract_color, + extract_contacts, + extract_logo_url, + find_residue, + is_thin, +) + +# WordPress 形态首页：script/style/nav/footer/header/注释各一段（埋独特词）， +# 内联 #FCCA03×3 / #ffffff×5 / #888888×2，logo 双版本（日期新者应胜出）， +# wa.me / mailto 联系方式，页脚模板残留三件套，同源产品/关于/联系页链接。 +FIXTURE_WP = """ + +Hesuu Building Materials + +
HEADJUNK promo bar text
+ + + + +
+

Welcome to Hesuu Building Materials

+

We are a leading supplier of construction products with over fifteen years + of export experience. Our catalogue covers steel pipes, fittings, and panels + for wholesalers and distributors worldwide, with stable quality and fast lead + times on every order of building products.

+
+ hero + logo +
+
+ Featured collections +

Mild steel pipes DN800 in stock.

+ All products | + Next page | + About us | + Contact us | + WhatsApp us | + Email sales | + External blog + old logo +
+
+
+

Card title two

+
+
+
+
+

FOOTJUNK widget area

+

Call us: 1-555-644-5566 © Zemez

+

Add Your Heading Text Here

+ white on white +
+ + +""" + +# 内联只剩 #ffffff（白系被排除）——CSS 文件的 #6b9ebd×4 应胜出 +FIXTURE_WP_ONLY_WHITE = """ +
+

Plain white headline

+
White box with some words.
+
+ +""" + +# 只有灰系——全被排除则 None +FIXTURE_ONLY_GRAY = """ +

Only gray text here.

+gray border + +""" + +# CSS 文件内容：#6b9ebd×4 + #ffffff×1 +FIXTURE_CSS = ( + ".btn{color:#6b9ebd}\n" + ".btn-primary{color:#6b9ebd}\n" + ".link{color:#6b9ebd}\n" + ".badge{color:#6b9ebd}\n" + ".wrap{background:#ffffff}\n" +) + +FIXTURE_THIN = "
" + +# 8K 截断用 9000 字符正文版本（>8192，避免测试文件巨大） +FIXTURE_9K = "" + "x" * 9000 + "" + +# 跨源 / 非 http 链接剔除 +FIXTURE_CROSS = """ +About +Evil +JS +""" + + +class TestCleanHtml: + def test_去script_style_nav_footer_header_comment(self): + t = clean_html(FIXTURE_WP) + for junk in ["SCRIPTJUNK", "STYLEJUNK", "NAVJUNK", "FOOTJUNK", "HEADJUNK", "COMMENTJUNK"]: + assert junk not in t + assert "products" in t # 正文保留 + + def test_8K字符截断(self): + t = clean_html(FIXTURE_9K) + assert len(t) <= 8192 + + +class TestExtractColor: + def test_排除白黑灰取最高频(self): + # 内联 #FCCA03×3 应胜出（#ffffff×5 与 #888888×2 被排除） + assert extract_color(FIXTURE_WP, None) == "#fcca03" + + def test_CSS文件参与统计(self): + # 内联只有 #ffffff；CSS 里 #6b9ebd×4 → 应取 #6b9ebd + assert extract_color(FIXTURE_WP_ONLY_WHITE, FIXTURE_CSS) == "#6b9ebd" + + def test_全被排除则None(self): + assert extract_color(FIXTURE_ONLY_GRAY, None) is None + + def test_三位hex展开计数(self): + html = 'xy' + assert extract_color(html, None) == "#ffcc33" + + +class TestExtractLogoUrl: + def test_取日期新者(self): + assert extract_logo_url(FIXTURE_WP) == ( + "https://ex.com/wp-content/uploads/2026/03/logo-new.png") + + def test_无logo返回None(self): + assert extract_logo_url(FIXTURE_THIN) is None + + +class TestExtractContacts: + def test_wa_me归一E164(self): + c = extract_contacts(FIXTURE_WP) + assert c["whatsapp"] == "+8618603858592" + assert c["email"] == "sales@hesuu.com" + + def test_无联系方式返回空(self): + assert extract_contacts(FIXTURE_THIN) == {"whatsapp": None, "email": None} + + +class TestFindResidue: + def test_常见模板残留命中(self): + hits = find_residue("page text", FIXTURE_WP) + assert any("555" in h for h in hits) + assert any("Zemez" in h for h in hits) + assert any("Heading" in h for h in hits) + + +class TestDiscoverLinks: + def test_同源产品关于联系页发现(self): + links = discover_links("https://ex.com/", FIXTURE_WP) + assert "https://ex.com/products/" in links + assert "https://ex.com/products/page/2/" in links + assert "https://ex.com/about/" in links + + def test_跨源链接剔除(self): + links = discover_links("https://ex.com/", FIXTURE_CROSS) + assert links == ["https://ex.com/about/"] + + +class TestIsThin: + def test_空壳为True(self): + assert is_thin(clean_html(FIXTURE_THIN)) is True + + def test_正常页为False(self): + assert is_thin(clean_html(FIXTURE_WP)) is False