#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Collect matched Facebook Messenger conversations through AdsPower + Playwright. This collector is read-only: it never sends messages, writes the workbook, or closes the user's AdsPower browser. Translation and intent analysis are performed by the agent in a separate JSON artifact before workbook write-back. """ from __future__ import annotations import argparse import hashlib import json import random import re import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple from urllib.parse import parse_qs, urlparse from openpyxl import load_workbook SCRIPT_DIR = Path(__file__).resolve().parent SKILL_ROOT = SCRIPT_DIR.parents[1] COMMON_DIR = SKILL_ROOT / "scripts" / "common" SCRAPER_DIR = SKILL_ROOT / "scripts" / "scraper" for directory in (COMMON_DIR, SCRAPER_DIR): if str(directory) not in sys.path: sys.path.insert(0, str(directory)) from ads_power_client import AdsPowerClient # type: ignore # noqa: E402 from artifact_manager import new_run_id, resolve_artifact_path # type: ignore # noqa: E402 from workbook_resolver import resolve_workbook_path # type: ignore # noqa: E402 FACEBOOK_SHEET = "Facebook" CONVERSATION_SHEET = "Facebook对话记录" DEFAULT_MAX_MESSAGES = 2000 MAJOR_WAIT = (90.0, 200.0) MINOR_WAIT = (30.0, 90.0) TECHNICAL_WAIT = (0.5, 8.0) HEADER_ALIASES = { "index": ["序号", "编号", "ID"], "company": ["客户姓名/公司", "公司名称", "公司姓名", "客户名称"], "facebook_link": ["主页/链接", "Facebook主页链接", "Facebook链接", "facebook链接"], "attribute": ["客户属性"], "customer_type": ["客户类型"], "business": ["主营业务", "公司主营业务"], "status": ["建联状态", "建联情况"], "note": ["备注", "说明"], } RISK_MARKERS = [ "temporarily blocked", "verification required", "confirm your identity", "suspicious activity", "try again later", "暂时封锁", "验证", "确认身份", "异常活动", ] OUTGOING_MARKERS = [ "you sent", "sent by you", "您发送", "你发送", "你已发送", "vous avez envoyé", "envoyé par vous", "لقد أرسلت", ] SYSTEM_MARKERS = [ "messages and calls are secured", "end-to-end encrypted", "created this group", "joined the conversation", "changed the theme", "missed a call", "通话和消息", "端到端加密", "加入了对话", "更改了主题", ] AUTO_REPLY_MARKERS = [ "automated response", "automatic reply", "auto-reply", "away message", "réponse automatique", "message automatique", "رسالة تلقائية", "自动回复", ] READ_ONLY_MARKERS = ["seen", "已读", "vu", "تمت المشاهدة"] def clean(value: Any) -> str: if value is None: return "" return re.sub(r"\s+", " ", str(value).strip()) def now_iso() -> str: return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") def normalize_name(value: str) -> str: value = clean(value).casefold() value = re.sub(r"https?://|www\.", " ", value) value = re.sub(r"[^0-9a-z\u00c0-\u024f\u0600-\u06ff\u4e00-\u9fff]+", " ", value) return " ".join(part for part in value.split() if len(part) > 1) def page_identity(url: str) -> Tuple[str, str]: parsed = urlparse(clean(url)) host = parsed.netloc.casefold().removeprefix("www.") if "facebook.com" not in host: return "", "" query = parse_qs(parsed.query) if parsed.path.rstrip("/").casefold() == "/profile.php" and query.get("id"): value = clean(query["id"][0]) return value, value parts = [part for part in parsed.path.split("/") if part] if not parts: return "", "" reserved = {"pages", "groups", "marketplace", "watch", "messages", "home.php"} if parts[0].casefold() in reserved: if parts[0].casefold() == "pages" and parts[-1].isdigit(): return parts[-1], parts[-1] return "", "" slug = parts[0] return slug, slug def thread_url_for(page_url: str) -> Tuple[str, str]: slug, page_id = page_identity(page_url) identity = page_id or slug if not identity: return "", "" return f"https://www.facebook.com/messages/t/{identity}", identity def header_map(ws) -> Dict[str, int]: raw = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)} mapped: Dict[str, int] = {} for key, aliases in HEADER_ALIASES.items(): for alias in aliases: if alias in raw: mapped[key] = raw[alias] break return mapped def cell_value(ws, row: int, columns: Dict[str, int], key: str) -> str: column = columns.get(key) return clean(ws.cell(row=row, column=column).value) if column else "" def read_customers(workbook_path: Path, sheet_name: str) -> List[Dict[str, str]]: wb = load_workbook(workbook_path, data_only=True, read_only=True) if sheet_name not in wb.sheetnames: raise KeyError(f"Sheet not found: {sheet_name}") ws = wb[sheet_name] columns = header_map(ws) missing = [key for key in ("index", "company", "facebook_link") if key not in columns] if missing: raise RuntimeError(f"Facebook sheet is missing required columns: {', '.join(missing)}") rows: List[Dict[str, str]] = [] for row in range(2, ws.max_row + 1): link = cell_value(ws, row, columns, "facebook_link") name = cell_value(ws, row, columns, "company") thread_url, identity = thread_url_for(link) if not name or not thread_url: continue rows.append( { "excel_row": str(row), "index": cell_value(ws, row, columns, "index"), "company": name, "facebook_link": link, "thread_url": thread_url, "expected_identity": identity, "attribute": cell_value(ws, row, columns, "attribute"), "customer_type": cell_value(ws, row, columns, "customer_type"), "business": cell_value(ws, row, columns, "business"), "status": cell_value(ws, row, columns, "status"), "note": cell_value(ws, row, columns, "note"), } ) wb.close() return rows def read_existing_record_ids(workbook_path: Path, sheet_name: str = CONVERSATION_SHEET) -> Set[str]: wb = load_workbook(workbook_path, data_only=True, read_only=True) if sheet_name not in wb.sheetnames: wb.close() return set() ws = wb[sheet_name] headers = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1)} record_col = headers.get("记录ID") if not record_col: wb.close() return set() result = { clean(ws.cell(row=row, column=record_col).value) for row in range(2, ws.max_row + 1) if clean(ws.cell(row=row, column=record_col).value) } wb.close() return result def paced_wait(label: str, bounds: Tuple[float, float], enabled: bool = True) -> float: if not enabled: return 0.0 seconds = random.uniform(*bounds) print(f" -> {label}: {seconds:.1f}s", flush=True) time.sleep(seconds) return seconds def risk_text(page) -> str: try: text = clean(page.locator("body").inner_text(timeout=5000)).casefold() except Exception: return "" return next((marker for marker in RISK_MARKERS if marker.casefold() in text), "") def visible_thread_title(page) -> str: selectors = [ "[role='main'] h1", "[role='main'] h2", "header h1", "header h2", "a[role='link'][href*='/messages/t/']", ] for selector in selectors: try: locator = page.locator(selector) for idx in range(min(locator.count(), 8)): text = clean(locator.nth(idx).inner_text(timeout=1000)) if text and len(text) <= 160: return text except Exception: continue return "" def name_match(expected: str, observed: str) -> bool: expected_norm = normalize_name(expected) observed_norm = normalize_name(observed) if not expected_norm or not observed_norm: return False if expected_norm in observed_norm or observed_norm in expected_norm: return True expected_tokens = set(expected_norm.split()) observed_tokens = set(observed_norm.split()) overlap = expected_tokens & observed_tokens return bool(overlap) and len(overlap) / max(1, min(len(expected_tokens), len(observed_tokens))) >= 0.6 def validate_thread(page, customer: Dict[str, str]) -> Dict[str, Any]: current_url = clean(page.url) expected_identity = customer["expected_identity"].casefold() parsed = urlparse(current_url) path_parts = [part.casefold() for part in parsed.path.split("/") if part] url_match = "messages" in path_parts and "t" in path_parts and expected_identity in path_parts title = visible_thread_title(page) title_match = name_match(customer["company"], title) matched = bool(url_match and title_match) return { "matched": matched, "url_match": url_match, "title_match": title_match, "observed_title": title, "current_url": current_url, "match_basis": "thread_url+title" if matched else "", } EXTRACT_SCRIPT = r""" () => { const main = document.querySelector('[role="main"]') || document.body; const selectors = [ '[role="row"]', '[data-testid*="message" i]', '[data-scope*="message" i]', 'div[aria-label*="message" i]', 'div[aria-label*="sent" i]' ]; const nodes = []; const seen = new Set(); for (const selector of selectors) { for (const node of main.querySelectorAll(selector)) { if (!(node instanceof HTMLElement)) continue; const rect = node.getBoundingClientRect(); if (rect.width < 24 || rect.height < 12) continue; const text = (node.innerText || '').trim(); const aria = (node.getAttribute('aria-label') || '').trim(); if (!text && !aria && !node.querySelector('img,video,audio,[role="img"]')) continue; const key = [text, aria, Math.round(rect.top), Math.round(rect.left)].join('|'); if (seen.has(key)) continue; seen.add(key); const timeNode = node.querySelector('time[datetime], abbr[data-tooltip-content], [data-tooltip-content]'); const datetime = timeNode ? (timeNode.getAttribute('datetime') || timeNode.getAttribute('data-tooltip-content') || timeNode.textContent || '').trim() : ''; const idNode = node.closest('[data-message-id], [data-testid], [id]') || node; const rawId = idNode.getAttribute('data-message-id') || idNode.getAttribute('id') || ''; const senderNode = node.querySelector('h3,h4,strong,a[role="link"]'); const sender = senderNode ? (senderNode.textContent || '').trim() : ''; const hrefs = Array.from(node.querySelectorAll('a[href]')).map(a => a.href).filter(Boolean).slice(0, 8); nodes.push({ text, aria, datetime, raw_id: rawId, sender, hrefs, left: rect.left, width: rect.width, viewport_width: window.innerWidth, image_count: node.querySelectorAll('img,[role="img"]').length, video_count: node.querySelectorAll('video').length, audio_count: node.querySelectorAll('audio').length, file_count: node.querySelectorAll('a[download],a[href*="/file/"],a[href*="attachment"]').length }); } } return nodes; } """ def message_type(item: Dict[str, Any]) -> str: if item.get("audio_count"): return "语音" if item.get("video_count"): return "视频" if item.get("file_count"): return "文件" if item.get("image_count") and not clean(item.get("text")): return "图片" return "文本" def classify_direction(item: Dict[str, Any]) -> str: combined = " ".join([clean(item.get("aria")), clean(item.get("text"))]).casefold() if any(marker.casefold() in combined for marker in OUTGOING_MARKERS): return "我方发送" viewport = float(item.get("viewport_width") or 0) left = float(item.get("left") or 0) width = float(item.get("width") or 0) if viewport and left + width / 2 >= viewport * 0.58: return "我方发送" return "客户回复" def classify_raw_kind(item: Dict[str, Any]) -> Tuple[str, List[str]]: combined = " ".join([clean(item.get("aria")), clean(item.get("text"))]).casefold() risks: List[str] = [] if any(marker.casefold() in combined for marker in SYSTEM_MARKERS): return "系统消息", risks if any(marker.casefold() in combined for marker in AUTO_REPLY_MARKERS): risks.append("auto_reply") return "自动回复", risks if any(marker.casefold() == combined or marker.casefold() in combined for marker in READ_ONLY_MARKERS): return "已读提示", risks return message_type(item), risks def stable_record_id(thread_id: str, item: Dict[str, Any], direction: str, ordinal: int) -> str: raw_id = clean(item.get("raw_id")) if raw_id and len(raw_id) >= 6: return f"fb:{raw_id}" basis = "|".join( [ thread_id, clean(item.get("datetime")), direction, clean(item.get("sender")), clean(item.get("text")), ";".join(item.get("hrefs") or []), str(ordinal), ] ) return "fbh:" + hashlib.sha256(basis.encode("utf-8")).hexdigest()[:32] def normalize_messages(raw_items: Iterable[Dict[str, Any]], thread_id: str) -> List[Dict[str, Any]]: messages: List[Dict[str, Any]] = [] seen: Set[str] = set() for ordinal, item in enumerate(raw_items): text = clean(item.get("text")) aria = clean(item.get("aria")) if not text and not aria and not any(item.get(key) for key in ("image_count", "video_count", "audio_count", "file_count")): continue direction = classify_direction(item) kind, risks = classify_raw_kind(item) record_id = stable_record_id(thread_id, item, direction, ordinal) if record_id in seen: continue seen.add(record_id) messages.append( { "record_id": record_id, "message_time_raw": clean(item.get("datetime")), "direction": direction, "sender": clean(item.get("sender")), "original_text": text or aria, "message_type": kind, "attachment_links": item.get("hrefs") or [], "raw_aria": aria, "risk_flags": risks, } ) return messages def find_scroll_container(page) -> bool: return bool( page.evaluate( r""" () => { const main = document.querySelector('[role="main"]') || document.body; const nodes = [main, ...main.querySelectorAll('*')].filter(el => { const style = getComputedStyle(el); return el.scrollHeight - el.clientHeight > 240 && ['auto','scroll'].includes(style.overflowY); }); nodes.sort((a, b) => (b.clientWidth * b.clientHeight) - (a.clientWidth * a.clientHeight)); const target = nodes[0]; if (!target) return false; target.dataset.wulingConversationScroller = '1'; target.scrollTop = 0; return true; } """ ) ) def scroll_history_to_start(page, max_messages: int, pacing: bool) -> Dict[str, Any]: stable_rounds = 0 previous_signature = "" estimated_count = 0 truncated = False if not find_scroll_container(page): return {"scroll_rounds": 0, "history_start_reached": False, "history_truncated": False} rounds = 0 while stable_rounds < 3: rounds += 1 raw = page.evaluate(EXTRACT_SCRIPT) estimated_count = len(raw) first = raw[0] if raw else {} signature = clean(first.get("raw_id")) or clean(first.get("datetime")) or clean(first.get("text"))[:120] stable_rounds = stable_rounds + 1 if signature == previous_signature else 0 previous_signature = signature if estimated_count >= max_messages: truncated = True break page.evaluate( r""" () => { const target = document.querySelector('[data-wuling-conversation-scroller="1"]'); if (target) target.scrollTop = 0; } """ ) paced_wait("minor history expansion wait", MINOR_WAIT, pacing) if rounds >= 200: truncated = True break return { "scroll_rounds": rounds, "history_start_reached": stable_rounds >= 3, "history_truncated": truncated, "estimated_loaded_nodes": estimated_count, } def collect_customer(page, customer: Dict[str, str], mode: str, max_messages: int, pacing: bool) -> Dict[str, Any]: page.goto(customer["thread_url"], wait_until="domcontentloaded", timeout=90000) paced_wait("technical thread readiness", TECHNICAL_WAIT, pacing) marker = risk_text(page) if marker: return {"customer": customer, "status": "risk_stop", "risk_flags": [f"facebook_risk:{marker}"], "messages": []} paced_wait("major new-thread wait", MAJOR_WAIT, pacing) match = validate_thread(page, customer) if not match["matched"]: return { "customer": customer, "status": "thread_match_failed", "thread_match": match, "risk_flags": ["thread_match_failed"], "messages": [], } history = {"scroll_rounds": 0, "history_start_reached": False, "history_truncated": False} if mode == "initial_full": history = scroll_history_to_start(page, max_messages, pacing) raw_items = page.evaluate(EXTRACT_SCRIPT) messages = normalize_messages(raw_items, customer["expected_identity"])[:max_messages] if len(messages) >= max_messages: history["history_truncated"] = True return { "customer": customer, "status": "collected", "thread_id": customer["expected_identity"], "thread_match": match, "history": history, "risk_flags": ["history_truncated"] if history.get("history_truncated") else [], "messages": messages, } def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Collect matched Facebook Messenger conversations via AdsPower + Playwright.") mode = parser.add_mutually_exclusive_group() mode.add_argument("--initial-full", action="store_true", help="Scroll each matched thread to the available history start.") mode.add_argument("--incremental", action="store_true", help="Collect only records not already stored in Facebook对话记录 (default).") parser.add_argument("--profile-id", required=True, help="AdsPower profile ID.") parser.add_argument("--adspower-url", default="http://127.0.0.1:50325") parser.add_argument("--api-key", default="") parser.add_argument("--excel", default="") parser.add_argument("--facebook-sheet", default=FACEBOOK_SHEET) parser.add_argument("--conversation-sheet", default=CONVERSATION_SHEET) parser.add_argument("--max-customers", type=int, default=0, help="0 means all matching Facebook rows.") parser.add_argument("--max-messages-per-thread", type=int, default=DEFAULT_MAX_MESSAGES) parser.add_argument("--output", default="") parser.add_argument("--run-id", default="") parser.add_argument("--no-pacing", action="store_true", help=argparse.SUPPRESS) return parser.parse_args(argv) def main(argv: Optional[Sequence[str]] = None) -> int: args = parse_args(argv) run_id = args.run_id or new_run_id("facebook_conversation_collect") resolved = resolve_workbook_path(args.excel, create_from_template=False) workbook = resolved.get("path") if not workbook: raise FileNotFoundError("No outreach workbook found. Pass --excel.") workbook_path = Path(workbook) customers = read_customers(workbook_path, args.facebook_sheet) if args.max_customers > 0: customers = customers[: args.max_customers] existing_ids = read_existing_record_ids(workbook_path, args.conversation_sheet) mode = "initial_full" if args.initial_full else "incremental" pacing = not args.no_pacing output_path = resolve_artifact_path( args.output, kind="facebook_conversations", default_name="facebook_conversations_raw.json", run_id=run_id, ) client = AdsPowerClient(args.adspower_url, args.api_key) threads: List[Dict[str, Any]] = [] stopped_for_risk = False try: browser = client.start_browser(args.profile_id) if not browser.contexts: raise RuntimeError("AdsPower browser has no Playwright context.") context = browser.contexts[0] page = context.pages[0] if context.pages else context.new_page() for index, customer in enumerate(customers): if index: paced_wait("major customer-switch wait", MAJOR_WAIT, pacing) print(f"[{index + 1}/{len(customers)}] {customer['company']}", flush=True) result = collect_customer(page, customer, mode, args.max_messages_per_thread, pacing) if mode == "incremental": result["messages"] = [ message for message in result.get("messages", []) if message.get("record_id") not in existing_ids ] threads.append(result) if result.get("status") == "risk_stop": stopped_for_risk = True break finally: client.detach() payload = { "schema_version": "4.26", "run_id": run_id, "created_at": now_iso(), "mode": mode, "profile_id": args.profile_id, "workbook": str(workbook_path), "facebook_sheet": args.facebook_sheet, "conversation_sheet": args.conversation_sheet, "read_only": True, "messages_sent": 0, "browser_left_open": True, "stopped_for_risk": stopped_for_risk, "summary": { "customers_selected": len(customers), "threads_attempted": len(threads), "threads_matched": sum(1 for thread in threads if thread.get("status") == "collected"), "thread_match_failed": sum(1 for thread in threads if thread.get("status") == "thread_match_failed"), "new_messages": sum(len(thread.get("messages", [])) for thread in threads), "history_truncated": sum(1 for thread in threads if thread.get("history", {}).get("history_truncated")), }, "threads": threads, "agent_next_step": "Translate both directions to Chinese, analyze only effective customer replies, preview in chat, then run write_facebook_conversations.py with the validated analysis JSON.", } output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"output": str(output_path), **payload["summary"], "browser_left_open": True}, ensure_ascii=False, indent=2)) return 2 if stopped_for_risk else 0 if __name__ == "__main__": raise SystemExit(main())