""" Collect Morocco auto-dealer leads from vertical auto sites and business directories. This script uses an already-open AdsPower browser profile, searches six approved source websites, deep-scrapes candidate/company pages, follows public merchant website contact pages for emails, and writes results to a dedicated workbook sheet. It intentionally does not use Moteur. """ import argparse import json import random import re import shutil import sys import time from dataclasses import dataclass, field 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, Iterable, List, Optional, Set, Tuple from urllib.parse import quote_plus, urljoin, urlparse import requests from openpyxl import Workbook, load_workbook from playwright.sync_api import BrowserContext, Page, sync_playwright try: from . import discovery_common as dc from ..common import resolve_workbook_path except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scraper import discovery_common as dc from common import resolve_workbook_path if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") DEFAULT_EXCEL_DIR = Path.cwd() DEFAULT_SHEET = "汽车网站精选线索" DEFAULT_COUNTRY = "摩洛哥" DEFAULT_CITY = "摩洛哥全国" DEFAULT_ADSPOWER_URL = "http://127.0.0.1:50325" HEADERS = [ "序号", "客户姓名/公司", "国家", "城市", "客户类型", "主页/链接", "来源网站", "联系人", "职位", "电话/WhatsApp", "邮箱", "主营业务", "建联状态", "下次跟进", "备注", ] CONTACT_LINK_HINTS = { "contact", "contactez", "nous contacter", "a propos", "à propos", "apropos", "qui sommes", "mentions", "legal", "devis", } HIGH_VALUE_TERMS = { "importation", "importateur", "importateurs", "véhicules neufs", "vehicules neufs", "voitures neuves", "professionnel", "professionnels", "concessionnaire", "concessionnaires", "distributeur", "distributeurs", "showroom", "stock", "parc auto", "location", "lld", "leasing", "flotte", "fleet", "utilitaire", "utilitaires", "camionnette", "fourgon", "pick-up", "pickup", "mpv", "minibus", "économique", "economique", } LOW_VALUE_TERMS = { "garage réparation", "garage reparation", "diagnostic", "pièces détachées", "pieces detachees", "assurance", "lavage", "car wash", "pare-brise", "immobilier", "emploi", } @dataclass class SourceConfig: name: str domains: Tuple[str, ...] queries: List[str] start_urls: List[str] = field(default_factory=list) SOURCES = [ SourceConfig( name="OtoMoto.ma", domains=("otomoto.ma",), start_urls=["https://otomoto.ma/guide/professionnels", "https://otomoto.ma/"], queries=[ "site:otomoto.ma Maroc professionnel concessionnaire automobile", "site:otomoto.ma Maroc revendeur automobile professionnel", "site:otomoto.ma Maroc importateur voiture professionnel", ], ), SourceConfig( name="Wandaloo", domains=("wandaloo.com",), start_urls=["https://www.wandaloo.com/neuf/maroc/concessionnaire.html"], queries=[ "site:wandaloo.com/neuf/maroc concessionnaire distributeur automobile Maroc", "site:wandaloo.com/neuf/maroc importateur showroom Maroc", "site:wandaloo.com/neuf/maroc utilitaire concessionnaire Maroc", ], ), SourceConfig( name="Kerix", domains=("kerix.net",), start_urls=["https://www.kerix.net/fr/annuaire-entreprise/automobiles.html"], queries=[ "site:kerix.net Maroc automobiles importation concessionnaire", "site:kerix.net Maroc concessionnaires régionaux automobiles", "site:kerix.net Maroc vehicules utilitaires importateur", ], ), SourceConfig( name="Kompass", domains=("kompass.com",), start_urls=["https://ma.kompass.com/y/importer/a/vehicules-utilitaires/66340/"], queries=[ "site:ma.kompass.com Maroc importateur véhicules utilitaires", "site:ma.kompass.com Maroc concessionnaire automobile", "site:ma.kompass.com Maroc distributeur véhicules automobiles", ], ), SourceConfig( name="Maroc Annuaire", domains=("marocannuaire.org",), start_urls=["https://marocannuaire.org/Annuaire/Activite.php?activite=Automobile+%28Concessionnaires%29"], queries=[ "site:marocannuaire.org Automobile Concessionnaires Maroc email", "site:marocannuaire.org importateur automobile Maroc email", "site:marocannuaire.org location voitures Maroc entreprise", ], ), SourceConfig( name="Telecontact", domains=("telecontact.ma",), start_urls=["https://www.telecontact.ma/villes/automobiles-agents-concessionnaires.php"], queries=[ "site:telecontact.ma automobiles agents concessionnaires Maroc", "site:telecontact.ma concessionnaire automobile Casablanca Maroc", "site:telecontact.ma véhicules utilitaires concessionnaire Maroc", ], ), ] def clean(value: Any) -> str: return re.sub(r"\s+", " ", str(value or "")).strip() def is_source_url(url: str, source: SourceConfig) -> bool: host = urlparse(str(url or "")).netloc.casefold() return any(domain in host for domain in source.domains) def is_html_candidate(url: str) -> bool: if not url.startswith(("http://", "https://")): return False lowered = url.casefold() blocked_parts = [ "facebook.com", "instagram.com", "linkedin.com", "youtube.com", "wa.me", "whatsapp", "google.", "bing.com", "/login", "/signup", "/privacy", ] if any(part in lowered for part in blocked_parts): return False if re.search(r"\.(pdf|jpg|jpeg|png|gif|webp|zip|rar)(?:$|\?)", lowered): return False return True def find_workbook(base_dir: Path = DEFAULT_EXCEL_DIR) -> Path: candidates: List[Path] = [] for path in base_dir.glob("*.xlsx"): if path.name.startswith("~$") or "_backup_" in path.name or "_with_" in path.name: continue try: workbook = load_workbook(path, read_only=True, data_only=False) if any(sheet in workbook.sheetnames for sheet in ["Facebook", "Google Maps", "LinkedIn", "本地汽车网站"]): candidates.append(path) workbook.close() except Exception: continue if not candidates: raise FileNotFoundError("未找到摩洛哥客户建联表") return sorted(candidates, key=lambda p: p.stat().st_mtime, reverse=True)[0] def get_active_profile(base_url: str) -> Tuple[str, str]: resp = requests.get(f"{base_url.rstrip('/')}/api/v1/browser/local-active", timeout=10) resp.raise_for_status() data = resp.json() active = ((data.get("data") or {}).get("list") or []) if not active: raise RuntimeError("未发现已打开的 AdsPower 浏览器配置") item = active[0] ws = (item.get("ws") or {}).get("puppeteer") or (item.get("ws") or {}).get("selenium") if not ws: raise RuntimeError(f"已打开配置没有 ws endpoint: {item}") return item.get("user_id", ""), ws def dismiss_dialogs(page: Page) -> None: labels = [ "Accept all", "Tout accepter", "J'accepte", "Accepter", "Reject all", "Plus tard", "Fermer", "Close", "OK", ] for label in labels: try: page.get_by_text(label, exact=False).first.click(timeout=800) time.sleep(0.3) except Exception: pass def safe_body_text(page: Page, timeout: int = 12000) -> str: try: return page.locator("body").inner_text(timeout=timeout) except Exception: return "" def page_links(page: Page) -> List[Dict[str, str]]: try: return page.evaluate( """ () => Array.from(document.querySelectorAll('a[href]')).map((a) => ({ href: a.href || '', raw: a.getAttribute('href') || '', text: (a.innerText || '').trim(), aria: (a.getAttribute('aria-label') || '').trim() })) """ ) except Exception: return [] def collect_bing_results(page: Page, source: SourceConfig, query: str, limit: int) -> List[Dict[str, Any]]: url = f"https://www.bing.com/search?q={quote_plus(query)}" print(f"搜索 {source.name}: {query}", flush=True) page.goto(url, wait_until="domcontentloaded", timeout=60000) time.sleep(random.uniform(2.5, 4.0)) dismiss_dialogs(page) for _ in range(2): page.mouse.wheel(0, 1000) time.sleep(random.uniform(0.7, 1.2)) links = page.evaluate( """ () => Array.from(document.querySelectorAll('li.b_algo h2 a, a[href]')).map((a) => ({ href: a.href || '', text: (a.innerText || a.getAttribute('aria-label') || '').trim() })) """ ) results: List[Dict[str, Any]] = [] seen: Set[str] = set() for item in links: href = item.get("href", "") if not is_source_url(href, source) or not is_html_candidate(href): continue key = dc.normalize_url(href) if not key or key in seen: continue seen.add(key) results.append({ "url": href, "title": clean(item.get("text")), "source": source.name, "source_queries": [query], "collection_method": "bing", }) if len(results) >= limit: break return results def collect_start_url_links(page: Page, source: SourceConfig, start_url: str, limit: int) -> List[Dict[str, Any]]: print(f"打开来源页 {source.name}: {start_url}", flush=True) page.goto(start_url, wait_until="domcontentloaded", timeout=60000) time.sleep(random.uniform(2.5, 4.0)) dismiss_dialogs(page) for _ in range(3): page.mouse.wheel(0, 1200) time.sleep(random.uniform(0.7, 1.2)) body = safe_body_text(page)[:3000] results: List[Dict[str, Any]] = [{ "url": page.url, "title": clean(page.title()), "text_hint": body, "source": source.name, "source_queries": [start_url], "collection_method": "start_url", }] seen = {dc.normalize_url(page.url)} for item in page_links(page): href = urljoin(page.url, item.get("href", "")) if not is_source_url(href, source) or not is_html_candidate(href): continue label = clean(" ".join([item.get("text", ""), item.get("aria", "")])) if not label or len(label) > 120: continue key = dc.normalize_url(href) if not key or key in seen: continue if not any(term in f"{label} {href}".casefold() for term in ["concession", "auto", "voiture", "garage", "import", "vehicule", "véhicule", "dealer", "annuaire", "societe", "entreprise"]): continue seen.add(key) results.append({ "url": href, "title": label, "source": source.name, "source_queries": [start_url], "collection_method": "start_url_link", }) if len(results) >= limit: break return results def merge_candidates(candidates: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: merged: Dict[str, Dict[str, Any]] = {} for candidate in candidates: key = dc.normalize_url(candidate.get("url", "")) or dc.normalized_company_key(candidate.get("title")) if not key: continue if key not in merged: merged[key] = candidate continue existing = merged[key] existing["source_queries"] = sorted(set(existing.get("source_queries", []) + candidate.get("source_queries", []))) if candidate.get("text_hint"): existing["text_hint"] = clean(" ".join([existing.get("text_hint", ""), candidate.get("text_hint", "")]))[:4000] return list(merged.values()) def extract_company_name(page: Page, candidate: Dict[str, Any], body: str) -> str: for selector in ["h1", "h2"]: try: value = clean(page.locator(selector).first.inner_text(timeout=1500)) if value and len(value) <= 100: return re.sub(r"\s*[-|].*$", "", value).strip() except Exception: pass title = clean(page.title() or candidate.get("title", "")) title = re.sub(r"\s*[-|]\s*(Kerix|Kompass|Telecontact|Wandaloo|OtoMoto|Maroc Annuaire).*$", "", title, flags=re.I) return title[:100] or clean(candidate.get("title", ""))[:100] def extract_external_websites(current_url: str, links: List[Dict[str, str]], source: SourceConfig) -> List[str]: websites: List[str] = [] source_hosts = set(source.domains) for item in links: href = urljoin(current_url, item.get("href") or item.get("raw") or "") if not dc.is_external_business_url(href): continue host = urlparse(href).netloc.casefold().removeprefix("www.") if any(domain in host for domain in source_hosts): continue label = clean(" ".join([item.get("text", ""), item.get("aria", ""), item.get("raw", "")])).casefold() if any(term in label for term in ["site web", "website", "web", "www", "visiter", "voir le site", "site internet"]) or len(websites) < 2: normalized = href.split("#")[0] if normalized not in websites: websites.append(normalized) return websites[:3] def same_host(url: str, base_url: str) -> bool: return urlparse(url).netloc.casefold().removeprefix("www.") == urlparse(base_url).netloc.casefold().removeprefix("www.") def scrape_website_email(context: BrowserContext, website_url: str, max_pages: int = 4) -> Dict[str, Any]: if not website_url or not dc.is_external_business_url(website_url): return {"email": "", "emails": [], "sources": [], "checked_urls": []} checked: List[str] = [] queue: List[str] = [website_url] emails: List[str] = [] sources: 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 (checked and not same_host(url, website_url)): continue checked.append(url) try: print(f" 检查官网邮箱: {url}", flush=True) page.goto(url, wait_until="domcontentloaded", timeout=35000) time.sleep(random.uniform(1.2, 2.2)) dismiss_dialogs(page) body = safe_body_text(page) links = page_links(page) mailto_text = " ".join(item.get("raw", "") for item in links if item.get("raw", "").casefold().startswith("mailto:")) for email in dc.extract_emails(body + " " + mailto_text): if email not in emails: emails.append(email) sources.append(url) if emails: break for item in links: label = clean(" ".join([item.get("raw", ""), 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") or item.get("raw") or "", page.url) if href and href not in checked and href not in queue and same_host(href, website_url): queue.append(href) except Exception: continue finally: try: page.close() except Exception: pass return {"email": emails[0] if emails else "", "emails": emails, "sources": sources, "checked_urls": checked} def score_record(name: str, url: str, body: str, source: str) -> Tuple[int, List[str], List[str]]: combined = f"{name} {url} {body}".casefold() score = 0 reasons: List[str] = [] risks: List[str] = [] sales_terms = [term for term in ["véhicules neufs", "vehicules neufs", "voitures neuves", "concessionnaire", "showroom", "vente automobile", "stock", "parc auto", "professionnel"] if term in combined] import_terms = [term for term in ["importation", "importateur", "importateurs", "distributeur", "distributeurs", "réseau", "reseau", "points de vente", "succursales", "agences"] if term in combined] multibrand_terms = [term for term in ["multimarque", "multi-brand", "multi brand", "plusieurs marques", "marques multiples"] if term in combined] rental_terms = [term for term in ["location", "lld", "leasing", "flotte", "fleet"] if term in combined] service_terms = [term for term in ["garage réparation", "garage reparation", "diagnostic", "pièces détachées", "pieces detachees", "pneus", "tires", "lavage", "car wash", "pare-brise", "assurance", "immobilier", "emploi"] if term in combined] has_actual_channel = bool(sales_terms or 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 source in {"Kerix", "Kompass", "Maroc Annuaire", "Telecontact"}: score += 1 reasons.append("annuaire professionnel") if re.search(r"(?:\+212|0)\s?\d[\d\s.-]{6,}\d", body): score += 2 reasons.append("可建联入口: 电话/WhatsApp") if dc.extract_emails(body): score += 2 reasons.append("可建联入口: email public") if any(term in combined for term in ["facebook.com", "linkedin.com", "whatsapp", "wa.me", "contact"]): score += 1 reasons.append("可建联入口: social/contact link") if any(term in combined for term in ["stock", "annonces", "véhicules disponibles", "vehicules disponibles", "parc"]): score += 1 reasons.append("stock véhicules") if service_terms: score -= 6 risks.append("纯维修/配件/轮胎/服务类,不纳入汽车渠道合作伙伴: " + ", ".join(service_terms[:5])) if rental_terms and not has_actual_channel: score += 1 risks.append("纯租赁/车队线索,转入批量采购与运营客户/汽车租赁公司: " + ", ".join(rental_terms[:4])) 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 new_vehicle_needs_review and dc.should_request_manual_review(score, "preview_only", reasons): dc.add_manual_review_flag(risks, "new_vehicle") return score, reasons, risks def summarize_record( source: str, score: int, reasons: List[str], risks: List[str], page_url: str, website: str, email_source: str, queries: List[str], ) -> str: capability = "批量采购能力判断:" reason_text = "、".join(reasons[:8]) if reasons else "证据不足,低优先级;如无更多渠道价值证据应跳过" if any(term in reason_text for term in ["multi-site", "stock", "importation", "distributeur", "utilitaire", "flotte", "location"]): capability += "有批量/车队/分销潜力" else: capability += "待确认" parts = [ f"来源线索:{source};详情页 {page_url}", f"主营业务判断:{reason_text}", capability, f"库存/门店/租赁/进口证据:{reason_text}", f"联系方式证据:官网 {website}" if website else "联系方式证据:未发现独立官网", f"邮箱来源:{email_source}" if email_source else "邮箱来源:未发现公开邮箱", f"来源搜索词:{', '.join(queries[:3])}", f"评分:{score}", ] if risks: parts.append("风险/待确认项:" + "、".join(risks[:5])) return " | ".join(parts)[:1400] def deep_scrape_candidate(page: Page, context: BrowserContext, candidate: Dict[str, Any], source: SourceConfig) -> Optional[Dict[str, Any]]: url = candidate["url"] print(f"深采 {source.name}: {url}", flush=True) try: page.goto(url, wait_until="domcontentloaded", timeout=60000) time.sleep(random.uniform(2.0, 3.4)) dismiss_dialogs(page) body = safe_body_text(page) except Exception as exc: return { "_skip": True, "_skip_reason": f"打开失败: {exc}", "url": url, "source": source.name, } if not body or len(body) < 80: return {"_skip": True, "_skip_reason": "页面正文过短", "url": url, "source": source.name} name = extract_company_name(page, candidate, body) if not name or len(name) < 2: return {"_skip": True, "_skip_reason": "未识别公司名", "url": url, "source": source.name} is_oem, brand = dc.looks_like_oem_local_branch(name, url, body) if is_oem: return {"_skip": True, "_skip_reason": f"疑似品牌官方国家页: {brand}", "url": url, "source": source.name} score, reasons, risks = score_record(name, url, body, source.name) if score < 3: return {"_skip": True, "_skip_reason": f"评分过低: {score}", "url": url, "source": source.name} links = page_links(page) websites = extract_external_websites(page.url, links, source) website = websites[0] if websites else "" page_emails = dc.extract_emails(body + " " + " ".join(item.get("raw", "") for item in links)) website_email_result = scrape_website_email(context, website) if website else {"email": "", "emails": [], "sources": [], "checked_urls": []} email = website_email_result.get("email") or (page_emails[0] if page_emails else "") email_source = "" if website_email_result.get("email"): email_source = "官网公开页面 " + ", ".join(website_email_result.get("sources", [])[:2]) elif email: email_source = "来源网站页面公开文本" combined_text = f"{name}\n{body}\n{' '.join(candidate.get('source_queries', []))}" phone = dc.extract_phone(body) city = dc.extract_city(combined_text, fallback=DEFAULT_CITY) business = dc.summarize_business(combined_text) customer_type = dc.classify_customer_type(combined_text) note = summarize_record( source=source.name, score=score, reasons=reasons, risks=risks, page_url=page.url, website=website, email_source=email_source, queries=candidate.get("source_queries", []), ) return { "序号": "", "客户姓名/公司": name, "国家": DEFAULT_COUNTRY, "城市": city, "客户类型": customer_type, "主页/链接": page.url, "来源网站": source.name, "联系人": "", "职位": "", "电话/WhatsApp": phone, "邮箱": email, "主营业务": business, "建联状态": "未联系", "下次跟进": "", "备注": note, "_score": score, "_score_reasons": reasons, "_risk_flags": risks, "_website": website, "_website_email_result": website_email_result, } def split_multi(value: str) -> List[str]: return [clean(part) for part in re.split(r"[;;|]+", str(value or "")) if clean(part)] def append_unique_text(old: str, addition: str, separator: str = " | ") -> str: old = clean(old) addition = clean(addition) if not addition: return old if not old: return addition if addition in old: return old return (old + separator + addition)[:3000] def merge_sources(old: str, addition: str) -> str: values = [] for item in split_multi(old) + split_multi(addition): if item and item not in values: values.append(item) return ";".join(values) def duplicate_key(record: Dict[str, Any]) -> List[str]: keys = [] website = record.get("_website") or "" if website: keys.append("website:" + dc.normalize_url(website)) name = record.get("客户姓名/公司") if name: keys.append("name:" + dc.normalized_company_key(name)) for field in ["电话/WhatsApp", "邮箱", "主页/链接"]: value = clean(record.get(field)) if value: normalized = dc.normalize_url(value) if field == "主页/链接" else value.casefold() keys.append(f"{field}:{normalized}") return [key for key in keys if key and not key.endswith(":")] def prepare_sheet(workbook: Workbook, sheet_name: str): if sheet_name in workbook.sheetnames: ws = workbook[sheet_name] for idx, header in enumerate(HEADERS, start=1): ws.cell(row=1, column=idx).value = header return ws ws = workbook.create_sheet(sheet_name) ws.append(HEADERS) return ws def read_existing_sheet_index(ws) -> Dict[str, int]: index: Dict[str, int] = {} header_map = {clean(ws.cell(1, col).value): col for col in range(1, ws.max_column + 1)} for row in range(2, ws.max_row + 1): record = {header: ws.cell(row, col).value for header, col in header_map.items()} for key in duplicate_key(record): index[key] = row return index def write_records_to_workbook(excel_path: Path, sheet_name: str, records: List[Dict[str, Any]]) -> Dict[str, Any]: locks = sorted(p.name for p in excel_path.parent.glob("~$*.xlsx")) if locks: raise PermissionError("检测到 Excel 临时锁文件: " + ", ".join(locks)) backup_path = create_backup_once(excel_path, purpose="auto_websites", run_id="auto_websites") wb = load_workbook(excel_path) ws = prepare_sheet(wb, sheet_name) header_map = {header: idx for idx, header in enumerate(HEADERS, start=1)} existing_index = read_existing_sheet_index(ws) appended = 0 merged = 0 no_email = 0 for record in records: if not record.get("邮箱"): no_email += 1 row = None for key in duplicate_key(record): if key in existing_index: row = existing_index[key] break if row: merged += 1 for field in ["来源网站", "备注"]: col = header_map[field] if field == "来源网站": ws.cell(row, col).value = merge_sources(ws.cell(row, col).value, record.get(field, "")) else: ws.cell(row, col).value = append_unique_text(ws.cell(row, col).value, record.get(field, "")) for field in ["电话/WhatsApp", "邮箱", "主页/链接", "主营业务", "客户类型", "城市"]: col = header_map[field] if not clean(ws.cell(row, col).value) and clean(record.get(field)): ws.cell(row, col).value = record.get(field) continue appended += 1 row = ws.max_row + 1 record["序号"] = row - 1 for header, col in header_map.items(): ws.cell(row, col).value = record.get(header, "") for key in duplicate_key(record): existing_index[key] = row wb.save(excel_path) wb.close() return { "backup_path": str(backup_path), "appended": appended, "merged": merged, "no_email": no_email, "sheet": sheet_name, "workbook": str(excel_path), } def collect_records( profile_id: str, ws_endpoint: str, max_per_source: int, max_candidates_per_source: int, queries_per_source: int, start_urls_per_source: int, ) -> Dict[str, Any]: 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}) source_logs: List[Dict[str, Any]] = [] records: List[Dict[str, Any]] = [] skipped: List[Dict[str, Any]] = [] try: for source in SOURCES: raw: List[Dict[str, Any]] = [] errors: List[str] = [] try: for start_url in source.start_urls[:start_urls_per_source]: raw.extend(collect_start_url_links(page, source, start_url, max_candidates_per_source)) time.sleep(random.uniform(1.2, 2.0)) for query in source.queries[:queries_per_source]: raw.extend(collect_bing_results(page, source, query, max_candidates_per_source)) time.sleep(random.uniform(1.5, 2.5)) except Exception as exc: errors.append(str(exc)) candidates = merge_candidates(raw)[:max_candidates_per_source] source_records: List[Dict[str, Any]] = [] for candidate in candidates: if len(source_records) >= max_per_source: break result = deep_scrape_candidate(page, context, candidate, source) if not result: continue if result.get("_skip"): skipped.append(result) continue source_records.append(result) records.append(result) time.sleep(random.uniform(1.4, 2.4)) source_logs.append({ "source": source.name, "raw_candidates": len(raw), "merged_candidates": len(candidates), "records": len(source_records), "errors": errors, }) finally: try: page.close() except Exception: pass playwright.stop() return { "generated_at": datetime.now(timezone.utc).isoformat(), "profile_id": profile_id, "records": records, "skipped": skipped, "source_logs": source_logs, } def main() -> None: parser = argparse.ArgumentParser(description="Collect Morocco auto website leads via AdsPower") parser.add_argument("--profile-id", default="", help="AdsPower profile ID; defaults to current local active browser") parser.add_argument("--ads-power-url", default=DEFAULT_ADSPOWER_URL) parser.add_argument("--excel", default="", help="Workbook path; defaults to detected Morocco outreach workbook") parser.add_argument("--sheet", default=DEFAULT_SHEET) parser.add_argument("--max-per-source", type=int, default=5) parser.add_argument("--max-candidates-per-source", type=int, default=12) parser.add_argument("--queries-per-source", type=int, default=3) parser.add_argument("--start-urls-per-source", type=int, default=1) parser.add_argument("--run-id", default="", help="Run ID used for runs/YYYYMMDD// artifacts.") parser.add_argument("--output", default="auto_website_leads.json") parser.add_argument("--write-excel", action="store_true") args = parser.parse_args() if args.profile_id: ws_endpoint = dc.get_active_ws_endpoint(args.ads_power_url, args.profile_id) profile_id = args.profile_id else: profile_id, ws_endpoint = get_active_profile(args.ads_power_url) workbook_info = resolve_workbook_path(args.excel, create_from_template=args.write_excel) excel_path = workbook_info.get("path") if excel_path and (args.write_excel or workbook_info.get("source") != "missing"): print(f"Workbook resolved: {excel_path} ({workbook_info['source']})", flush=True) result = collect_records( profile_id=profile_id, ws_endpoint=ws_endpoint, max_per_source=args.max_per_source, max_candidates_per_source=args.max_candidates_per_source, queries_per_source=max(0, args.queries_per_source), start_urls_per_source=max(0, args.start_urls_per_source), ) result["summary"] = { "record_count": len(result["records"]), "records_with_email": len([r for r in result["records"] if r.get("邮箱")]), "sources": {log["source"]: log["records"] for log in result["source_logs"]}, "skipped_count": len(result["skipped"]), "excel": str(excel_path) if excel_path else "", "sheet": args.sheet, } 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"已保存采集结果: {output_path}", flush=True) if args.write_excel: if not excel_path: raise SystemExit("使用 --write-excel 时必须提供 --excel 工作簿路径,或在当前目录放置可识别的建联表。") write_result = write_records_to_workbook(excel_path, args.sheet, result["records"]) result["write_result"] = write_result output_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print("Excel 写入结果: " + json.dumps(write_result, ensure_ascii=False), flush=True) else: print("预览模式:未写入 Excel。", flush=True) if __name__ == "__main__": main()