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

"""
Facebook 经销商搜索与采集
使用 Playwright + AdsPower 指纹浏览器
"""
import re
import time
import random
import argparse
import json
from typing import List, Dict, Any, Optional
from urllib.parse import urlencode, urlparse, parse_qs
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 playwright.sync_api import Page, TimeoutError as PlaywrightTimeout

try:
    from deep_translator import GoogleTranslator
    TRANSLATOR_AVAILABLE = True
except ImportError:
    TRANSLATOR_AVAILABLE = False
    GoogleTranslator = None

# 导入同目录或上级 common 模块
try:
    from .ads_power_client import AdsPowerClient
    from .website_deep_scraper import build_chinese_notes, classify_customer_type_cn, extract_company_website_from_facebook, scrape_public_website, summarize_business_cn
    from ..common import deduplicate_records, enrich_record_with_brands, append_records, resolve_workbook_path
except ImportError:
    import sys
    sys.path.insert(0, str(Path(__file__).parent.parent))
    from scraper.ads_power_client import AdsPowerClient
    from scraper.website_deep_scraper import build_chinese_notes, classify_customer_type_cn, extract_company_website_from_facebook, scrape_public_website, summarize_business_cn
    from common import deduplicate_records, enrich_record_with_brands, append_records, resolve_workbook_path


def is_mostly_chinese(text: str) -> bool:
    """判断文本是否主要为中文"""
    if not text:
        return False
    chinese_chars = sum(1 for c in text if "\u4e00" <= c <= "\u9fff")
    return chinese_chars / max(len(text), 1) > 0.3


def translate_to_chinese(text: str) -> str:
    """将文本翻译为中文；失败或已是中文时返回原文"""
    if not text or len(text.strip()) < 3:
        return text
    if is_mostly_chinese(text):
        return text
    if not TRANSLATOR_AVAILABLE:
        return text
    try:
        translated = GoogleTranslator(source="auto", target="zh-CN").translate(text)
        time.sleep(0.3)
        return translated or text
    except Exception as exc:
        print(f"翻译失败，保留原文: {exc}")
        return text


class FacebookScraper:
    """Facebook 经销商搜索采集器"""

    def __init__(
        self,
        ads_power_url: str = "http://127.0.0.1:50325",
        api_key: str = "",
        profile_id: str = "",
        max_results: int = 20,
        scroll_attempts: int = 5,
        headless: bool = False
    ):
        self.ads_power_url = ads_power_url
        self.api_key = api_key
        self.profile_id = profile_id
        self.max_results = max_results
        self.scroll_attempts = scroll_attempts
        self.headless = headless
        self.client: Optional[AdsPowerClient] = None
        self.page: Optional[Page] = None

    def start(self):
        """启动浏览器"""
        self.client = AdsPowerClient(self.ads_power_url, api_key=self.api_key)
        self.client.start_browser(self.profile_id, headless=self.headless)
        self.page = self.client.get_open_page() or self.client.new_page()
        self.page.set_viewport_size({"width": 1280, "height": 800})

    def stop(self):
        """断开 Playwright 连接，浏览器保持打开"""
        if self.client:
            self.client.close_browser()

    def search(self, query: str, region: str = "") -> List[Dict[str, Any]]:
        """
        在 Facebook 搜索经销商主页
        """
        search_query = f"{query} {region}".strip() if region else query
        encoded_query = urlencode({"q": search_query})
        search_url = f"https://www.facebook.com/search/pages/?{encoded_query}"

        print(f"访问搜索页: {search_url}")
        self.page.goto(search_url, wait_until="domcontentloaded", timeout=60000)

        # 等待页面加载
        time.sleep(random.uniform(3, 5))

        # 处理可能的登录弹窗或提示
        self._dismiss_dialogs()

        # 滚动加载结果
        page_links = self._collect_page_links()

        print(f"收集到 {len(page_links)} 个主页链接")

        # 采集每个主页详情
        records = []
        for idx, link in enumerate(page_links[:self.max_results], 1):
            print(f"[{idx}/{min(len(page_links), self.max_results)}] 采集: {link}")
            try:
                record = self._scrape_page(link)
                if record.get("客户姓名/公司"):
                    record["主页/链接"] = link
                    record["国家"] = "摩洛哥"  # 默认，后续可按 region 推断
                    record["城市"] = region
                    record = enrich_record_with_brands(record)
                    records.append(record)
            except Exception as e:
                print(f"采集失败 {link}: {e}")
            # 随机间隔，降低风控
            time.sleep(random.uniform(2, 4))

        return deduplicate_records(records)

    def _dismiss_dialogs(self):
        """尝试关闭可能的弹窗"""
        try:
            # 常见的 cookie/登录提示关闭按钮
            selectors = [
                '[aria-label="关闭"]',
                '[aria-label="Close"]',
                '[role="button"][tabindex="0"]',
            ]
            for sel in selectors:
                buttons = self.page.query_selector_all(sel)
                for btn in buttons:
                    text = btn.inner_text() if btn else ""
                    if any(k in text.lower() for k in ["close", "关闭", "not now", "以后再说", "ok"]):
                        btn.click()
                        time.sleep(0.5)
        except Exception:
            pass

    def _collect_page_links(self) -> List[str]:
        """从搜索结果页收集主页链接"""
        links = set()

        for attempt in range(self.scroll_attempts):
            # 获取当前页所有可能的页面链接
            elements = self.page.query_selector_all('a[href*="/"]')
            for el in elements:
                href = el.get_attribute("href")
                if href and self._is_page_link(href):
                    links.add(self._normalize_facebook_url(href))

            print(f"滚动 {attempt + 1}/{self.scroll_attempts}, 当前收集 {len(links)} 个链接")

            if len(links) >= self.max_results:
                break

            # 滚动到底部
            self.page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            time.sleep(random.uniform(2, 4))

        return list(links)

    def _is_page_link(self, href: str) -> bool:
        """判断是否是 Facebook 主页链接"""
        parsed = urlparse(href)
        if parsed.netloc not in ["www.facebook.com", "facebook.com", "m.facebook.com"]:
            return False
        # 排除搜索、个人资料、帖子等
        excluded_patterns = [
            "/search/", "/watch", "/groups", "/marketplace",
            "/messages", "/notifications", "/settings",
            "/photo.php", "/videos", "/posts", "/reel",
            "/profile.php"
        ]
        for pattern in excluded_patterns:
            if pattern in parsed.path:
                return False
        # 主页链接通常是 /xxx/ 形式
        path = parsed.path.strip("/")
        if not path or "/" in path:
            return False
        return True

    def _normalize_facebook_url(self, url: str) -> str:
        """标准化 Facebook URL"""
        parsed = urlparse(url)
        path = parsed.path.strip("/").split("/")[0]
        return f"https://www.facebook.com/{path}/"

    def _scrape_page(self, page_url: str) -> Dict[str, Any]:
        """采集单个 Facebook 主页信息"""
        record = {
            "客户姓名/公司": "",
            "客户类型": "",
            "公司官网": "",
            "联系人": "",
            "职位": "",
            "电话/WhatsApp": "",
            "邮箱": "",
            "主营业务": "",
            "建联状态": "未联系",
            "下次跟进": "",
            "备注": "",
        }

        self.page.goto(page_url, wait_until="domcontentloaded", timeout=60000)
        time.sleep(random.uniform(3, 5))

        # 1. 先读取主页联络资料：官网、电话、邮箱、粉丝数和简介。About 只能补充，不覆盖主页已确认字段。
        home_page_text = ""
        try:
            home_page_text = self.page.inner_text("body")
        except Exception:
            pass
        website_from_home = extract_company_website_from_facebook(self.page)
        if website_from_home:
            record["公司官网"] = website_from_home
            print(f"从主页联络资料提取到公司官网: {website_from_home}")
        home_phone = self._extract_phone(home_page_text)
        if home_phone:
            record["电话/WhatsApp"] = home_phone
            print(f"从主页联络资料提取到电话: {home_phone}")
        home_email = self._extract_email(home_page_text)
        if home_email:
            record["邮箱"] = home_email
            print(f"从主页联络资料提取到邮箱: {home_email}")
        followers_from_home = self._extract_followers(home_page_text)

        # 2. 页面标题作为公司名称
        try:
            title = self.page.title()
            # Facebook 标题通常是 "Page Name | Facebook"
            record["客户姓名/公司"] = title.split("|")[0].strip() if "|" in title else title.replace("Facebook", "").strip()
        except Exception:
            pass

        # 3. 尝试多种方式获取页面名称
        name_selectors = [
            'h1',
            '[role="main"] h2',
            '[data-pagelet="ProfileActions"] h1',
            'a[aria-label]',
        ]
        for sel in name_selectors:
            try:
                el = self.page.query_selector(sel)
                if el:
                    text = el.inner_text().strip()
                    if text and len(text) < 100 and "Facebook" not in text:
                        record["客户姓名/公司"] = text
                        break
            except Exception:
                continue

        # 4. 主页联络资料读取完成后，再获取 About 信息
        about_url = f"{page_url.rstrip('/')}/about/"
        self.page.goto(about_url, wait_until="domcontentloaded", timeout=60000)
        time.sleep(random.uniform(2, 4))

        page_text = ""
        try:
            page_text = self.page.inner_text("body")
        except Exception:
            pass

        about_website = extract_company_website_from_facebook(self.page)
        if about_website:
            if not record.get("公司官网"):
                record["公司官网"] = about_website
            elif about_website not in record["公司官网"]:
                record["公司官网"] = record["公司官网"] + "；" + about_website
        company_website = record.get("公司官网", "").split("；")[0].strip()

        website_result = {"email": "", "phone": "", "text": "", "business_summary": "", "evidence_notes": []}
        if company_website:
            try:
                print(f"深搜公司官网: {company_website}")
                website_result = scrape_public_website(self.page.context, company_website, max_pages=5)
            except Exception as exc:
                website_result = {"email": "", "phone": "", "text": "", "business_summary": "", "evidence_notes": [f"官网深搜失败：{str(exc)[:80]}"]}

        # 4. 提取电话
        phone = self._extract_phone(page_text)
        if phone and not record.get("电话/WhatsApp"):
            record["电话/WhatsApp"] = phone

        # 5. 提取邮箱，About 只补充空字段
        email = self._extract_email(page_text)
        if email and not record.get("邮箱"):
            record["邮箱"] = email

        if not record.get("邮箱") and website_result.get("email"):
            record["邮箱"] = website_result.get("email", "")
        if not record.get("电话/WhatsApp") and website_result.get("phone"):
            record["电话/WhatsApp"] = website_result.get("phone", "")

        # 6. 提取主营业务/描述并翻译为中文
        description = self._extract_description(page_text)
        if description:
            record["主营业务"] = translate_to_chinese(description)

        # 7. 综合 Facebook 与官网证据推断客户类型和主营业务
        combined_business_text = "\n".join([home_page_text, page_text, website_result.get("text", "")])
        record["客户类型"] = classify_customer_type_cn(combined_business_text)
        record["主营业务"] = summarize_business_cn(combined_business_text)

        # 8. 备注：中文结构化摘要
        followers = followers_from_home or self._extract_followers(page_text)
        record["备注"] = build_chinese_notes(
            facebook_url=page_url,
            facebook_text="\n".join(["证据采集顺序：主页联络资料 -> About -> 官网深搜", home_page_text, page_text]),
            website_url=record.get("公司官网", ""),
            website_result=website_result,
            followers=followers,
            post_analysis="",
            detected_brands=[],
            exclusivity="",
        )

        if record.get("备注"):
            record["备注"] = "证据采集顺序：主页联络资料 -> About -> 官网深搜 | " + record["备注"]
        return record

    def _extract_phone(self, text: str) -> str:
        """从文本中提取电话号码"""
        if not text:
            return ""
        patterns = [
            r'\+212[\s\-]?\d[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}[\s\-]?\d{2}',
            r'\+212[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}[\s\-]?\d{2}',
            r'0\d[\s\-]?\d{4}[\s\-]?\d{4}',
            r'0\d{3}[\s\-]?\d{2}[\s\-]?\d{2}[\s\-]?\d{2}',
        ]
        for pattern in patterns:
            match = re.search(pattern, text)
            if match:
                return match.group(0).strip()
        return ""

    def _extract_email(self, text: str) -> str:
        """从文本中提取邮箱"""
        if not text:
            return ""
        pattern = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
        matches = re.findall(pattern, text)
        # 过滤掉常见非企业邮箱
        excluded = ["@facebook.com", "@fb.com", "@example.com"]
        for m in matches:
            if all(e not in m.lower() for e in excluded):
                return m.strip()
        return ""

    def _extract_description(self, text: str) -> str:
        """提取页面描述/主营业务"""
        if not text:
            return ""
        lines = [line.strip() for line in text.split("\n") if line.strip()]
        # 找包含汽车相关关键词的短句
        keywords = ["car", "auto", "voiture", "vehicle", "motor", "dealer", "occasion", "automotive"]
        for line in lines[:50]:
            if any(k in line.lower() for k in keywords) and len(line) < 200:
                return line
        return ""

    def _extract_followers(self, text: str) -> str:
        """提取粉丝数"""
        if not text:
            return ""
        patterns = [
            r'(\d+[\d\s,.]*[KkMm]?)\s*(followers|fans|abonnés|likes)',
            r'(\d+[\d\s,.]*)\s*(followers|fans|abonnés|likes)',
        ]
        for pattern in patterns:
            match = re.search(pattern, text, re.IGNORECASE)
            if match:
                return f"{match.group(1).strip()} {match.group(2)}"
        return ""

    def _infer_dealer_type(self, text: str) -> str:
        """根据页面文本推断客户类型"""
        if not text:
            return "待判断"
        text_lower = text.lower()
        if any(k in text_lower for k in ["used car", "occasion", "二手车", "occaz"]):
            return "二手车商"
        if any(k in text_lower for k in ["dealer", "concessionnaire", "distributor", "经销商"]):
            return "经销商"
        if any(k in text_lower for k in ["rental", "location", "租车"]):
            return "租车公司"
        if any(k in text_lower for k in ["repair", "garage", "维修"]):
            return "汽车维修"
        return "待判断"


def main():
    parser = argparse.ArgumentParser(description="Facebook 经销商搜索采集")
    parser.add_argument("--query", required=True, help="搜索关键词")
    parser.add_argument("--region", default="", help="地区过滤")
    parser.add_argument("--profile-id", required=True, help="AdsPower profile ID")
    parser.add_argument("--ads-power-url", default="http://127.0.0.1:50325", help="AdsPower API URL")
    parser.add_argument("--api-key", default="", help="AdsPower API Key（如已开启认证）")
    parser.add_argument("--max-results", type=int, default=10, help="最大采集数")
    parser.add_argument("--run-id", default="", help="Run ID used for runs/YYYYMMDD/<run_id>/ artifacts.")
    parser.add_argument("--output", default="facebook_scraped.json", help="输出 JSON 文件")
    parser.add_argument("--excel", default="", help="建联表路径；不传时按项目优先级自动查找")
    parser.add_argument("--sheet", default="Facebook", help="回写 Sheet 名")
    parser.add_argument("--write-excel", action="store_true", help="确认写入建联表；否则只输出 JSON 预览")
    parser.add_argument("--headless", action="store_true", help="无头模式")

    args = parser.parse_args()

    scraper = FacebookScraper(
        ads_power_url=args.ads_power_url,
        api_key=args.api_key,
        profile_id=args.profile_id,
        max_results=args.max_results,
        headless=args.headless
    )

    records = []
    try:
        scraper.start()
        records = scraper.search(args.query, args.region)
    finally:
        scraper.stop()

    # 保存 JSON
    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)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(records, f, ensure_ascii=False, indent=2)
    print(f"已保存 {len(records)} 条记录到 {output_path}")

    # 可选回写 Excel
    if args.write_excel:
        workbook_info = resolve_workbook_path(args.excel, create_from_template=True)
        args.excel = str(workbook_info["path"])
        print(f"Workbook for writing: {args.excel} ({workbook_info['source']})")
        result = append_records(
            excel_path=args.excel,
            sheet_name=args.sheet,
            records=records,
            dedup_keys=["客户姓名/公司", "城市", "主页/链接"]
        )
        print(f"Excel 回写结果: {result}")
    else:
        print("默认预览模式：未写入 Excel。确认要入表时再使用 --write-excel。")


if __name__ == "__main__":
    main()