import os, json, asyncio, re, shutil, sqlite3, traceback, urllib.parse, random, time, logging 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 logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger(__name__) 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', '') FX_COOKIE_DB = os.path.join(FX_PROFILE, 'cookies.sqlite') if FX_PROFILE else '' 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 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", ] async def get_fb_cookies(): if not FX_COOKIE_DB or not os.path.exists(FX_COOKIE_DB): logger.warning("FX_COOKIE_DB not found or FX_PROFILE not set") return [] tmp = os.path.join(os.path.dirname(FX_COOKIE_DB), f'cookies_tmp_{os.getpid()}.sqlite') for attempt in range(3): try: shutil.copy2(FX_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 [] 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 _extract_posts_from_text(raw: str, url: str) -> list[dict]: lines = [l.strip() for l in raw.split('\n')] 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: 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 kw_indices = [i for i, l in enumerate(block) if kw_match(l)] if not kw_indices: continue 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": _parse_fb_date(block), }) return posts 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) await page.wait_for_timeout(6000) except Exception as e: logger.warning("Facebook search navigation failed for '%s': %s", query, e) return [] try: raw = await page.evaluate('document.body.innerText') except Exception as e: logger.warning("Failed to evaluate page text: %s", e) return [] return _extract_posts_from_text(raw, url) async def scrape_facebook() -> list[dict]: all_posts = [] fb_cookies = await get_fb_cookies() if not fb_cookies: logger.warning("No Facebook cookies available") 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 Exception as e: logger.warning("Failed to inject cookie %s: %s", c.get('name'), e) 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(): logger.warning("Facebook login page detected — cookies may be expired") return [] for query in FB_SEARCHES[:6]: try: posts = await search_facebook(page, query) all_posts.extend(posts) delay = random.uniform(3, 7) await page.wait_for_timeout(int(delay * 1000)) except Exception as e: logger.warning("Facebook search '%s' failed: %s", query, e) except Exception as e: logger.error("Facebook scrape failed: %s", e) return [] seen = set() deduped = [] for p in all_posts: key = p.get('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() data = r.json() return data["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: date_str = datetime.utcfromtimestamp(int(m.group(1))).strftime('%Y-%m-%d') if not href.startswith('http'): href = 'https://www.warriorforum.com' + href results.append({ "title": title, "url": href, "author": author, "content": title[:300], "source": "warriorforum", "date": date_str }) except Exception: logger.error("WarriorForum scrape failed", exc_info=True) return results 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) raw = raw.strip() # Strip markdown code fences if raw.startswith("```"): # Remove opening fence first_nl = raw.find('\n') if first_nl != -1: raw = raw[first_nl + 1:] # Remove closing fence 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 = [] for i, ans in enumerate(answers): if isinstance(ans, str) and ans.lower() == 'yes': filtered.append(results[i]) if filtered: return filtered[:10] # AI successfully classified but returned all "no" — respect that decision logger.info("AI classified all %d items as NOT LEAD — returning empty", len(results)) return [] except Exception as e: logger.warning("AI classification failed, falling back to keyword filter: %s", e) # Fallback: only use keyword filter when AI failed if not ai_succeeded: filtered = [] for r in results: t = r['title'].lower() if not kw_match_strict(t): continue 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] return [] @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 Exception as e: logger.error("WarriorForum scrape in /scrape/all failed: %s", e) try: fb = await scrape_facebook() if fb: fb = await classify_leads(fb) results.extend(fb[:8]) except Exception as e: logger.error("Facebook scrape in /scrape/all failed: %s", e) return results[:12] if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=PORT)