CRM_ENVR/browser-use-service/main.py

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)