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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Semi-automated Facebook follow/message execution through AdsPower.

Default behavior is dry-run. Real execution requires --confirm plus
--batch-confirmed after the whole preview has been shown to the user. Messenger
sending is deliberately scoped to the Facebook Page header Message button and
the matching Messenger dialog only.
"""
from __future__ import annotations

import argparse
import json
import random
import re
import sys
import time
from datetime import datetime, timedelta
from math import ceil
from pathlib import Path

import sys
sys.path.append(str(Path(__file__).resolve().parents[1]))
from common.artifact_manager import resolve_artifact_path
from typing import Any, Dict, List, Optional, Sequence

from openpyxl import load_workbook

SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parents[1]
sys.path.insert(0, str(SKILL_ROOT))
from scripts.scraper.ads_power_client import AdsPowerClient  # noqa: E402

DEFAULT_ADS_POWER_URL = "http://127.0.0.1:50325"
MAX_PER_RUN = 3
DEFAULT_DAILY_FOLLOW_LIMIT = 20
DEFAULT_DAILY_DM_LIMIT = 20
DEFAULT_SESSION_MAX = 3
FAILURE_STOP_THRESHOLD = 3
DEFAULT_RISK_COOLDOWN_MINUTES = 24 * 60
WAIT_TIERS = {
    "major": (90, 200),
    "minor": (30, 90),
    "technical": (0.5, 8),
}
PACING_STAGE_TIERS = {
    "page_open": "major",
    "after_follow": "major",
    "after_send": "major",
    "between_customers": "major",
    "session_cooldown": "major",
    "after_message_open": "minor",
    "before_send": "minor",
}
PACING_PROFILES = {
    "very_conservative": {
        "page_open": WAIT_TIERS["major"],
        "after_follow": WAIT_TIERS["major"],
        "after_message_open": WAIT_TIERS["minor"],
        "before_send": WAIT_TIERS["minor"],
        "after_send": WAIT_TIERS["major"],
        "between_customers": WAIT_TIERS["major"],
        "session_cooldown": (1200, 3600),
    },
    "conservative": {
        "page_open": WAIT_TIERS["major"],
        "after_follow": WAIT_TIERS["major"],
        "after_message_open": WAIT_TIERS["minor"],
        "before_send": WAIT_TIERS["minor"],
        "after_send": WAIT_TIERS["major"],
        "between_customers": WAIT_TIERS["major"],
        "session_cooldown": (600, 1800),
    },
    "standard": {
        "page_open": WAIT_TIERS["major"],
        "after_follow": WAIT_TIERS["major"],
        "after_message_open": WAIT_TIERS["minor"],
        "before_send": WAIT_TIERS["minor"],
        "after_send": WAIT_TIERS["major"],
        "between_customers": WAIT_TIERS["major"],
        "session_cooldown": (300, 900),
    },
}
RISK_STOP_PHRASES = [
    "we limit",
    "you are temporarily blocked",
    "you're temporarily blocked",
    "action blocked",
    "confirm your identity",
    "suspicious activity",
    "temporarily unavailable",
    "identity confirmation",
    "verification required",
]
STATUS_VALUES = {
    "friend_success": "\u5df2\u52a0\u597d\u53cb,\u5f85\u79c1\u4fe1",
    "follow_success": "\u5df2\u5173\u6ce8,\u5f85\u79c1\u4fe1",
    "dm_success": "\u5df2\u53d1\u79c1\u4fe1",
    "both_success": "\u5df2\u5173\u6ce8,\u5df2\u53d1\u79c1\u4fe1",
    "friend_failed": "\u52a0\u597d\u53cb\u5931\u8d25",
    "follow_failed": "\u5173\u6ce8\u5931\u8d25",
    "dm_failed": "\u53d1\u9001\u5931\u8d25",
}
HEADER_ALIASES = {
    "index": ["\u5e8f\u53f7", "Index", "No."],
    "name": ["\u516c\u53f8\u59d3\u540d", "\u5ba2\u6237\u59d3\u540d/\u516c\u53f8", "\u516c\u53f8\u540d\u79f0", "Name", "Company"],
    "facebook_link": ["Facebook\u4e3b\u9875\u94fe\u63a5", "Facebook\u94fe\u63a5", "facebook\u94fe\u63a5", "\u4e3b\u9875/\u94fe\u63a5", "Link", "URL"],
    "status": ["\u5efa\u8054\u72b6\u6001", "\u5efa\u8054\u60c5\u51b5", "\u72b6\u6001", "Status"],
    "note": ["\u5907\u6ce8", "\u8bf4\u660e", "Notes"],
}
COMMENT_HINTS = ["comment", "留言", "reply", "回覆", "回复", "write a comment", "撰寫留言"]
MESSAGE_BUTTON_RE = re.compile(r"(^|\s)(message|訊息)(\s|$)|send message|發送訊息|发送讯息|发送消息", re.I)
FOLLOW_BUTTON_RE = re.compile(r"(^|\s)(follow|追蹤|关注)(\s|$)", re.I)
FOLLOWING_BUTTON_RE = re.compile(r"following|追蹤中|已关注|已追蹤", re.I)


def now_iso() -> str:
    return datetime.now().isoformat(timespec="seconds")


def today_key() -> str:
    return datetime.now().strftime("%Y%m%d")


def ledger_path(profile_id: str) -> Path:
    safe_profile = re.sub(r"[^0-9A-Za-z_.-]+", "_", profile_id.strip() or "unknown")
    return Path("logs") / f"facebook_account_ledger_{safe_profile}_{today_key()}.json"


def empty_ledger(profile_id: str) -> Dict[str, Any]:
    return {
        "profile_id": profile_id,
        "date": today_key(),
        "follow_count": 0,
        "dm_count": 0,
        "failure_count": 0,
        "risk_events": [],
        "events": [],
        "last_run_at": "",
        "cooldown_until": "",
    }


def load_ledger(profile_id: str) -> Dict[str, Any]:
    path = ledger_path(profile_id)
    if not path.exists():
        return empty_ledger(profile_id)
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        data = empty_ledger(profile_id)
    defaults = empty_ledger(profile_id)
    for key, value in defaults.items():
        data.setdefault(key, value)
    return data


def save_ledger(profile_id: str, ledger: Dict[str, Any]) -> None:
    path = ledger_path(profile_id)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(ledger, ensure_ascii=False, indent=2), encoding="utf-8")


def ledger_event(ledger: Dict[str, Any], event_type: str, dealer_name: str = "", detail: str = "") -> None:
    ledger["last_run_at"] = now_iso()
    ledger.setdefault("events", []).append({
        "timestamp": now_iso(),
        "type": event_type,
        "dealer_name": dealer_name,
        "detail": detail,
    })
    if event_type == "follow":
        ledger["follow_count"] = int(ledger.get("follow_count", 0)) + 1
    elif event_type == "dm":
        ledger["dm_count"] = int(ledger.get("dm_count", 0)) + 1
    elif event_type == "failure":
        ledger["failure_count"] = int(ledger.get("failure_count", 0)) + 1
    elif event_type == "risk":
        ledger.setdefault("risk_events", []).append({"timestamp": now_iso(), "dealer_name": dealer_name, "detail": detail})
        ledger["cooldown_until"] = (datetime.now() + timedelta(minutes=DEFAULT_RISK_COOLDOWN_MINUTES)).isoformat(timespec="seconds")


def cooldown_active(ledger: Dict[str, Any]) -> str:
    raw = clean(ledger.get("cooldown_until", ""))
    if not raw:
        return ""
    try:
        until = datetime.fromisoformat(raw)
    except ValueError:
        return ""
    if until > datetime.now():
        return raw
    return ""


def action_needs_follow(action: str) -> bool:
    return action in {"follow", "follow_dm"}


def action_needs_dm(action: str) -> bool:
    return action in {"dm", "both", "follow_dm"}


def remaining_capacity(ledger: Dict[str, Any], daily_follow_limit: int, daily_dm_limit: int) -> Dict[str, int]:
    return {
        "follow": max(0, daily_follow_limit - int(ledger.get("follow_count", 0))),
        "dm": max(0, daily_dm_limit - int(ledger.get("dm_count", 0))),
    }


def allowed_item_count(action: str, requested: int, ledger: Dict[str, Any], daily_follow_limit: int, daily_dm_limit: int, session_max: int) -> int:
    capacity = remaining_capacity(ledger, daily_follow_limit, daily_dm_limit)
    limits = [requested, session_max]
    if action_needs_follow(action):
        limits.append(capacity["follow"])
    if action_needs_dm(action):
        limits.append(capacity["dm"])
    return max(0, min(limits))


def pacing_range(profile: Dict[str, tuple[int, int]], stage: str) -> tuple[int, int]:
    return profile.get(stage, (10, 20))


def pacing_stage_tier(stage: str) -> str:
    return PACING_STAGE_TIERS.get(stage, "technical")


def sample_wait(profile: Dict[str, tuple[int, int]], stage: str) -> float:
    low, high = pacing_range(profile, stage)
    return random.uniform(low, high)


def pause(profile: Dict[str, tuple[int, int]], stage: str, confirm: bool, reason: str) -> float:
    wait = sample_wait(profile, stage)
    print(f"  -> pacing {reason}: {wait:.1f}s")
    if confirm:
        time.sleep(wait)
    return wait


def estimate_schedule(items: List[Dict[str, Any]], action: str, profile: Dict[str, tuple[int, int]]) -> Dict[str, Any]:
    stages = ["page_open", "between_customers"]
    if action_needs_follow(action):
        stages.append("after_follow")
    if action_needs_dm(action):
        stages.extend(["after_message_open", "before_send", "after_send"])
    per_customer_min = sum(pacing_range(profile, stage)[0] for stage in stages)
    per_customer_max = sum(pacing_range(profile, stage)[1] for stage in stages)
    cooldown = pacing_range(profile, "session_cooldown")
    return {
        "customers": len(items),
        "estimated_seconds_min": per_customer_min * len(items) + cooldown[0],
        "estimated_seconds_max": per_customer_max * len(items) + cooldown[1],
        "estimated_minutes_min": ceil((per_customer_min * len(items) + cooldown[0]) / 60),
        "estimated_minutes_max": ceil((per_customer_max * len(items) + cooldown[1]) / 60),
        "stages": {stage: pacing_range(profile, stage) for stage in stages + ["session_cooldown"]},
        "stage_tiers": {stage: pacing_stage_tier(stage) for stage in stages + ["session_cooldown"]},
        "tier_ranges": WAIT_TIERS,
    }


def build_schedule_preview(items: List[Dict[str, Any]], args: argparse.Namespace, ledger: Dict[str, Any], profile: Dict[str, tuple[int, int]]) -> Dict[str, Any]:
    return {
        "mode": "confirmed" if args.confirm else "dry_run",
        "risk_profile": args.risk_profile,
        "profile_id": args.profile_id,
        "today_used": {
            "follow": int(ledger.get("follow_count", 0)),
            "dm": int(ledger.get("dm_count", 0)),
            "failures": int(ledger.get("failure_count", 0)),
            "risk_events": len(ledger.get("risk_events", [])),
        },
        "today_remaining": remaining_capacity(ledger, args.daily_follow_limit, args.daily_dm_limit),
        "limits": {
            "daily_follow_limit": args.daily_follow_limit,
            "daily_dm_limit": args.daily_dm_limit,
            "session_max": args.session_max,
            "failure_stop_threshold": FAILURE_STOP_THRESHOLD,
        },
        "schedule": estimate_schedule(items, args.action, profile),
        "customers": [{"dealer_name": item.get("dealer_name", ""), "page_url": item.get("page_url", "")} for item in items],
        "stop_rules": [
            "Stop immediately on Facebook verification, temporary block, suspicious activity, or identity prompt.",
            "Stop when the account daily Follow/DM limit is exhausted.",
            "Stop when failure count reaches threshold.",
        ],
    }



def detect_risk(page) -> str:
    try:
        text = page.content().casefold()
    except Exception:
        return ""
    for phrase in RISK_STOP_PHRASES:
        if phrase.casefold() in text:
            return phrase
    return ""


def clean(value: Any) -> str:
    return "" if value is None else str(value).strip()


def norm(value: str) -> str:
    return re.sub(r"\s+", " ", clean(value)).casefold()


def name_tokens(name: str) -> List[str]:
    return [part for part in re.split(r"[^0-9a-zA-Z]+", name.casefold()) if len(part) >= 3]


def get_element_text(locator) -> str:
    try:
        return clean(locator.inner_text(timeout=1000))
    except Exception:
        try:
            return clean(locator.get_attribute("aria-label"))
        except Exception:
            return ""


def first_visible(locator, timeout_ms: int = 1800):
    try:
        count = locator.count() if locator else 0
        for idx in range(count):
            item = locator.nth(idx)
            if item.is_visible(timeout=timeout_ms):
                return item
    except Exception:
        return None
    return None


def find_page_header_button(page, include_re: re.Pattern, exclude_re: Optional[re.Pattern] = None):
    """Find a visible action button in the Page header, not a post/comment area."""
    try:
        viewport = page.viewport_size or {"width": 1600, "height": 1000}
        max_y = viewport["height"] * 0.76
        min_y = 72
        min_x = viewport["width"] * 0.45
        candidates = []
        locator = page.locator("div[role='button'], a[role='button'], button")
        for idx in range(locator.count()):
            item = locator.nth(idx)
            if not item.is_visible(timeout=500):
                continue
            box = item.bounding_box()
            if not box:
                continue
            if box.get("y", 0) < min_y or box.get("y", 0) > max_y or box.get("x", 0) < min_x:
                continue
            text = get_element_text(item)
            aria = clean(item.get_attribute("aria-label"))
            combined = f"{text} {aria}".strip()
            if not include_re.search(combined):
                continue
            if exclude_re and exclude_re.search(combined):
                continue
            candidates.append((box.get("y", 0), box.get("x", 0), item, combined))
        if candidates:
            candidates.sort(key=lambda row: (row[0], row[1]))
            return candidates[0][2], candidates[0][3]
    except Exception:
        return None, ""
    return None, ""


def click_follow(page, confirm: bool) -> Dict[str, Any]:
    result = {"action": "follow", "clicked": False, "already_active": False, "error": ""}
    try:
        active_button, active_label = find_page_header_button(page, FOLLOWING_BUTTON_RE)
        if active_button is not None:
            result["already_active"] = True
            print(f"  -> Follow already active: {active_label}")
            return result

        button, label = find_page_header_button(page, FOLLOW_BUTTON_RE, exclude_re=FOLLOWING_BUTTON_RE)
        if button is None:
            result["error"] = "Follow button not found in Page header"
            return result
        print(f"  [Follow] button is visible: {label}")
        if confirm:
            button.click()
            result["clicked"] = True
            print("  -> Follow clicked")
        else:
            print("  -> dry-run: follow not clicked")
    except Exception as exc:  # pragma: no cover - browser dependent
        result["error"] = str(exc)
    return result


def click_add_friend(page, confirm: bool) -> Dict[str, Any]:
    result = {"action": "add_friend", "clicked": False, "error": ""}
    try:
        button, label = find_page_header_button(page, re.compile(r"add\s*friend|ajouter|connect", re.I))
        if button is None:
            result["error"] = "Add Friend button not found in Page header"
            return result
        print(f"  [Add Friend] button is visible: {label}")
        if confirm:
            button.click()
            result["clicked"] = True
            print("  -> clicked Add Friend")
        else:
            print("  -> dry-run: no click")
    except Exception as exc:  # pragma: no cover - browser dependent
        result["error"] = str(exc)
    return result


def open_message_dialog(page, confirm: bool) -> Dict[str, Any]:
    result = {"action": "open_message", "clicked": False, "error": ""}
    button, label = find_page_header_button(page, MESSAGE_BUTTON_RE)
    if button is None:
        result["error"] = "Message button not found in Page header"
        return result
    print(f"  [Message] button is visible in Page header: {label}")
    if confirm:
        button.click()
        result["clicked"] = True
        print("  -> Message button clicked")
    else:
        print("  -> dry-run: message dialog not opened")
    return result


def dialog_matches_customer(dialog, dealer_name: str) -> bool:
    text = norm(get_element_text(dialog))
    tokens = name_tokens(dealer_name)
    if not tokens:
        return False
    return sum(1 for token in tokens if token in text) >= min(2, len(tokens))


def find_matching_messenger_dialog(page, dealer_name: str):
    """Return a right-side Messenger dialog whose visible title matches dealer_name."""
    try:
        viewport = page.viewport_size or {"width": 1600, "height": 1000}
        min_x = viewport["width"] * 0.42
        min_y = viewport["height"] * 0.28
        candidates = []
        containers = page.locator("div[role='dialog'], div[aria-label*='Messenger' i], div[aria-label*='訊息' i], div[aria-label*='Message' i]")
        for idx in range(containers.count()):
            dialog = containers.nth(idx)
            if not dialog.is_visible(timeout=500):
                continue
            box = dialog.bounding_box()
            if not box:
                continue
            if box.get("x", 0) < min_x or box.get("y", 0) < min_y:
                continue
            if not dialog_matches_customer(dialog, dealer_name):
                continue
            dialog_text = norm(get_element_text(dialog))
            if any(hint in dialog_text for hint in COMMENT_HINTS):
                continue
            candidates.append((box.get("x", 0), box.get("y", 0), dialog))
        if candidates:
            candidates.sort(key=lambda row: (row[0], row[1]), reverse=True)
            return candidates[0][2]
    except Exception:
        return None
    return None


def find_dialog_textbox(dialog):
    selectors = [
        'div[contenteditable="true"][role="textbox"]',
        'div[contenteditable="true"][data-lexical-editor="true"]',
        '[aria-label="Aa"][contenteditable="true"]',
        '[aria-label*="Message" i][contenteditable="true"]',
        '[aria-label*="訊息" i][contenteditable="true"]',
    ]
    for selector in selectors:
        textbox = first_visible(dialog.locator(selector), timeout_ms=800)
        if textbox is not None:
            label = clean(textbox.get_attribute("aria-label"))
            text = norm(get_element_text(textbox))
            if any(hint in norm(label) or hint in text for hint in COMMENT_HINTS):
                return None, "Blocked: detected comment composer, not Messenger dialog"
            return textbox, ""
    return None, "Message textbox not found inside matching Messenger dialog"


def facebook_thread_url(page_url: str) -> str:
    match = re.search(r"facebook\.com/([^/?#]+)", page_url or "", re.I)
    if not match:
        return ""
    slug = match.group(1).strip("/")
    if not slug or slug in {"messages", "profile.php", "pages"}:
        return ""
    return f"https://www.facebook.com/messages/t/{slug}"


def messenger_body_matches_customer(body: str, dealer_name: str) -> bool:
    if not dealer_name:
        return False
    body_norm = norm(body)
    direct = norm(dealer_name)
    if direct and direct in body_norm:
        return True
    tokens = name_tokens(dealer_name)
    if not tokens:
        return False
    return sum(1 for token in tokens if token in body_norm) >= min(2, len(tokens))


def find_messenger_thread_textbox(thread_page, dealer_name: str):
    try:
        body = thread_page.locator("body").inner_text(timeout=3000)
    except Exception:
        body = ""
    if not messenger_body_matches_customer(body, dealer_name):
        return None, f"Messenger thread title mismatch for {dealer_name}"

    locator = thread_page.locator('[role="textbox"], div[contenteditable="true"]')
    for idx in range(locator.count()):
        textbox = locator.nth(idx)
        try:
            if not textbox.is_visible(timeout=500):
                continue
            label = clean(textbox.get_attribute("aria-label"))
            nearby = norm(label + " " + get_element_text(textbox))
            if any(hint in nearby for hint in COMMENT_HINTS):
                return None, "Blocked: detected comment composer, not Messenger thread"
            box = textbox.bounding_box()
            if not box or box.get("y", 0) < 400:
                continue
            if dealer_name and dealer_name.casefold() in label.casefold():
                return textbox, ""
            if label or box:
                return textbox, ""
        except Exception:
            continue
    return None, "Messenger thread textbox not found"


def send_dm_via_messenger_thread(page, dealer_name: str, text: str, confirm: bool, before_send_wait: tuple[int, int] = (0, 0)) -> Dict[str, Any]:
    result = {"action": "send_dm_thread", "clicked": False, "error": ""}
    thread_url = facebook_thread_url(page.url)
    if not thread_url:
        result["error"] = "Could not build Messenger thread URL from Facebook Page URL"
        return result
    if not confirm:
        print(f"  -> dry-run: Messenger thread fallback available: {thread_url}")
        return result

    thread_page = page.context.new_page()
    thread_page.goto(thread_url, wait_until="domcontentloaded", timeout=60000)
    thread_page.wait_for_timeout(8000)
    thread_page.bring_to_front()

    textbox, error = find_messenger_thread_textbox(thread_page, dealer_name)
    if textbox is None:
        result["error"] = error
        return result

    textbox.click()
    try:
        existing_text = textbox.evaluate("el => (el.innerText || el.textContent || '').trim()") or ""
    except Exception:
        existing_text = ""
    if text[:80] not in existing_text:
        textbox.fill(text)
        print("  -> Message typed into Messenger thread")
    else:
        print("  -> existing drafted message detected; not inserting duplicate text")
    if before_send_wait != (0, 0):
        delay = random.uniform(*before_send_wait)
        print(f"  -> pacing before send: {delay:.1f}s")
        time.sleep(delay)
    thread_page.keyboard.press("Enter")
    result["clicked"] = True
    print("  -> Message sent by Enter in Messenger thread")
    return result


def send_dm(page, dealer_name: str, text: str, confirm: bool, message_open_wait: tuple[int, int] = (0, 0), before_send_wait: tuple[int, int] = (0, 0)) -> Dict[str, Any]:
    result = {"action": "send_dm", "clicked": False, "error": ""}
    try:
        dialog = find_matching_messenger_dialog(page, dealer_name)
        if dialog is None:
            opened = open_message_dialog(page, confirm=confirm)
            if opened.get("error"):
                print(f"  ! Message button issue: {opened['error']}; trying Messenger thread fallback")
                fallback = send_dm_via_messenger_thread(page, dealer_name, text, confirm, before_send_wait=before_send_wait)
                if fallback.get("clicked"):
                    result["clicked"] = True
                    return result
                result["error"] = fallback.get("error") or opened["error"]
                return result
            if not confirm:
                print("  -> dry-run: no message sent")
                return result
            if message_open_wait != (0, 0):
                delay = random.uniform(*message_open_wait)
                print(f"  -> pacing after message open: {delay:.1f}s")
                time.sleep(delay)
            for _ in range(10):
                time.sleep(0.8)
                dialog = find_matching_messenger_dialog(page, dealer_name)
                if dialog is not None:
                    break
        if dialog is None:
            print(f"  ! Messenger dialog for {dealer_name} not found; trying Messenger thread fallback")
            fallback = send_dm_via_messenger_thread(page, dealer_name, text, confirm, before_send_wait=before_send_wait)
            if fallback.get("clicked"):
                result["clicked"] = True
                return result
            result["error"] = fallback.get("error") or f"Messenger dialog for {dealer_name} not found or title mismatch"
            return result
        print(f"  -> Messenger dialog matched: {dealer_name}")

        textbox, error = find_dialog_textbox(dialog)
        if textbox is None:
            result["error"] = error
            return result
        if not confirm:
            print("  -> dry-run: Messenger textbox found; no message typed")
            return result

        textbox.click()
        existing_text = ""
        try:
            existing_text = textbox.evaluate("el => (el.innerText || el.textContent || '').trim()") or ""
        except Exception:
            existing_text = ""
        if text[:80] not in existing_text:
            page.keyboard.insert_text(text)
            print("  -> Message typed into Messenger dialog")
        else:
            print("  -> existing drafted message detected; not inserting duplicate text")

        if before_send_wait != (0, 0):
            delay = random.uniform(*before_send_wait)
            print(f"  -> pacing before send: {delay:.1f}s")
            time.sleep(delay)
        page.keyboard.press("Enter")
        result["clicked"] = True
        print("  -> Message sent by Enter")
    except Exception as exc:  # pragma: no cover - browser dependent
        result["error"] = str(exc)
    return result


def header_map(ws) -> Dict[str, int]:
    raw = {clean(cell.value): idx for idx, cell in enumerate(ws[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 update_workbook(excel_path: str, sheet_name: str, updates: List[Dict[str, str]]) -> int:
    wb = load_workbook(excel_path)
    if sheet_name not in wb.sheetnames:
        summary = "\u5ba2\u6237\u4fe1\u606f\u6c47\u603b\u8868"
        if summary in wb.sheetnames:
            sheet_name = summary
        else:
            raise KeyError(f"Sheet not found: {sheet_name}")
    ws = wb[sheet_name]
    columns = header_map(ws)
    if "status" not in columns:
        raise RuntimeError("Workbook is missing outreach status column.")

    update_by_index = {str(item.get("index", "")).strip(): item for item in updates if str(item.get("index", "")).strip()}
    update_by_url = {norm(item.get("page_url", "")): item for item in updates if norm(item.get("page_url", ""))}
    update_by_name = {norm(item.get("dealer_name", "")): item for item in updates if norm(item.get("dealer_name", ""))}
    updated = 0
    for row_number, row in enumerate(ws.iter_rows(min_row=2, values_only=False), start=2):
        update = None
        if str(row_number) in update_by_index:
            update = update_by_index[str(row_number)]
        if update is None and "index" in columns:
            row_index = clean(row[columns["index"]].value)
            update = update_by_index.get(row_index)
        if update is None and "facebook_link" in columns:
            row_url = norm(row[columns["facebook_link"]].value)
            update = update_by_url.get(row_url)
        if update is None and "name" in columns:
            row_name = norm(row[columns["name"]].value)
            update = update_by_name.get(row_name)
        if update is None:
            continue
        if update.get("status"):
            row[columns["status"]].value = update["status"]
        if update.get("note_append") and "note" in columns:
            existing = clean(row[columns["note"]].value)
            separator = " | " if existing else ""
            row[columns["note"]].value = existing + separator + update["note_append"]
        updated += 1
    wb.save(excel_path)
    return updated

def write_audit_log(log_path: Path, entries: List[Dict[str, Any]]) -> None:
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with log_path.open("a", encoding="utf-8") as handle:
        for entry in entries:
            handle.write(json.dumps(entry, ensure_ascii=False) + "\n")


def select_english_dm(item: Dict[str, Any]) -> str:
    message = item.get("messages", {}).get("dm", {}).get("en", "")
    if not message:
        raise ValueError(f"Missing English DM text for {item.get('dealer_name', '(unknown)')}")
    return message


def determine_status(action: str, friend_result: Optional[Dict[str, Any]], follow_result: Optional[Dict[str, Any]], dm_result: Optional[Dict[str, Any]]) -> tuple[str, str]:
    friend_ok = bool(friend_result and friend_result.get("clicked"))
    follow_ok = bool(follow_result and (follow_result.get("clicked") or follow_result.get("already_active")))
    dm_ok = bool(dm_result and dm_result.get("clicked"))
    if action == "friend":
        return (STATUS_VALUES["friend_success"], "") if friend_ok else (STATUS_VALUES["friend_failed"], friend_result.get("error", "unknown") if friend_result else "unknown")
    if action == "follow":
        return (STATUS_VALUES["follow_success"], "") if follow_ok else (STATUS_VALUES["follow_failed"], follow_result.get("error", "unknown") if follow_result else "unknown")
    if action == "dm":
        return (STATUS_VALUES["dm_success"], "") if dm_ok else (STATUS_VALUES["dm_failed"], dm_result.get("error", "unknown") if dm_result else "unknown")
    if action == "follow_dm":
        if follow_ok and dm_ok:
            return STATUS_VALUES["both_success"], ""
        if follow_ok:
            return STATUS_VALUES["follow_success"], dm_result.get("error", "DM not sent") if dm_result else "DM not sent"
        if dm_ok:
            return STATUS_VALUES["dm_success"], follow_result.get("error", "follow not completed") if follow_result else "follow not completed"
        return STATUS_VALUES["dm_failed"], "follow and DM were not completed"
    if friend_ok and dm_ok:
        return STATUS_VALUES["both_success"], ""
    if friend_ok:
        return STATUS_VALUES["friend_success"], dm_result.get("error", "DM not sent") if dm_result else "DM not sent"
    if dm_ok:
        return STATUS_VALUES["dm_success"], friend_result.get("error", "friend request not sent") if friend_result else "friend request not sent"
    return STATUS_VALUES["dm_failed"], "friend request and DM were not completed"


def should_navigate(page, target_url: str, use_open_page: bool) -> bool:
    if not use_open_page:
        return True
    current = clean(getattr(page, "url", ""))
    if not current:
        return True
    return target_url.rstrip("/") not in current.rstrip("/")


def find_open_page_for_target(browser, target_url: str):
    """Pick the most recently opened tab that already matches the target URL."""
    target = target_url.rstrip("/")
    fallback = None
    for context in browser.contexts:
        for candidate in context.pages:
            fallback = candidate
            try:
                current = clean(candidate.url).rstrip("/")
            except Exception:
                continue
            if target and target in current:
                fallback = candidate
    return fallback


def main(argv: Optional[Sequence[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Run dry-run or confirmed English Facebook outreach in AdsPower.")
    parser.add_argument("--preview", required=True, help="JSON preview generated by prepare_facebook_outreach.py")
    parser.add_argument("--profile-id", required=True, help="AdsPower profile ID")
    parser.add_argument("--ads-power-url", default=DEFAULT_ADS_POWER_URL, help="AdsPower local API URL")
    parser.add_argument("--action", choices=["friend", "follow", "dm", "both", "follow_dm"], default="follow_dm", help="Action to perform")
    parser.add_argument("--max-per-run", type=int, default=MAX_PER_RUN, help="Maximum customers per run. Default: 3.")
    parser.add_argument("--confirm", action="store_true", help="Allow real clicks/sends after the whole preview has been shown and approved")
    parser.add_argument("--batch-confirmed", action="store_true", help="Required with --confirm; means the full batch preview was shown in chat and approved once")
    parser.add_argument("--headless", action="store_true", help="Start AdsPower browser in headless mode if supported")
    parser.add_argument("--keep-browser-open", action="store_true", default=True, help="Compatibility flag; AdsPower browser is always kept open")
    parser.add_argument("--no-write-workbook", action="store_true", help="Do not write outreach status back to the workbook; default behavior unless --write-workbook is set")
    parser.add_argument("--write-workbook", action="store_true", help="Explicitly write outreach status back to the workbook")
    parser.add_argument("--use-open-page", action="store_true", help="Use the first currently open AdsPower page instead of creating a new tab")
    parser.add_argument("--auto-confirm-each", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument("--risk-profile", choices=sorted(PACING_PROFILES), default="very_conservative", help="Pacing policy for compliant low-frequency outreach. Default: very_conservative.")
    parser.add_argument("--daily-follow-limit", type=int, default=DEFAULT_DAILY_FOLLOW_LIMIT, help="Per-profile daily Follow cap. Default: 20.")
    parser.add_argument("--daily-dm-limit", type=int, default=DEFAULT_DAILY_DM_LIMIT, help="Per-profile daily DM/customer outreach cap. Default: 20.")
    parser.add_argument("--session-max", type=int, default=DEFAULT_SESSION_MAX, help="Maximum customers per execution session. Default: 3.")
    parser.add_argument("--dry-run-schedule", action="store_true", help="Print the pacing schedule, limits, and customer list without opening AdsPower.")
    args = parser.parse_args(argv)
    if args.auto_confirm_each:
        print("--auto-confirm-each is deprecated; v4.1 uses one batch-level confirmation via --batch-confirmed.")
        args.batch_confirmed = True
    if args.session_max < 1:
        raise ValueError("--session-max must be at least 1")
    if args.daily_follow_limit < 0 or args.daily_dm_limit < 0:
        raise ValueError("Daily limits cannot be negative")
    if args.confirm and not args.batch_confirmed:
        raise ValueError("Real Facebook sending requires --confirm --batch-confirmed after the full preview is shown in chat and approved.")

    preview_path = Path(args.preview).expanduser()
    if not preview_path.is_absolute():
        preview_path = Path.cwd() / preview_path
    if not preview_path.exists():
        raise FileNotFoundError(f"Preview file not found: {preview_path}")

    preview = json.loads(preview_path.read_text(encoding="utf-8"))
    language_policy = preview.get("language_policy", {})
    if language_policy.get("customer_facing") and language_policy.get("customer_facing") != "English":
        raise ValueError("Preview is not marked as English customer-facing. Regenerate with the updated prepare script.")

    all_items = preview.get("items", [])
    requested_count = max(0, args.max_per_run)
    pacing_profile = PACING_PROFILES[args.risk_profile]
    ledger = load_ledger(args.profile_id)
    active_cooldown = cooldown_active(ledger)
    if active_cooldown:
        raise RuntimeError(f"Profile {args.profile_id} is cooling down until {active_cooldown}; stop Facebook outreach for this account.")
    if ledger.get("risk_events"):
        raise RuntimeError(f"Profile {args.profile_id} has risk events today; stop Facebook outreach for this account.")
    if int(ledger.get("failure_count", 0)) >= FAILURE_STOP_THRESHOLD:
        raise RuntimeError(f"Profile {args.profile_id} reached today's failure threshold; stop Facebook outreach for this account.")
    allowed_count = allowed_item_count(args.action, requested_count, ledger, args.daily_follow_limit, args.daily_dm_limit, args.session_max)
    items = all_items[:allowed_count]
    if requested_count and allowed_count < requested_count:
        print(f"Pacing limit reduced this run from {requested_count} to {allowed_count} customers.")
    schedule_preview = build_schedule_preview(items, args, ledger, pacing_profile)
    print(json.dumps({"schedule_preview": schedule_preview}, ensure_ascii=False, indent=2))
    if args.dry_run_schedule:
        print("Dry-run schedule only: browser execution skipped.")
        return 0
    if requested_count and not items:
        raise RuntimeError("No Facebook outreach capacity remains for this profile today.")
    source = preview.get("source", {})
    excel_path = source.get("excel", "")
    sheet_name = source.get("sheet", "Facebook")
    confirm_mode = bool(args.confirm)
    write_workbook = bool(args.write_workbook and not args.no_write_workbook)

    print(f"Mode: {'CONFIRMED batch execution' if confirm_mode else 'dry-run only'}")
    print("Customer-facing language: English")
    print(f"AdsPower profile: {args.profile_id}")
    print(f"Items: {len(items)} / {len(all_items)}")
    print(f"Workbook write: {'enabled' if write_workbook else 'disabled'}")
    print("-" * 60)

    client = AdsPowerClient(base_url=args.ads_power_url)
    audit_entries: List[Dict[str, Any]] = []
    status_updates: List[Dict[str, str]] = []
    stopped_early = False

    try:
        browser = client.start_browser(args.profile_id, headless=args.headless)
        page = client.get_open_page() if args.use_open_page else None
        page = page or client.new_page()

        for offset, item in enumerate(items, start=1):
            dm_text = select_english_dm(item)
            dealer_name = clean(item.get("dealer_name", ""))
            target_url = item.get("page_url", "")
            print(f"\n[{offset}/{len(items)}] {dealer_name} | {target_url}")
            print(f"    Current status: {item.get('status', '未联系')} | Suggested action: {item.get('suggested_action', '')}")
            print(f"    English DM preview:\n    {dm_text[:500]}{'...' if len(dm_text) > 500 else ''}")

            if args.use_open_page:
                matched_page = find_open_page_for_target(browser, target_url)
                if matched_page is not None:
                    page = matched_page

            if should_navigate(page, target_url, args.use_open_page):
                page.goto(target_url, wait_until="domcontentloaded", timeout=60000)
                pause(pacing_profile, "page_open", confirm=confirm_mode, reason="after page open")
            else:
                print("  -> using current open Page tab")
                pause(pacing_profile, "page_open", confirm=confirm_mode, reason="after existing page focus")

            try:
                page.bring_to_front()
                time.sleep(1)
            except Exception:
                pass

            risk = detect_risk(page)
            if risk:
                print(f"\nRisk prompt detected: {risk}. Stopping immediately.")
                ledger_event(ledger, "risk", dealer_name, risk)
                save_ledger(args.profile_id, ledger)
                stopped_early = True
                break

            if confirm_mode:
                print("  -> batch preview already confirmed; executing this customer without per-row prompt")
            else:
                print("  -> dry-run: no click/send and no per-row prompt")

            friend_result = None
            follow_result = None
            dm_result = None
            if args.action in {"friend", "both"}:
                friend_result = click_add_friend(page, confirm=confirm_mode)
                if friend_result.get("error"):
                    print(f"  ! Add Friend: {friend_result['error']}")
            if args.action in {"follow", "follow_dm"}:
                follow_result = click_follow(page, confirm=confirm_mode)
                if follow_result.get("error"):
                    print(f"  ! Follow: {follow_result['error']}")
                elif confirm_mode and follow_result.get("clicked"):
                    ledger_event(ledger, "follow", dealer_name)
                    save_ledger(args.profile_id, ledger)
                pause(pacing_profile, "after_follow", confirm=confirm_mode, reason="after Follow")
            if args.action in {"dm", "both", "follow_dm"}:
                dm_result = send_dm(
                    page,
                    dealer_name,
                    dm_text,
                    confirm=confirm_mode,
                    message_open_wait=pacing_range(pacing_profile, "after_message_open"),
                    before_send_wait=pacing_range(pacing_profile, "before_send"),
                )
                if dm_result.get("error"):
                    print(f"  ! Message: {dm_result['error']}")
                elif confirm_mode and dm_result.get("clicked"):
                    ledger_event(ledger, "dm", dealer_name)
                    save_ledger(args.profile_id, ledger)
                pause(pacing_profile, "after_send", confirm=confirm_mode, reason="after DM send")

            new_status = ""
            error_note = ""
            if confirm_mode:
                new_status, error_note = determine_status(args.action, friend_result, follow_result, dm_result)
                status_updates.append({
                    "index": str(item.get("row_number") or item.get("index", "")),
                    "dealer_name": dealer_name,
                    "page_url": target_url,
                    "status": new_status,
                    "note_append": f"{now_iso()} Facebook {args.action}->{new_status}" + (f" err={error_note}" if error_note else ""),
                })
                if error_note:
                    ledger_event(ledger, "failure", dealer_name, error_note)
                    save_ledger(args.profile_id, ledger)

            audit_entries.append({
                "timestamp": now_iso(),
                "profile_id": args.profile_id,
                "target_url": target_url,
                "dealer_name": dealer_name,
                "action": args.action,
                "confirm_mode": confirm_mode,
                "auto_confirm_each": bool(args.auto_confirm_each),
                "language": "English",
                "new_status": new_status,
                "result": "dry_run" if not confirm_mode else ("success" if not error_note else "partial_or_failed"),
                "error": error_note,
            })

            pause(pacing_profile, "between_customers", confirm=confirm_mode, reason="between customers")
            if int(ledger.get("failure_count", 0)) >= FAILURE_STOP_THRESHOLD:
                print("Failure threshold reached. Stopping remaining customers for today.")
                stopped_early = True
                break
            capacity_now = remaining_capacity(ledger, args.daily_follow_limit, args.daily_dm_limit)
            if (action_needs_follow(args.action) and capacity_now["follow"] <= 0) or (action_needs_dm(args.action) and capacity_now["dm"] <= 0):
                print("Daily Follow/DM capacity exhausted. Stopping remaining customers.")
                stopped_early = True
                break

    except KeyboardInterrupt:
        print("\nInterrupted by user.")
    finally:
        print("Browser left open")
        try:
            if client.playwright:
                client.playwright.stop()
        except Exception:
            pass

    log_path = Path("logs") / f"facebook_social_{datetime.now().strftime('%Y%m%d')}.jsonl"
    if audit_entries:
        write_audit_log(log_path, audit_entries)
        print(f"\nAudit log written: {log_path}")
    save_ledger(args.profile_id, ledger)
    if args.confirm:
        cooldown = sample_wait(pacing_profile, "session_cooldown")
        print(f"Session cooldown recommendation: {cooldown / 60:.1f} minutes before the next Facebook outreach run.")

    if confirm_mode and status_updates and write_workbook:
        if not excel_path:
            backup = resolve_artifact_path("", kind="facebook_status_updates", default_name=f"facebook_social_status_updates_{now_iso().replace(':', '-')}.json")
            backup.write_text(json.dumps(status_updates, ensure_ascii=False, indent=2), encoding="utf-8")
            print(f"Workbook path missing in preview. Status updates saved to: {backup}")
        else:
            try:
                updated = update_workbook(excel_path, sheet_name, status_updates)
                print(f"Workbook updated: {updated} rows ({excel_path})")
            except Exception as exc:
                backup = resolve_artifact_path("", kind="facebook_status_updates", default_name=f"facebook_social_status_updates_{now_iso().replace(':', '-')}.json")
                backup.write_text(json.dumps(status_updates, ensure_ascii=False, indent=2), encoding="utf-8")
                print(f"Workbook update failed: {exc}")
                print(f"Status updates saved to: {backup}")

    if confirm_mode and status_updates and not write_workbook:
        print("Workbook update skipped")

    print("\nDone.")
    if stopped_early:
        print("Stopped early because a Facebook risk prompt was detected.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())