""" Facebook 高活跃度摩洛哥经销商搜索与采集 - 使用多个本地化的搜索关键词 - 过滤个人资料/群组/非主页链接 - 先生成候选预览并评分 - 仅在显式 --deep-scrape 时打开高分主页深采 - 仅在显式 --write-excel 时写入 Excel """ import re import sys import time import random import argparse import json 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 List, Dict, Any, Set, Tuple, Optional from urllib.parse import urlencode, urlparse # Windows 控制台输出中文/阿拉伯文时避免 GBK 编码错误 if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") import pandas as pd from playwright.sync_api import Page try: from .ads_power_client import AdsPowerClient from .scrape_single_page import scrape_page_record from . import discovery_common as dc from ..common import 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.scrape_single_page import scrape_page_record from scraper import discovery_common as dc from common import append_records, resolve_workbook_path PAGE_NAME_BLACKLIST = { "通知", "个人资料", "帖子", "简介", "提及", "好友", "照片", "视频", "Notifications", "Profile", "Posts", "Photos", "Videos", "About", "Friends", "Home", "Page", "Facebook", "Unknown", } DEFAULT_CITY_SCOPE = "摩洛哥全国" MOROCCO_TARGET_CITIES = [ "Casablanca", "Rabat", "Marrakech", "Tanger", "Fes", "Agadir", "Meknes", "Oujda", "Kenitra", "Tetouan", "Nador", "Safi", ] NATIONWIDE_KEYWORDS = [ "concessionnaire voiture chinoise Maroc", "distributeur voitures chinoises Maroc", "importateur voitures chinoises Maroc", "showroom voiture chinoise Maroc", "groupe automobile marques chinoises Maroc", "concessionnaire multimarque voitures chinoises Maroc", "concessionnaire utilitaire chinois Maroc", "importateur camion chinois Maroc", "distributeur camion chinois Maroc", "concessionnaire camion Maroc", "concessionnaire Chery Maroc", "concessionnaire DFSK Maroc", "concessionnaire Foton Maroc", "concessionnaire JAC Maroc", "concessionnaire Jetour Maroc", "concessionnaire Geely Maroc", "concessionnaire SITRAK Maroc", "distributeur Sinotruk Maroc", "distributeur Shacman Maroc", "camion chinois Maroc", "import voiture chine Maroc", "voiture chinoise Maroc", "importateur auto chine Maroc", "groupe automobile Maroc", "concessionnaire automobile Maroc", "concessionnaire multimarque Maroc", "importateur voiture occasion Maroc", "voiture occasion Maroc", "showroom auto Maroc", ] CITY_KEYWORD_PATTERNS = [ "concessionnaire automobile {city}", "voiture occasion {city}", "showroom auto {city}", ] def build_default_keywords() -> List[str]: """Build nationwide Morocco queries plus city waves for large and small dealers.""" keywords = list(NATIONWIDE_KEYWORDS) for city in 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() PATH_BLACKLIST = { "photo", "photos", "video", "videos", "watch", "home", "messages", "notifications", "settings", "events", "marketplace", "friends", "friend", "login", "logout", "recover", "help", "support", "privacy", "terms", "ads", "business", "creatorstudio", "gaming", "jobs", "weather", "places", "explore", "bookmarks", "memories", "saved", "reels", "stories", "shop", "donate", "fundraisers", "pages", "groups", "profile.php", } BUSINESS_SUFFIXES = { "ma", "com", "net", "org", "fr", "uk", "eu", "info", "shop", "store", "auto", "cars", "voiture", "maroc", "morocco", "casa", "group", } CHINA_BRAND_TERMS = { "baic", "byd", "changan", "chery", "dfsk", "dongfeng", "foton", "gac", "geely", "great wall", "haval", "jac", "jetour", "maxus", "mg", "omoda", "saic", "wuling", "sitrak", "sinotruk", "shacman", "faw", "yutong", "king long", "golden dragon", "higer", "forland", } NON_CHINA_BRAND_TERMS = { "audi", "bmw", "chevrolet", "citroen", "dacia", "daf", "fiat", "ford", "hino", "honda", "hyundai", "isuzu", "iveco", "jeep", "kia", "man", "mazda", "mercedes", "mitsubishi", "nissan", "opel", "peugeot", "renault", "scania", "seat", "skoda", "suzuki", "toyota", "volkswagen", "volvo", } OEM_BRANCH_BRAND_TERMS = CHINA_BRAND_TERMS | NON_CHINA_BRAND_TERMS OEM_BRANCH_COUNTRY_SUFFIXES = {"maroc", "morocco", "ma"} OEM_BRANCH_FILLER_TERMS = { "auto", "autos", "car", "cars", "motor", "motors", "truck", "trucks", "vehicle", "vehicles", "vehicule", "vehicules", "official", "officiel", } COMMERCIAL_VEHICLE_TERMS = { "truck", "trucks", "camion", "camions", "utilitaire", "utilitaires", "heavy", "light truck", "pickup", "van", "mpv", "bus", "fleet", "flotte", } HIGH_INTENT_TERMS = { "import", "importateur", "importation", "chine", "chinoise", "chinese", "concessionnaire", "dealer", "distributeur", "distribution", "multimarque", "groupe", "group", "showroom", "occasion", "voiture occasion", "automobile", "auto", "revendeur", "vendeur", "vente", "stock", "parc auto", } AUTOMOTIVE_URL_TERMS = { "auto", "cars", "car", "moteur", "motor", "voiture", "vehicule", "vehicle", "maroc", "morocco", "casa", "casablanca", "rabat", "marrakech", "tanger", "fes", "agadir", "meknes", "oujda", "kenitra", "tetouan", "nador", "safi", "group", "groupe", "garage", "truck", "camion", "sitrak", "sinotruk", "foton", "chery", "dfsk", "byd", "haval", "jac", } NEW_VEHICLE_SALES_TERMS = { "v?hicules neufs", "vehicules neufs", "voitures neuves", "new vehicle", "new cars", "concessionnaire", "dealer", "showroom", "vente", "stock", "parc auto", } IMPORT_DISTRIBUTION_TERMS = { "import", "importateur", "importation", "distributeur", "distribution", "groupe", "group", "r?seau", "reseau", "dealer network", } MULTIBRAND_TERMS = { "multimarque", "multi-brand", "multi brand", "plusieurs marques", "marques multiples", } CONTACT_ENTRY_TERMS = { "facebook.com", "whatsapp", "wa.me", "tel", "phone", "telephone", "t?l?phone", "email", "mail", "contact", "linkedin", "+212", "@", } PURE_SERVICE_TERMS = { "garage r?paration", "garage reparation", "diagnostic", "diag", "pieces", "pi?ces", "spare", "parts", "pneus", "tires", "tyres", "lavage", "wash", "repair", "reparation", } RENTAL_FLEET_TERMS = {"location", "rental", "rent car", "rentcar", "fleet", "flotte", "leasing", "lld"} PERSONAL_SELLER_TERMS = {"particulier", "personal seller", "vendeur particulier", "profile.php"} DIRECT_BRANCH_TERMS = {"succursale", "branch", "filiale", "subsidiary", "officiel", "official", "page officielle", "official page"} LOW_VALUE_TERMS = PURE_SERVICE_TERMS | { "assurance", "insurance", "immobilier", "emploi", "marketing", "software", } def compact_text(value: str) -> str: """Normalize a short page/name hint for brand-branch detection.""" return re.sub(r"[^a-z0-9]+", "", value.lower()) def looks_like_oem_local_branch(slug: str) -> Tuple[bool, str]: """ Detect official brand-country pages such as BYD Maroc or BMW Morocco. These are usually OEM local branches or brand pages, not independent dealer channels. Independent entities with owner/channel names should pass through. """ compact = compact_text(slug) if not compact: return False, "" fillers = {compact_text(term) for term in OEM_BRANCH_FILLER_TERMS} countries = {compact_text(term) for term in OEM_BRANCH_COUNTRY_SUFFIXES} for brand in OEM_BRANCH_BRAND_TERMS: brand_key = compact_text(brand) if not brand_key or brand_key not in compact: continue for country in countries: direct_variants = { f"{brand_key}{country}", f"{country}{brand_key}", f"{brand_key}official{country}", f"{brand_key}officiel{country}", f"{brand_key}{country}official", f"{brand_key}{country}officiel", } if compact in direct_variants: return True, brand if compact.startswith(brand_key) and compact.endswith(country): middle = compact[len(brand_key):-len(country)] if country else "" if not middle or middle in fillers: return True, brand if compact.startswith(country) and compact.endswith(brand_key): middle = compact[len(country):-len(brand_key)] if not middle or middle in fillers: return True, brand return False, "" def normalize_link(value: Any) -> str: """Normalize page links for pre-scrape duplicate checks.""" if value is None: return "" link = str(value).strip() if not link or link.lower() == "nan": return "" link = link.split("#", 1)[0].split("?", 1)[0].rstrip("/") replacements = { "https://m.facebook.com/": "https://www.facebook.com/", "http://m.facebook.com/": "https://www.facebook.com/", "http://facebook.com/": "https://www.facebook.com/", "https://facebook.com/": "https://www.facebook.com/", } for src, dst in replacements.items(): link = link.replace(src, dst) return link.lower() def load_existing_links(excel_path: str, sheet_name: str) -> Set[str]: """Load existing homepage links from the target workbook before scraping.""" if not excel_path: return set() path = Path(excel_path) if not path.exists(): raise FileNotFoundError(f"Excel 文件不存在: {excel_path}") df = pd.read_excel(path, sheet_name=sheet_name) if "主页/链接" not in df.columns: return set() return { normalized for normalized in (normalize_link(value) for value in df["主页/链接"].tolist()) if normalized } def load_blocklist_json(blocklist_path: Optional[str]) -> Set[str]: """Load additional Facebook page URLs to skip from a JSON blocklist file.""" if not blocklist_path: return set() path = Path(blocklist_path) if not path.exists(): return set() try: data = json.loads(path.read_text(encoding="utf-8")) except Exception: return set() if isinstance(data, list): return {normalized for normalized in (normalize_link(value) for value in data) if normalized} if isinstance(data, dict) and "blocklist" in data: values = data["blocklist"] return {normalized for normalized in (normalize_link(value) for value in values) if normalized} return set() def is_page_link(href: str) -> bool: """判断是否是 Facebook 公开主页链接，排除个人资料、群组、搜索等""" parsed = urlparse(href) if parsed.netloc not in ["www.facebook.com", "facebook.com", "m.facebook.com"]: return False path = parsed.path.strip("/").lower() if not path or "/" in path: return False if path in PATH_BLACKLIST: return False if not re.match(r"^[A-Za-z0-9._\-]+$", path): return False if "." in path: parts = path.split(".") if len(parts) == 2 and len(parts[0]) >= 2 and len(parts[1]) >= 2: suffix = parts[1].rstrip("0123456789").lower() if suffix not in BUSINESS_SUFFIXES: return False return True def normalize_facebook_url(url: str) -> str: """标准化为 https://www.facebook.com/{pagename}/""" parsed = urlparse(url) path = parsed.path.strip("/").split("/")[0] return f"https://www.facebook.com/{path}/" def page_name_hint(url: str) -> str: """Return the Facebook slug as a weak candidate name hint.""" parsed = urlparse(url) return parsed.path.strip("/").split("/")[0] def matched_terms(text: str, terms: Set[str]) -> List[str]: text_lower = text.lower() return sorted(term for term in terms if term in text_lower) def score_candidate(url: str, source_queries: List[str]) -> Dict[str, Any]: """Score a candidate before opening its homepage.""" source_queries = sorted(set(q for q in source_queries if q)) query_text = " ".join(source_queries).lower() slug = page_name_hint(url) slug_text = re.sub(r"[._\-]+", " ", slug).lower() combined = f"{query_text} {slug_text} {url}".casefold() score = 0 reasons: List[str] = [] risks: List[str] = [] recommended_action = "preview_only" is_oem_branch, oem_brand = looks_like_oem_local_branch(slug) brand_terms = matched_terms(combined, OEM_BRANCH_BRAND_TERMS) direct_branch_terms = matched_terms(combined, DIRECT_BRANCH_TERMS) if is_oem_branch or (brand_terms and direct_branch_terms): score -= 10 risks.append(f"疑似汽车品牌当地官方分公司/官方主页/直营网点，非目标经销渠道: {oem_brand or ', '.join(brand_terms[:2])}") recommended_action = "skip_brand_branch" service_terms = matched_terms(combined, PURE_SERVICE_TERMS) if service_terms: score -= 6 risks.append("纯维修/配件/轮胎/服务类，不纳入汽车渠道合作伙伴: " + ", ".join(service_terms[:5])) if recommended_action == "preview_only": recommended_action = "skip_non_channel" personal_terms = matched_terms(combined, PERSONAL_SELLER_TERMS) if personal_terms: score -= 5 risks.append("疑似个人卖家，不纳入汽车渠道合作伙伴: " + ", ".join(personal_terms[:4])) if recommended_action == "preview_only": recommended_action = "skip_non_channel" sales_terms = matched_terms(combined, NEW_VEHICLE_SALES_TERMS) import_terms = matched_terms(combined, IMPORT_DISTRIBUTION_TERMS) multibrand_terms = matched_terms(combined, MULTIBRAND_TERMS) rental_terms = matched_terms(combined, RENTAL_FLEET_TERMS) contact_terms = matched_terms(combined, CONTACT_ENTRY_TERMS) china_terms = matched_terms(combined, CHINA_BRAND_TERMS) commercial_terms = matched_terms(combined, COMMERCIAL_VEHICLE_TERMS) high_intent_terms = matched_terms(query_text, HIGH_INTENT_TERMS) slug_terms = matched_terms(slug_text, AUTOMOTIVE_URL_TERMS) has_actual_channel = bool(sales_terms or import_terms or multibrand_terms) ownership_needs_review = brand_terms and recommended_action != "skip_brand_branch" and not (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 china_terms and recommended_action != "skip_brand_branch": score += 1 reasons.append("命中中国品牌/商用车品牌: " + ", ".join(china_terms[:5])) if commercial_terms: score += 1 reasons.append("命中商用车/车队相关词: " + ", ".join(commercial_terms[:5])) if high_intent_terms: score += min(3, len(high_intent_terms)) reasons.append("搜索词意图强: " + ", ".join(high_intent_terms[:5])) if slug_terms: score += 1 reasons.append("主页链接像汽车业务: " + ", ".join(slug_terms[:5])) if contact_terms or "facebook.com" in url.casefold(): score += 2 reasons.append("可建联入口: Facebook Page") if len(source_queries) > 1: extra = min(2, len(source_queries) - 1) score += extra reasons.append(f"被 {len(source_queries)} 个搜索词重复命中") if rental_terms and not has_actual_channel: score += 1 risks.append("纯租赁/车队线索，转入批量采购与运营客户/汽车租赁公司，不作为汽车渠道合作伙伴: " + ", ".join(rental_terms[:4])) if brand_terms and sales_terms and not (import_terms or multibrand_terms) and recommended_action != "skip_brand_branch": score += 1 risks.append("独立单品牌新车经销商可保留；排他协议及新增品牌权限待确认") new_vehicle_needs_review = False if not has_actual_channel and not rental_terms: score -= 3 risks.append("缺少新整车销售、进口、分销或 showroom 证据，建议先人工看预览") new_vehicle_needs_review = True if ownership_needs_review and dc.should_request_manual_review(score, recommended_action, reasons): dc.add_manual_review_flag(risks, "ownership") if new_vehicle_needs_review and dc.should_request_manual_review(score, recommended_action, reasons): dc.add_manual_review_flag(risks, "new_vehicle") if not reasons: reasons.append("仅从 Facebook 搜索结果获得，需人工判断") return {"url": url, "normalized_url": normalize_link(url), "candidate_name_hint": slug, "source_queries": source_queries, "score": score, "score_reasons": reasons, "risk_flags": risks, "recommended_action": recommended_action} def enrich_record_with_candidate_signals(record: Dict[str, Any], candidate: Dict[str, Any]) -> Dict[str, Any]: """Apply the local expanded brand terms even if the shared detector is not writable.""" combined = " ".join([ str(record.get("客户姓名/公司", "")), str(record.get("主页/链接", "")), str(record.get("主营业务", "")), str(record.get("备注", "")), str(record.get("candidate_reasons", "")), " ".join(candidate.get("score_reasons", [])), " ".join(candidate.get("source_queries", [])), ]) china_terms = matched_terms(combined, CHINA_BRAND_TERMS) if china_terms: existing = record.get("detected_brands", []) or [] if not isinstance(existing, list): existing = [str(existing)] record["detected_brands"] = sorted(set(existing + china_terms)) record["exclusivity_assessment"] = "已代理中国品牌，需评估" return record def collect_page_links( page: Page, query: str, max_links: int = 10, scroll_attempts: int = 3, ) -> List[str]: """访问 Facebook search/top 并收集主页链接""" encoded = urlencode({"q": query}) search_url = f"https://www.facebook.com/search/top/?{encoded}" print(f"\n搜索关键词: {query}", flush=True) print(f"访问: {search_url}", flush=True) try: page.goto(search_url, wait_until="domcontentloaded", timeout=30000) time.sleep(random.uniform(2, 4)) except Exception as e: print(f" 搜索页加载失败: {e}", flush=True) return [] try: for sel in ['[aria-label="关闭"]', '[aria-label="Close"]']: for btn in page.query_selector_all(sel): try: btn.click() time.sleep(0.5) except Exception: pass except Exception: pass links: Set[str] = set() start_time = time.time() for attempt in range(scroll_attempts): selectors = [ 'a[href*="facebook.com/"]', '[role="article"] a', 'a[href*="/"]', ] for sel in selectors: try: for el in page.query_selector_all(sel): try: href = el.get_attribute("href") or "" if is_page_link(href): links.add(normalize_facebook_url(href)) except Exception: continue except Exception as e: print(f" 选择器 {sel} 提取失败: {e}", flush=True) continue print(f" 滚动 {attempt + 1}/{scroll_attempts}, 已收集 {len(links)} 个主页链接", flush=True) if len(links) >= max_links: break if time.time() - start_time > 30: print(" 搜索时间超过 30 秒，提前结束", flush=True) break page.evaluate("window.scrollTo(0, document.body.scrollHeight)") time.sleep(random.uniform(1, 2)) return list(links)[:max_links] def looks_old_from_time_text(time_text: str) -> bool: """根据帖子时间文本判断是否明显陈旧。""" if not time_text: return False text = time_text.lower() if re.search(r"(201[0-9]|202[0-3])", text): return True if re.search(r"\d+\s*(year|years|yr|yrs|年)", text): return True month_match = re.search(r"(\d+)\s*(month|months|mo|月)", text) if month_match and int(month_match.group(1)) > 3: return True return False def looks_recently_active(time_text: str) -> bool: """判断时间文本是否暗示近期活跃（90 天内）。""" if not time_text: return False text = time_text.lower() recent_units = [ "h", "hr", "hrs", "hour", "hours", "小时", "min", "mins", "minute", "minutes", "分钟", "d", "day", "days", "天", "w", "week", "weeks", "周", "just now", "刚刚", "now", "现在", "昨天", "yesterday", ] if any(unit in text for unit in recent_units): return True month_match = re.search(r"\b(\d+)\s*(month|months|mo|月)\b", text) if month_match and int(month_match.group(1)) <= 3: return True if re.search(r"(2024|2025|2026)", text): return True return False def extract_latest_time_from_notes(notes: str) -> str: """从备注字段中解析最近发帖时间""" if not notes: return "" match = re.search(r"最近发帖时间：([^|]+)", notes) if match: return match.group(1).strip() return "" def build_candidates( all_page_links: List[str], link_sources: Dict[str, Set[str]], existing_links: Set[str], min_candidate_score: int, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """Create scored candidates and skipped-existing records.""" candidates: List[Dict[str, Any]] = [] skipped_existing: List[Dict[str, Any]] = [] for link in all_page_links: normalized = normalize_link(link) source_queries = sorted(link_sources.get(link, set())) candidate = score_candidate(link, source_queries) if normalized and normalized in existing_links: candidate["recommended_action"] = "skip_existing" skipped_existing.append(candidate) print(f" 已在表中，跳过不访问主页: {link}") continue if candidate.get("recommended_action") == "skip_brand_branch": print(f" 疑似品牌当地分公司/官方页，跳过深采: {link}") elif candidate["score"] >= min_candidate_score: candidate["recommended_action"] = "deep_scrape" candidates.append(candidate) candidates.sort(key=lambda c: (-int(c.get("score", 0)), c.get("url", ""))) skipped_existing.sort(key=lambda c: c.get("url", "")) return candidates, skipped_existing def search_active_dealers( profile_id: str, keywords: List[str] = None, city: str = DEFAULT_CITY_SCOPE, country: str = "摩洛哥", max_links_per_keyword: int = 5, max_pages_to_scrape: int = 20, min_active_score: int = 1, min_candidate_score: int = 3, ads_power_url: str = "http://127.0.0.1:50325", api_key: str = "", headless: bool = False, existing_links: Set[str] = None, deep_scrape: bool = False, ) -> Dict[str, Any]: """ 执行多关键词搜索。默认只生成候选预览；显式 deep_scrape=True 才打开高分主页。 min_active_score: 0=不过滤, 1=过滤明显陈旧的, 2=仅保留明确近期活跃的 """ if keywords is None: keywords = DEFAULT_KEYWORDS if existing_links is None: existing_links = set() client = AdsPowerClient(ads_power_url, api_key=api_key) client.start_browser(profile_id, headless=headless) page = client.get_open_page() or client.new_page() page.set_viewport_size({"width": 1280, "height": 800}) all_page_links: List[str] = [] link_sources: Dict[str, Set[str]] = {} records: List[Dict[str, Any]] = [] candidates: List[Dict[str, Any]] = [] skipped_existing: List[Dict[str, Any]] = [] try: for query in keywords: links = collect_page_links( page=page, query=query, max_links=max_links_per_keyword, scroll_attempts=5, ) print(f" 关键词 '{query}' 收集到 {len(links)} 个主页链接") for link in links: if link not in all_page_links: all_page_links.append(link) link_sources.setdefault(link, set()).add(query) time.sleep(random.uniform(2, 4)) print(f"\n去重后共 {len(all_page_links)} 个搜索到主页，开始候选评分") candidates, skipped_existing = build_candidates( all_page_links=all_page_links, link_sources=link_sources, existing_links=existing_links, min_candidate_score=min_candidate_score, ) if skipped_existing: print(f"已按 Excel 现有主页链接跳过 {len(skipped_existing)} 个重复主页") selected_candidates = [c for c in candidates if c.get("recommended_action") == "deep_scrape"] if len(selected_candidates) > max_pages_to_scrape: print(f"高分候选 {len(selected_candidates)} 个，限制深采前 {max_pages_to_scrape} 个") selected_candidates = selected_candidates[:max_pages_to_scrape] if not deep_scrape: print("预览模式：已生成候选评分，不打开主页深采。需要深采时加 --deep-scrape。") else: for idx, candidate in enumerate(selected_candidates, 1): link = candidate["url"] print(f"\n[{idx}/{len(selected_candidates)}] 深采主页: {link} (score={candidate['score']})") try: record = scrape_page_record( page=page, page_url=link, city=city, country=country, ) name = record.get("客户姓名/公司", "") dealer_type = record.get("客户类型", "") if name and name != "Unknown" and name not in PAGE_NAME_BLACKLIST and dealer_type != "个人用户": record["candidate_score"] = candidate.get("score", 0) record["candidate_reasons"] = "; ".join(candidate.get("score_reasons", [])) record["risk_flags"] = "; ".join(candidate.get("risk_flags", [])) record = enrich_record_with_candidate_signals(record, candidate) records.append(record) normalized = normalize_link(link) if normalized: existing_links.add(normalized) else: print(f" 跳过无效主页: {name} ({dealer_type})") except Exception as e: print(f"采集失败 {link}: {e}", flush=True) time.sleep(random.uniform(2, 4)) finally: print("\n断开 Playwright 连接，浏览器保持打开...", flush=True) client.close_browser() if records: print(f"\n深采完成，原始记录 {len(records)} 条") if min_active_score >= 1: filtered = [] for r in records: latest_time = extract_latest_time_from_notes(r.get("备注", "")) if not looks_old_from_time_text(latest_time): filtered.append(r) else: print(f" 过滤陈旧账号: {r.get('客户姓名/公司')} ({latest_time})") records = filtered print(f"过滤明显陈旧后剩余 {len(records)} 条") def sort_key(r): latest_time = extract_latest_time_from_notes(r.get("备注", "")) if looks_recently_active(latest_time): return 0 if not latest_time or latest_time == "未识别": return 1 if looks_old_from_time_text(latest_time): return 3 return 2 records.sort(key=sort_key) return { "summary": { "keywords": keywords, "found_links": len(all_page_links), "candidate_count": len(candidates), "skipped_existing": len(skipped_existing), "selected_for_deep_scrape": len([c for c in candidates if c.get("recommended_action") == "deep_scrape"][:max_pages_to_scrape]), "deep_scrape": deep_scrape, "scraped_records": len(records), "min_candidate_score": min_candidate_score, "max_pages_to_scrape": max_pages_to_scrape, }, "candidates": candidates, "skipped_existing": skipped_existing, "records": records, } def main(): parser = argparse.ArgumentParser(description="Facebook 摩洛哥经销商候选搜索、评分与可选深采") 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("--keywords", default="", help="逗号分隔的搜索关键词，留空使用默认列表") parser.add_argument("--city", default=DEFAULT_CITY_SCOPE, help="城市或覆盖范围，默认摩洛哥全国") parser.add_argument("--country", default="摩洛哥", help="国家") parser.add_argument("--max-links-per-keyword", type=int, default=5, help="每个关键词最多收集链接数；全国搜索默认较小以控制账号风险") parser.add_argument("--max-pages", type=int, default=20, help="--deep-scrape 时最多打开深采的新主页数") parser.add_argument("--min-candidate-score", type=int, default=3, help="候选进入深采队列的最低分") parser.add_argument("--min-active-score", type=int, default=1, choices=[0, 1, 2], help="活跃度过滤: 0=不过滤, 1=过滤明显陈旧, 2=仅保留明确近期活跃") parser.add_argument("--run-id", default="", help="Run ID used for runs/YYYYMMDD// artifacts.") parser.add_argument("--output", default="facebook_candidate_preview.json", help="候选预览/深采输出 JSON 文件") parser.add_argument("--excel", default="", help="用于预去重和可选回写的 Excel 路径；写表时必须显式提供") parser.add_argument("--sheet", default="Facebook", help="Sheet 名") parser.add_argument("--headless", action="store_true", help="无头模式") parser.add_argument("--deep-scrape", action="store_true", help="打开高分候选主页并深采 About + 最近帖子") parser.add_argument("--write-excel", action="store_true", help="确认后将深采记录写入 Excel；默认只输出 JSON 预览") parser.add_argument("--blocklist-json", default="", help="可选 JSON 文件，包含额外要跳过的 Facebook 主页链接") parser.add_argument("--no-excel", action="store_true", help="不读取也不写入 Excel，只输出 JSON") args = parser.parse_args() keywords = None if args.keywords: keywords = [k.strip() for k in args.keywords.split(",") if k.strip()] existing_links: Set[str] = set() 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']})") if not args.no_excel and args.excel: try: existing_links = load_existing_links(args.excel, args.sheet) print(f"已从 Excel 加载 {len(existing_links)} 个现有主页链接，采集前将直接跳过") except FileNotFoundError as e: print(f"未读取到现有 Excel，跳过预去重: {e}") blocklist_links = load_blocklist_json(args.blocklist_json or None) if blocklist_links: existing_links = existing_links | blocklist_links print(f"已从黑名单 JSON 加载 {len(blocklist_links)} 个额外链接，合并后去重库共 {len(existing_links)} 个") result = search_active_dealers( profile_id=args.profile_id, keywords=keywords, city=args.city, country=args.country, max_links_per_keyword=args.max_links_per_keyword, max_pages_to_scrape=args.max_pages, min_active_score=args.min_active_score, min_candidate_score=args.min_candidate_score, ads_power_url=args.ads_power_url, api_key=args.api_key, headless=args.headless, existing_links=existing_links, deep_scrape=args.deep_scrape, ) 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(result, f, ensure_ascii=False, indent=2) print(f"\n已保存候选/采集结果到 {output_path}") if args.write_excel and not args.no_excel: write_workbook_info = resolve_workbook_path(args.excel, create_from_template=True) args.excel = str(write_workbook_info["path"]) if write_workbook_info.get("created"): print(f"Created blank workbook from skill template: {args.excel}", flush=True) records = result.get("records", []) if not records: print("No Excel written: no deep-scraped records. Use --deep-scrape first.") else: try: write_result = append_records( excel_path=args.excel, sheet_name=args.sheet, records=records, dedup_keys=["客户姓名/公司", "城市", "主页/链接"], ) print(f"Excel write result: {write_result}", flush=True) except PermissionError as e: print(f"Excel write failed. Close the workbook and retry: {e}", flush=True) else: print("Preview mode: Excel was not written. Use --deep-scrape --write-excel after confirmation.") if __name__ == "__main__": main()