428 lines
18 KiB
Python
428 lines
18 KiB
Python
import os, json, asyncio, re, shutil, sqlite3, urllib.parse, random, logging, httpx
|
|
from datetime import datetime, timedelta
|
|
from dotenv import load_dotenv
|
|
|
|
dotenv_path = os.path.join(os.path.dirname(__file__), '..', '.env.local')
|
|
if os.path.exists(dotenv_path):
|
|
load_dotenv(dotenv_path)
|
|
from fastapi import FastAPI, Query
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import uvicorn
|
|
from playwright.async_api import async_playwright
|
|
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI()
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["http://localhost:3006", "http://127.0.0.1:3006"],
|
|
allow_methods=["POST"],
|
|
allow_headers=["*"],
|
|
)
|
|
PORT = int(os.getenv("PORT", "3008"))
|
|
|
|
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
|
|
CLASSIFY_MODEL = os.getenv("CLASSIFY_MODEL", "dolphin-llama3:8b")
|
|
FX_PROFILE = os.getenv('FX_PROFILE', '')
|
|
BU_API_KEY = os.getenv('BU_API_KEY', '')
|
|
|
|
BU_API_BASE = "https://api.browser-use.com/api/v3"
|
|
|
|
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"\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",
|
|
]
|
|
|
|
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",
|
|
]
|
|
|
|
FB_SEARCHES = [
|
|
"looking for web developer", "need a website designed",
|
|
"need website built South Africa", "need someone to build my website",
|
|
"need web designer", "looking for someone to create website",
|
|
"need ecommerce website built", "want to hire web developer",
|
|
"need wordpress website", "I need a website for my business",
|
|
"need a site for my business", "looking for website designer South Africa",
|
|
"need website for my small business", "who can build me a website",
|
|
"want to create a website for my business", "looking for affordable web design",
|
|
"need a web developer South Africa", "help me build a website",
|
|
"need someone to design my website", "looking for web designer near me",
|
|
"need a website for my startup",
|
|
]
|
|
|
|
def is_offer(text: str) -> bool:
|
|
return any(re.search(p, text.lower()) for p in OFFER_PATTERNS)
|
|
|
|
def is_request(text: str) -> bool:
|
|
return any(re.search(p, text.lower()) for p in REQUEST_PATTERNS)
|
|
|
|
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:
|
|
d = datetime.now() - timedelta(hours=int(m.group(1)))
|
|
return d.strftime('%Y-%m-%d')
|
|
if l.lower() == 'yesterday':
|
|
return (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
|
low = l.lower()
|
|
if low in WEEKDAY_ORDER:
|
|
day_idx = WEEKDAY_ORDER.index(low)
|
|
today_idx = datetime.now().weekday()
|
|
d = datetime.now() - timedelta(days=(today_idx - day_idx) % 7)
|
|
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:
|
|
return datetime.strptime(l, fmt).strftime('%Y-%m-%d')
|
|
except ValueError:
|
|
pass
|
|
return datetime.now().strftime('%Y-%m-%d')
|
|
|
|
def _clean_fb_text(text: str) -> str:
|
|
cleaned = []
|
|
for l in text.split('\n'):
|
|
s = l.strip()
|
|
if len(s) < 3 or all(not c.isalpha() for c in s):
|
|
continue
|
|
cleaned.append(s)
|
|
return '\n'.join(cleaned)
|
|
|
|
def _extract_posts_from_elements(elements: list[dict], base_url: str) -> list[dict]:
|
|
posts = []
|
|
seen = set()
|
|
for el in elements:
|
|
raw = el.get('text', '')
|
|
if len(raw) < 40:
|
|
continue
|
|
text = _clean_fb_text(raw)
|
|
if len(text) < 40 or is_offer(text):
|
|
continue
|
|
dekey = text[:80]
|
|
if dekey in seen:
|
|
continue
|
|
seen.add(dekey)
|
|
score = 2 if is_request(text) else 0
|
|
posts.append({
|
|
"title": text[:300], "content": text[:1000],
|
|
"author": el.get('author', ''), "url": el.get('url') or base_url,
|
|
"source": "facebook", "date": _parse_fb_date(text.split('\n')),
|
|
"_score": score,
|
|
})
|
|
posts.sort(key=lambda p: p['_score'], 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 = [], set()
|
|
cur = []
|
|
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):
|
|
snippet = ' '.join(cur[-3:] + [l]) if cur else l
|
|
if len(snippet) >= 40 and not is_offer(snippet) and snippet[:80] not in seen:
|
|
seen.add(snippet[:80])
|
|
posts.append({"title": snippet[:300], "content": snippet[:1000],
|
|
"author": '', "url": url, "source": "facebook",
|
|
"date": _parse_fb_date(cur + [l] if cur else [l])})
|
|
cur = []
|
|
cur.append(l)
|
|
if len(cur) > 10:
|
|
cur.pop(0)
|
|
return posts
|
|
|
|
async def _get_article_elements(page) -> list[dict]:
|
|
return await page.evaluate('''() => {
|
|
const results = [];
|
|
const seenTexts = new Set();
|
|
const selectors = ['div[role="article"]', 'div[role="feed"] > div', 'div.x1yztbdb', 'div[data-pagelet]'];
|
|
for (const sel of selectors) {
|
|
const els = document.querySelectorAll(sel);
|
|
for (const el of els) {
|
|
const text = (el.innerText || '').trim();
|
|
if (text.length < 40) continue;
|
|
const key = text.substring(0, 80);
|
|
if (seenTexts.has(key)) continue;
|
|
seenTexts.add(key);
|
|
const links = el.querySelectorAll('a');
|
|
let author = '', postUrl = '';
|
|
for (const a of links) {
|
|
const t = (a.innerText || '').trim();
|
|
if (t && t.length > 2 && t.length < 80 && /^[a-zA-ZÀ-ÿ][a-zA-ZÀ-ÿ .'-]+$/.test(t)) {
|
|
author = t; break;
|
|
}
|
|
}
|
|
for (const a of links) {
|
|
const h = (a.getAttribute('href') || '');
|
|
if (h.includes('/posts/') || h.includes('/photo/') || h.includes('/video/') || h.includes('/groups/')) {
|
|
postUrl = h.startsWith('http') ? h : 'https://www.facebook.com' + h; break;
|
|
}
|
|
}
|
|
results.push({ text, author, url: postUrl });
|
|
}
|
|
if (results.length > 0) break;
|
|
}
|
|
return results;
|
|
}''')
|
|
|
|
async def get_fb_cookies(profile_path: str | None = None):
|
|
cookie_db_path = profile_path or FX_PROFILE
|
|
if not cookie_db_path:
|
|
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:
|
|
return []
|
|
try:
|
|
conn = sqlite3.connect(tmp)
|
|
rows = conn.cursor().execute(
|
|
"SELECT name, value, host, path FROM moz_cookies WHERE host LIKE '%facebook.com'"
|
|
).fetchall()
|
|
conn.close()
|
|
os.remove(tmp)
|
|
return [{"name": n, "value": v, "domain": h if h.startswith('.') else f'.{h}',
|
|
"path": p, "httpOnly": True, "secure": True, "sameSite": "Lax"}
|
|
for n, v, h, p in rows]
|
|
except Exception as e:
|
|
logger.error("Cookie DB error: %s", e)
|
|
try: os.remove(tmp)
|
|
except: pass
|
|
return []
|
|
|
|
async def create_browser_session():
|
|
if not BU_API_KEY:
|
|
raise RuntimeError("BU_API_KEY not set")
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
r = await client.post(
|
|
f"{BU_API_BASE}/browsers",
|
|
headers={"X-Browser-Use-API-Key": BU_API_KEY},
|
|
json={"timeout": 60, "proxyCountryCode": "za"}
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
return data["id"], data["cdpUrl"]
|
|
|
|
async def stop_browser_session(session_id: str):
|
|
if not BU_API_KEY:
|
|
return
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
await client.patch(
|
|
f"{BU_API_BASE}/browsers/{session_id}",
|
|
headers={"X-Browser-Use-API-Key": BU_API_KEY, "Content-Type": "application/json"},
|
|
json={"action": "stop"}
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Failed to stop BU session: %s", e)
|
|
|
|
async def search_facebook(page, query: str) -> list[dict]:
|
|
url = f'https://www.facebook.com/search/posts/?q={urllib.parse.quote(query)}'
|
|
try:
|
|
await page.goto(url, wait_until='domcontentloaded', timeout=30000)
|
|
try:
|
|
await page.wait_for_selector('div[role="article"], div[role="feed"]', timeout=10000)
|
|
except Exception:
|
|
pass
|
|
await page.wait_for_timeout(2000)
|
|
await page.evaluate("window.scrollBy(0, 600)")
|
|
await page.wait_for_timeout(1500)
|
|
raw_articles = await _get_article_elements(page)
|
|
posts = _extract_posts_from_elements(raw_articles, url) if raw_articles else []
|
|
if not posts:
|
|
raw = await page.evaluate('document.body.innerText')
|
|
posts = _extract_posts_from_text(raw, url)
|
|
return posts
|
|
except Exception as e:
|
|
logger.warning("Facebook search '%s' failed: %s", query, e)
|
|
return []
|
|
|
|
async def scrape_facebook(profile_path: str | None = None, force: bool = False) -> dict:
|
|
fb_cookies = await get_fb_cookies(profile_path)
|
|
if not fb_cookies:
|
|
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": "No cookies available"}
|
|
|
|
session_id = None
|
|
try:
|
|
session_id, cdp_url = await create_browser_session()
|
|
logger.info("BU session %s created, connecting via CDP", session_id)
|
|
except Exception as e:
|
|
logger.error("Failed to create BU session: %s", e)
|
|
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": f"BU session failed: {e}"}
|
|
|
|
try:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.connect_over_cdp(cdp_url)
|
|
ctx = browser.contexts[0] if browser.contexts else await browser.new_context()
|
|
if not browser.contexts:
|
|
for c in fb_cookies:
|
|
try: await ctx.add_cookies([c])
|
|
except Exception: pass
|
|
|
|
page = ctx.pages[0] if ctx.pages else await ctx.new_page()
|
|
await page.goto('https://www.facebook.com/', wait_until='domcontentloaded', timeout=30000)
|
|
await page.wait_for_timeout(2000)
|
|
|
|
url = page.url
|
|
if '/login' in url.lower():
|
|
logger.warning("Facebook login page detected")
|
|
await browser.close()
|
|
await stop_browser_session(session_id)
|
|
return {"success": False, "leads": [], "flagged": True, "flag_reason": "login_page", "error": "Facebook login page detected"}
|
|
|
|
all_posts = []
|
|
searches = random.sample(FB_SEARCHES, k=random.randint(5, 8))
|
|
for query in searches:
|
|
posts = await search_facebook(page, query)
|
|
all_posts.extend(posts)
|
|
await page.wait_for_timeout(1000)
|
|
|
|
seen = set()
|
|
deduped = []
|
|
for p in all_posts:
|
|
key = p.get('content', '')[:100]
|
|
if key not in seen:
|
|
seen.add(key)
|
|
deduped.append(p)
|
|
|
|
leads = deduped[:20]
|
|
if leads:
|
|
leads = await classify_leads(leads)
|
|
|
|
await browser.close()
|
|
await stop_browser_session(session_id)
|
|
|
|
return {"success": True, "leads": leads[:15], "flagged": False, "flag_reason": None, "error": None}
|
|
|
|
except Exception as e:
|
|
logger.error("Facebook scrape failed: %s", e)
|
|
await stop_browser_session(session_id)
|
|
return {"success": False, "leads": [], "flagged": False, "flag_reason": None, "error": str(e)}
|
|
|
|
async def ask_ollama(prompt: str) -> str:
|
|
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()
|
|
return r.json()["message"]["content"]
|
|
|
|
async def classify_leads(results: list[dict]) -> list[dict]:
|
|
if not results:
|
|
return []
|
|
briefs = [r["title"][:200] for r in results]
|
|
prompt = f"""Classify each post as LEAD or NOT.
|
|
LEAD = 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:
|
|
- Offering web design services: "I build websites", "I offer web design", "Affordable web design packages"
|
|
- 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
|
|
- "Need web hosting", "Looking for a partner", "Looking for content writer", "Video spokesperson"
|
|
|
|
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_succeeded = False
|
|
try:
|
|
raw = (await ask_ollama(prompt)).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):
|
|
ai_succeeded = True
|
|
filtered = [results[i] for i, ans in enumerate(answers) if isinstance(ans, str) and ans.lower() == 'yes']
|
|
if filtered:
|
|
return filtered[:10]
|
|
except Exception as e:
|
|
logger.warning("AI classification failed, falling back to keyword filter: %s", e)
|
|
|
|
if not ai_succeeded:
|
|
strict = ["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", "need someone", "hire a",
|
|
"want someone", "need help with", "would like", "build me", "design my",
|
|
"make me a", "create my"]
|
|
filtered = []
|
|
for r in results:
|
|
t = r['title'].lower()
|
|
if not any(kw in t for kw in strict):
|
|
continue
|
|
if any(kw in t for kw in ['i build website', 'i offer web', 'i am a web developer',
|
|
'affordable web design package', 'web hosting']):
|
|
continue
|
|
filtered.append(r)
|
|
return filtered[:10]
|
|
return []
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
@app.post("/scrape/facebook")
|
|
async def scrape_facebook_endpoint(profile_path: str | None = Query(None), force: bool = Query(False)):
|
|
result = await scrape_facebook(profile_path, force)
|
|
return result
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=PORT)
|