"""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]