""" Shared helpers for Morocco dealer discovery. The scrapers in this folder are preview-first lead research tools. This module keeps platform-specific scripts aligned on duplicate handling, OEM brand-page rejection, customer classification, and public contact extraction. """ import json import re from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple from urllib.parse import urljoin, urlparse, urlunparse import pandas as pd import requests try: from ..common.customer_taxonomy import classify_from_text except ImportError: # pragma: no cover - direct script execution import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from common.customer_taxonomy import classify_from_text DEFAULT_CITY_SCOPE = "摩洛哥全国" MOROCCO_TARGET_CITIES = [ "Casablanca", "Rabat", "Marrakech", "Tanger", "Fes", "Agadir", "Meknes", "Oujda", "Kenitra", "Tetouan", "Nador", "Safi", ] MOROCCO_TERMS = { "maroc", "morocco", "casablanca", "rabat", "marrakech", "tanger", "fes", "fès", "agadir", "meknes", "oujda", "kenitra", "tetouan", "tétouan", "nador", "safi", "settat", "dar bouazza", "el jadida", } CHINA_BRAND_TERMS = { "baic", "byd", "changan", "chery", "dfsk", "dongfeng", "foton", "forland", "gac", "geely", "great wall", "haval", "jac", "jetour", "maxus", "mg", "omoda", "saic", "sitrak", "sinotruk", "shacman", "faw", "faw trucks", "yutong", "king long", "golden dragon", "higer", "wuling", } NON_CHINA_BRAND_TERMS = { "audi", "bmw", "chevrolet", "citroen", "citroën", "dacia", "daf", "fiat", "ford", "hino", "honda", "hyundai", "isuzu", "iveco", "jeep", "kia", "man", "mazda", "mercedes", "mercedes-benz", "mitsubishi", "nissan", "opel", "peugeot", "renault", "scania", "seat", "skoda", "suzuki", "toyota", "volkswagen", "volvo", "volvo trucks", } OEM_BRANCH_BRANDS = CHINA_BRAND_TERMS | NON_CHINA_BRAND_TERMS INDEPENDENT_CHANNEL_CLUES = { "auto hall": "autohall", "smaa": "smaa", "smeia": "smeia", "auto nejma": "autonejma", "la continentale": "lacontinentale", "prince auto": "princeauto", "kifal": "kifal", "autochek": "autochek", "bugshan": "bugshan", "bamotors": "bamotors", "sopriam": "sopriam", "cac": "cac", "centrale automobile": "centraleautomobile", "univers motors": "universmotors", "m-automotiv": "mautomotiv", "m automotiv": "mautomotiv", } INDEPENDENT_CHANNEL_DISPLAY = { "auto hall": "Auto Hall", "smaa": "SMAA", "smeia": "SMEIA", "auto nejma": "Auto Nejma", "la continentale": "La Continentale Auto", "prince auto": "Prince Auto", "kifal": "Kifal Auto", "autochek": "Autochek Morocco", "bugshan": "Bugshan Automotive", "bamotors": "Bamotors Maroc", "sopriam": "Sopriam", "centrale automobile": "Centrale Automobile Chérifienne", "univers motors": "Univers Motors", "m-automotiv": "M-AUTOMOTIV", "m automotiv": "M-AUTOMOTIV", } HIGH_INTENT_TERMS = { "concessionnaire", "dealer", "distributeur", "distribution", "importateur", "importation", "showroom", "groupe", "group", "automobile", "auto", "retail", "vente", "voiture", "vehicule", "v?hicule", "stock", "parc auto", "multimarque", "multi-brand", "v?hicules neufs", "vehicules neufs", "voitures neuves", "new vehicle", "new cars", "r?seau", "reseau", "dealer network", "fleet", "flotte", "utilitaire", "camion", } NEW_VEHICLE_SALES_TERMS = { "v?hicules neufs", "vehicules neufs", "voitures neuves", "new vehicle", "new vehicles", "new cars", "concessionnaire", "dealer", "showroom", "vente automobile", "vente de voitures", "stock", "parc auto", "retail", } IMPORT_DISTRIBUTION_TERMS = { "importateur", "importation", "importer", "distributeur", "distribution", "r?seau", "reseau", "dealer network", "national distributor", "regional distributor", "agent", "groupe", "group", } MULTIBRAND_TERMS = { "multimarque", "multi-brand", "multi brand", "multi marques", "plusieurs marques", "marques multiples", "showroom multimarque", } CONTACT_ENTRY_TERMS = { "whatsapp", "wa.me", "tel", "t?l", "telephone", "t?l?phone", "phone", "email", "mail", "contact", "linkedin.com", "facebook.com", "site web", "website", "+212", "@", } PURE_SERVICE_TERMS = { "garage r?paration", "garage reparation", "r?paration", "reparation", "repair", "diagnostic", "diag", "pi?ces d?tach?es", "pieces detachees", "spare parts", "pieces auto", "pi?ces auto", "pneus", "tires", "tyres", "tuning", "lavage", "car wash", "pare-brise", "assurance", "insurance", } RENTAL_FLEET_TERMS = { "location", "rental", "rent car", "rentcar", "flotte", "fleet", "leasing", "lld", } PERSONAL_SELLER_TERMS = { "particulier", "personal seller", "vendeur particulier", "marketplace seller", "profile.php", } DIRECT_BRANCH_TERMS = { "succursale", "branch", "filiale", "subsidiary", "official dealer", "concession officielle", "showroom officiel", "page officielle", "official page", "officiel", "official", } LOW_VALUE_TERMS = PURE_SERVICE_TERMS | { "aerospace", "a?ronautique", "software", "marketing", "emailing", "real estate", "immobilier", "location uniquement", "rental only", } CITY_PATTERNS = [ "Casablanca", "Rabat", "Marrakech", "Tanger", "Fes", "Fès", "Agadir", "Meknes", "Oujda", "Kenitra", "Tetouan", "Tétouan", "Nador", "Safi", "Settat", "Dar Bouazza", "El Jadida", ] GENERIC_COMPANY_WORDS = { "maroc", "morocco", "ma", "officiel", "official", "page", "company", "automobile", "auto", "cars", "motors", "motor", "groupe", "group", "sarl", "sa", "llc", "ltd", } OEM_BRANCH_FILLER_TERMS = { "auto", "autos", "car", "cars", "motor", "motors", "truck", "trucks", "vehicle", "vehicles", "vehicule", "vehicules", "official", "officiel", } MANUAL_REVIEW_FLAGS = { "ownership": "主体归属待确认：无法判断是独立公司，还是品牌官方主体、进口商直营网点或普通分店。", "new_vehicle": "新车业务待确认：无法判断主营新整车，还是二手车、维修、配件、轮胎等业务。", "platform_entity": "平台与行业渠道主体待确认：无法确认是真实机构主体，还是普通个人页、内容号或非正式资源页。", "platform_role": "平台与行业渠道作用待确认：无法确认其是否具备汽车行业资源、渠道引荐、媒体传播或行业活动组织作用。", "phone_only": "仅电话/WhatsApp待人工确认：线索已有目标价值信号，但只有电话或 WhatsApp，缺少官网、Facebook、LinkedIn、Google Maps 等可复核资料，需人工联系确认主体和实际业务。", "detail": "详细信息待确认：因登录限制、页面屏蔽、地区限制或网站无法访问，无法读取 Facebook、LinkedIn、官网等内容。", } def add_manual_review_flag(risks: List[str], key: str) -> None: flag = MANUAL_REVIEW_FLAGS.get(key, "") if flag and flag not in risks: risks.append(flag) def should_request_manual_review(score: int, recommended_action: str, reasons: Sequence[str]) -> bool: if recommended_action in {"skip_brand_branch", "skip_non_channel", "skip_existing"}: return False reason_text = " ".join(str(item) for item in reasons) value_markers = ( '实际新整车销售', 'showroom', '库存', '进口/分销', '网络能力', '多品牌', '已知独立', '中国品牌', '商用车', '车队', '汽车行业平台', '汽车媒体', '行业活动', '协会', '商会', '车商联盟', '经销商资源', '平台与行业渠道', '搜索词意图强', '主页链接像汽车业务', 'stock', 'dealer', 'media', 'event', 'association', 'chamber', 'platform', '可建联入口', ) return score >= 4 or any(marker in reason_text for marker in value_markers) def clean_space(value: Any) -> str: return re.sub(r"\s+", " ", str(value or "")).strip() def compact_text(value: Any) -> str: return re.sub(r"[^a-z0-9]+", "", str(value or "").casefold()) def matched_terms(text: str, terms: Iterable[str]) -> List[str]: text_lower = str(text or "").casefold() return sorted(term for term in terms if term.casefold() in text_lower) def normalize_url(value: Any) -> str: link = str(value or "").strip() if not link or link.casefold() == "nan": return "" parsed = urlparse(link) if not parsed.scheme or not parsed.netloc: return compact_text(link) netloc = parsed.netloc.casefold() if netloc.startswith("m."): netloc = "www." + netloc[2:] path = re.sub(r"/+", "/", parsed.path).rstrip("/") normalized = urlunparse((parsed.scheme.casefold(), netloc, path, "", "", "")) return normalized.rstrip("/").casefold() def normalize_linkedin_url(url: str) -> str: parsed = urlparse(str(url or "").strip()) if "linkedin.com" not in parsed.netloc.casefold(): return "" parts = [part for part in parsed.path.split("/") if part] if len(parts) < 2 or parts[0] != "company": return "" return f"https://www.linkedin.com/company/{parts[1]}/".casefold() def normalize_google_maps_url(url: str) -> str: parsed = urlparse(str(url or "").strip()) if "google." not in parsed.netloc.casefold() or "/maps/" not in parsed.path: return normalize_url(url) path = parsed.path.rstrip("/") return urlunparse(("https", parsed.netloc.casefold(), path, "", "", "")).casefold() def normalized_company_key(name: Any) -> str: text = clean_space(name).casefold() if not text or text == "nan": return "" tokens = re.findall(r"[a-z0-9]+", text) tokens = [token for token in tokens if token not in GENERIC_COMPANY_WORDS] return "".join(tokens) or compact_text(text) def independent_channel_display_name(text: str) -> str: combined = str(text or "").casefold() for clue, display in INDEPENDENT_CHANNEL_DISPLAY.items(): if clue in combined: return display return "" def canonical_dealer_name(name: str, text: str = "") -> str: combined = f"{name} {text}" display = independent_channel_display_name(combined) if not display: return clean_space(name) brand_terms = matched_terms(combined, OEM_BRANCH_BRANDS) name_key = normalized_company_key(name) display_key = normalized_company_key(display) if brand_terms and name_key != display_key: return display return clean_space(name) or display def dealer_group_key(candidate: Dict[str, Any]) -> str: name = clean_space(candidate.get("name") or candidate.get("公司名称") or candidate.get("客户姓名/公司")) combined = f"{name} {candidate.get('href', '')} {candidate.get('url', '')} {candidate.get('text', '')}".casefold() for clue, key in INDEPENDENT_CHANNEL_CLUES.items(): if clue in combined: return key return normalized_company_key(name) def load_existing_identity( excel_path: str, sheet_name: str, link_columns: Sequence[str], name_columns: Sequence[str], ) -> Dict[str, Set[str]]: path = Path(excel_path) if not excel_path or not path.exists(): return {"links": set(), "names": set()} try: df = pd.read_excel(path, sheet_name=sheet_name) except Exception: return {"links": set(), "names": set()} links: Set[str] = set() for col in link_columns: if col not in df.columns: continue for value in df[col].tolist(): normalized = normalize_url(value) if "linkedin.com" in normalized: normalized = normalize_linkedin_url(str(value)) elif "google." in normalized and "/maps/" in normalized: normalized = normalize_google_maps_url(str(value)) if normalized: links.add(normalized) names: Set[str] = set() for col in name_columns: if col not in df.columns: continue for value in df[col].tolist(): key = normalized_company_key(value) if key: names.add(key) return {"links": links, "names": names} def load_blocklist_json(blocklist_path: Optional[str], normalizer=normalize_url) -> Set[str]: if not blocklist_path: return set() path = Path(blocklist_path) if not path.exists(): return set() try: data = json.loads(path.read_text(encoding="utf-8")) except Exception: return set() values: Iterable[Any] if isinstance(data, list): values = data elif isinstance(data, dict) and "blocklist" in data: values = data["blocklist"] else: return set() return {normalized for normalized in (normalizer(str(value).strip()) for value in values) if normalized} def get_active_ws_endpoint(base_url: str, profile_id: str) -> str: resp = requests.get( f"{base_url.rstrip('/')}/api/v1/browser/active", params={"user_id": profile_id}, timeout=10, ) resp.raise_for_status() data = resp.json() if data.get("code") != 0: raise RuntimeError(f"获取活动浏览器失败: {data}") ws = (data.get("data") or {}).get("ws") or {} endpoint = ws.get("puppeteer") or ws.get("selenium") if not endpoint: raise RuntimeError(f"未找到已打开浏览器 ws endpoint，请确认 AdsPower profile 已打开: {profile_id}") return endpoint def looks_like_oem_local_branch(name: str, url: str = "", text: str = "") -> Tuple[bool, str]: combined = f"{name} {url} {text}".casefold() name_clean = clean_space(name).casefold() compact_name = compact_text(name) slug = urlparse(str(url or "")).path.strip("/").split("/")[-1] compact_slug = compact_text(slug) if any(clue in combined for clue in INDEPENDENT_CHANNEL_CLUES): return False, "" official_marker = re.search(r"\b(page officielle|official page|officiel|official)\b", combined) country_marker = re.search(r"\b(maroc|morocco|ma)\b", combined) for brand in OEM_BRANCH_BRANDS: brand_key = compact_text(brand) if not brand_key: continue direct_variants = { f"{brand_key}maroc", f"{brand_key}morocco", f"{brand_key}ma", f"maroc{brand_key}", f"morocco{brand_key}", } if compact_name in direct_variants or compact_slug in direct_variants: return True, brand filler_keys = {compact_text(term) for term in OEM_BRANCH_FILLER_TERMS} for country in ("maroc", "morocco", "ma"): if compact_name.startswith(brand_key) and compact_name.endswith(country): middle = compact_name[len(brand_key):-len(country)] if not middle or middle in filler_keys: return True, brand if compact_slug.startswith(brand_key) and compact_slug.endswith(country): middle = compact_slug[len(brand_key):-len(country)] if not middle or middle in filler_keys: return True, brand if re.fullmatch(rf"{re.escape(brand.casefold())}\s+(maroc|morocco)", name_clean): return True, brand if brand.casefold() in name_clean and official_marker and country_marker: return True, brand return False, "" def score_dealer_candidate( candidate: Dict[str, Any], existing_links: Optional[Set[str]] = None, existing_names: Optional[Set[str]] = None, platform: str = "", ) -> Dict[str, Any]: existing_links = existing_links or set() existing_names = existing_names or set() name = clean_space(candidate.get("name") or candidate.get("title") or "") href = clean_space(candidate.get("href") or candidate.get("url") or "") text = clean_space(candidate.get("text") or "") queries = candidate.get("source_queries", []) or [] combined = f"{name} {href} {text} {' '.join(queries)}" combined_lower = combined.casefold() normalized_link = normalize_url(href) if "linkedin.com" in normalized_link: normalized_link = normalize_linkedin_url(href) elif "google." in normalized_link and "/maps/" in normalized_link: normalized_link = normalize_google_maps_url(href) name_key = normalized_company_key(name) if normalized_link and normalized_link in existing_links: return {**candidate, "score": -10, "score_reasons": ["Already exists in workbook"], "risk_flags": ["duplicate_existing_link"], "recommended_action": "skip_existing"} if name_key and name_key in existing_names: return {**candidate, "score": -9, "score_reasons": ["Company name already exists in workbook"], "risk_flags": ["duplicate_existing_name"], "recommended_action": "skip_existing"} score = 0 reasons: List[str] = [] risks: List[str] = [] recommended_action = "preview_only" is_oem, brand = looks_like_oem_local_branch(name, href, text) brand_terms = matched_terms(combined, OEM_BRANCH_BRANDS) direct_branch_terms = matched_terms(combined, DIRECT_BRANCH_TERMS) if is_oem or (brand_terms and direct_branch_terms and not any(clue in combined_lower for clue in INDEPENDENT_CHANNEL_CLUES)): score -= 10 label = brand or ", ".join(brand_terms[:2]) or "brand" risks.append(f"疑似品牌当地分公司/官方主页/直营网点，排除: {label}") recommended_action = "skip_brand_branch" service_terms = matched_terms(combined, PURE_SERVICE_TERMS) if service_terms: score -= 6 risks.append("纯维修/配件/轮胎/服务类，不纳入汽车渠道合作伙伴: " + ", ".join(service_terms[:5])) if recommended_action == "preview_only": recommended_action = "skip_non_channel" personal_terms = matched_terms(combined, PERSONAL_SELLER_TERMS) if personal_terms: score -= 5 risks.append("疑似个人卖家，不纳入汽车渠道合作伙伴: " + ", ".join(personal_terms[:4])) if recommended_action == "preview_only": recommended_action = "skip_non_channel" sales_terms = matched_terms(combined, NEW_VEHICLE_SALES_TERMS) import_terms = matched_terms(combined, IMPORT_DISTRIBUTION_TERMS) multibrand_terms = matched_terms(combined, MULTIBRAND_TERMS) rental_terms = matched_terms(combined, RENTAL_FLEET_TERMS) contact_terms = matched_terms(combined, CONTACT_ENTRY_TERMS) morocco_terms = matched_terms(combined, MOROCCO_TERMS) has_actual_channel = bool(sales_terms or import_terms or multibrand_terms) ownership_needs_review = brand_terms and recommended_action != "skip_brand_branch" and not any(clue in combined_lower for clue in INDEPENDENT_CHANNEL_CLUES) and not (import_terms or multibrand_terms) if sales_terms: score += 3 reasons.append("实际新整车销售/showroom/库存信号: " + ", ".join(sales_terms[:5])) if import_terms: score += 3 reasons.append("进口/分销/网络能力信号: " + ", ".join(import_terms[:5])) if multibrand_terms: score += 2 reasons.append("多品牌经营信号: " + ", ".join(multibrand_terms[:4])) if morocco_terms: score += 1 reasons.append("Morocco signal: " + ", ".join(morocco_terms[:4])) if any(clue in combined_lower for clue in INDEPENDENT_CHANNEL_CLUES): score += 3 reasons.append("已知独立汽车渠道主体") if href and any(host in href.casefold() for host in ["facebook.com", "linkedin.com", "google."]): score += 2 reasons.append("可建联入口: 来源主页/平台页") elif contact_terms: score += 2 reasons.append("可建联入口: " + ", ".join(contact_terms[:4])) china_terms = matched_terms(combined, CHINA_BRAND_TERMS) if china_terms and recommended_action != "skip_brand_branch": score += 1 reasons.append("中国品牌渠道语境: " + ", ".join(china_terms[:4])) commercial_terms = matched_terms(combined, {"utilitaire", "camion", "truck", "fleet", "flotte"}) if commercial_terms: score += 1 reasons.append("商用车/车队相关信号: " + ", ".join(commercial_terms[:4])) if rental_terms and not has_actual_channel: score += 1 risks.append("纯租赁/车队线索，转入批量采购与运营客户/汽车租赁公司，不作为汽车渠道合作伙伴: " + ", ".join(rental_terms[:4])) if brand_terms and sales_terms and not (import_terms or multibrand_terms) and recommended_action != "skip_brand_branch": risks.append("独立单品牌新车经销商可保留；排他协议及新增品牌权限待确认") score += 1 new_vehicle_needs_review = False if not has_actual_channel and not rental_terms: score -= 3 risks.append("未发现新整车销售、进口、分销或 showroom 证据") new_vehicle_needs_review = True if ownership_needs_review and should_request_manual_review(score, recommended_action, reasons): add_manual_review_flag(risks, "ownership") if new_vehicle_needs_review and should_request_manual_review(score, recommended_action, reasons): add_manual_review_flag(risks, "new_vehicle") canonical_name = canonical_dealer_name(name, text) if canonical_name and canonical_name != name: candidate = {**candidate, "canonical_name": canonical_name} reasons.append(f"映射到独立经销集团: {canonical_name}") if not reasons: reasons.append(f"{platform or '搜索'} 信号较弱，低优先级；如无更多渠道价值证据应跳过") threshold = 4 if platform.casefold() == "google maps" else 5 if recommended_action == "preview_only" and score >= threshold: recommended_action = "deep_scrape" return {**candidate, "score": score, "score_reasons": reasons, "risk_flags": risks, "recommended_action": recommended_action} def merge_candidates_by_url(items: List[Dict[str, Any]], normalizer=normalize_url) -> List[Dict[str, Any]]: by_key: Dict[str, Dict[str, Any]] = {} for item in items: url = item.get("href") or item.get("url") or "" key = normalizer(url) or normalized_company_key(item.get("name", "")) if not key: continue if key not in by_key: by_key[key] = item continue existing = by_key[key] existing["source_queries"] = sorted(set(existing.get("source_queries", []) + item.get("source_queries", []))) snippets = existing.setdefault("snippets", [existing.get("text", "")]) if item.get("text") and item["text"] not in snippets: snippets.append(item["text"]) return list(by_key.values()) def dedupe_dealer_groups(candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]: grouped: Dict[str, Dict[str, Any]] = {} for candidate in candidates: key = dealer_group_key(candidate) if not key: continue current = grouped.get(key) if not current: grouped[key] = candidate continue current_score = int(current.get("score", 0)) candidate_score = int(candidate.get("score", 0)) current_name = clean_space(current.get("name", "")) candidate_name = clean_space(candidate.get("name", "")) prefer_candidate = ( candidate.get("recommended_action") != "skip_brand_branch" and ( current.get("recommended_action") == "skip_brand_branch" or candidate_score > current_score or (candidate_score == current_score and len(candidate_name) < len(current_name)) ) ) keeper = candidate if prefer_candidate else current merged = current if prefer_candidate else candidate keeper["source_queries"] = sorted(set(keeper.get("source_queries", []) + merged.get("source_queries", []))) keeper.setdefault("merged_duplicate_candidates", []) keeper["merged_duplicate_candidates"].append({ "name": merged.get("name", ""), "url": merged.get("href") or merged.get("url", ""), "score": merged.get("score"), "recommended_action": merged.get("recommended_action"), }) risks = set(keeper.get("risk_flags", [])) risks.add("同一集团/重复名称候选已合并") keeper["risk_flags"] = sorted(risks) grouped[key] = keeper return sorted(grouped.values(), key=lambda c: (-int(c.get("score", 0)), clean_space(c.get("name", "")))) def extract_urls(text: str) -> List[str]: urls = re.findall(r"https?://[^\s<>()\"']+", str(text or "")) cleaned: List[str] = [] for url in urls: url = url.rstrip(".,，。;；") if url not in cleaned: cleaned.append(url) return cleaned def extract_emails(text: str) -> List[str]: value = str(text or "") value = re.sub(r"\s*(?:\[at\]|\(at\)|\sat\s)\s*", "@", value, flags=re.I) value = re.sub(r"\s*(?:\[dot\]|\(dot\)|\sdot\s)\s*", ".", value, flags=re.I) emails = re.findall(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", value) filtered: List[str] = [] for email in emails: email = email.strip(".,;:()[]<>").casefold() if email.endswith((".png", ".jpg", ".jpeg", ".webp", ".gif")): continue if email not in filtered: filtered.append(email) return filtered def extract_email(text: str) -> str: emails = extract_emails(text) return emails[0] if emails else "" def extract_phone(text: str) -> str: value = str(text or "") explicit = re.search(r"(?:Téléphone|Telephone|电话|Tél|Tel|WhatsApp|Phone)[:：]?\s*([+()\d][+()\d\s.-]{6,}\d)", value, re.I) if explicit: return re.sub(r"\s+", " ", explicit.group(1)).strip() match = re.search(r"(?:\+212|0)\s?\d[\d\s.-]{6,}\d", value) return re.sub(r"\s+", " ", match.group(0)).strip() if match else "" def extract_city(text: str, fallback: str = DEFAULT_CITY_SCOPE) -> str: hits = [] for city in CITY_PATTERNS: if re.search(rf"\b{re.escape(city)}\b", str(text or ""), re.I) and city not in hits: hits.append(city) return " / ".join(hits[:4]) if hits else fallback def classify_customer_type(text: str) -> str: _attribute, customer_type = classify_from_text(str(text or "")) return customer_type def summarize_business(text: str) -> str: lower = str(text or "").casefold() parts: List[str] = [] if any(term in lower for term in ["occasion", "reprise", "voiture d'occasion"]): parts.append("二手车买卖/置换") if any(term in lower for term in ["concessionnaire", "showroom", "vente automobile", "汽车零售"]): parts.append("汽车销售/showroom") if any(term in lower for term in ["importateur", "distribution", "distributeur"]): parts.append("进口/分销") if any(term in lower for term in ["utilitaire", "camion", "truck", "fleet", "flotte"]): parts.append("商用车/车队") china = matched_terms(text, CHINA_BRAND_TERMS) non_china = matched_terms(text, NON_CHINA_BRAND_TERMS) if china: parts.append("涉及中国品牌: " + ", ".join(china[:4])) elif non_china: parts.append("主要品牌信号: " + ", ".join(non_china[:5])) return "；".join(parts) if parts else "汽车渠道线索，主营业务证据不足，需深搜确认" def parse_keywords(value: str) -> Optional[List[str]]: if not value.strip(): return None return [item.strip() for item in value.split(",") if item.strip()] def is_external_business_url(url: str) -> bool: parsed = urlparse(str(url or "")) if parsed.scheme not in {"http", "https"} or not parsed.netloc: return False host = parsed.netloc.casefold() blocked = [ "google.", "gstatic.", "facebook.com", "instagram.com", "linkedin.com", "tiktok.com", "youtube.com", "youtu.be", "twitter.com", "x.com", "wa.me", "whatsapp.com", "maps.apple.com", ] return not any(term in host for term in blocked) def looks_like_oem_brand_country_url(url: str) -> Tuple[bool, str]: parsed = urlparse(str(url or "")) compact = compact_text(" ".join([parsed.netloc, parsed.path])) if not compact: return False, "" for brand in OEM_BRANCH_BRANDS: brand_key = compact_text(brand) if not brand_key: continue variants = { f"{brand_key}maroc", f"{brand_key}morocco", f"{brand_key}ma", f"maroc{brand_key}", f"morocco{brand_key}", } if any(variant in compact for variant in variants): return True, brand return False, "" def normalize_website_url(url: str, base_url: str = "") -> str: url = clean_space(url) if not url: return "" if base_url: url = urljoin(base_url, url) parsed = urlparse(url) if not parsed.scheme and parsed.netloc: url = "https:" + url elif not parsed.scheme: url = "https://" + url return url