354 lines
12 KiB
Python
354 lines
12 KiB
Python
import os, json, asyncio, re, shutil, sqlite3, traceback, urllib.parse
|
|
from datetime import datetime, timedelta
|
|
from bs4 import BeautifulSoup
|
|
import httpx
|
|
from fastapi import FastAPI
|
|
from pydantic import BaseModel
|
|
import uvicorn
|
|
from playwright.async_api import async_playwright
|
|
|
|
app = FastAPI()
|
|
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', r'C:\Users\USER-PC\AppData\Roaming\Mozilla\Firefox\Profiles\h8p11vlj.default-release')
|
|
FX_COOKIE_DB = os.path.join(FX_PROFILE, 'cookies.sqlite')
|
|
|
|
HEADERS = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
|
|
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
|
|
}
|
|
|
|
last_scrape_time: datetime | None = None
|
|
_pw_browser = None
|
|
|
|
STRICT_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",
|
|
]
|
|
|
|
BROAD_KEYWORDS = STRICT_KEYWORDS + [
|
|
"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",
|
|
]
|
|
|
|
def kw_match_strict(text: str) -> bool:
|
|
t = text.lower()
|
|
return any(kw in t for kw in STRICT_KEYWORDS)
|
|
|
|
def kw_match(text: str) -> bool:
|
|
t = text.lower()
|
|
return any(kw in t for kw in BROAD_KEYWORDS)
|
|
|
|
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",
|
|
"website quote please",
|
|
"need wordpress website",
|
|
"I need a website for my business",
|
|
"need a site for my business",
|
|
]
|
|
|
|
def get_fb_cookies():
|
|
"""Copy Firefox cookie DB and extract Facebook cookies for Chromium injection."""
|
|
tmp = os.path.join(os.path.dirname(FX_COOKIE_DB), 'cookies_tmp.sqlite')
|
|
try:
|
|
if not os.path.exists(FX_COOKIE_DB):
|
|
return []
|
|
shutil.copy2(FX_COOKIE_DB, tmp)
|
|
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()
|
|
os.remove(tmp)
|
|
return [{
|
|
"name": name, "value": value,
|
|
"domain": host if host.startswith('.') else f'.{host}',
|
|
"path": path,
|
|
"httpOnly": False, "secure": False, "sameSite": "Lax",
|
|
} for name, value, host, path in rows]
|
|
except Exception as e:
|
|
return []
|
|
|
|
def _is_real_text(line: str) -> bool:
|
|
if not line:
|
|
return False
|
|
# Count words (sequences of letters)
|
|
words = re.findall(r'[A-Za-z]{2,}', line)
|
|
return len(words) >= 2
|
|
|
|
def _extract_posts_from_text(raw: str, url: str) -> list[dict]:
|
|
"""Parse Facebook search results page text into structured posts."""
|
|
lines = [l.strip() for l in raw.split('\n')]
|
|
|
|
# Split into blocks separated by "Facebook" boilerplate lines (2+ consecutive)
|
|
blocks = []
|
|
cur = []
|
|
fb_run = 0
|
|
for l in lines:
|
|
if 'facebook' in l.lower():
|
|
fb_run += 1
|
|
if cur and fb_run >= 2:
|
|
blocks.append(cur)
|
|
cur = []
|
|
else:
|
|
fb_run = 0
|
|
if l and len(l) >= 15:
|
|
# Check for real words (not scattered reaction chars)
|
|
words = re.findall(r'[A-Za-z]{2,}', l)
|
|
if len(words) >= 2:
|
|
cur.append(l)
|
|
if cur:
|
|
blocks.append(cur)
|
|
|
|
posts = []
|
|
seen_texts = set()
|
|
for block in blocks:
|
|
if not block:
|
|
continue
|
|
# Find keyword-matched content in this block
|
|
kw_indices = [i for i, l in enumerate(block) if kw_match(l)]
|
|
if not kw_indices:
|
|
continue
|
|
# Use first keyword match as anchor, grab context
|
|
i = kw_indices[0]
|
|
start = max(0, i - 2)
|
|
end = min(len(block), i + 5)
|
|
snippet = ' '.join(block[start:end])
|
|
if len(snippet) < 40:
|
|
continue
|
|
dekey = snippet[:80]
|
|
if dekey in seen_texts:
|
|
continue
|
|
seen_texts.add(dekey)
|
|
posts.append({
|
|
"title": snippet[:300],
|
|
"content": snippet[:1000],
|
|
"author": block[start] if start < i else '',
|
|
"url": url,
|
|
"source": "facebook",
|
|
"date": datetime.now().strftime('%Y-%m-%d'),
|
|
})
|
|
|
|
return posts
|
|
|
|
async def search_facebook(page, query: str) -> list[dict]:
|
|
"""Search Facebook and extract post leads from raw page text."""
|
|
url = f'https://www.facebook.com/search/posts/?q={urllib.parse.quote(query)}'
|
|
try:
|
|
await page.goto(url, wait_until='domcontentloaded', timeout=30000)
|
|
# Wait for the page content to fully render (React/XHR loads)
|
|
await page.wait_for_timeout(6000)
|
|
except:
|
|
return []
|
|
|
|
raw = await page.evaluate('document.body.innerText')
|
|
return _extract_posts_from_text(raw, url)
|
|
|
|
async def scrape_facebook() -> list[dict]:
|
|
"""Launch headless Chromium, inject Firefox Facebook cookies, search for leads."""
|
|
all_posts = []
|
|
fb_cookies = get_fb_cookies()
|
|
if not fb_cookies:
|
|
return []
|
|
try:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(headless=True)
|
|
context = await browser.new_context(
|
|
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:150.0) Gecko/20100101 Firefox/150.0',
|
|
viewport={'width': 1280, 'height': 800},
|
|
)
|
|
for c in fb_cookies:
|
|
try:
|
|
await context.add_cookies([c])
|
|
except:
|
|
pass
|
|
|
|
page = await context.new_page()
|
|
|
|
await page.goto('https://www.facebook.com/', wait_until='domcontentloaded', timeout=30000)
|
|
await page.wait_for_timeout(5000)
|
|
|
|
url = page.url
|
|
if '/login' in url.lower():
|
|
return []
|
|
|
|
for query in FB_SEARCHES[:6]:
|
|
try:
|
|
posts = await search_facebook(page, query)
|
|
all_posts.extend(posts)
|
|
await page.wait_for_timeout(2000)
|
|
except:
|
|
continue
|
|
except Exception:
|
|
return []
|
|
|
|
seen = set()
|
|
deduped = []
|
|
for p in all_posts:
|
|
key = p['content'][:100]
|
|
if key not in seen:
|
|
seen.add(key)
|
|
deduped.append(p)
|
|
return deduped[:20]
|
|
|
|
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 scrape_warriorforum() -> list[dict]:
|
|
results = []
|
|
try:
|
|
async with httpx.AsyncClient(headers=HEADERS, timeout=15, follow_redirects=True) as c:
|
|
r = await c.get('https://www.warriorforum.com/wanted-members-looking-hire-you/')
|
|
soup = BeautifulSoup(r.text, 'lxml')
|
|
for row in soup.find_all('td', class_='FlexTable-item--title'):
|
|
a = row.find('a', href=True)
|
|
if not a:
|
|
continue
|
|
href = a['href']
|
|
title = a.text.strip()
|
|
if not title or len(title) < 15 or not kw_match_strict(title):
|
|
continue
|
|
author_div = row.find('div', class_='FlexTable-item-author')
|
|
author = author_div.get_text(strip=True).lstrip('by') if author_div else ''
|
|
date_str = ''
|
|
date_div = row.find('div', class_=lambda c: c and 'media--available' in c)
|
|
if date_div:
|
|
txt = date_div.get_text(strip=True)
|
|
m = re.search(r'(\d{10})', txt)
|
|
if m:
|
|
from datetime import datetime as dtdt
|
|
date_str = dtdt.utcfromtimestamp(int(m.group(1))).strftime('%Y-%m-%d')
|
|
results.append({
|
|
"title": title, "url": href,
|
|
"author": author,
|
|
"content": title[:300],
|
|
"source": "warriorforum",
|
|
"date": date_str
|
|
})
|
|
except Exception:
|
|
traceback.print_exc()
|
|
return results
|
|
|
|
async def classify_leads(results: list[dict]) -> list[dict]:
|
|
if not results:
|
|
return []
|
|
filtered = []
|
|
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."""
|
|
try:
|
|
raw = await ask_ollama(prompt)
|
|
raw = raw.strip()
|
|
if raw.startswith("```json"): raw = raw[7:]
|
|
if raw.startswith("```"): raw = raw.split("\n", 1)[-1].rsplit("```", 1)[0]
|
|
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':
|
|
filtered.append(results[i])
|
|
except:
|
|
pass
|
|
if not filtered:
|
|
for r in results:
|
|
t = r['title'].lower()
|
|
# MUST match a web-dev keyword to be a lead
|
|
if not kw_match_strict(t):
|
|
continue
|
|
# But NOT these (offerings, not requests)
|
|
if any(kw in t for kw in ['i build', 'i offer', 'i create', 'i am a', 'web developer available',
|
|
'affordable web', 'web hosting', 'free website',
|
|
'limited time', 'special offer', 'sign up now',
|
|
'offer']):
|
|
continue
|
|
filtered.append(r)
|
|
return filtered[:10]
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
class ScrapeRequest(BaseModel):
|
|
query: str = ""
|
|
|
|
@app.post("/scrape/requests")
|
|
async def scrape_requests(req: ScrapeRequest):
|
|
global last_scrape_time
|
|
now = datetime.now()
|
|
if last_scrape_time and (now - last_scrape_time) < timedelta(seconds=15):
|
|
return {"error": "rate_limited", "retry_after": 15}
|
|
last_scrape_time = now
|
|
results = await scrape_warriorforum()
|
|
if results:
|
|
results = await classify_leads(results)
|
|
return results[:10]
|
|
|
|
@app.post("/scrape/facebook")
|
|
async def scrape_facebook_endpoint():
|
|
results = await scrape_facebook()
|
|
if results:
|
|
results = await classify_leads(results)
|
|
return results[:15]
|
|
|
|
@app.post("/scrape/all")
|
|
async def scrape_all():
|
|
results = []
|
|
try:
|
|
wf = await scrape_warriorforum()
|
|
if wf:
|
|
wf = await classify_leads(wf)
|
|
results.extend(wf[:5])
|
|
except:
|
|
pass
|
|
try:
|
|
fb = await scrape_facebook()
|
|
if fb:
|
|
fb = await classify_leads(fb)
|
|
results.extend(fb[:8])
|
|
except:
|
|
pass
|
|
return results[:12]
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=PORT)
|