""" Google Maps Morocco dealer discovery with public website email extraction. Preview-first workflow: - Connect to an already-open AdsPower browser by profile ID. - Search Google Maps with Morocco-wide dealer/importer/showroom keywords. - Score and de-duplicate candidates before opening place detail pages. - Reject OEM local brand-country pages such as BYD Maroc or BMW Maroc. - Optionally deep-scrape place details and merchant websites for public emails. - Write to Excel only when --write-excel is explicitly passed. """ import argparse import json import random import re import sys import time from datetime import datetime, timezone from pathlib import Path import sys sys.path.append(str(Path(__file__).resolve().parents[1])) from common.artifact_manager import resolve_artifact_path, create_backup_once from typing import Any, Dict, List, Optional, Set from urllib.parse import parse_qs, quote_plus, urlparse if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") from playwright.sync_api import BrowserContext, Page, sync_playwright try: from ..common import append_records, resolve_workbook_path from . import discovery_common as dc except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from common import append_records, resolve_workbook_path from scraper import discovery_common as dc DEFAULT_EXCEL = "" DEFAULT_SHEET = "Google Maps" DEFAULT_CITY_SCOPE = "摩洛哥全国" NATIONWIDE_KEYWORDS = [ "concessionnaire automobile Maroc", "concessionnaire multimarque Maroc", "showroom auto Maroc", "voiture occasion Maroc", "importateur automobile Maroc", "distributeur automobile Maroc", "groupe automobile Maroc", "concessionnaire utilitaire Maroc", "camion Maroc concessionnaire", "voiture chinoise Maroc showroom", ] CITY_KEYWORD_PATTERNS = [ "concessionnaire automobile {city}", "showroom auto {city}", "voiture occasion {city}", "importateur automobile {city}", ] CONTACT_LINK_HINTS = { "contact", "nous contacter", "contactez", "about", "a-propos", "apropos", "à propos", "qui sommes", "mentions", "legal", "devis", "service client", } def build_default_keywords() -> List[str]: keywords = list(NATIONWIDE_KEYWORDS) for city in dc.MOROCCO_TARGET_CITIES: for pattern in CITY_KEYWORD_PATTERNS: keywords.append(pattern.format(city=city)) deduped: List[str] = [] seen: Set[str] = set() for keyword in keywords: key = keyword.casefold() if key not in seen: deduped.append(keyword) seen.add(key) return deduped DEFAULT_KEYWORDS = build_default_keywords() def dismiss_google_dialogs(page: Page) -> None: labels = [ "Accept all", "I agree", "Tout accepter", "J'accepte", "Accepter", "Reject all", "Plus tard", "Not now", "Fermer", "Close", ] for label in labels: try: page.get_by_text(label, exact=False).first.click(timeout=1200) time.sleep(0.4) except Exception: pass def clean_place_name(value: str) -> str: text = dc.clean_space(value) text = re.sub(r"\b(Directions|Itinéraire|Website|Site Web|Call|Appeler)\b.*$", "", text, flags=re.I).strip() lines = [dc.clean_space(line) for line in text.splitlines() if dc.clean_space(line)] if lines: text = lines[0] text = re.sub(r"\s+\d(?:[.,]\d)?\s*\(?\d*\)?$", "", text).strip() return text def collect_place_results(page: Page, query: str, max_results: int) -> List[Dict[str, Any]]: search_url = f"https://www.google.com/maps/search/{quote_plus(query)}" print(f"搜索 Google Maps: {query}", flush=True) page.goto(search_url, wait_until="domcontentloaded", timeout=60000) time.sleep(random.uniform(3, 5)) dismiss_google_dialogs(page) for _ in range(5): try: feed = page.locator('div[role="feed"]').first if feed.count(): feed.evaluate("el => el.scrollBy(0, 1400)") else: page.mouse.wheel(0, 1200) except Exception: page.mouse.wheel(0, 1200) time.sleep(random.uniform(1, 1.7)) raw = page.evaluate( """ () => { const anchors = Array.from(document.querySelectorAll('a[href*="/maps/place/"], a[href*="google.com/maps/place/"]')); return anchors.map((a) => { const node = a.closest('[role="article"]') || a.closest('.Nv2PK') || a.parentElement; const text = (node && node.innerText ? node.innerText : a.innerText || '').trim(); const label = (a.getAttribute('aria-label') || '').trim(); return {href: a.href || '', name: label || '', text}; }); } """ ) results: List[Dict[str, Any]] = [] seen: Set[str] = set() for item in raw: href = dc.normalize_google_maps_url(item.get("href", "")) if not href or href in seen: continue name = clean_place_name(item.get("name") or item.get("text", "")) text = re.sub(r"\n{2,}", "\n", item.get("text", "")).strip() if not name or name.casefold() in {"results", "google maps", "directions"}: continue seen.add(href) results.append({ "name": name, "href": item.get("href", ""), "text": text[:1200], "source_queries": [query], }) if len(results) >= max_results: break print(f" 收集到 {len(results)} 个地图候选", flush=True) return results def unwrap_google_redirect(url: str) -> str: parsed = urlparse(str(url or "")) if "google." in parsed.netloc.casefold() and parsed.path == "/url": target = parse_qs(parsed.query).get("q", [""])[0] if target: return target return url def extract_website_link(page: Page) -> str: raw_links = page.evaluate( """ () => Array.from(document.querySelectorAll('a[href]')).map((a) => ({ href: a.href || '', text: (a.innerText || '').trim(), aria: (a.getAttribute('aria-label') || '').trim(), data: (a.getAttribute('data-item-id') || '').trim() })) """ ) for item in raw_links: href = unwrap_google_redirect(item.get("href", "")) label = " ".join([item.get("text", ""), item.get("aria", ""), item.get("data", "")]).casefold() if not dc.is_external_business_url(href): continue if "authority" in label or "website" in label or "site web" in label: return href for item in raw_links: href = unwrap_google_redirect(item.get("href", "")) if dc.is_external_business_url(href): return href return "" def extract_maps_phone(page: Page, text: str) -> str: try: phone = page.evaluate( """ () => { const nodes = Array.from(document.querySelectorAll('button, a')); for (const el of nodes) { const href = el.href || ''; const aria = el.getAttribute('aria-label') || ''; const text = el.innerText || ''; if (href.startsWith('tel:')) return href.replace(/^tel:/, ''); const combined = `${aria} ${text}`; const match = combined.match(/(?:\+212|0)\s?\d[\d\s.-]{6,}\d/); if (match) return match[0]; } return ''; } """ ) if phone: return re.sub(r"\s+", " ", phone).strip() except Exception: pass return dc.extract_phone(text) def extract_rating(text: str) -> str: match = re.search(r"(\d[.,]\d)\s*\(?\s*(\d+[\d\s,.]*)?\s*(avis|reviews)?", text, re.I) if not match: return "" rating = match.group(1).replace(",", ".") reviews = dc.clean_space(match.group(2) or "") return f"{rating} ({reviews} avis)" if reviews else rating def same_site(url: str, base_url: str) -> bool: host = urlparse(url).netloc.casefold().removeprefix("www.") base_host = urlparse(base_url).netloc.casefold().removeprefix("www.") return bool(host and base_host and host == base_host) def scrape_public_email_from_website(context: BrowserContext, website_url: str, max_pages: int = 4) -> Dict[str, Any]: result = scrape_public_website_enrichment(context, website_url, max_pages=max_pages) return { "email": result.get("email", ""), "emails": result.get("emails", []), "sources": result.get("checked_urls", []), "checked_urls": result.get("checked_urls", []), "error": "" if result.get("checked_urls") else "no readable public website pages", } def extract_place_name(page: Page, fallback: str) -> str: for selector in ["h1", '[role="main"] h1']: try: value = page.locator(selector).first.inner_text(timeout=2500) value = clean_place_name(value) if value: return value except Exception: pass title = page.title().replace(" - Google Maps", "") return clean_place_name(title) or fallback def deep_scrape_place(page: Page, context: BrowserContext, candidate: Dict[str, Any], country: str) -> Optional[Dict[str, Any]]: print(f"深采 Google Maps: {candidate.get('name')}", flush=True) page.goto(candidate["href"], wait_until="domcontentloaded", timeout=60000) time.sleep(random.uniform(3, 5)) dismiss_google_dialogs(page) try: body_text = page.locator("body").inner_text(timeout=15000) except Exception: body_text = "" name = extract_place_name(page, candidate.get("name", "")) name = dc.canonical_dealer_name(name, body_text + "\n" + candidate.get("text", "")) place_url = page.url or candidate.get("href", "") website = extract_website_link(page) phone = extract_maps_phone(page, body_text) rating = extract_rating(body_text) city = dc.extract_city(body_text, fallback=DEFAULT_CITY_SCOPE) combined = "\n".join([name, place_url, body_text, candidate.get("text", ""), " ".join(candidate.get("source_queries", []))]) is_oem, brand = dc.looks_like_oem_local_branch(name, place_url, body_text) if is_oem: candidate["recommended_action"] = "skip_brand_branch" candidate.setdefault("risk_flags", []).append(f"深采确认疑似品牌官方页: {brand}") dc.add_manual_review_flag(candidate["risk_flags"], "ownership") return None website_is_oem, website_brand = dc.looks_like_oem_brand_country_url(website) if not body_text.strip(): candidate.setdefault("risk_flags", []) dc.add_manual_review_flag(candidate["risk_flags"], "detail") website_email = scrape_public_email_from_website(context, website) if website else {"email": "", "emails": [], "sources": [], "checked_urls": []} maps_email = dc.extract_email(body_text) if website_is_oem and website_email.get("email"): email = maps_email if maps_email and website_email.get("email") != maps_email else "" else: email = website_email.get("email") or maps_email customer_type = dc.classify_customer_type(combined) business = dc.summarize_business(combined) email_note = "" if email: if website_email.get("email") and not website_is_oem: email_note = "邮箱来源：官网公开页面 " + ", ".join(website_email.get("sources", [])[:2]) else: email_note = "邮箱来源：Google Maps 页面公开文本" elif website_is_oem and website_email.get("email"): email_note = f"发现 {website_brand} 品牌官网邮箱，未写入客户邮箱；需人工确认独立经销主体联系方式" candidate.setdefault("risk_flags", []) dc.add_manual_review_flag(candidate["risk_flags"], "ownership") elif website: email_note = "官网未发现公开邮箱" else: email_note = "无官网，未发现公开邮箱" note_parts = [ "Google Maps深采", f"官网：{website}" if website else "无官网", f"电话：{phone}" if phone else "未发现电话", f"评分：{rating}" if rating else "", email_note, f"官网检查页：{', '.join(website_email.get('checked_urls', [])[:3])}" if website_email.get("checked_urls") else "", f"来源搜索词：{', '.join(candidate.get('source_queries', []))}", f"评分：{candidate.get('score')}；原因：{'; '.join(candidate.get('score_reasons', []))}", ] if candidate.get("risk_flags"): note_parts.append("风险：" + "; ".join(candidate["risk_flags"])) return { "客户姓名/公司": name, "国家": country, "城市": city, "客户类型": customer_type, "主页/链接": place_url, "联系人": "", "职位": "", "电话/WhatsApp": phone, "邮箱": email, "主营业务": business, "建联状态": "未联系", "下次跟进": "", "备注": " | ".join(part for part in note_parts if part)[:1200], "_website": website, "_website_email_result": website_email, } def search_google_maps_dealers( profile_id: str, keywords: Optional[List[str]], ads_power_url: str, excel_path: str, sheet_name: str, max_results: int, max_results_per_query: int, min_score: int, deep_scrape: bool, country: str, no_excel: bool = False, ) -> Dict[str, Any]: if no_excel: existing_links: Set[str] = set() existing_names: Set[str] = set() else: existing_identity = dc.load_existing_identity( excel_path=excel_path, sheet_name=sheet_name, link_columns=["主页/链接"], name_columns=["客户姓名/公司"], ) existing_links = existing_identity["links"] existing_names = existing_identity["names"] print(f"已从 {sheet_name} Sheet 加载 {len(existing_links)} 个 Google Maps 现有链接、{len(existing_names)} 个公司名用于去重", flush=True) ws_endpoint = dc.get_active_ws_endpoint(ads_power_url, profile_id) playwright = sync_playwright().start() browser = playwright.chromium.connect_over_cdp(ws_endpoint) context = browser.contexts[0] if browser.contexts else browser.new_context() page = context.new_page() page.set_viewport_size({"width": 1366, "height": 850}) queries = keywords or DEFAULT_KEYWORDS raw_candidates: List[Dict[str, Any]] = [] search_log: List[Dict[str, Any]] = [] records: List[Dict[str, Any]] = [] selected: List[Dict[str, Any]] = [] scored: List[Dict[str, Any]] = [] skipped_existing: List[Dict[str, Any]] = [] try: for query in queries: items = collect_place_results(page, query, max_results_per_query) raw_candidates.extend(items) search_log.append({"query": query, "found": len(items), "url": page.url, "title": page.title()}) time.sleep(random.uniform(2, 4)) merged = dc.merge_candidates_by_url(raw_candidates, normalizer=dc.normalize_google_maps_url) scored_all = [ dc.score_dealer_candidate(item, existing_links, existing_names, platform="Google Maps") for item in merged ] skipped_existing = [c for c in scored_all if c.get("recommended_action") == "skip_existing"] active_scored = [c for c in scored_all if c.get("recommended_action") != "skip_existing"] scored = dc.dedupe_dealer_groups(active_scored) scored.sort(key=lambda c: (-int(c.get("score", 0)), c.get("name", ""))) selected = [ c for c in scored if c.get("recommended_action") == "deep_scrape" and int(c.get("score", 0)) >= min_score ][:max_results] if deep_scrape: for candidate in selected: record = deep_scrape_place(page, context, candidate, country=country) if record: records.append(record) normalized = dc.normalize_google_maps_url(record.get("主页/链接", "")) if normalized: existing_links.add(normalized) time.sleep(random.uniform(2, 4)) else: print("预览模式：已生成候选评分，不打开地图详情或官网。需要邮箱提取时加 --deep-scrape。", flush=True) finally: try: page.close() except Exception: pass playwright.stop() return { "generated_at": datetime.now(timezone.utc).isoformat(), "source": "Google Maps search via AdsPower active profile", "summary": { "profile_id": profile_id, "queries": queries, "raw_candidates": len(raw_candidates), "candidate_count": len(scored), "skipped_existing": len(skipped_existing), "selected_for_deep_scrape": len(selected), "deep_scrape": deep_scrape, "record_count": len(records), "records_with_email": len([r for r in records if r.get("邮箱")]), "min_score": min_score, "max_results": max_results, "existing_links": len(existing_links), "existing_names": len(existing_names), }, "search_log": search_log, "candidates": scored, "skipped_existing": skipped_existing, "selected_candidates": selected, "records": records, } def main() -> None: parser = argparse.ArgumentParser(description="Google Maps Morocco dealer discovery with public email extraction") parser.add_argument("--profile-id", required=True, help="AdsPower profile ID; must already be open") parser.add_argument("--ads-power-url", default="http://127.0.0.1:50325", help="AdsPower local API URL") parser.add_argument("--keywords", default="", help="Comma-separated Google Maps search keywords") parser.add_argument("--excel", default=DEFAULT_EXCEL, help="Workbook for duplicate checking and optional write-back") parser.add_argument("--sheet", default=DEFAULT_SHEET, help="Target sheet, normally Google Maps") parser.add_argument("--country", default="摩洛哥", help="Country value for records") parser.add_argument("--max-results", type=int, default=10, help="Maximum deep-scraped records") parser.add_argument("--max-results-per-query", type=int, default=6, help="Place links collected per query") parser.add_argument("--min-score", type=int, default=4, help="Minimum score for deep-scrape selection") parser.add_argument("--deep-scrape", action="store_true", help="Open selected place pages and merchant websites for emails") parser.add_argument("--write-excel", action="store_true", help="Write deep-scraped records to the workbook") parser.add_argument("--no-excel", action="store_true", help="Do not read or write Excel; pure JSON preview") parser.add_argument("--run-id", default="", help="Run ID used for runs/YYYYMMDD// artifacts.") parser.add_argument("--output", default="google_maps_candidate_preview.json", help="Output JSON path") args = parser.parse_args() if not args.no_excel: read_workbook_info = resolve_workbook_path(args.excel, create_from_template=False) if read_workbook_info.get("path"): args.excel = str(read_workbook_info["path"]) print(f"Workbook for duplicate checking: {args.excel} ({read_workbook_info['source']})", flush=True) result = search_google_maps_dealers( profile_id=args.profile_id, keywords=dc.parse_keywords(args.keywords), ads_power_url=args.ads_power_url, excel_path=args.excel, sheet_name=args.sheet, max_results=args.max_results, max_results_per_query=args.max_results_per_query, min_score=args.min_score, deep_scrape=args.deep_scrape, country=args.country, no_excel=args.no_excel, ) output_path = resolve_artifact_path(args.output, kind="scraper_preview", default_name=Path(args.output).name, run_id=args.run_id or None) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(f"已保存 Google Maps 候选结果: {output_path}", flush=True) if args.write_excel and not args.no_excel: write_workbook_info = resolve_workbook_path(args.excel, create_from_template=True) if write_workbook_info.get("path"): args.excel = str(write_workbook_info["path"]) print(f"Workbook for writing: {args.excel} ({write_workbook_info['source']})", flush=True) if not args.deep_scrape: print("未写入 Excel：需要先使用 --deep-scrape 生成 records。", flush=True) elif not result.get("records"): print("未写入 Excel：没有可写入记录。", flush=True) else: write_result = append_records( excel_path=args.excel, sheet_name=args.sheet, records=result["records"], dedup_keys=["客户姓名/公司", "主页/链接"], ) print(f"Excel 回写结果: {write_result}", flush=True) else: print("默认预览模式：未写入 Excel。确认要入表时再使用 --deep-scrape --write-excel。", flush=True) if __name__ == "__main__": main()