#!/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": "已加好友,待私信", "follow_success": "已关注,待私信", "dm_success": "已发私信", "both_success": "已关注,已发私信", "friend_failed": "加好友失败", "follow_failed": "关注失败", "dm_failed": "发送失败", } HEADER_ALIASES = { "index": ["序号", "Index", "No."], "status": ["建联状态", "建联情况", "状态", "Status"], "note": ["备注", "说明", "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: raise KeyError(f"Sheet not found: {sheet_name}") ws = wb[sheet_name] columns = header_map(ws) if "index" not in columns or "status" not in columns: raise RuntimeError("Workbook is missing 序号 or 建联状态 columns.") update_by_index = {str(item.get("index", "")).strip(): item for item in updates if str(item.get("index", "")).strip()} updated = 0 for row in ws.iter_rows(min_row=2, values_only=False): row_index = clean(row[columns["index"]].value) if row_index not in update_by_index: continue update = update_by_index[row_index] 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("index", "")), "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())