wuling-overseas-dealer-expansion-skill/scripts/scraper/website_deep_scraper.py

"""Public website enrichment helpers for dealer discovery.

The helpers only use public pages. They do not guess emails, prices, MOQ,
inventory, or cooperation terms.
"""
from __future__ import annotations

import re
import time
from typing import Any, Dict, List, Set
from urllib.parse import parse_qs, parse_qsl, urlencode, unquote, urlparse, urlunparse

from playwright.sync_api import BrowserContext, Page

try:
    from . import discovery_common as dc
except ImportError:  # pragma: no cover - direct script execution
    import sys
    from pathlib import Path

    sys.path.insert(0, str(Path(__file__).resolve().parent))
    import discovery_common as dc


CONTACT_LINK_HINTS = {
    "contact",
    "nous contacter",
    "contactez",
    "about",
    "a-propos",
    "apropos",
    "à propos",
    "qui sommes",
    "service",
    "services",
    "vehicule",
    "véhicule",
    "vehicules",
    "véhicules",
    "occasion",
    "location",
}

SOCIAL_OR_PLATFORM_HOSTS = {
    "facebook.com",
    "m.facebook.com",
    "fb.com",
    "instagram.com",
    "linkedin.com",
    "tiktok.com",
    "youtube.com",
    "youtu.be",
    "wa.me",
    "whatsapp.com",
    "google.com",
    "maps.google.com",
    "bing.com",
    "maps.bing.com",
    "gmail.com",
    "mail.google.com",
    "outlook.com",
    "hotmail.com",
    "yahoo.com",
    "goo.gl",
}

BARE_DOMAIN_RE = re.compile(
    r"\b(?:https?://)?(?:www\.)?([a-z0-9][a-z0-9-]{1,63}(?:\.[a-z0-9][a-z0-9-]{1,63})+\b(?:/[^\s<>'\"]*)?)",
    re.IGNORECASE,
)

GENERIC_NON_WEBSITE_DOMAINS = {
    "facebook.com",
    "fb.com",
    "instagram.com",
    "youtube.com",
    "youtu.be",
    "tiktok.com",
    "wa.me",
    "whatsapp.com",
    "google.com",
    "bing.com",
    "gmail.com",
    "outlook.com",
    "hotmail.com",
    "yahoo.com",
    "goo.gl",
}

BUSINESS_SIGNALS = {
    "二手车销售/置换": ["occasion", "voiture d'occasion", "reprise", "used car", "achat", "vente"],
    "汽车销售/showroom": ["showroom", "concessionnaire", "vente automobile", "véhicules neufs", "vehicules neufs", "vehicle sales"],
    "进口/分销": ["importation", "importateur", "distribution", "distributeur", "import"],
    "租赁/车队服务": ["location", "rental", "fleet", "flotte", "lld", "leasing"],
    "商用车/工具车": ["utilitaire", "camion", "truck", "van", "mpv", "minibus", "bus"],
    "售后/维修支持": ["service après-vente", "après-vente", "sav", "garage", "réparation", "maintenance"],
}

BATCH_CAPACITY_SIGNALS = {
    "多门店/区域覆盖": ["réseau", "reseau", "succursale", "points de vente", "agence", "branches", "network"],
    "库存/展厅信号": ["stock", "parc auto", "showroom", "disponible", "inventory"],
    "进口/分销能力": ["importation", "importateur", "distribution", "distributeur"],
    "车队/租赁需求": ["fleet", "flotte", "location", "lld", "leasing"],
    "商用车适配": ["utilitaire", "camion", "truck", "van", "mpv", "minibus"],
}


def normalize_host(url: str) -> str:
    return urlparse(str(url or "")).netloc.casefold().removeprefix("www.")


def is_social_or_platform_url(url: str) -> bool:
    host = normalize_host(url)
    return any(host == blocked or host.endswith("." + blocked) for blocked in SOCIAL_OR_PLATFORM_HOSTS)


def is_company_website_url(url: str) -> bool:
    if not dc.is_external_business_url(url):
        return False
    if is_social_or_platform_url(url):
        return False
    parsed = urlparse(str(url or ""))
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        return False
    host = normalize_host(url)
    labels = host.split(".")
    if not labels:
        return False
    tld = labels[-1]
    if len(tld) > 6:
        return False
    if not (len(tld) == 2 or tld in {"com", "net", "org", "info", "biz", "auto", "cars", "shop", "store"}):
        return False
    return True


def _same_site(url: str, base_url: str) -> bool:
    return normalize_host(url) == normalize_host(base_url)


def _strip_tracking_query(url: str) -> str:
    parsed = urlparse(str(url or ""))
    if not parsed.scheme or not parsed.netloc:
        return url
    blocked_prefixes = ("utm_",)
    blocked_keys = {"fbclid", "gclid", "yclid", "mc_cid", "mc_eid", "igshid"}
    query_items = [
        (key, value)
        for key, value in parse_qsl(parsed.query, keep_blank_values=True)
        if key.casefold() not in blocked_keys and not key.casefold().startswith(blocked_prefixes)
    ]
    return urlunparse(parsed._replace(query=urlencode(query_items, doseq=True), fragment=""))


def _unwrap_facebook_redirect(url: str) -> str:
    """Return the real destination when Facebook wraps an external link."""
    parsed = urlparse(str(url or ""))
    host = parsed.netloc.casefold().removeprefix("www.")
    if host not in {"facebook.com", "m.facebook.com", "l.facebook.com"}:
        return url
    query = parse_qs(parsed.query)
    for key in ("u", "url", "href"):
        values = query.get(key)
        if values:
            return unquote(values[0])
    return url


def _candidate_urls_from_visible_text(text: str) -> List[str]:
    """Extract bare domains visible in Facebook contact/about sections."""
    candidates: List[str] = []
    for match in BARE_DOMAIN_RE.finditer(str(text or "")):
        raw = match.group(1).rstrip(".,;:)")
        if not raw or "@" in raw:
            continue
        url = raw if raw.startswith(("http://", "https://")) else f"https://{raw}"
        host = normalize_host(url)
        if host in GENERIC_NON_WEBSITE_DOMAINS:
            continue
        if is_company_website_url(url) and url not in candidates:
            candidates.append(url)
    return candidates


def _extract_external_links(page: Page) -> List[Dict[str, str]]:
    try:
        return page.evaluate(
            """
            () => Array.from(document.querySelectorAll('a[href]')).map((a) => ({
              href: a.href || '',
              text: (a.innerText || '').trim(),
              aria: (a.getAttribute('aria-label') || '').trim()
            }))
            """
        )
    except Exception:
        return []


def extract_company_website_from_facebook(page: Page) -> str:
    """Extract the best external company website visible on a Facebook page."""
    candidates: List[tuple[int, str]] = []
    for item in _extract_external_links(page):
        href = _strip_tracking_query(dc.normalize_website_url(_unwrap_facebook_redirect(item.get("href", ""))))
        if not is_company_website_url(href):
            visible_candidates = _candidate_urls_from_visible_text(
                " ".join([item.get("text", ""), item.get("aria", "")])
            )
            if not visible_candidates:
                continue
            href = _strip_tracking_query(visible_candidates[0])
        label = " ".join([item.get("text", ""), item.get("aria", ""), href]).casefold()
        score = 0
        if any(term in label for term in ["website", "site web", "site", "www.", ".ma", ".com", ".net", ".org"]):
            score += 2
        if any(term in label for term in ["contact", "official", "officiel"]):
            score += 1
        candidates.append((score, href))
    try:
        body_text = page.locator("body").inner_text(timeout=5000)
    except Exception:
        body_text = ""
    for href in _candidate_urls_from_visible_text(body_text):
        candidates.append((1, _strip_tracking_query(href)))
    if not candidates:
        return ""
    candidates.sort(key=lambda item: (-item[0], len(item[1])))
    return candidates[0][1]


def _score_terms(text: str, signal_map: Dict[str, List[str]]) -> List[str]:
    lower = str(text or "").casefold()
    hits: List[str] = []
    for label, terms in signal_map.items():
        if any(term.casefold() in lower for term in terms):
            hits.append(label)
    return hits


def summarize_business_cn(combined_text: str) -> str:
    business = _score_terms(combined_text, BUSINESS_SIGNALS)
    capacity = _score_terms(combined_text, BATCH_CAPACITY_SIGNALS)
    parts: List[str] = []
    if business:
        parts.append("主营业务判断：" + "；".join(business[:4]))
    else:
        parts.append("主营业务判断：汽车渠道线索，需人工确认")
    if capacity:
        parts.append("批量采购能力判断：发现" + "、".join(capacity[:4]) + "，可评估小批量采购/分销潜力")
    else:
        parts.append("批量采购能力判断：暂未发现明确批量采购证据")
    return "；".join(parts)


def classify_customer_type_cn(combined_text: str) -> str:
    return dc.classify_customer_type(str(combined_text or ""))


def scrape_public_website(context: BrowserContext, website_url: str, max_pages: int = 5) -> Dict[str, Any]:
    """Scrape a public company website for contact and business evidence."""
    if not is_company_website_url(website_url):
        return {"website": "", "email": "", "emails": [], "phone": "", "checked_urls": [], "business_summary": "", "evidence_notes": ["未发现可用公司官网"], "text": ""}

    checked: List[str] = []
    queue: List[str] = [website_url]
    emails: List[str] = []
    phones: List[str] = []
    text_chunks: List[str] = []
    errors: List[str] = []
    page = context.new_page()
    page.set_viewport_size({"width": 1280, "height": 850})

    try:
        while queue and len(checked) < max_pages:
            url = dc.normalize_website_url(queue.pop(0))
            if url in checked or not is_company_website_url(url):
                continue
            if checked and not _same_site(url, website_url):
                continue
            checked.append(url)
            try:
                page.goto(url, wait_until="domcontentloaded", timeout=35000)
                time.sleep(1.2)
                body_text = page.locator("body").inner_text(timeout=12000)
                text_chunks.append(body_text[:5000])
                links = _extract_external_links(page)
                mailto_text = " ".join(item.get("href", "") for item in links if item.get("href", "").casefold().startswith("mailto:"))
                for email in dc.extract_emails(body_text + " " + mailto_text):
                    if email not in emails:
                        emails.append(email)
                phone = dc.extract_phone(body_text)
                if phone and phone not in phones:
                    phones.append(phone)
                for item in links:
                    label = " ".join([item.get("href", ""), item.get("text", ""), item.get("aria", "")]).casefold()
                    if not any(hint in label for hint in CONTACT_LINK_HINTS):
                        continue
                    href = dc.normalize_website_url(item.get("href", ""), page.url)
                    if href and href not in checked and href not in queue and is_company_website_url(href) and _same_site(href, website_url):
                        queue.append(href)
            except Exception as exc:
                errors.append(str(exc)[:120])
                continue
    finally:
        try:
            page.close()
        except Exception:
            pass

    combined_text = "\n".join(text_chunks)
    summary = summarize_business_cn(combined_text) if combined_text else "主营业务判断：官网未能读取正文，详细信息待确认；批量采购能力判断：暂未发现明确证据"
    evidence: List[str] = []
    if checked:
        evidence.append("官网检查页：" + "；".join(checked[:3]))
    if emails:
        evidence.append("官网发现公开邮箱")
    else:
        evidence.append("官网未发现公开邮箱")
    if errors and not text_chunks:
        evidence.append("官网读取失败：" + "；".join(errors[:2]))
        evidence.append(dc.MANUAL_REVIEW_FLAGS["detail"])

    return {
        "website": _strip_tracking_query(dc.normalize_website_url(website_url)),
        "email": emails[0] if emails else "",
        "emails": emails,
        "phone": phones[0] if phones else "",
        "checked_urls": checked,
        "business_summary": summary,
        "evidence_notes": evidence,
        "text": combined_text[:12000],
    }


def build_chinese_notes(
    facebook_url: str,
    facebook_text: str,
    website_url: str,
    website_result: Dict[str, Any],
    followers: str = "",
    post_analysis: str = "",
    detected_brands: List[str] | None = None,
    exclusivity: str = "",
) -> str:
    combined = "\n".join([facebook_text or "", website_result.get("text", "") or "", post_analysis or ""])
    fb_summary = summarize_business_cn(facebook_text or post_analysis or "")
    website_summary = website_result.get("business_summary", "")
    contact_parts: List[str] = []
    if website_result.get("email"):
        contact_parts.append("官网公开邮箱：" + website_result["email"])
    if website_result.get("phone"):
        contact_parts.append("官网公开电话：" + website_result["phone"])
    if not contact_parts:
        contact_parts.append("未发现新的官网公开联系方式")

    notes = [
        "来源线索：Facebook主页 " + facebook_url,
        "Facebook证据：" + ("粉丝量：" + followers if followers else "已检查主页/About/近期帖子"),
        "官网证据：" + (website_url if website_url else "Facebook 未发现公司官网"),
        "主营业务判断：" + summarize_business_cn(combined).replace("主营业务判断：", "", 1),
        "联系方式证据：" + "；".join(contact_parts),
    ]
    if website_result.get("evidence_notes"):
        notes.append("官网检查：" + "；".join(website_result["evidence_notes"]))
    if post_analysis:
        notes.append("近期动态：" + post_analysis[:260])
    if detected_brands:
        notes.append("品牌信号：涉及中国品牌 " + "、".join(detected_brands))
    if exclusivity:
        notes.append("风险/待确认项：" + exclusivity)
    return " | ".join(part for part in notes if part)[:1500]