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

"""
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/<run_id>/ 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()