#!/usr/bin/env python3 """ Facebook Group Lead Capture — Playwright Automation For 24/7 headless operation, run on a dedicated always-on Mac. See tony.austinvisuals.com/group-leads.html for setup wizard. Requirements: - Install Playwright first: pip install playwright && playwright install chromium - Provide credentials in /root/.openclaw/secrets/fb-credentials.json with: {"email": "...", "password": "...", "groups": ["GROUP_ID_1", "GROUP_ID_2"]} - Designed to run on a Mac Mini or another always-on computer - Connection status is visible at tony.austinvisuals.com/group-leads.html """ from __future__ import annotations import json import logging import os import platform import random import re import sys import time from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any import requests from playwright.sync_api import BrowserContext, Page, TimeoutError as PlaywrightTimeoutError, sync_playwright _HOME = Path.home() CREDENTIALS_PATH = _HOME / ".av-group-leads" / "fb-credentials.json" SESSION_PATH = _HOME / ".av-group-leads" / "fb-session.json" OPENCLAW_CONFIG = _HOME / ".av-group-leads" / "openclaw.json" SERVICE_URL = os.getenv("GROUP_LEADS_SERVICE_URL", "https://tony.austinvisuals.com/api/group-leads") INGEST_URL = f"{SERVICE_URL}/ingest" HEARTBEAT_URL = f"{SERVICE_URL}/heartbeat" SCRAPE_INTERVAL_SECONDS = int(os.getenv("FB_SCRAPE_INTERVAL_SECONDS", str(3 * 60 * 60))) HEARTBEAT_INTERVAL_SECONDS = int(os.getenv("FB_HEARTBEAT_INTERVAL_SECONDS", str(15 * 60))) ALERT_AFTER_SECONDS = int(os.getenv("FB_ALERT_AFTER_SECONDS", str(6 * 60 * 60))) PAGE_LOAD_TIMEOUT_MS = 60_000 MAC_CHROME_UA = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ) EMAIL_RE = re.compile(r"([A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,})", re.I) NAME_SELECTORS = [ '[data-visualcompletion="ignore-dynamic"] a[role="link"]', 'h3 a', '[data-testid="request-name"] a', ] CARD_SELECTORS = [ 'div[role="main"] div[role="article"]', 'div[role="feed"] > div', '[data-pagelet] div[role="article"]', ] logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) log = logging.getLogger("fb-group-leads") def ensure_credentials_file() -> dict[str, Any]: CREDENTIALS_PATH.parent.mkdir(parents=True, exist_ok=True) if not CREDENTIALS_PATH.exists(): CREDENTIALS_PATH.write_text( json.dumps({"email": "", "password": "", "groups": []}, indent=2) + "\n" ) log.warning("Created placeholder credentials file at %s", CREDENTIALS_PATH) data = json.loads(CREDENTIALS_PATH.read_text()) data.setdefault("email", "") data.setdefault("password", "") data.setdefault("groups", []) return data def load_bot_token() -> str | None: if not OPENCLAW_CONFIG.exists(): return None try: cfg = json.loads(OPENCLAW_CONFIG.read_text()) tokens = re.findall(r'[0-9]{8,10}:[A-Za-z0-9_-]{30,}', json.dumps(cfg)) return tokens[0] if tokens else None except Exception as exc: log.warning("Could not read bot token: %s", exc) return None def random_delay(min_seconds: float = 1.0, max_seconds: float = 4.0) -> None: time.sleep(random.uniform(min_seconds, max_seconds)) def send_telegram_alert(message: str) -> None: token = load_bot_token() if not token: log.warning("Skipping Telegram alert, bot token not found") return try: requests.post( f"https://api.telegram.org/bot{token}/sendMessage", json={"chat_id": 613132769, "text": message}, timeout=10, ) except Exception as exc: log.warning("Telegram alert failed: %s", exc) def post_json(url: str, payload: dict[str, Any]) -> bool: try: r = requests.post(url, json=payload, timeout=20) r.raise_for_status() return True except Exception as exc: log.warning("POST failed to %s: %s", url, exc) return False def extract_email(questions: list[dict[str, str]]) -> str | None: for qa in questions: answer = qa.get("answer", "") match = EMAIL_RE.search(answer) if match: return match.group(1) return None def parse_mutual_friends(text: str) -> int: match = re.search(r"(\d+)\s+mutual friend", text, re.I) return int(match.group(1)) if match else 0 def extract_questions(card_text: str) -> list[dict[str, str]]: lines = [ln.strip() for ln in card_text.splitlines() if ln.strip()] questions: list[dict[str, str]] = [] i = 0 while i < len(lines) - 1 and len(questions) < 3: current = lines[i] nxt = lines[i + 1] lowered = current.lower() if ( "question" in lowered or current.endswith("?") or "why" in lowered or "experience" in lowered or "website" in lowered ): if nxt != current and len(nxt) > 1: questions.append({"question": current[:300], "answer": nxt[:1000]}) i += 2 continue i += 1 return questions def safe_text(locator) -> str: try: return locator.inner_text(timeout=1500).strip() except Exception: return "" def find_first_text(page_or_card, selectors: list[str]) -> tuple[str, str]: for selector in selectors: try: loc = page_or_card.locator(selector).first if loc.count() > 0: text = safe_text(loc) href = "" try: href = loc.get_attribute("href", timeout=1000) or "" except Exception: pass if text: return text, href except Exception: continue return "", "" def detect_logged_out(page: Page) -> bool: body = page.locator("body") text = safe_text(body)[:5000].lower() return any(token in text for token in ["log in", "login", "password", "forgot password"]) def login_if_needed(page: Page, creds: dict[str, Any]) -> None: page.goto("https://www.facebook.com/", wait_until="domcontentloaded", timeout=PAGE_LOAD_TIMEOUT_MS) random_delay() if not detect_logged_out(page): log.info("Facebook session already active") return email = creds.get("email", "").strip() password = creds.get("password", "").strip() if not email or not password: raise RuntimeError(f"Facebook credentials missing in {CREDENTIALS_PATH}") log.info("Logging into Facebook") page.locator('input[name="email"]').fill(email) random_delay(0.5, 1.2) page.locator('input[name="pass"]').fill(password) random_delay(0.5, 1.0) page.locator('button[name="login"], input[name="login"]').first.click() page.wait_for_load_state("networkidle", timeout=PAGE_LOAD_TIMEOUT_MS) random_delay(2, 4) if detect_logged_out(page): raise RuntimeError("Facebook login appears unsuccessful, check credentials or checkpoint prompts") def build_context(playwright) -> BrowserContext: SESSION_PATH.parent.mkdir(parents=True, exist_ok=True) launch_args = [ "--disable-blink-features=AutomationControlled", "--no-sandbox", "--disable-dev-shm-usage", ] browser = playwright.chromium.launch(headless=False, args=launch_args) context = browser.new_context( user_agent=MAC_CHROME_UA, viewport={"width": 1440, "height": 960}, locale="en-US", timezone_id="America/Chicago", storage_state=str(SESSION_PATH) if SESSION_PATH.exists() else None, ) context.add_init_script( """ Object.defineProperty(navigator, 'webdriver', {get: () => undefined}); window.chrome = { runtime: {} }; Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3]}); Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']}); """ ) return context def get_group_name(page: Page, group_id: str) -> str: for selector in ["h1", '[role="main"] h2', 'title']: try: if selector == 'title': title = page.title().strip() if title: return title.replace(" | Facebook", "") else: text = safe_text(page.locator(selector).first) if text: return text except Exception: continue return group_id def normalize_profile_url(href: str) -> str: if not href: return "" if href.startswith("http"): return href return f"https://www.facebook.com{href}" def parse_request_time(text: str) -> str: match = re.search(r"Requested[^\n.]*", text, re.I) return match.group(0).strip() if match else "" def parse_join_date(text: str) -> str: match = re.search(r"Joined Facebook[^\n.]*", text, re.I) return match.group(0).strip() if match else "" def collect_cards(page: Page): for selector in CARD_SELECTORS: loc = page.locator(selector) try: count = loc.count() if count: return [loc.nth(i) for i in range(count)] except Exception: continue return [] def approve_member(card, group_id: str) -> bool: """Click the Approve button on a member request card.""" selectors = [ 'div[aria-label="Approve"]', 'button[aria-label="Approve"]', 'div[aria-label="Confirm"]', 'button:has-text("Approve")', 'div[role="button"]:has-text("Approve")', 'div[role="button"]:has-text("Confirm")', ] for selector in selectors: try: btn = card.locator(selector).first if btn.count() > 0: btn.scroll_into_view_if_needed(timeout=2000) btn.click(timeout=3000) random_delay(0.8, 2.0) log.info("Approved member in group %s (selector: %s)", group_id, selector) return True except Exception: continue log.warning("Could not find Approve button in group %s", group_id) return False def scrape_group(page: Page, group_id: str) -> tuple[str, list[dict[str, Any]]]: # Try member-requests first, fall back to participant_requests for url_suffix in ["member-requests", "participant_requests"]: url = f"https://www.facebook.com/groups/{group_id}/{url_suffix}" page.goto(url, wait_until="domcontentloaded", timeout=PAGE_LOAD_TIMEOUT_MS) random_delay(2, 4) page.wait_for_timeout(random.randint(1000, 2500)) if detect_logged_out(page): raise RuntimeError("Facebook session is logged out or blocked") body = page.locator("body").inner_text(timeout=3000)[:2000].lower() if "isn't available" not in body and "not available" not in body: break log.info("URL %s not available, trying alternate for group %s", url_suffix, group_id) group_name = get_group_name(page, group_id) cards = collect_cards(page) log.info("Found %s candidate cards for group %s", len(cards), group_id) scraped: list[dict[str, Any]] = [] for card in cards: try: card_text = safe_text(card) if not card_text or "Requested" not in card_text: continue name, href = find_first_text(card, NAME_SELECTORS) if not name: continue questions = extract_questions(card_text) lead = { "name": name, "profile_url": normalize_profile_url(href), "request_time": parse_request_time(card_text), "join_date": parse_join_date(card_text), "mutual_friends": parse_mutual_friends(card_text), "questions": questions, "email": extract_email(questions), "group_id": group_id, "group_name": group_name, } scraped.append(lead) # Auto-approve after capturing data approve_member(card, group_id) random_delay(1.0, 2.5) except Exception as exc: log.warning("Card parse error in group %s: %s", group_id, exc) return group_name, scraped def persist_session(context: BrowserContext) -> None: try: context.storage_state(path=str(SESSION_PATH)) except Exception as exc: log.warning("Failed to persist session: %s", exc) def process_group(page: Page, group_id: str, captured_today: dict[str, int]) -> tuple[bool, str]: try: group_name, leads = scrape_group(page, group_id) success_count = 0 for lead in leads: random_delay(1, 2) if post_json(INGEST_URL, lead): success_count += 1 captured_today[group_id] = captured_today.get(group_id, 0) + success_count post_json( HEARTBEAT_URL, { "group_id": group_id, "group_name": group_name, "status": "alive", "captured_today": captured_today[group_id], "device_name": platform.node() or "Automation Device", }, ) log.info("Group %s (%s): scraped=%s ingested=%s", group_id, group_name, len(leads), success_count) return True, group_name except Exception as exc: log.exception("Group scrape failed for %s: %s", group_id, exc) return False, group_id def run() -> int: creds = ensure_credentials_file() groups = [str(g).strip() for g in creds.get("groups", []) if str(g).strip()] if not groups: log.warning("No Facebook group IDs configured in %s", CREDENTIALS_PATH) captured_today: dict[str, int] = {} last_success_at: datetime | None = None last_alert_at: datetime | None = None last_heartbeat_at: datetime | None = None today_key = datetime.now().strftime("%Y-%m-%d") with sync_playwright() as playwright: context = build_context(playwright) page = context.new_page() page.set_default_timeout(PAGE_LOAD_TIMEOUT_MS) while True: now = datetime.now(timezone.utc) current_day = datetime.now().strftime("%Y-%m-%d") if current_day != today_key: captured_today = {} today_key = current_day try: login_if_needed(page, creds) persist_session(context) except Exception as exc: log.exception("Login/setup failed: %s", exc) any_group_success = False latest_group_name = "" for group_id in groups: random_delay() ok, group_name = process_group(page, group_id, captured_today) latest_group_name = group_name or latest_group_name if ok: any_group_success = True last_success_at = datetime.now(timezone.utc) persist_session(context) heartbeat_due = ( last_heartbeat_at is None or (now - last_heartbeat_at).total_seconds() >= HEARTBEAT_INTERVAL_SECONDS ) if heartbeat_due and groups: for group_id in groups: post_json( HEARTBEAT_URL, { "group_id": group_id, "group_name": latest_group_name or group_id, "status": "alive", "captured_today": captured_today.get(group_id, 0), "device_name": platform.node() or "Automation Device", }, ) last_heartbeat_at = now if not any_group_success and last_success_at and (now - last_success_at).total_seconds() >= ALERT_AFTER_SECONDS: if last_alert_at is None or (now - last_alert_at).total_seconds() >= ALERT_AFTER_SECONDS: send_telegram_alert( "Facebook Group Leads alert: no group has scraped successfully for more than 6 hours." ) last_alert_at = now sleep_until = time.time() + SCRAPE_INTERVAL_SECONDS while time.time() < sleep_until: remaining = sleep_until - time.time() chunk = min(60, max(5, int(remaining))) time.sleep(chunk) tick_now = datetime.now(timezone.utc) if groups and ( last_heartbeat_at is None or (tick_now - last_heartbeat_at).total_seconds() >= HEARTBEAT_INTERVAL_SECONDS ): for group_id in groups: post_json( HEARTBEAT_URL, { "group_id": group_id, "group_name": latest_group_name or group_id, "status": "alive", "captured_today": captured_today.get(group_id, 0), "device_name": platform.node() or "Automation Device", }, ) last_heartbeat_at = tick_now if not groups: break return 0 if __name__ == "__main__": try: raise SystemExit(run()) except KeyboardInterrupt: log.info("Stopped by user") raise SystemExit(0)