CRM_ENVIROMENT/CRM_ENVR
7
0
Fork 0
mirror of https://git.coastit.co.za/caitlin/CRM_ENVR.git synced 2026-07-10 11:15:43 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
d793604e92c670bab8ccc3f6c88ebbdf95aae3fd
CRM_ENVR/browser-use-service/main.py
T

2160 lines
100 KiB
Python
Raw Blame History

# ── Imports ──────────────────────────────────────────────────────────
import os, json, asyncio, re, shutil, sqlite3, urllib.parse, random, logging, tempfile
from datetime import datetime, timedelta
from fastapi import FastAPI, Query, Body
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from playwright.async_api import async_playwright
# ── Lazy Imports for browser-use (not always available) ──────────────
def _make_ollama(model: str | None = None, **kwargs):
from langchain_ollama import ChatOllama
llm = ChatOllama(model=model or CLASSIFY_MODEL, **kwargs)
object.__setattr__(llm, 'provider', 'ollama')
object.__setattr__(llm, 'name', 'ChatOllama')
object.__setattr__(llm, 'model_name', llm.model)
return llm
def _make_browser(**kwargs):
from browser_use import Browser
return Browser(**kwargs)
def _make_agent(task: str, llm, browser, **kwargs):
from browser_use import Agent
return Agent(task=task, llm=llm, browser=browser, **kwargs)
# ── Logging & App Setup ──────────────────────────────────────────────
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=os.getenv("CORS_ORIGINS", "http://localhost:3006,http://127.0.0.1:3006").split(","),
allow_methods=["POST"],
allow_headers=["*"],
)
PORT = int(os.getenv("PORT", "3008"))
# ── AI / Ollama Config ───────────────────────────────────────────────
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
CLASSIFY_MODEL = os.getenv("CLASSIFY_MODEL", "dolphin-llama3:8b")
# ── Env Vars (set via .env.local) ──────────────────────────────────────
# These store detected browser profile paths from the setup wizard.
FX_PROFILE = os.getenv('FX_PROFILE', '')
CHROME_PROFILE = os.getenv('CHROME_PROFILE', '')
OPERA_PROFILE = os.getenv('OPERA_PROFILE', '')
EDGE_PROFILE = os.getenv('EDGE_PROFILE', '')
# SELECTED_BROWSER determines which browser the scrape dispatcher tries first.
# Values: "firefox", "chrome", "opera", "edge", or "" (auto-detect all).
SELECTED_BROWSER = os.getenv('SELECTED_BROWSER', '')
# ── Chromium Launch Args ──────────────────────────────────────────────
# These flags reduce detectable automation signals in Chrome/Edge/Opera.
# They disable background services, sync, updates, throttling, popups, etc.
CHROME_LAUNCH_ARGS = [
'--headless=new',
'--disable-blink-features=AutomationControlled',
'--disable-background-networking',
'--no-first-run',
'--disable-sync',
'--disable-field-trial-config',
'--disable-background-timer-throttling',
'--disable-backgrounding-occluded-windows',
'--disable-breakpad',
'--disable-component-update',
'--disable-default-apps',
'--disable-dev-shm-usage',
'--disable-popup-blocking',
'--disable-prompt-on-repost',
'--disable-renderer-backgrounding',
'--metrics-recording-only',
'--no-default-browser-check',
'--disable-extensions-http-throttling',
'--disable-hang-monitor',
'--disable-ipc-flooding-protection',
]
# ── Profile Detection ────────────────────────────────────────────────
# All find_*_profile() functions return the parent directory (User Data or equivalent),
# NOT the "Default" subdirectory. This is important because copy_chrome_profile()
# appends "Default" internally, so returning the full Default path would cause
# duplicated path segments (e.g. ".../User Data/Default/Default").
def detect_browser_from_profile(profile_path: str) -> str | None:
"""
Infer browser type from the profile path string.
Checks for keywords like "firefox", "opera", "edge", "chrome" in the path.
Returns "firefox", "chrome", "opera", "edge", or None.
"""
if not profile_path:
return None
p = profile_path.lower().replace('\\', '/')
if 'firefox' in p or '.default' in p or '.dev-edition' in p:
return 'firefox'
if 'opera' in p:
return 'opera'
if 'edge' in p:
return 'edge'
if 'chrome' in p or 'chromium' in p:
return 'chrome'
return None
def find_firefox_profile() -> str | None:
"""Auto-detect Firefox profile directory cross-platform.
Scans the Profiles dir and returns the first .default or .dev-edition folder found,
preferring "default-release" profiles."""
import sys
home = os.path.expanduser("~")
candidates = []
if sys.platform == "win32":
appdata = os.environ.get("APPDATA", "")
if appdata:
profiles_dir = os.path.join(appdata, "Mozilla", "Firefox", "Profiles")
else:
profiles_dir = os.path.join(home, "AppData", "Roaming", "Mozilla", "Firefox", "Profiles")
elif sys.platform == "darwin":
profiles_dir = os.path.join(home, "Library", "Application Support", "Firefox", "Profiles")
else:
profiles_dir = os.path.join(home, ".mozilla", "firefox")
if os.path.isdir(profiles_dir):
for entry in os.listdir(profiles_dir):
full = os.path.join(profiles_dir, entry)
if os.path.isdir(full) and (".default" in entry.lower() or ".dev-edition" in entry.lower()):
candidates.append(full)
candidates.sort(key=lambda p: "default-release" in p.lower(), reverse=True)
if candidates:
logger.info("Auto-detected Firefox profile: %s", candidates[0])
return candidates[0]
return None
def find_chrome_profile() -> str | None:
"""Auto-detect Chrome profile (User Data dir) cross-platform.
Returns the User Data parent dir (NOT the Default subdirectory)."""
import sys
home = os.path.expanduser("~")
if sys.platform == "win32":
base = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Google", "Chrome", "User Data")
elif sys.platform == "darwin":
base = os.path.join(home, "Library", "Application Support", "Google", "Chrome")
else:
base = os.path.join(home, ".config", "google-chrome")
if os.path.isdir(os.path.join(base, "Default")):
logger.info("Auto-detected Chrome profile: %s", base)
return base
return None
def find_opera_profile() -> str | None:
"""Auto-detect Opera profile directory cross-platform.
Opera stores its profile at the "Opera Stable" level (no "User Data/Default"
nesting like Chrome/Edge), so this returns the profile root directly."""
import sys
home = os.path.expanduser("~")
if sys.platform == "win32":
base = os.path.join(os.environ.get("APPDATA", ""), "Opera Software", "Opera Stable")
elif sys.platform == "darwin":
base = os.path.join(home, "Library", "Application Support", "com.operasoftware.Opera")
else:
base = os.path.join(home, ".config", "opera")
if os.path.isdir(base) and os.listdir(base):
logger.info("Auto-detected Opera profile: %s", base)
return base
return None
def find_edge_profile() -> str | None:
"""Auto-detect Edge profile (User Data dir) cross-platform.
Returns the User Data parent dir (NOT the Default subdirectory)."""
import sys
home = os.path.expanduser("~")
if sys.platform == "win32":
base = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Microsoft", "Edge", "User Data")
elif sys.platform == "darwin":
base = os.path.join(home, "Library", "Application Support", "Microsoft Edge")
else:
base = os.path.join(home, ".config", "microsoft-edge")
if os.path.isdir(os.path.join(base, "Default")):
logger.info("Auto-detected Edge profile: %s", base)
return base
return None
# ── Profile Copying ─────────────────────────────────────────────────
# Both functions copy essential cookies/login/storage files from the user's
# real browser profile to a temp directory. Playwright then uses these temp
# profiles via launch_persistent_context(), preserving Facebook login state
# without modifying the user's actual profile.
def copy_firefox_profile(src_path: str) -> str:
"""Copy essential Firefox profile files (cookies, localStorage, permissions) to a temp dir for Playwright.
Also creates a minimal profiles.ini so Firefox treats it as a valid profile."""
essential = ['cookies.sqlite', 'webappsstore.sqlite', 'permissions.sqlite']
dst = tempfile.mkdtemp(prefix='fb_fx_profile_')
for f_name in essential:
s = os.path.join(src_path, f_name)
if os.path.exists(s):
shutil.copy2(s, os.path.join(dst, f_name))
with open(os.path.join(dst, 'profiles.ini'), 'w') as fh:
fh.write("[Profile0]\nName=default\nIsRelative=0\nPath=.\nDefault=yes\n")
logger.info("Copied Firefox profile to %s", dst)
return dst
def copy_chrome_profile(user_data_dir: str, profile_dir: str = 'Default') -> str:
"""Copy Chromium-based profile (Chrome/Edge/Opera) to a temp dir.
Copies Cookies, Login Data, Bookmarks, Web Data, Local Storage, Session Storage, and Local State.
Args:
user_data_dir: Parent directory (e.g. ".../Chrome/User Data")
profile_dir: Subdirectory name, typically "Default" (default)
"""
essential = ['Cookies', 'Login Data', 'Bookmarks', 'Web Data']
dst = tempfile.mkdtemp(prefix='fb_chrome_udir_')
src_profile = os.path.join(user_data_dir, profile_dir)
os.makedirs(dst, exist_ok=True)
os.makedirs(os.path.join(dst, profile_dir), exist_ok=True)
dst_profile = os.path.join(dst, profile_dir)
for f_name in essential:
s = os.path.join(src_profile, f_name)
if os.path.exists(s):
shutil.copy2(s, os.path.join(dst_profile, f_name))
for sub in ['Local Storage', 'Session Storage']:
s = os.path.join(src_profile, sub)
if os.path.isdir(s):
shutil.copytree(s, os.path.join(dst_profile, sub), dirs_exist_ok=True)
local_state_src = os.path.join(user_data_dir, 'Local State')
local_state_dst = os.path.join(dst, 'Local State')
if os.path.exists(local_state_src):
shutil.copy2(local_state_src, local_state_dst)
logger.info("Copied Chrome profile from %s to %s", src_profile, dst)
return dst
def ensure_ublock_extension() -> str | None:
"""Download and extract uBlock Origin extension for ad blocking via Chromium.
Cached after first download to avoid re-downloading on every scrape."""
ext_dir = os.path.join(tempfile.gettempdir(), 'fb_ublock_origin')
manifest = os.path.join(ext_dir, 'manifest.json')
if os.path.exists(manifest):
return ext_dir
import urllib.request, zipfile
crx_url = ('https://clients2.google.com/service/update2/crx'
'?response=redirect&prodversion=133&acceptformat=crx3'
'&x=id%3Dddkjiahejlhfcafbddmgiahcphecmpfh%26uc')
crx_path = os.path.join(tempfile.gettempdir(), 'ublock_origin.crx')
try:
if not os.path.exists(crx_path):
logger.info("Downloading uBlock Origin extension...")
urllib.request.urlretrieve(crx_url, crx_path)
with zipfile.ZipFile(crx_path, 'r') as z:
z.extractall(ext_dir)
logger.info("uBlock Origin extension ready at %s", ext_dir)
return ext_dir
except Exception as e:
logger.warning("Failed to load uBlock Origin extension: %s", e)
return None
# ── Agent Output Parsing ──────────────────────────────────────────────
# When the browser-use Agent fallback is used, it returns natural language
# output containing post listings. This function tries to extract structured
# data from that output, first attempting JSON parsing, then falling back to
# a line-by-line heuristic (numbered or bulleted items).
def extract_agent_posts(agent_result: str, page_text: str) -> list[dict]:
"""Parse posts from agent output or page text fallback."""
posts = []
json_match = re.search(r'\[.*?\]', agent_result, re.DOTALL)
if json_match:
try:
parsed = json.loads(json_match.group())
if isinstance(parsed, list):
for item in parsed:
if isinstance(item, dict) and item.get('content'):
posts.append({
'content': item.get('content', ''),
'author': item.get('author', '') or item.get('publisher', '') or '',
'url': item.get('url', ''),
'date': item.get('date', ''),
})
except json.JSONDecodeError:
pass
if not posts:
lines = [l.strip() for l in agent_result.split('\n') if l.strip()]
cur = []
for l in lines:
if l.startswith(('1.', '2.', '3.', '4.', '5.', '6.', '7.', '8.', '9.', '- ', '* ')):
if cur:
combined = ' '.join(cur)
if len(combined) > 30:
posts.append({'content': combined[:500], 'author': '', 'url': '', 'date': ''})
cur = []
cur.append(l)
if cur:
combined = ' '.join(cur)
if len(combined) > 30:
posts.append({'content': combined[:500], 'author': '', 'url': '', 'date': ''})
return posts
# ── Keyword Filters for Lead Detection ──────────────────────────────
# BROAD_KEYWORDS — catch any post that might relate to web development needs.
# OFFER_PATTERNS — regex patterns matching service offers (which we ignore).
# REQUEST_PATTERNS — regex patterns matching people asking for help/services.
# Together these form a two-pass filter + AI classification pipeline.
BROAD_KEYWORDS = [
"website", "web design", "web develop", "web dev",
"build my website", "build a website", "create a website",
"need web", "looking for web", "new website",
"landing page", "wordpress",
"need a website", "my website", "business website",
"need a designer", "help with my website",
"redesign", "update my website",
"looking for", "need a", "need an", "looking to",
"help me build", "create my", "build me",
"design my", "make me a", "would like",
"need someone", "hire a", "looking to hire",
"want someone", "need help with",
]
OFFER_PATTERNS = [
r"\bfree\s+domain\b",
r"\bweb\s+hosting\b",
r"\bfree\s+website\b",
r"\bfree\s+\w+\s+website\b",
r"\bprofessional\s+website",
r"\baffordable\s+web\b",
r"\bget\s+a\s+free\b",
r"\bsee\s+more\b",
r"\bbook\s+your\b",
r"\bhire\s+a\s+freelancer\b",
r"\bcontent\s+strategy\b",
r"\bdigital\s+marketing\b",
r"\bseo\s+service",
r"\bsocial\s+media\b",
r"\blimited\s+time\s+offer\b",
r"\bspecial\s+offer\b",
r"\bsign\s+up\s+now\b",
r"\br\d[\d,.]",
r"\bstarting\s+at\b",
r"\bper\s+month\b",
r"\bmonthly\b",
r"\bfreelancer\s+marketplace\b",
r"\bfirst\s+order\b",
r"\bdomain\s+name\b",
r"\bregister\s+your\s+domain\b",
r"\bsponsored\b",
r"\bpromoted\b",
r"\badvertisement\b",
r"\bdo\s+you\s+(need|want)\b",
r"\bwe\s+(can|will)\s+(build|design|create|develop|do)\b",
r"\bi\s+(can|will)\s+(build|design|create|develop|do)\b",
r"\bmy\s+portfolio\b",
r"\breasonable\s+(price|pricing|rate)\b",
r"\bwhatsapp\s+(me|us|at)\b",
r"\bwhatsapp\b",
r"\blooking\s+for\s+(a\s+)?(business|client|customer)\b",
r"\bhelp\s+your\s+business\b",
r"\bi\s+am\s+a\s+(web|freelance|designer|developer)\b",
r"\bcontact\s+me\b",
r"\bwe\s+(are|offer|provide)\s+(web|website|design|service)\b",
r"\btake\s+(the\s+|this\s+)?quiz\b",
r"\bhomeschool\b",
r"\byour\s+(home\s+)?tutor\b",
r"\blink\s+(in\s+)?(bio|comment|below)\b",
r"\bapply\s+now\b",
r"\bget\s+started\b",
r"\bfor\s+only\b",
r"\blow\s+(price|cost|rate)\b",
r"\bhit\s+me\s+up\b",
r"\bsend\s+me\s+(a\s+)?(message|dm|pm)\b",
r"\bi\s+do\s+(web|website|design|development)\b",
r"\bwe\s+do\s+(web|website|design|development)\b",
r"\bi'm\s+offering\b",
r"\bwe're\s+offering\b",
r"\bstarting\s+a\s+\w+\s+(service|business)\b",
r"\blooking\s+for\s+a\s+few\s+businesses?\b",
r"\blet('?)s\s+connect\b",
r"\bi\s+have\s+projects?\b",
r"\byour\s+business\b",
r"\bwordpress\s+development\b",
r"\be.?commerce\s+website\b",
r"\bwebsite\s+builder\b",
r"\bai\s+studio\b",
r"\bpage\s+speed\b",
r"\bguest\s+post",
r"\bguest\s+blog",
r"\bfor\s+sale\b",
r"\bselling\s+(my|a|the|this)\b",
r"\bpremium\b",
r"\bi'm\s+selling\b",
r"\bgroup\b",
r"\bi\s+can\s+help\b",
r"\binbox\s+me\b",
r"\bpm\s+me\b",
r"\bdm\s+me\b",
r"\bmessage\s+me\s+for\b",
r"\bquote\b",
r"\bdiscount\b",
r"\bbest\s+(deal|price|offer|service)\b",
r"\bprice\s+(start|begin|include|range)\b",
r"\breach\s+out\b",
r"\bclick\s+(the\s+)?link\b",
r"\bcheck\s+(out|this|my)\b",
r"\bwebsite\s+for\s+your\b",
r"\bprobono\b",
r"\bfreelance\s+opportunit",
r"\bfull.?stack\b",
r"\bresponsive\s+(web)?sites?\b",
r"\blooking\s+for\s+(new\s+)?(projects?|work)\b",
r"\bmern\s+stack\b",
r"\bexpress\s+js\b",
r"\breact\s+js\b",
r"\bnode\s+js\b",
r"\bwebsite\s+for\s+\$\b",
r"\bi\s+build\s+(websites?|shopify|wordpress)\b",
r"\bwe\s+build\s+(websites?|shopify|wordpress)\b",
r"\bi\s+create\s+(websites?|shopify|wordpress)\b",
r"\bwe\s+create\s+(websites?|shopify|wordpress)\b",
r"\bfull.?time\s+(position|job|role|work)\b",
r"\bremote\s+(position|job|role|work)\b",
r"\bhelp\s+wanted\b",
r"\bjob\s+(opening|vacancy|opportunity)\b",
r"\bnow\s+hiring\b",
r"\bpart.?time\s+(position|job|role|work)\b",
r"\byears?\s+of\s+(teaching|experience)\b",
r"\bsend\s+(you|me)\s+(the\s+)?link\b",
r"\bcomment\s+.*\bsend\b",
r"\bfor\s+free\b",
r"\bmake\s+money\b",
r"\bno\s+coding\b",
]
REQUEST_PATTERNS = [
r"\bi\s+need\b",
r"\bi\s+need\s+someone\b",
r"\bi\s+need\s+a\b",
r"\bi\s+need\s+an\b",
r"\bi\s+want\b",
r"\bi\s+would\s+like\b",
r"\bi'm\s+looking\b",
r"\bi\s+am\s+looking\b",
r"\blooking\s+for\b",
r"\blooking\s+to\b",
r"\bwho\s+can\b",
r"\bcan\s+someone\b",
r"\bhelp\s+me\b",
r"\bneed\s+help\b",
r"\bwould\s+like\s+to\b",
r"\bwant\s+to\b",
r"\banyone\s+know\b",
r"\bdoes\s+anyone\b",
r"\brecommend\b",
r"\bsuggestion\b",
r"\bquote\s+(please|for|to)\b",
]
# ── Anti-Detection Scripts ────────────────────────────────────────────
# These JS scripts are injected into every page via add_init_script().
# They override navigator properties (webdriver, userAgent, plugins, etc.)
# and WebGL/canvas/audio fingerprints to make Playwright harder to detect.
# Firefox and Chromium variants differ slightly (e.g. Chrome needs the
# "window.chrome" object; Firefox must NOT have it).
def firefox_init_script() -> str:
return r"""// Firefox Anti-Detection
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
Object.defineProperty(navigator, 'platform', { get: () => 'Win32' });
Object.defineProperty(navigator, 'hardwareConcurrency', { get: () => 8 });
Object.defineProperty(navigator, 'maxTouchPoints', { get: () => 0 });
Object.defineProperty(navigator, 'oscpu', { get: () => 'Windows NT 10.0; Win64; x64' });
Object.defineProperty(navigator, 'vendor', { get: () => '' });
Object.defineProperty(navigator, 'vendorSub', { get: () => '' });
Object.defineProperty(navigator, 'productSub', { get: () => '20100101' });
Object.defineProperty(navigator, 'product', { get: () => 'Gecko' });
Object.defineProperty(navigator, 'appCodeName', { get: () => 'Mozilla' });
Object.defineProperty(navigator, 'appName', { get: () => 'Netscape' });
Object.defineProperty(navigator, 'appVersion', { get: () => '5.0 (Windows NT 10.0; Win64; x64; rv:130.0) Gecko/20100101 Firefox/130.0' });
Object.defineProperty(navigator, 'userAgent', { get: () => 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:130.0) Gecko/20100101 Firefox/130.0' });
Object.defineProperty(navigator, 'buildID', { get: () => '20250101000000' });
if (window.chrome !== undefined) { window.chrome = undefined; }
if (navigator.permissions) {
const origQ = navigator.permissions.query.bind(navigator.permissions);
navigator.permissions.query = (p) => origQ(p).then(r => (['camera','microphone','geolocation','notifications'].includes(p.name) ? Object.assign(r, { state: 'prompt' }) : r));
}
const _gEC = HTMLCanvasElement.prototype.getContext;
HTMLCanvasElement.prototype.getContext = function(t, ...a) {
const ctx = _gEC.call(this, t, ...a);
if (ctx && (t === 'webgl' || t === 'webgl2')) {
const _gP = ctx.getParameter.bind(ctx);
ctx.getParameter = function(p) { if (p===37445) return 'Intel Inc.'; if (p===37446) return 'Intel Iris OpenGL Engine'; if (p===7936||p===7937||p===35724) return ''; return _gP(p); };
const _gE = ctx.getExtension.bind(ctx);
ctx.getExtension = function(n) { if (n==='WEBGL_debug_renderer_info') return null; return _gE(n); };
}
return ctx;
};
const _tDU = HTMLCanvasElement.prototype.toDataURL;
HTMLCanvasElement.prototype.toDataURL = function(...a) {
const img = _tDU.apply(this, a);
if (img.includes('image/png') && img.length > 5000) { const n = btoa(String.fromCharCode(Math.floor(Math.random()*256))); return img.slice(0,-100)+n+img.slice(-100); }
return img;
};
const _gFD = AnalyserNode.prototype.getFloatFrequencyData;
if (_gFD) { AnalyserNode.prototype.getFloatFrequencyData = function(a) { _gFD.call(this,a); for(let i=0;i<a.length;i++) a[i]+=(Math.random()-0.5)*0.02; }; }
Object.defineProperty(screen, 'colorDepth', { get: () => 24 });
Object.defineProperty(screen, 'pixelDepth', { get: () => 24 });
Object.defineProperty(navigator, 'plugins', { get: () => [{name:'Widevine Content Decryption Module',filename:'widevinecdm.dll',length:1,description:'Enables Widevine licenses'},{name:'OpenH264 Video Codec',filename:'openh264.dll',length:1,description:'H.264 video codec'}] });
Object.defineProperty(navigator, 'mimeTypes', { get: () => [{type:'application/pdf',suffixes:'pdf',description:'Portable Document Format',enabledPlugin:navigator.plugins[0]}] });"""
def chromium_init_script() -> str:
return r"""// Enhanced Chromium Anti-Detection
Object.defineProperty(navigator, 'webdriver', { get: () => false });
Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
Object.defineProperty(navigator, 'platform', { get: () => 'Win32' });
Object.defineProperty(navigator, 'hardwareConcurrency', { get: () => 8 });
Object.defineProperty(navigator, 'deviceMemory', { get: () => 8 });
Object.defineProperty(navigator, 'maxTouchPoints', { get: () => 0 });
Object.defineProperty(navigator, 'oscpu', { get: () => 'Windows NT 10.0; Win64; x64' });
Object.defineProperty(navigator, 'vendor', { get: () => 'Google Inc.' });
Object.defineProperty(navigator, 'vendorSub', { get: () => '' });
Object.defineProperty(navigator, 'productSub', { get: () => '20030107' });
Object.defineProperty(navigator, 'product', { get: () => 'Gecko' });
Object.defineProperty(navigator, 'appCodeName', { get: () => 'Mozilla' });
Object.defineProperty(navigator, 'appName', { get: () => 'Netscape' });
Object.defineProperty(navigator, 'appVersion', { get: () => '5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36' });
Object.defineProperty(navigator, 'userAgent', { get: () => 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36' });
window.chrome = { runtime: { onConnect: { addListener: () => {} }, onMessage: { addListener: () => {} }, sendMessage: () => {} }, app: { isInstalled: false, InstallState: { DISABLED: 'disabled', INSTALLED: 'installed', NOT_INSTALLED: 'not_installed' }, getDetails: () => null, getIsInstalled: () => false }, csi: () => {}, loadTimes: () => {} };
if (navigator.permissions) {
const origQ = navigator.permissions.query.bind(navigator.permissions);
navigator.permissions.query = (p) => origQ(p).then(r => (['camera','microphone','geolocation','notifications'].includes(p.name) ? Object.assign(r, { state: 'prompt' }) : r));
}
const _gEC = HTMLCanvasElement.prototype.getContext;
HTMLCanvasElement.prototype.getContext = function(t, ...a) {
const ctx = _gEC.call(this, t, ...a);
if (ctx && (t === 'webgl' || t === 'webgl2')) {
const _gP = ctx.getParameter.bind(ctx);
ctx.getParameter = function(p) { if (p===37445) return 'Intel Inc.'; if (p===37446) return 'Intel Iris OpenGL Engine'; if (p===7936||p===7937||p===35724) return ''; return _gP(p); };
const _gE = ctx.getExtension.bind(ctx);
ctx.getExtension = function(n) { if (n==='WEBGL_debug_renderer_info') return null; return _gE(n); };
}
return ctx;
};
const _tDU = HTMLCanvasElement.prototype.toDataURL;
HTMLCanvasElement.prototype.toDataURL = function(...a) {
const img = _tDU.apply(this, a);
if (img.includes('image/png') && img.length > 5000) { const n = btoa(String.fromCharCode(Math.floor(Math.random()*256))); return img.slice(0,-100)+n+img.slice(-100); }
return img;
};
const _gFD = AnalyserNode.prototype.getFloatFrequencyData;
if (_gFD) { AnalyserNode.prototype.getFloatFrequencyData = function(a) { _gFD.call(this,a); for(let i=0;i<a.length;i++) a[i]+=(Math.random()-0.5)*0.02; }; }
Object.defineProperty(screen, 'colorDepth', { get: () => 24 });
Object.defineProperty(screen, 'pixelDepth', { get: () => 24 });
Object.defineProperty(navigator, 'plugins', { get: () => [1,2,3,4,5].map(i => ({name:['Chrome PDF Plugin','Chrome PDF Viewer','Native Client','Widevine Content Decryption Module','Chromium PDF Viewer'][i-1],filename:'internal-pdf-viewer',length:1,description:['Portable Document Format','','Native Client Executable','Enables Widevine licenses',''][i-1]})) });
Object.defineProperty(navigator, 'mimeTypes', { get: () => [{type:'application/pdf',suffixes:'pdf',description:'Portable Document Format',enabledPlugin:navigator.plugins[0]},{type:'text/pdf',suffixes:'pdf',description:'Portable Document Format',enabledPlugin:navigator.plugins[0]}] });"""
# ── Text Classification Utilities ────────────────────────────────────
def kw_match(text: str) -> bool:
t = text.lower()
return any(kw in t for kw in BROAD_KEYWORDS)
def is_request(text: str) -> bool:
"""Detect if text contains a request for services (e.g. "I need", "looking for")."""
t = text.lower()
return any(re.search(p, t) for p in REQUEST_PATTERNS)
def is_offer(text: str) -> bool:
"""Detect if text is an offer/advertisement (we skip these, not leads)."""
t = text.lower()
return any(re.search(p, t) for p in OFFER_PATTERNS)
# ── Facebook Search Queries ───────────────────────────────────────────
# These are the search terms the scraper will use on Facebook.
# They target people looking for web development / website services.
# Each run picks 2-4 random queries to vary behavior and reduce detection.
FB_SEARCHES = [
"looking for web developer",
"need someone to build my website",
"need website for my small business",
"who can build me a website",
"recommend a web developer",
"I need a website for my",
"need a site for my business",
"looking for web designer",
"help me build a website",
"need someone to design my website",
"anyone know a web developer",
"need a web developer for my project",
"need website for my startup",
"want a website for my company",
"need a new website for my",
"looking for web developer for my business",
"need ecommerce website for my",
"looking for affordable web design",
"need wordpress website",
"who can design a website for my",
"looking for someone to create a website",
"recommendations for a web designer",
"I need a website",
"I need a website built",
"need wordpress site",
"need a custom website",
"I need a new website",
"need an online store",
"need an online shop",
"need website help",
"who can help me with my website",
"looking for a website designer",
"need a shopify website",
"looking for website design",
"want to build a website for my",
]
TUTORING_SEARCHES = [
"need a tutor",
"looking for a tutor",
"tutor for my child",
"need help with homework",
"private tutor needed",
"looking for tutoring",
"need a math tutor",
"english tutor needed",
"online tutor for",
"lessons for my child",
"help my child with",
"need someone to tutor",
"looking for someone to teach",
"tutoring for my",
"need help learning",
"recommend a tutor",
"anyone know a tutor",
"need a private tutor",
"tutoring services for my child",
"need academic help",
"need a reading tutor",
"looking for piano teacher",
"need help with maths",
"need a science tutor",
"looking for language tutor",
"need programming tutor",
"coding tutor for my",
"sat tutor needed",
"exam preparation tutor",
"need a homeschool tutor",
"tutor near me for",
]
def _search_list_for_query(query: str) -> list[str]:
"""Pick the appropriate search query pool based on the search term."""
tl = query.lower()
tutoring_terms = ["tutor", "tutoring", "lessons", "homework", "teach", "learning", "child", "math", "english", "science", "exam", "homeschool", "coding", "programming", "piano", "reading"]
if any(t in tl for t in tutoring_terms):
return TUTORING_SEARCHES
return FB_SEARCHES
# ── South African Multi-Language Queries ──────────────────────────────
# 4 most spoken SA languages: English, Afrikaans, isiXhosa, isiZulu.
# Each scrape searches ALL 4 languages to catch leads across all
# language communities on Facebook.
SA_WEBSITE_QUERIES = [
"I need a website for my business", # English
"ek benodig n webwerf", # Afrikaans
"ek soek iemand om n webwerf te bou", # Afrikaans
"ndidinga iwebhusayithi yeshishini", # isiXhosa
"ndifuna umntu owakha iwebhusayithi", # isiXhosa
"ngidinga iwebhusayithi yebhizinisi", # isiZulu
"ngifuna umuntu owakha iwebhusayithi", # isiZulu
]
SA_TUTOR_QUERIES = [
"I need a tutor for my child", # English
"ek benodig n tutor vir my kind", # Afrikaans
"ek soek n privaat onderwyser", # Afrikaans
"ndifuna utitshala womntwana wam", # isiXhosa
"ndidinga umfundisi-ntsapho", # isiXhosa
"ngidinga uthisha wengane yami", # isiZulu
"ngifuna umfundisi wengane", # isiZulu
]
async def _quick_search(page, context, query: str) -> tuple:
"""Fast search — load search results page, wait for render, extract visible posts.
No scrolling or extra human-like delays. Used for non-English language queries."""
page = await _ensure_page(page, context)
url = f'https://www.facebook.com/search/posts/?q={urllib.parse.quote(query)}'
try:
await page.goto(url, wait_until='domcontentloaded', timeout=20000)
current_url = page.url
if '/login' in current_url.lower():
logger.warning("Quick search redirected to login for '%s'", query[:40])
return page, []
await page.wait_for_timeout(random.randint(4000, 6000))
await page.evaluate("window.scrollBy(0, 600)")
await page.wait_for_timeout(random.randint(2000, 4000))
await page.evaluate("window.scrollBy(0, 600)")
await page.wait_for_timeout(random.randint(2000, 3000))
raw_articles = await _get_article_elements(page)
posts = _extract_posts_from_elements(raw_articles, url) if raw_articles else []
raw = await page.evaluate('document.body.innerText')
text_posts = _extract_posts_from_text(raw, url)
existing = {(p.get('title') or p.get('content',''))[:80] for p in posts}
for tp in text_posts:
key = (tp.get('title') or tp.get('content',''))[:80]
if key not in existing:
posts.append(tp)
if posts:
try:
profiles = await page.evaluate(r'''() => {
const out = [];
const seenTxt = new Set();
for (const a of document.querySelectorAll('a[href*="/profile.php"], a[href*="/user/"], a[href*="/people/"], a[href*="/me/"]')) {
const name = (a.innerText || '').trim();
if (!name || name.length < 3 || name.length > 60) continue;
const words = name.split(' ');
if (words.length < 2 || words.length > 6) continue;
if (!/^[A-Z]/.test(name)) continue;
if (name.includes('facebook') || name.includes('/')) continue;
const cell = a.closest('div[style]') || a.parentElement;
const txt = cell ? (cell.innerText || '').substring(0, 200) : '';
if (!txt) continue;
const key2 = txt.substring(0, 80);
if (seenTxt.has(key2)) continue;
seenTxt.add(key2);
out.push({ name, textKey: key2 });
}
return out;
}''')
if profiles:
for p in posts:
pk = (p.get('content') or p.get('title') or '')[:80].strip()
if not pk: continue
for pr in profiles:
if pk[:30] in pr['textKey'] or pr['textKey'][:30] in pk:
if not p.get('author'):
p['author'] = pr['name']
break
except Exception:
pass
for p in posts:
if p.get('author'):
a = p['author']
al = a.lower()
if any(kw in al for kw in BROAD_KEYWORDS) or is_offer(a) or len(a.split()) < 2 or any(w in _NON_NAMES for w in al.split()):
p['author'] = ''
posts = [p for p in posts if not (
'/groups/' in p.get('url', '') or '/group/' in p.get('url', '')
or '/pages/' in p.get('url', '')
or ' / ' in (p.get('title') or p.get('content') or '')
)]
except Exception as e:
logger.warning("Quick search '%s' failed: %s", query, e)
return page, []
return page, posts
VIEWPORTS = [
{'width': 1280, 'height': 800},
{'width': 1366, 'height': 768},
{'width': 1440, 'height': 900},
{'width': 1536, 'height': 864},
{'width': 1920, 'height': 1080},
]
# ── Cookie Extraction ────────────────────────────────────────────────
# Reads Facebook cookies from the browser profile's SQLite database.
# Firefox stores cookies in cookies.sqlite; Chromium stores them in Cookies
# (a SQLite DB as well). This function handles Firefox's format.
# The cookies are copied to a temp file first to avoid locking issues.
async def get_fb_cookies(profile_path: str | None = None):
cookie_db_path = profile_path or FX_PROFILE
if not cookie_db_path:
logger.warning("No profile path provided")
return []
cookie_db = os.path.join(cookie_db_path, 'cookies.sqlite')
if not os.path.exists(cookie_db):
logger.warning("Cookie DB not found at %s", cookie_db)
return []
tmp = os.path.join(os.path.dirname(cookie_db), f'cookies_tmp_{os.getpid()}.sqlite')
for attempt in range(3):
try:
shutil.copy2(cookie_db, tmp)
break
except PermissionError:
if attempt < 2:
await asyncio.sleep(0.5)
else:
logger.error("Failed to copy cookie DB after 3 attempts")
return []
try:
conn = sqlite3.connect(tmp)
c = conn.cursor()
c.execute("SELECT name, value, host, path FROM moz_cookies WHERE host LIKE '%facebook.com'")
rows = c.fetchall()
conn.close()
try:
os.remove(tmp)
except Exception:
pass
return [{
"name": name, "value": value,
"domain": host if host.startswith('.') else f'.{host}',
"path": path,
"httpOnly": True, "secure": True, "sameSite": "Lax",
} for name, value, host, path in rows]
except Exception as e:
logger.error("Cookie DB read error: %s", e)
try:
os.remove(tmp)
except Exception:
pass
return []
# ── Date Parsing ─────────────────────────────────────────────────────
# Facebook displays dates in relative format ("2h ago", "Yesterday", "Monday")
# or various absolute formats. These functions normalize them to YYYY-MM-DD.
WEEKDAY_ORDER = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday']
def _parse_fb_date(block: list[str]) -> str:
for l in block:
l = l.strip()
m = re.match(r'(\d+)\s*(h|hr|hrs|hour|hours)\s*ago', l)
if m:
hours = int(m.group(1))
d = datetime.now() - timedelta(hours=hours)
return d.strftime('%Y-%m-%d')
if l.lower() == 'yesterday':
d = datetime.now() - timedelta(days=1)
return d.strftime('%Y-%m-%d')
low = l.lower()
if low in WEEKDAY_ORDER:
day_idx = WEEKDAY_ORDER.index(low)
today_idx = datetime.now().weekday()
days_ago = (today_idx - day_idx) % 7
d = datetime.now() - timedelta(days=days_ago)
return d.strftime('%Y-%m-%d')
for fmt in ['%Y-%m-%d', '%d %B %Y', '%B %d %Y', '%d %b %Y', '%b %d %Y', '%d %b', '%b %d']:
try:
dt = datetime.strptime(l, fmt)
return dt.strftime('%Y-%m-%d')
except ValueError:
pass
return datetime.now().strftime('%Y-%m-%d')
def _is_within_days(date_str: str, max_days: int = 3) -> bool:
"""Check if date is within max_days from now. Empty/unparseable = keep."""
if not date_str:
return True
try:
dt = datetime.strptime(date_str.strip(), '%Y-%m-%d')
return (datetime.now() - dt).days <= max_days
except ValueError:
return True
def _clean_fb_text(text: str) -> str:
cleaned_lines = []
for l in text.split('\n'):
stripped = l.strip()
if not stripped:
continue
if len(stripped) < 3:
continue
if all(not c.isalpha() for c in stripped):
continue
cleaned_lines.append(stripped)
return '\n'.join(cleaned_lines)
# Words that should never be treated as person names
_NON_NAMES = {"online","tutor","tutoring","school","academy","learning","urgently","looking","south","africa","philippines","australia","creative","professional","digital","services","solutions","statistics","actuarial","homeschooling","homeschool","reading","math","english","science","grade","group","page","website","design","development","agency","company","studio","graphic","technical","support","customer","guest","post","fashion","wordpress","shopify","ecommerce","business","marketing","social","media","content","strategy","seo","free","domain","hosting","sponsored","promoted","advertisement","need","want","help","please","contact","send","message","whatsapp","hire","freelance","writer","blogger","expert","specialist","consultant","freelancer","software","creators","village","town","city","university","college","institute","evolve","aesthetics","sale","selling","premium","builder","pro","people","ecommerce","press","anonymous","tuition","parents","tutors","advanced","custom","fields","speed","optimization","elementor","woocommerce","acf","availability","january","february","march","april","may","june","july","august","september","october","november","december","mums","dads","uae","sharjah","dubai","abu","dhabi","hong","kong","canada","usa","united","kingdom","aunty","piano","plus","recommendations","needed"}
# ── Post Extraction ──────────────────────────────────────────────────
# Two strategies for extracting posts from page content:
# 1. _extract_posts_from_elements — uses structured DOM data from _get_article_elements()
# 2. _extract_posts_from_text — fallback that parses raw page text line by line
# Both apply dedup (seen_texts), offer filtering, and request scoring.
_GROUP_SUFFIXES = [" - South Africa", " - Australia", " - Philippines", " PH", "Group", "help group", "info group", "public group", "private group", "group for", "community", "creators", "marketplace"]
def _extract_posts_from_elements(elements: list[dict], base_url: str) -> list[dict]:
posts = []
seen_texts = set()
now = datetime.utcnow()
for el in elements:
# Reject group posts immediately
if el.get('isGroup'):
continue
# Reject posts older than 2 days
ts = el.get('ts', 0)
if ts:
age_seconds = (now.timestamp() * 1000 - ts) / 1000
if age_seconds > 172800: # 2 days
continue
raw_text = el.get('text', '')
if len(raw_text) < 40:
continue
text = _clean_fb_text(raw_text)
if len(text) < 40:
continue
if is_offer(text):
continue
lines = text.split('\n')
dekey = text[:80]
if dekey in seen_texts:
continue
seen_texts.add(dekey)
# Reject if text contains group-like patterns
lower = text.lower()
if ' / ' in lower:
continue
skip = False
for suff in _GROUP_SUFFIXES:
if suff.lower() in lower:
skip = True
break
if skip:
continue
request_score = 2 if is_request(text) else 0
post_url = el.get('url') or base_url
# Prefer JS-extracted date, fall back to text parsing
js_date = (el.get('date') or '').strip()
if js_date:
try:
dt = datetime.fromisoformat(js_date.replace('Z', '+00:00'))
date_str = dt.strftime('%Y-%m-%d')
except (ValueError, TypeError):
date_str = _parse_fb_date([js_date])
else:
date_str = _parse_fb_date(lines)
posts.append({
"title": text[:300],
"content": text[:1000],
"author": el.get('author', ''),
"url": post_url,
"source": "facebook",
"date": date_str,
"_score": request_score,
})
posts.sort(key=lambda p: p.get('_score', 0), reverse=True)
for p in posts:
p.pop('_score', None)
return posts
def _extract_posts_from_text(raw: str, url: str) -> list[dict]:
lines = [l.strip() for l in raw.split('\n') if l.strip() and len(l.strip()) >= 15]
posts = []
seen_texts = set()
cur = []
def _find_author(candidate_lines: list[str]) -> str:
for line in reversed(candidate_lines):
ln = line.strip()
if not ln or len(ln) > 60 or len(ln) < 3:
continue
words = ln.split()
if not (1 < len(words) < 8):
continue
if not all(w[0].isalpha() and w[0].isupper() for w in words if w):
continue
lower_ln = ln.lower()
if any(kw in lower_ln for kw in BROAD_KEYWORDS) or is_offer(ln):
continue
if 'http' in ln or '/' in ln or '@' in ln:
continue
return ln
return ''
for l in lines:
words = re.findall(r'[A-Za-z]{2,}', l)
if len(words) < 2:
continue
lower = l.lower()
if any(kw in lower for kw in BROAD_KEYWORDS) and not is_offer(l):
if cur:
snippet = ' '.join(cur[-3:] + [l])
if len(snippet) >= 40 and not is_offer(snippet):
dekey = snippet[:80]
if dekey not in seen_texts:
seen_texts.add(dekey)
posts.append({
"title": snippet[:300],
"content": snippet[:1000],
"author": _find_author(cur[-3:]),
"url": url,
"source": "facebook",
"date": _parse_fb_date(cur + [l]),
})
cur = []
else:
if not is_offer(l):
snippet = l
if len(snippet) >= 40:
dekey = snippet[:80]
if dekey not in seen_texts:
seen_texts.add(dekey)
posts.append({
"title": snippet[:300],
"content": snippet[:1000],
"author": '',
"url": url,
"source": "facebook",
"date": _parse_fb_date([l]),
})
cur.append(l)
if len(cur) > 10:
cur.pop(0)
# Post-process: scan each post's text for inline names
def _scan_inline_name(text: str) -> str:
for m in re.finditer(r'([A-Z][a-z]{2,}(?:\s+[A-Z][a-z]{2,}){1,3})[\s,;:]', text):
cand = m.group(1).strip()
if cand and 3 <= len(cand) <= 60:
lower = cand.lower()
words = lower.split()
if not any(kw in lower for kw in BROAD_KEYWORDS) and not is_offer(cand):
if not any(w in _NON_NAMES for w in words):
return cand
return ''
for p in posts:
if not p.get('author'):
txt = p.get('content') or p.get('title') or ''
name = _scan_inline_name(txt)
if name:
p['author'] = name
# Remove group posts: any post whose text looks like it came from a group
filtered = []
for p in posts:
t = (p.get('title') or p.get('content') or '')
if ' / ' in t:
continue
lower = t.lower()
skip = False
for suff in _GROUP_SUFFIXES:
if suff.lower() in lower:
skip = True
break
if skip:
continue
filtered.append(p)
# Dedup: merge posts where one is a suffix of another (same post split)
merged = []
for p in filtered:
t = (p.get('title') or p.get('content') or '').lower()
# Check if this post is substantially contained in an already-merged post
found = False
for i, m in enumerate(merged):
mt = (m.get('title') or m.get('content') or '').lower()
# Word-level overlap: count shared words
twords = set(t.split())
mwords = set(mt.split())
if len(twords) > 3 and len(mwords) > 3:
overlap = len(twords & mwords)
smaller = min(len(twords), len(mwords))
if overlap / smaller >= 0.6:
# Keep the longer one
if len(t) > len(mt):
merged[i] = p
found = True
break
if not found:
merged.append(p)
return merged
# ── Human-like Behavior Simulation ──────────────────────────────────
# These functions add random delays, mouse movements, and scroll patterns
# to make automated browsing look more like a real human user.
# This is critical for avoiding Facebook's bot detection.
async def human_scroll(page, steps: int = None, total_delay: float = None):
steps = steps or random.randint(2, 5)
total_delay = total_delay or random.uniform(6, 18)
step_delay = total_delay / steps
for i in range(steps):
scroll_dist = random.randint(200, 600)
await page.evaluate(f"window.scrollBy(0, {scroll_dist})")
try:
vp = await page.evaluate('({w: window.innerWidth, h: window.innerHeight})')
x = random.randint(100, vp['w'] - 100)
y = random.randint(100, vp['h'] - 100)
await page.mouse.move(x, y, steps=random.randint(15, 35))
except Exception:
pass
await page.wait_for_timeout(int(step_delay * 1000))
if random.random() < 0.3:
await page.evaluate(f"window.scrollBy(0, -{random.randint(100, 300)})")
await page.wait_for_timeout(random.randint(1000, 3000))
async def random_idle(page):
action = random.random()
if action < 0.15:
try:
elems = await page.query_selector_all('a, span, div[role="button"]')
if elems:
el = random.choice(elems[:20])
box = await el.bounding_box()
if box:
x = box['x'] + box['width'] / 2 + random.randint(-20, 20)
y = box['y'] + box['height'] / 2 + random.randint(-10, 10)
await page.mouse.move(x, y, steps=random.randint(20, 40))
await page.wait_for_timeout(random.randint(500, 2000))
except Exception:
pass
# ── Facebook DOM Parsing ─────────────────────────────────────────────
# This JS function runs inside the page context to extract structured post
# data (text, author, URL, date) from Facebook's complex DOM structure.
# It tries multiple selectors (article, feed, etc.) and uses the first
# one that returns results, since Facebook's DOM changes frequently.
async def _get_article_elements(page) -> list[dict]:
return await page.evaluate('''() => {
const results = [];
const seenTexts = new Set();
const terms = ["website","web design","web develop","need a","looking for","build my","create a","wordpress","landing page","ecommerce","tutor","tutoring","homeschool","math","reading","english","science","grade"];
const now = Date.now();
const DAY_MS = 86400000;
const anchors = document.querySelectorAll('a[href*="/posts/"], a[href*="/story.php"], a[href*="/photo.php"], a[href*="/watch"], a[href*="/reel/"], a[href*="/permalink/"]');
for (const a of anchors) {
const h = a.getAttribute('href') || '';
if (!h) continue;
const cell = a.closest('div[style]') || a.parentElement;
if (!cell) continue;
const txt = (cell.innerText || '').trim();
if (txt.length < 40) continue;
const lower = txt.toLowerCase();
let matched = false;
for (const t of terms) { if (lower.includes(t)) { matched = true; break; } }
if (!matched) continue;
const key = txt.substring(0, 80);
if (seenTexts.has(key)) continue;
seenTexts.add(key);
const isGroup = h.includes('/groups/');
let author = '';
const nameEl = cell.querySelector('h4, strong, a[role="link"], span[dir="auto"]');
if (nameEl) author = (nameEl.innerText || '').trim();
if (!author || author.length > 60 || author.length < 2) author = '';
let postUrl = h.startsWith('http') ? h : 'https://www.facebook.com' + h;
let date = '';
let ts = 0;
const timeEl = cell.querySelector('time');
if (timeEl) {
date = timeEl.getAttribute('datetime') || timeEl.getAttribute('aria-label') || '';
const dtVal = timeEl.getAttribute('datetime');
if (dtVal) { ts = new Date(dtVal).getTime(); }
}
results.push({ text: txt, author, url: postUrl, date, isGroup, ts });
}
return results;
}''')
async def _ensure_page(page, context):
"""Check if the current page is still alive. If closed (e.g. by a popup),
create a fresh page and navigate to Facebook."""
try:
await page.evaluate('1')
return page
except Exception:
logger.warning("Page was closed mid-scrape, creating a fresh page")
page = await context.new_page()
try:
await page.goto('https://www.google.com/', wait_until='domcontentloaded', timeout=15000)
await page.wait_for_timeout(random.randint(1000, 3000))
except Exception:
logger.warning("Google navigation failed during page recreation")
await page.goto('https://www.facebook.com/', wait_until='domcontentloaded', timeout=30000)
await page.wait_for_timeout(random.randint(3000, 8000))
return page
# ── Facebook Search & Scrape ─────────────────────────────────────────
# Performs a Facebook search for a given query, scrolls through results,
# extracts posts, and returns them. Includes randomization of scroll behavior,
# mouse movements, and idle actions to mimic human browsing patterns.
async def search_facebook(page, context, query: str):
page = await _ensure_page(page, context)
url = f'https://www.facebook.com/search/posts/?q={urllib.parse.quote(query)}'
try:
await page.goto(url, wait_until='domcontentloaded', timeout=30000)
await page.wait_for_timeout(random.randint(5000, 10000))
await human_scroll(page, steps=random.randint(4, 7), total_delay=random.uniform(12, 25))
if random.random() < 0.3:
await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
await page.wait_for_timeout(random.randint(1000, 3000))
await page.evaluate("window.scrollBy(0, -300)")
await page.wait_for_timeout(random.randint(1000, 3000))
if random.random() < 0.2:
await page.evaluate("window.scrollTo(0, 0)")
await page.wait_for_timeout(random.randint(2000, 5000))
if random.random() < 0.3:
await random_idle(page)
raw_articles = await _get_article_elements(page)
posts = _extract_posts_from_elements(raw_articles, url) if raw_articles else []
raw = await page.evaluate('document.body.innerText')
text_posts = _extract_posts_from_text(raw, url)
existing = {(p.get('title') or p.get('content',''))[:80] for p in posts}
for tp in text_posts:
key = (tp.get('title') or tp.get('content',''))[:80]
if key not in existing:
posts.append(tp)
if posts:
# Extract author names from profile links in the DOM
try:
profiles = await page.evaluate(r'''() => {
const out = [];
const seenTxt = new Set();
for (const a of document.querySelectorAll('a[href*="/profile.php"], a[href*="/user/"], a[href*="/people/"], a[href*="/me/"]')) {
const name = (a.innerText || '').trim();
if (!name || name.length < 3 || name.length > 60) continue;
const words = name.split(' ');
if (words.length < 2 || words.length > 6) continue;
if (!/^[A-Z]/.test(name)) continue;
if (name.includes('facebook') || name.includes('/')) continue;
const cell = a.closest('div[style]') || a.parentElement;
const txt = cell ? (cell.innerText || '').substring(0, 200) : '';
if (!txt) continue;
const key = txt.substring(0, 80);
if (seenTxt.has(key)) continue;
seenTxt.add(key);
out.push({ name, textKey: key });
}
return out;
}''')
if profiles:
for p in posts:
pk = (p.get('content') or p.get('title') or '')[:80].strip()
if not pk: continue
for pr in profiles:
if pk[:30] in pr['textKey'] or pr['textKey'][:30] in pk:
if not p.get('author'):
p['author'] = pr['name']
break
except Exception:
pass
try:
meta = await page.evaluate(r'''() => {
const out = [];
const seenHref = new Set();
const all = document.querySelectorAll('a[href]');
for (const a of all) {
const h = a.href;
if (!h || seenHref.has(h)) continue;
const u = new URL(h);
if (!u.hostname.includes('facebook.com') && !u.hostname.includes('fb.com')) continue;
const path = u.pathname;
if (path === '/' || path === '/search/' || path === '/marketplace/' || path.startsWith('/messages/') || path.startsWith('/notifications/') || path.startsWith('/settings/')) continue;
const isPost = path.includes('/posts/') || path.includes('/videos/') || path.includes('/photos/') || path.startsWith('/story.php') || path.startsWith('/photo.php') || path.startsWith('/watch') || path.includes('/reel/') || path.startsWith('/permalink/');
if (!isPost) continue;
const cell = a.closest('[style*="position"], [role="article"], div[data-pagelet], div.x1yztbdb, div.x78zum5, div.x1n2onr6') || a.parentElement;
if (!cell) continue;
const txt = (cell.innerText || '').trim();
if (txt.length < 40) continue;
let author = '';
const nameEl = cell.querySelector('h4, strong, a[role="link"], span[dir="auto"]');
if (nameEl) author = (nameEl.innerText || '').trim();
if (!author || author.length > 60) author = '';
const key = txt.substring(0, 100);
seenHref.add(h);
out.push({ key, href: h, author, textMap: txt.substring(0,200) });
}
return out;
}''')
if meta:
for p in posts:
pk = (p.get('content') or p.get('title') or '')[:100].strip()
if not pk: continue
for m in meta:
mk = m.get('key') or ''
if pk[:40] in mk or mk[:40] in pk:
if m.get('href') and 'search/' not in m['href']:
p['url'] = m['href']
if m.get('author') and not p.get('author'):
p['author'] = m['author']
break
except Exception:
pass
# Clean up: remove group names passed off as authors
for p in posts:
if p.get('author'):
a = p['author']
al = a.lower()
if any(kw in al for kw in BROAD_KEYWORDS) or is_offer(a) or len(a.split()) < 2 or any(w in _NON_NAMES for w in al.split()):
p['author'] = ''
# Final filter: strip any remaining group posts
posts = [p for p in posts if not (
'/groups/' in p.get('url', '') or '/group/' in p.get('url', '')
or '/pages/' in p.get('url', '')
or ' / ' in (p.get('title') or p.get('content') or '')
)]
except Exception as e:
logger.warning("Facebook search '%s' failed: %s", query, e)
return page, []
return page, posts
# ── Detection Signals ────────────────────────────────────────────────
# Keywords/phrases in page URL or body text that indicate Facebook's
# security/bot detection systems have flagged the session.
DETECTION_SIGNALS = [
'/checkpoint/', '/login.php?', 'action=security_check',
'unusual activity', 'suspicious login', 'suspicious activity',
'temporarily blocked', 'temporarily restricted',
'confirm your identity', 'enter the code we sent',
'we noticed unusual', 'help us confirm',
'security check', 'challenge', 'please verify',
]
def check_detection_signals(page_url: str, page_text: str = '') -> str | None:
"""Check page for Facebook detection signals. Returns signal text or None."""
combined = (page_url + ' ' + page_text).lower()
for signal in DETECTION_SIGNALS:
if signal in combined:
return signal
return None
def cleanup_chrome():
"""Kill orphaned Chrome headless shell processes. These sometimes linger
after failed scrapes and interfere with subsequent runs."""
import subprocess, signal
try:
subprocess.run(["taskkill", "/F", "/IM", "chrome-headless-shell.exe"], capture_output=True, timeout=5)
except Exception:
pass
# ── Main Scrape Dispatcher ────────────────────────────────────────────
# scrape_facebook() is the main entry point. It:
# 1. Resolves the browser profile path (from SELECTED_BROWSER env var or auto-detect)
# 2. Dispatches to the correct browser-specific scraper
# 3. If the browser scrape is flagged by Facebook, falls through to the Agent fallback
#
# Priority order for browser selection:
# - SELECTED_BROWSER env var (set by setup wizard) → that browser's scraper
# - Auto-detect: Firefox → Opera → Chrome → Edge (first found wins)
# - If no profile found → Agent (browser-use + ChatOllama) fallback
async def scrape_facebook(profile_path: str | None = None, force: bool = False, query: str | None = None) -> dict:
"""Dispatcher — respect SELECTED_BROWSER, fall through on flag, then Agent."""
effective_path = profile_path or ""
browser_type = ""
# Resolve profile and browser type
if effective_path:
browser_type = detect_browser_from_profile(effective_path) or ""
elif SELECTED_BROWSER == "firefox":
effective_path = FX_PROFILE or find_firefox_profile() or ""
browser_type = "firefox"
elif SELECTED_BROWSER == "opera":
effective_path = OPERA_PROFILE or find_opera_profile() or ""
browser_type = "opera"
elif SELECTED_BROWSER == "edge":
effective_path = EDGE_PROFILE or find_edge_profile() or ""
browser_type = "edge"
elif SELECTED_BROWSER == "chrome":
effective_path = CHROME_PROFILE or find_chrome_profile() or ""
browser_type = "chrome"
else:
# Auto-detect — try all in priority order
for name, env_var, find_fn in [
("firefox", "FX_PROFILE", find_firefox_profile),
("opera", "OPERA_PROFILE", find_opera_profile),
("chrome", "CHROME_PROFILE", find_chrome_profile),
("edge", "EDGE_PROFILE", find_edge_profile),
]:
p = os.getenv(env_var, "") or find_fn() or ""
if p:
effective_path = p
browser_type = name
break
if not browser_type or not effective_path:
logger.info("No profile found — falling back to Agent")
return await _scrape_with_agent(force)
logger.info("Selected browser: %s (profile: %s)", browser_type, effective_path)
# Firefox path
if browser_type == "firefox":
result = await _scrape_with_firefox(effective_path, force, query)
if result.get("success") or not result.get("flagged"):
return result
logger.warning("Firefox flagged (%s), trying Agent", result.get("flag_reason", "unknown"))
return await _scrape_with_agent(force)
# Chromium-based (chrome / opera / edge)
result = await _scrape_with_chromium(effective_path, browser_type, force, query)
if result.get("success") or not result.get("flagged"):
return result
logger.warning("%s flagged (%s), trying Agent", browser_type, result.get("flag_reason", "unknown"))
return await _scrape_with_agent(force)
# ── Firefox Scraper ──────────────────────────────────────────────────
# Launches a headless Firefox with the user's real profile cookies (copied
# to a temp dir). Uses firefox_user_prefs to disable automation flags and
# tracking protection. Injects anti-detection script via add_init_script().
# If force=False, randomly skips 30% of runs as a decoy (looks like random
# human browsing). If flagged by Facebook, returns flagged=True for the
# dispatcher to fall through to the Agent.
async def _scrape_with_firefox(profile_path: str, force: bool, query: str | None = None) -> dict:
"""Scrape Facebook using Firefox + persistent real profile (no cookie injection)."""
if not profile_path:
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": "No profile path"}
profile_dir = None
try:
profile_dir = copy_firefox_profile(profile_path)
async with async_playwright() as pw:
context = await pw.firefox.launch_persistent_context(
user_data_dir=profile_dir,
headless=True,
firefox_user_prefs={
"dom.webdriver.enabled": False,
"dom.webdriver.timeout": 0,
"useAutomationExtension": False,
"privacy.trackingprotection.enabled": False,
"privacy.trackingprotection.fingerprinting.enabled": False,
"privacy.trackingprotection.cryptomining.enabled": False,
},
)
pages = context.pages
page = pages[0] if pages else await context.new_page()
await context.add_init_script(firefox_init_script())
try:
await page.goto('https://www.google.com/', wait_until='domcontentloaded', timeout=15000)
await page.wait_for_timeout(random.randint(1000, 3000))
except Exception:
logger.warning("Google navigation failed, trying Facebook directly")
await page.goto('https://www.facebook.com/', wait_until='domcontentloaded', timeout=30000)
await page.wait_for_timeout(random.randint(3000, 8000))
url = page.url
page_text = await page.evaluate('document.body.innerText') if '/login' in url.lower() else ''
det = check_detection_signals(url, page_text)
if det or '/login' in url.lower():
logger.warning("Facebook login page detected — flag: %s", det or "login_page")
await context.close()
return {"success": False, "leads": [], "flagged": True, "flag_reason": det or "login_page", "error": "Facebook login page detected"}
await human_scroll(page, steps=random.randint(2, 4), total_delay=random.uniform(8, 20))
if random.random() < 0.25:
await page.evaluate("window.scrollTo(0, 0)")
await page.wait_for_timeout(random.randint(2000, 5000))
await human_scroll(page, steps=random.randint(1, 2))
if not force and random.random() < 0.3:
await page.wait_for_timeout(random.randint(8000, 20000))
await context.close()
return {"success": True, "leads": [], "flagged": False, "flag_reason": None, "error": None}
all_posts = []
tutoring = False
if query:
tl = query.lower()
tutoring = any(t in tl for t in ["tutor", "tutoring", "lessons", "homework", "teach", "learning", "child"])
lang_pool = SA_TUTOR_QUERIES if tutoring else SA_WEBSITE_QUERIES
non_english = [q for q in lang_pool if q.strip().lower() != tl]
query_pool = _search_list_for_query(query)
supp_k = random.randint(3, 4) if tutoring else random.randint(2, 3)
supplement = random.sample(query_pool, k=supp_k)
searches = [query] + supplement + non_english
english_count = 1 + len(supplement)
else:
searches = random.sample(FB_SEARCHES + SA_WEBSITE_QUERIES + SA_TUTOR_QUERIES, k=random.randint(4, 7))
for i, sq in enumerate(searches):
page, posts = await search_facebook(page, context, sq) if i == 0 else await _quick_search(page, context, sq)
all_posts.extend(posts)
if not posts:
continue
if i > 0:
await page.wait_for_timeout(random.uniform(3000, 7000))
continue
if random.random() < 0.4:
await page.evaluate(f"window.scrollBy(0, {random.randint(-300, 300)})")
delay = random.uniform(8, 25)
await page.wait_for_timeout(int(delay * 1000))
if i == random.randint(0, 1) and random.random() < 0.15:
new_page = await context.new_page()
try:
await new_page.goto('https://www.facebook.com/groups/', wait_until='domcontentloaded', timeout=15000)
await new_page.wait_for_timeout(random.randint(3000, 8000))
except Exception:
pass
await new_page.close()
page = await _ensure_page(page, context)
if random.random() < 0.5:
await page.wait_for_timeout(random.randint(3000, 10000))
await context.close()
seen = set()
deduped = []
for p in all_posts:
key = p.get('content', '')[:100]
if key not in seen:
seen.add(key)
deduped.append(p)
# Filter to last 3 days only
deduped = [p for p in deduped if _is_within_days(p.get('date', ''), 7)]
leads = deduped[:20]
if leads:
leads = await classify_leads(leads, tutoring=tutoring)
return {"success": True, "leads": leads[:15], "flagged": False, "flag_reason": None, "error": None}
except Exception as e:
logger.error("Firefox scrape failed: %s", e)
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": str(e)}
finally:
if profile_dir and os.path.exists(profile_dir):
try:
shutil.rmtree(profile_dir, ignore_errors=True)
except Exception:
pass
# ── Chromium Scraper ─────────────────────────────────────────────────
# ── Chromium Scraper ─────────────────────────────────────────────────
# ── Chromium Scraper ─────────────────────────────────────────────────
# Generic scraper for Chrome/Edge/Opera. Uses the same structure as the
# Firefox scraper but with Chromium-specific launch config:
# - Chrome: channel="chrome"
# - Edge: channel="msedge"
# - Opera: executable_path from shutil.which("opera")
# Cookies come from copy_chrome_profile (temp copy of user's profile data).
# Anti-detection script differs slightly from Firefox variant.
# Same decoy-skip and flagged-fallback behavior as Firefox.
async def _scrape_with_chromium(profile_path: str, browser: str, force: bool = False, query: str | None = None) -> dict:
"""Scrape Facebook using a Chromium-based browser profile (Chrome/Edge/Opera)."""
if not profile_path:
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": "No profile path"}
channel = None
executable_path = None
if browser == "chrome":
channel = "chrome"
elif browser == "edge":
channel = "msedge"
elif browser == "opera":
executable_path = shutil.which("opera") or shutil.which("Opera")
profile_dir = None
try:
profile_dir = copy_chrome_profile(profile_path)
async with async_playwright() as pw:
launch_kwargs = dict(
user_data_dir=profile_dir,
headless=True,
args=CHROME_LAUNCH_ARGS,
)
if channel:
launch_kwargs["channel"] = channel
if executable_path:
launch_kwargs["executable_path"] = executable_path
context = await pw.chromium.launch_persistent_context(**launch_kwargs)
pages = context.pages
page = pages[0] if pages else await context.new_page()
await context.add_init_script(chromium_init_script())
try:
await page.goto('https://www.google.com/', wait_until='domcontentloaded', timeout=15000)
await page.wait_for_timeout(random.randint(1000, 3000))
except Exception:
logger.warning("Google navigation failed, trying Facebook directly")
await page.goto('https://www.facebook.com/', wait_until='domcontentloaded', timeout=30000)
await page.wait_for_timeout(random.randint(3000, 8000))
url = page.url
page_text = await page.evaluate('document.body.innerText') if '/login' in url.lower() else ''
det = check_detection_signals(url, page_text)
if det or '/login' in url.lower():
logger.warning("Facebook login page detected — flag: %s", det or "login_page")
await context.close()
return {"success": False, "leads": [], "flagged": True, "flag_reason": det or "login_page", "error": "Facebook login page detected"}
await human_scroll(page, steps=random.randint(2, 4), total_delay=random.uniform(8, 20))
if random.random() < 0.25:
await page.evaluate("window.scrollTo(0, 0)")
await page.wait_for_timeout(random.randint(2000, 5000))
await human_scroll(page, steps=random.randint(1, 2))
if not force and random.random() < 0.3:
await page.wait_for_timeout(random.randint(8000, 20000))
await context.close()
return {"success": True, "leads": [], "flagged": False, "flag_reason": None, "error": None}
all_posts = []
tutoring = False
if query:
tl = query.lower()
tutoring = any(t in tl for t in ["tutor", "tutoring", "lessons", "homework", "teach", "learning", "child"])
lang_pool = SA_TUTOR_QUERIES if tutoring else SA_WEBSITE_QUERIES
non_english = [q for q in lang_pool if q.strip().lower() != tl]
query_pool = _search_list_for_query(query)
supp_k = random.randint(3, 4) if tutoring else random.randint(2, 3)
supplement = random.sample(query_pool, k=supp_k)
searches = [query] + supplement + non_english
english_count = 1 + len(supplement)
else:
searches = random.sample(FB_SEARCHES + SA_WEBSITE_QUERIES + SA_TUTOR_QUERIES, k=random.randint(4, 7))
for i, sq in enumerate(searches):
page, posts = await search_facebook(page, context, sq) if i == 0 else await _quick_search(page, context, sq)
all_posts.extend(posts)
if not posts:
continue
if i > 0:
await page.wait_for_timeout(random.uniform(3000, 7000))
continue
if random.random() < 0.4:
await page.evaluate(f"window.scrollBy(0, {random.randint(-300, 300)})")
delay = random.uniform(8, 25)
await page.wait_for_timeout(int(delay * 1000))
if i == random.randint(0, 1) and random.random() < 0.15:
new_page = await context.new_page()
try:
await new_page.goto('https://www.facebook.com/groups/', wait_until='domcontentloaded', timeout=15000)
await new_page.wait_for_timeout(random.randint(3000, 8000))
except Exception:
pass
await new_page.close()
page = await _ensure_page(page, context)
if random.random() < 0.5:
await page.wait_for_timeout(random.randint(3000, 10000))
await context.close()
seen = set()
deduped = []
for p in all_posts:
key = p.get('content', '')[:100]
if key not in seen:
seen.add(key)
deduped.append(p)
deduped = [p for p in deduped if _is_within_days(p.get('date', ''), 7)]
leads = deduped[:20]
if leads:
leads = await classify_leads(leads, tutoring=tutoring)
return {"success": True, "leads": leads[:15], "flagged": False, "flag_reason": None, "error": None}
except Exception as e:
logger.error("%s scrape failed: %s", browser, e)
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": str(e)}
finally:
if profile_dir and os.path.exists(profile_dir):
try:
shutil.rmtree(profile_dir, ignore_errors=True)
except Exception:
pass
# ── Agent Fallback ──────────────────────────────────────────────────
# When all browser-based scrapers fail/are flagged, this fallback uses
# browser-use Agent with ChatOllama (local, free) to navigate Facebook
# autonomously via GPT-style prompting. No API keys needed.
# Uses Chromium headless with the same launch args as _scrape_with_chromium.
# The Agent is prompted to extract structured post data and return JSON.
async def _scrape_with_agent(force: bool = False) -> dict:
"""Fallback scraper — browser-use Agent + ChatOllama (free/local, Chromium)."""
cleanup_chrome()
profile_dir = None
try:
user_data_dir = None
if CHROME_PROFILE:
profile_dir = copy_chrome_profile(CHROME_PROFILE)
user_data_dir = profile_dir
browser = _make_browser(
headless=True,
args=CHROME_LAUNCH_ARGS,
user_data_dir=user_data_dir,
allowed_domains=["*.facebook.com", "*.messenger.com", "www.google.com"],
)
await browser.start()
all_posts = []
pool = FB_SEARCHES + random.sample(SA_WEBSITE_QUERIES, k=min(4, len(SA_WEBSITE_QUERIES)))
for query in random.sample(pool, k=random.randint(2, 4)):
agent = _make_agent(
task=f"""You are logged into Facebook. Do the following:
1. Navigate to facebook.com and make sure you are on the homepage
2. Use the Facebook search to find: {query}
3. Scroll through the results
4. For each relevant post, extract and list:
- The publisher/author name (who posted it)
- The post text content
- The post URL (if visible)
- The post date
5. ONLY include posts from the last 3 days
6. Collect as many posts as you can (aim for 5-10 per search)
When done, return the data as a JSON list with keys: content, author, url, date.""",
llm=_make_ollama(num_ctx=32000, temperature=0.3),
browser=browser,
max_actions_per_step=3,
use_vision=False,
max_failures=6,
step_timeout=180,
)
history = await agent.run()
result = history.final_result() or ""
posts = extract_agent_posts(result, "")
all_posts.extend(posts)
if random.random() < 0.3:
await asyncio.sleep(random.randint(5, 15))
await browser.close()
seen = set()
deduped = []
for p in all_posts:
key = p.get('content', '')[:100]
if key not in seen:
seen.add(key)
deduped.append(p)
# Filter to last 3 days only
deduped = [p for p in deduped if _is_within_days(p.get('date', ''), 7)]
leads = deduped[:20]
if leads:
leads = await classify_leads(leads, tutoring=tutoring)
return {"success": True, "leads": leads[:15], "flagged": False, "flag_reason": None, "error": None}
except Exception as e:
logger.error("Agent scrape failed: %s", e)
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": str(e)}
finally:
if profile_dir and os.path.isdir(profile_dir):
try:
shutil.rmtree(profile_dir, ignore_errors=True)
except Exception as e:
logger.warning("Failed to clean up Chrome profile %s: %s", profile_dir, e)
# ── AI Classification ────────────────────────────────────────────────
# Uses the local Ollama model to classify scraped posts as LEAD or NOT LEAD.
# Falls back to keyword-based filtering if the AI is unavailable.
# ask_ollama() sends a raw prompt to Ollama's /api/chat endpoint.
# classify_leads() builds a classification prompt and parses the JSON response.
async def ask_ollama(prompt: str) -> str:
import httpx
async with httpx.AsyncClient(timeout=120) as c:
r = await c.post(f"{OLLAMA_URL}/api/chat", json={
"model": CLASSIFY_MODEL,
"messages": [
{"role": "system", "content": "You classify forum posts. Return only valid JSON."},
{"role": "user", "content": prompt}
],
"stream": False,
"options": {"temperature": 0.05, "num_predict": 1024},
})
r.raise_for_status()
data = r.json()
return data["message"]["content"]
async def classify_leads(results: list[dict], tutoring: bool = False) -> list[dict]:
if not results:
return []
# ── 1. AI classification ─────────────────────────────────────────
briefs = [r["title"][:200] for r in results]
if tutoring:
lead_desc = "someone REQUESTING/LOOKING FOR/WANTING a tutor, teacher, or lessons for their child or themselves"
lead_examples = '"Looking for a tutor for my child", "Need a math tutor for my son", "Need help with homework", "Looking for piano lessons for my daughter", "Need a reading tutor"'
not_lead_examples = '"I offer tutoring services", "I am a tutor with experience", "Affordable tutoring packages", "Online tutor available"'
extra_terms = '- Posts about homeschooling resources, curriculum sales, or educational products\n- Posts asking for study tips or general academic advice without requesting a tutor'
else:
lead_desc = "someone REQUESTING/POSTING/WANTING a website built, designed, or created for them"
lead_examples = '"Need a website for my business", "Looking for web developer to build my site", "I need someone to create my website", "Want a new website for my company", "Looking for someone to design my WordPress site"'
not_lead_examples = '"I build websites", "I offer web design", "Affordable web design packages"'
extra_terms = '- "Need web hosting", "Looking for a partner", "Looking for content writer", "Video spokesperson"'
prompt = f"""Classify each post as LEAD or NOT.
LEAD = {lead_desc}.
LEAD examples: {lead_examples}
NOT LEAD:
- Offering services: {not_lead_examples}
- Already have a website and need marketing, SEO, content, video, link building, email marketing, affiliates
- Recruiting employees, hiring staff, looking for business partners
- Selling products, promoting services, affiliate offers
{extra_terms}
- Posts from groups, communities, or pages (group announcements, group posts, page posts)
- Posts containing the word "group", "page", "community", "creators" — these are NEVER individual leads
- Vague questions or general recommendations without a clear intent to buy or hire
- People asking how to learn or do it themselves (not looking to hire someone)
- Posts about existing website issues like speed, SEO, errors, redesign advice — NOT a lead
For each numbered post, answer ONLY "yes" (LEAD) or "no" (NOT LEAD):
{chr(10).join(f'{i+1}. {t}' for i, t in enumerate(briefs))}
Return a JSON array like ["yes","no","yes"] matching the order above."""
ai_leads: list[dict] = []
try:
raw = await ask_ollama(prompt)
raw = raw.strip()
if raw.startswith("```"):
first_nl = raw.find('\n')
if first_nl != -1:
raw = raw[first_nl + 1:]
if raw.endswith("```"):
raw = raw[:-3]
raw = raw.strip()
answers = json.loads(raw)
if isinstance(answers, list) and len(answers) == len(results):
for i, ans in enumerate(answers):
if isinstance(ans, str) and ans.lower() == 'yes':
ai_leads.append(results[i])
logger.info("AI classified %d/%d as LEAD", len(ai_leads), len(results))
except Exception as e:
logger.warning("AI classification failed: %s", e)
# ── 2. Keyword supplement (never overrides AI, only adds missing leads) ──
if tutoring:
target_terms = [
"tutor", "tutoring", "tutor for", "private tutor",
"math tutor", "english tutor", "reading tutor",
"science tutor", "online tutor", "home tutor",
"lessons for", "lessons for my", "piano lessons",
"swimming lessons", "music lessons",
"help with homework", "homework help",
"teacher for", "teacher for my",
"need help learning", "need help with",
"exam prep", "exam preparation",
"homeschool", "homeschool tutor",
"tuition",
"coding for my", "programming for my",
"looking for a tutor", "need a tutor",
"tutor needed", "tutoring for",
"private lessons", "private tuition",
"afterschool", "after school",
"extra classes", "extra lessons",
]
offer_reject_tutor = [
'i am a tutor', "i'm a tutor", 'i offer tutoring',
'online tutor available', 'tutor available',
'i teach', 'i provide tutoring',
'affordable tutoring', 'tutoring services',
'experienced tutor', 'qualified tutor',
'your child', 'your kids', 'your children',
'enroll your', 'sign up',
'free trial', 'first lesson free',
'group lessons', 'group class',
'limited spots', 'book now',
'curriculum', 'workbook', 'worksheet',
'educational program',
'homeschool program', 'home school program',
]
else:
target_terms = [
"website", "web design", "web develop", "web dev",
"web designer", "web developer",
"build my website", "build a website", "create a website",
"landing page", "wordpress", "ecommerce",
"my website", "business website",
"site for my", "site for my business",
"new website", "redesign my website",
"help with my website", "update my website",
"make a website", "make my website",
"website for my",
"online store", "online shop",
"build my site", "build a site",
"set up a website", "set up my website",
"custom website",
"shopify",
"my site",
"webpage", "web page",
"who can build", "who can design",
"create my website", "create my site",
]
offer_reject_tutor = []
request_terms = [
"looking for", "need a", "need an", "looking to",
"need someone", "hire a", "want someone",
"need help with", "would like", "build me",
"design my", "make me a", "create my",
"who can", "i need",
"recommend", "anyone know", "anyone recommend",
"know a", "know any", "recommendation",
"suggest", "looking for recommendations",
"can anyone", "does anyone",
"dm me", "pm me", "message me",
"quote for",
"can you help",
"how much",
"price for",
]
keyword_leads: list[dict] = []
offer_reject = [
'i build website', 'i offer web', 'i am a web developer',
'affordable web design package', 'web hosting',
'i do web design', 'i develop websites',
'do you need a', 'do you want a', 'we can build',
'i can build', 'my portfolio', 'reasonable price',
'whatsapp me', 'looking for a business', 'looking for client',
'help your business', 'i am a web', 'contact me',
'we offer web', 'we provide web',
'take the quiz',
'link in bio', 'apply now', 'get started',
'for only', 'low price', 'hit me up',
'send me a message', 'i do website', 'we do website',
'we do web', 'i do web',
'website designer / web developer', 'website & software creators',
'website builders for small businesses',
'wix website design',
'for sale', 'selling my', 'premium',
'i\'m selling', 'i\'m offering', 'we\'re offering',
'free ecommerce', 'free website design',
'starting a', 'looking for a few businesses',
# Group-related rejections
'group', ' i need a website group',
'i can help', 'inbox me', 'message me for',
'best price', 'discount', 'reach out', 'check out my', 'check this',
'website for your', 'price start', 'price begin', 'website creator',
'website & software', 'creators &', 'creators marketplace',
'website group', 'page group',
'south africa web', 'philippines web', 'australia web',
'nigerian web', 'kenya web', 'india web',
# Self-promotion rejections
'i\'m a web', "i'm a web", 'i am a full stack', "i'm a full stack",
'freelance opportunity', 'looking for new project', 'looking for new work',
'full stack web', 'mern stack', 'responsive business website',
'i build website', 'i build shopify', 'i build wordpress',
'we build website', 'we build shopify', 'we build wordpress',
'i create website', 'we create website',
'full time position', 'full time job', 'remote position', 'remote job',
'help wanted', 'now hiring', 'job opening',
'website speed', 'speed optimization', 'on page seo', 'off page seo',
'elementor pro', 'woocommerce', 'acf', 'custom post type',
'send you the link', 'comment website',
'for free', 'no coding', 'make money', 'website for free',
'part time job', 'part time position',
'years of experience', 'years of teaching',
# Service offers that slip through two-word check
'i am a full stack', 'i am a developer',
'i will design', 'i will build', 'i will create',
'i can design', 'i can create',
'we will design', 'we will build',
'hire me', 'i am available for',
'available for work', 'freelance web',
'i specialize in', 'we specialize in',
"here's my portfolio", 'check my portfolio',
'see my work', 'view my work',
'we have a team', 'my team',
'i am looking for clients', 'i am looking for work',
'looking for web development work',
'looking for new clients',
# People learning / doing it themselves (not hiring)
'learn web development', 'learn to code',
'how to build a website', 'how to create a website',
'how to make a website', 'how to design a website',
'where to start', 'online course',
'want to learn', 'learning web',
'best platform for', 'which platform',
# Existing website issues (not new build)
'my website is down', 'website not loading',
'website error', 'website problem',
'website troubleshooting',
'need website advice', 'website tips',
'help with seo', 'google ranking',
'website design ideas', 'website inspiration',
]
for r in results:
t = r['title'].lower()
has_target = any(kw in t for kw in target_terms)
has_request = any(kw in t for kw in request_terms)
if not has_target or not has_request:
continue
if any(kw in t for kw in offer_reject):
continue
if any(kw in t for kw in offer_reject_tutor):
continue
keyword_leads.append(r)
# ── 3. Merge: prefer AI leads, supplement with keywords ──
seen_titles: set[int] = set()
merged: list[dict] = []
for r in ai_leads + keyword_leads:
key = hash(r.get('title', ''))
if key not in seen_titles:
seen_titles.add(key)
merged.append(r)
# Final sweep: strip any remaining offers or group posts from merged
group_words = ['group', ' groups', 'page:', 'page |', 'community', 'creators', 'marketplace']
merged = [r for r in merged if not any(kw in (r.get('title','') or '').lower() for kw in offer_reject)]
merged = [r for r in merged if not any(gw in (r.get('title','') or '').lower() for gw in group_words)]
logger.info("classify_leads: %d merged (%d AI + %d keyword) from %d raw", len(merged), len(ai_leads), len(keyword_leads), len(results))
return merged[:10]
# ══════════════════════════════════════════════════════════════════════
# FastAPI Endpoints
# ══════════════════════════════════════════════════════════════════════
@app.get("/health")
async def health():
"""Simple health check for the splash page status polling."""
return {"status": "ok"}
def _detect_all_profiles() -> dict:
"""Return all detected browser profiles."""
return {
"firefox": FX_PROFILE or find_firefox_profile() or "",
"opera": OPERA_PROFILE or find_opera_profile() or "",
"chrome": CHROME_PROFILE or find_chrome_profile() or "",
"edge": EDGE_PROFILE or find_edge_profile() or "",
}
@app.get("/setup/profile")
async def setup_profile():
"""Auto-detect all browser profiles."""
return _detect_all_profiles()
async def _check_browser_login(profile_path: str, browser: str) -> dict:
"""Check Facebook login for a specific browser type."""
profile_dir = None
try:
if browser == "firefox":
profile_dir = copy_firefox_profile(profile_path)
async with async_playwright() as pw:
context = await pw.firefox.launch_persistent_context(
user_data_dir=profile_dir, headless=True,
firefox_user_prefs={"dom.webdriver.enabled": False},
)
page = context.pages[0] if context.pages else await context.new_page()
await page.goto("https://www.facebook.com/", wait_until="domcontentloaded", timeout=30000)
await page.wait_for_timeout(3000)
logged_in = "/login" not in page.url.lower()
await context.close()
shutil.rmtree(profile_dir, ignore_errors=True)
return {"logged_in": logged_in, "browser": browser, "path": profile_path}
else:
channel = None
exe = None
if browser == "chrome":
channel = "chrome"
elif browser == "edge":
channel = "msedge"
elif browser == "opera":
exe = shutil.which("opera") or shutil.which("Opera")
profile_dir = copy_chrome_profile(profile_path)
async with async_playwright() as pw:
kw = dict(user_data_dir=profile_dir, headless=True, args=CHROME_LAUNCH_ARGS)
if channel:
kw["channel"] = channel
if exe:
kw["executable_path"] = exe
context = await pw.chromium.launch_persistent_context(**kw)
page = context.pages[0] if context.pages else await context.new_page()
await page.goto("https://www.facebook.com/", wait_until="domcontentloaded", timeout=30000)
await page.wait_for_timeout(3000)
logged_in = "/login" not in page.url.lower()
await context.close()
shutil.rmtree(profile_dir, ignore_errors=True)
return {"logged_in": logged_in, "browser": browser, "path": profile_path}
except Exception as e:
if profile_dir:
try: shutil.rmtree(profile_dir, ignore_errors=True)
except: pass
return {"logged_in": False, "browser": browser, "path": profile_path, "error": str(e)}
@app.post("/setup/check-login")
async def setup_check_login(body: dict):
"""Check if Facebook is logged in. Accepts optional 'browser' param, tries all if not given."""
browser = (body.get("browser") or "").strip().lower()
profile_path = (body.get("profile_path") or "").strip()
if browser and profile_path:
return await _check_browser_login(profile_path, browser)
# No specific browser — try all detected
profiles = _detect_all_profiles()
priority = ["firefox", "opera", "chrome", "edge"]
for b in priority:
p = profiles.get(b, "")
if p:
result = await _check_browser_login(p, b)
if result.get("logged_in"):
return result
logger.info("%s not logged in, trying next", b)
return {"logged_in": False, "reason": "none_logged_in", "message": "No browser with Facebook login found"}
@app.post("/agent/run")
async def agent_run(task: str = Body(..., embed=True)):
"""Run a browser-use Agent with ChatOllama (free/local)."""
try:
browser = _make_browser(headless=True, args=CHROME_LAUNCH_ARGS)
await browser.start()
agent = _make_agent(
task=task,
llm=_make_ollama(num_ctx=32000, temperature=0.3),
browser=browser,
use_vision=False,
max_actions_per_step=5,
max_failures=8,
step_timeout=180,
)
history = await agent.run()
await browser.close()
return {"success": True, "result": history.final_result()}
except Exception as e:
logger.error("Agent run failed: %s", e)
return {"success": False, "error": str(e)}
@app.post("/scrape/facebook")
async def scrape_facebook_endpoint(profile_path: str | None = Query(None), force: bool = Query(False), query: str | None = Query(None)):
result = await scrape_facebook(profile_path, force, query)
return result
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=PORT)
Reference in New Issue View Git Blame Copy Permalink