ai-kefu-admin/app/platform/site_import/extract.py

"""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,
)
# 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"""<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 截断。

    先删配对块、再兜底删 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:
    """<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