wuling-overseas-dealer-expansion-skill/scripts/social/collect_facebook_conversations.py

#!/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())