Merge branch 'main' of https://git.coastit.co.za/caitlin/CRM_ENVR

browser-use-service/main.py

@@ -0,0 +1,353 @@
+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)

browser-use-service/run_log.txt

@@ -0,0 +1,7 @@
+python : INFO:     Started server process [20044]
+    + CategoryInfo          : NotSpecified: (INFO:     Started server process [20044]:String) [], RemoteException
+    + FullyQualifiedErrorId : NativeCommandError
+ 
+INFO:     Waiting for application startup.
+INFO:     Application startup complete.
+INFO:     Uvicorn running on http://0.0.0.0:3008 (Press CTRL+C to quit)

browser-use-service/scrape_log.txt

@@ -0,0 +1,6 @@
+[2026-06-22T19:24:48.030482] Browser Use service starting on port 3008
+[2026-06-22T19:27:19.231701] Browser Use service starting on port 3008
+[2026-06-22T19:28:28.335765] Browser Use service starting on port 3008
+[2026-06-22T19:29:05.796265] Browser Use service starting on port 3008
+[2026-06-22T19:29:17.042807] Scraping WarriorForum...
+[2026-06-22T19:29:19.075166] WarriorForum: 55 results

browser-use-service/stderr.txt

@@ -0,0 +1,4 @@
+INFO:     Started server process [24268]
+INFO:     Waiting for application startup.
+INFO:     Application startup complete.
+INFO:     Uvicorn running on http://0.0.0.0:3008 (Press CTRL+C to quit)

browser-use-service/stdout.txt

@@ -0,0 +1,5 @@
+Browser Use service starting on port 3008
+INFO:     127.0.0.1:65003 - "GET /health HTTP/1.1" 200 OK
+Scraping WarriorForum...
+WarriorForum: 55 results
+INFO:     127.0.0.1:54587 - "POST /scrape/all HTTP/1.1" 200 OK

fix.md

@@ -1,60 +1,71 @@
-# Facebook Scraper - Configuration Needed
+# Scrapers - Configuration & Status
 
-## Proxy Configuration
+## Facebook Scraper
 
 **File:** `rust-ai/src/main.rs`  
-**Lines:** 25-27
-
-The scraper needs real proxy URLs. The dummy placeholder `"http://0.0.0.0:0"` will fail to connect (expected — no crash, just error logs):
+**Lines:** ~70-85
 
+Target URL (line 72):
 ```rust
-const PROXIES: &[&str] = &[
-    "http://0.0.0.0:0",  // <- Replace with real proxy(es)
-];
+let url = "https://www.facebook.com/search/top/?q=need%20website%20create";
 ```
 
-Replace with your actual proxies, e.g.:
-```rust
-const PROXIES: &[&str] = &[
-    "http://user:pass@192.168.1.1:8080",
-    "http://user:pass@192.168.1.2:8080",
-];
+**Status:** Uses direct connection (no proxy) — your home IP. Facebook blocks datacenter IPs. May work from a residential connection.
+
+**Test with curl:**
+```
+curl.exe -s "https://www.facebook.com/search/top/?q=need%20website%20create" -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" --max-time 10 2>$null
 ```
 
-## User Agents (Optional)
+**Expected log output when blocked:**
+```
+ERROR crm_ai: Facebook scraper error: error sending request for url (https://www.facebook.com/search/top/?q=need%20website%20create)
+```
+
+## Reddit Scraper
 
 **File:** `rust-ai/src/main.rs`  
-**Lines:** 32-36
+**Lines:** ~90-120
 
-Add more realistic user agents here if needed.
+Uses `old.reddit.com` (older design that allows scraping without captchas).
 
-## Scraper Target URL
+Search queries (currently 6, lines 93-100):
+- `r/southafrica` — "need website", "web developer"
+- Global — '"need a website"', "website quote"
+- `r/forhire` — "website"
+- `r/smallbusiness` — "website"
+
+**Status:** Working. Reddit results appear as `INFO LEAD:` entries in the server log.
+
+**Test with curl:**
+```
+curl.exe -s "https://old.reddit.com/r/southafrica/search?q=need+website&sort=new&restrict_sr=on&t=week" -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+```
+
+**Expected log output when working:**
+```
+INFO crm_ai: LEAD: [Post Title] -> https://old.reddit.com/r/.../...
+```
+
+## Background Loop
 
 **File:** `rust-ai/src/main.rs`  
-**Line:** 68
+**Lines:** ~370-383
 
-Currently fetches `https://www.facebook.com/search/top/?q=need%20website%20create`. Change if needed.
+Both scrapers run together every 60-180 seconds on a `spawn_blocking` thread.
 
-## Background Thread
+## What You Need to Do
 
-**File:** `rust-ai/src/main.rs`  
-**Lines:** 355-371
+### Facebook
+- Try running from your home PC instead — your residential IP may not be blocked
+- If still blocked, you need residential proxies (BrightData, IPRoyal, Oxylabs)
+- Configure them in the `PROXIES` array at line 25
 
-Runs in a `tokio::task::spawn_blocking` thread (changed from `tokio::spawn` because the scraper uses `reqwest::blocking::Client` + `thread::sleep`).
+### Reddit  
+- Works out of the box via `old.reddit.com`
+- If it stops working, Reddit IP-blocked you — use proxies or switch to Playwright
 
-## Expected Behavior Until Configured
+### To add more search queries
+Edit the `searches` array in `run_reddit_scraper()` (line 93).
 
-Until real proxies are set, the Rust server will log this every 1-5 seconds (not a crash):
-```
-ERROR crm_ai: Facebook scraper error: error sending request for url (https://www.facebook.com/messages)
-```
-Once you add working proxies, these errors stop and actual scraping begins.
-
-## How it works
-
-- Runs every 1-5 seconds on a background blocking thread (line 355-371)
-- Rotates through proxies and user agents for each request
-- Parses conversation HTML via `scraper` crate
-- Designed to detect job posts like "need a website" and notify sales reps
-- Error handling added to prevent server crash (uses `match` instead of `.unwrap()` at line 69)
 

package.json

@@ -4,11 +4,12 @@
   "private": true,
   "scripts": {
     "dev": "npm run dev:precheck & npm run dev:ollama & npm run dev:start",
-    "dev:start": "concurrently -n AI,NEXT -c cyan,green \"npm run dev:rust\" \"npm run dev:next\"",
+    "dev:start": "concurrently -n AI,BROWSE,NEXT -c cyan,magenta,green \"npm run dev:rust\" \"npm run dev:browser-use\" \"npm run dev:next\"",
     "dev:next": "next dev -p 3006",
-    "dev:precheck": "powershell -NoProfile -Command \"$targetPorts=3001,3006; netstat -ano | Select-String LISTENING | ForEach-Object { $line=$_.ToString(); foreach($p in $targetPorts){ if($line -match ('[:]'+$p+'\\s')){ $foundPid=($line -split '\\s+')[-1]; try{ Stop-Process -Id $foundPid -Force -ErrorAction SilentlyContinue; Write-Host ('Freed port '+$p) }catch{} } } }; exit 0\"",
+    "dev:precheck": "powershell -NoProfile -Command \"$targetPorts=3001,3006,3008; netstat -ano | Select-String LISTENING | ForEach-Object { $line=$_.ToString(); foreach($p in $targetPorts){ if($line -match ('[:]'+$p+'\\s')){ $foundPid=($line -split '\\s+')[-1]; try{ Stop-Process -Id $foundPid -Force -ErrorAction SilentlyContinue; Write-Host ('Freed port '+$p) }catch{} } } }; exit 0\"",
     "dev:ollama": "powershell -NoProfile -Command \"if (-not (Get-Process ollama -ErrorAction SilentlyContinue)) { Start-Process ollama -ArgumentList 'serve' -WindowStyle Hidden; Start-Sleep 3 }; exit 0\"",
     "dev:rust": "cd rust-ai && cargo run",
+    "dev:browser-use": "cd browser-use-service && python main.py",
     "build": "next build",
     "start": "npm run dev:next",
     "lint": "eslint"

rust-ai/src/main.rs

@@ -11,84 +11,9 @@ use std::sync::{Arc, Mutex};
 use tower_http::cors::{Any, CorsLayer};
 use tracing::{error, info};
 use uuid::Uuid;
-use reqwest::blocking::Client;
-use scraper::{Html, Selector};
-use rand::rngs::StdRng;
-use rand::SeedableRng;
 use rand::Rng;
 use std::time::Duration;
-use std::thread;
-
-// ── Facebook Scraper ───────────────────────────────────────────
-
-// List of proxies
-const PROXIES: &[&str] = &[
-    "http://user:pass@192.168.1.1:8080",
-    "http://user:pass@192.168.1.2:8080",
-    "http://user:pass@192.168.1.3:8080",
-    "http://user:pass@192.168.1.4:8080",
-    "http://user:pass@192.168.1.5:8080",
-];
-
-// List of user agents
-const USER_AGENTS: &[&str] = &[
-  "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3",
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.2 Safari/605.1.15",
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:61.0) Gecko/20100101 Firefox/61.0",
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.2 Safari/605.1.15",
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36",
-    // Add more user agents here
-];
-
-fn get_random_proxy() -> String {
-    let mut rng = rand::thread_rng();
-    PROXIES[rng.gen_range(0..PROXIES.len())].to_string()
-}
-
-fn get_random_user_agent() -> String {
-    let mut rng = rand::thread_rng();
-    USER_AGENTS[rng.gen_range(0..USER_AGENTS.len())].to_string()
-}
-
-fn scrape_conversations(url: &str) -> Result<Html, Box<dyn std::error::Error>> {
-    let proxy = get_random_proxy();
-    let user_agent = get_random_user_agent();
-    let client = Client::builder()
-        .proxy(reqwest::Proxy::all(proxy)?)
-        .build()?;
-
-    let response = client.get(url)
-        .header("User-Agent", user_agent)
-        .send()?;
-
-    if response.status().is_success() {
-        let html = response.text()?;
-        Ok(Html::parse_document(&html))
-    } else {
-        Err(format!("Failed to retrieve the page: {}", response.status()).into())
-    }
-}
-
-fn run_facebook_scraper() {
-    let url = "https://www.facebook.com/search/top/?q=need%20website%20create";
-    match scrape_conversations(url) {
-        Ok(soup) => {
-            // Example: Extract conversation data
-            let selector = Selector::parse("div.conversation").unwrap();
-            for element in soup.select(&selector) {
-                println!("Conversation: {}", element.inner_html());
-            }
-        }
-        Err(e) => {
-            error!("Facebook scraper error: {}", e);
-        }
-    }
-
-    // Introduce random delay
-    let mut rng = rand::thread_rng();
-    let delay = rng.gen_range(1..5); // Delay between 1 to 5 seconds
-    thread::sleep(Duration::from_secs(delay));
-}
+use std::time::{SystemTime, UNIX_EPOCH};
 
 // ── Shared state ───────────────────────────────────────────────
 
@@ -97,6 +22,46 @@ struct AppState {
     ollama_url: String,
     model: String,
     jobs: Vec<Job>,
+    leads: Arc<Mutex<LeadStore>>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+struct Lead {
+    title: String,
+    url: String,
+    source: String,
+    found_at: u64,
+    author: String,
+    date: String,
+    content: String,
+}
+
+struct LeadStore {
+    leads: Vec<Lead>,
+    max_size: usize,
+}
+
+impl LeadStore {
+    fn new(max_size: usize) -> Self {
+        Self { leads: Vec::new(), max_size }
+    }
+
+    fn push(&mut self, lead: Lead) {
+        // Deduplicate by URL
+        if !self.leads.iter().any(|l| l.url == lead.url) {
+            self.leads.insert(0, lead);
+            self.leads.truncate(self.max_size);
+        }
+    }
+
+    fn recent(&self, max_age_secs: u64, limit: usize) -> Vec<Lead> {
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+        self.leads.iter()
+            .filter(|l| now - l.found_at <= max_age_secs)
+            .take(limit)
+            .cloned()
+            .collect()
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -164,21 +129,57 @@ struct OllamaResponseMessage {
 
 // ── System prompt builder ─────────────────────────────────────
 
-fn build_system_prompt(jobs: &[Job]) -> String {
+fn format_leads_output(leads: &[Lead]) -> String {
+    if leads.is_empty() {
+        return "No new requests found yet.".to_string();
+    }
+    leads
+        .iter()
+        .enumerate()
+        .map(|(i, l)| {
+            let author = if l.author.is_empty() { "Unknown" } else { &l.author };
+            let date = if l.date.is_empty() { "Unknown" } else { &l.date[..l.date.len().min(10)] };
+            format!(
+                "{}. {}\n   {}\n   {}\n   {}",
+                i + 1,
+                author,
+                date,
+                l.title,
+                l.url
+            )
+        })
+        .collect::<Vec<_>>()
+        .join("\n")
+}
+
+fn build_system_prompt(jobs: &[Job], leads: &[Lead]) -> String {
     let job_list: Vec<String> = jobs
         .iter()
         .map(|j| format!("- {} ({}): {}", j.job_title, j.industry, j.description))
         .collect();
     let job_list_str = job_list.join("\n");
 
+    let lead_summary: Vec<String> = leads
+        .iter()
+        .map(|l| {
+            format!("{} | {} | {}", l.author, l.title, l.url)
+        })
+        .collect();
+    let lead_summary_str = if lead_summary.is_empty() {
+        "None yet.".to_string()
+    } else {
+        lead_summary.join("\n")
+    };
+
     format!(
-        "You are a Sales AI Assistant for Coast IT CRM. Your role is to help salespeople \
-         with tips, strategies, and guidance.\n\n\
+        "You are a Sales AI Assistant for Coast IT CRM.\n\n\
          Available job categories to target:\n{}\n\n\
-         Provide concise, actionable sales advice. When asked about a specific job category, \
-         give targeted tips on finding and engaging prospects in that field. \
-         Keep responses under 300 words unless asked for detail.",
-        job_list_str
+         Recent leads context:\n{}\n\n\
+         Rules:\n\
+         - When asked about leads, answer concisely under 150 words.\n\
+         - If asked to suggest a sales strategy, give brief actionable advice.\n\
+         - Be direct and professional. No fluff.",
+        job_list_str, lead_summary_str
     )
 }
 
@@ -194,7 +195,64 @@ async fn handle_chat(
         _ => return Err((axum::http::StatusCode::FORBIDDEN, "Forbidden".to_string())),
     }
 
-    let system_prompt = build_system_prompt(&state.jobs);
+    // Intercept lead listing requests — scrape on demand then return
+    let msg_lower = req.message.to_lowercase();
+    let msg_words: Vec<&str> = msg_lower.split_whitespace().collect();
+    let has_listing = msg_words.iter().any(|w| ["listings", "listing", "leads", "links", "lists"].contains(w));
+    let has_show = msg_lower.contains("show me") || msg_lower.contains("give me") || msg_lower.contains("pull");
+    let has_job = msg_words.contains(&"jobs") || msg_words.contains(&"job");
+    if has_listing || (has_show && has_job) || (has_show && msg_lower.contains("links")) || msg_lower.contains("recent leads") {
+        // Scrape fresh leads now — call both services
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+        let client = reqwest::Client::new();
+        let services = [
+            "http://localhost:3008/scrape/all".to_string(),
+        ];
+        for service_url in &services {
+            if let Ok(resp) = client
+                .post(service_url)
+                .timeout(Duration::from_secs(120))
+                .send()
+                .await
+            {
+                if let Ok(leads_data) = resp.json::<Vec<serde_json::Value>>().await {
+                    info!("Scraped {} leads from {}", leads_data.len(), service_url);
+                    let mut store = state.leads.lock().unwrap();
+                    for item in &leads_data {
+                        store.push(Lead {
+                            title: item["title"].as_str().unwrap_or("").chars().take(120).collect(),
+                            url: item["url"].as_str().unwrap_or("").to_string(),
+                            source: item["source"].as_str().unwrap_or("unknown").to_string(),
+                            found_at: now,
+                            author: item["author"].as_str().unwrap_or("").chars().take(60).collect(),
+                            date: item["date"].as_str().unwrap_or("").chars().take(30).collect(),
+                            content: item["content"].as_str().unwrap_or("").chars().take(300).collect(),
+                        });
+                    }
+                }
+            } else {
+                error!("Scraper service unreachable: {}", service_url);
+            }
+        }
+
+        let recent_leads = state.leads.lock().unwrap().recent(604800, 20);
+        let response = format_leads_output(&recent_leads);
+        let _ = sqlx::query(
+            "INSERT INTO ai_conversations (id, user_id, role, message, response) VALUES ($1, $2, $3, $4, $5)",
+        )
+        .bind(Uuid::new_v4())
+        .bind(Uuid::parse_str(&req.user_id).unwrap_or(Uuid::nil()))
+        .bind(&req.user_role)
+        .bind(&req.message)
+        .bind(&response)
+        .execute(&state.db)
+        .await;
+        return Ok(Json(ChatResponse { response }));
+    }
+
+    let recent_leads = state.leads.lock().unwrap().recent(604800, 20);
+
+    let system_prompt = build_system_prompt(&state.jobs, &recent_leads);
 
     let message_text = req.message.clone();
 
@@ -300,7 +358,7 @@ async fn main() {
         std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
     let ollama_url =
         std::env::var("OLLAMA_BASE_URL").unwrap_or_else(|_| "http://localhost:11434".to_string());
-    let model = std::env::var("AI_MODEL").unwrap_or_else(|_| "sam860/dolphin3-llama3.2:3b".to_string());
+    let model = std::env::var("AI_MODEL").unwrap_or_else(|_| "dolphin-phi".to_string());
     let host = std::env::var("AI_HOST").unwrap_or_else(|_| "0.0.0.0".to_string());
     let port: u16 = std::env::var("AI_PORT")
         .unwrap_or_else(|_| "3001".to_string())
@@ -332,11 +390,14 @@ async fn main() {
 
     info!("Connected to PostgreSQL");
 
+    let lead_store = Arc::new(Mutex::new(LeadStore::new(100)));
+
     let state = Arc::new(AppState {
         db,
         ollama_url,
         model,
         jobs,
+        leads: lead_store.clone(),
     });
 
     // CORS layer
@@ -359,20 +420,39 @@ async fn main() {
         .await
         .expect("Failed to bind address");
 
-    // Start Facebook scraper in a separate thread
-    let rng = Arc::new(Mutex::new(StdRng::from_entropy()));
+    // Start background scrapers
+    let bg_leads = lead_store.clone();
     tokio::task::spawn_blocking(move || {
+        let bg_client = reqwest::blocking::Client::builder()
+            .timeout(Duration::from_secs(120))
+            .build().ok();
         loop {
-            {
-                let _lock = rng.lock().unwrap();
-                run_facebook_scraper();
+            if let Some(ref c) = bg_client {
+                let urls = vec![
+                    "http://localhost:3008/scrape/all".to_string(),
+                ];
+                let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+                for url in &urls {
+                    if let Ok(resp) = c.post(url).send() {
+                        if let Ok(data) = resp.json::<Vec<serde_json::Value>>() {
+                            let mut store = bg_leads.lock().unwrap();
+                            for item in &data {
+                                store.push(Lead {
+                                    title: item["title"].as_str().unwrap_or("").chars().take(120).collect(),
+                                    url: item["url"].as_str().unwrap_or("").to_string(),
+                                    source: item["source"].as_str().unwrap_or("unknown").to_string(),
+                                    found_at: now,
+                                    author: item["author"].as_str().unwrap_or("").chars().take(60).collect(),
+                                    date: item["date"].as_str().unwrap_or("").chars().take(30).collect(),
+                                    content: item["content"].as_str().unwrap_or("").chars().take(300).collect(),
+                                });
+                            }
+                        }
+                    }
+                }
             }
-            // Introduce random delay
-            let delay = {
-                let mut rng = rng.lock().unwrap();
-                rng.gen_range(1..5)
-            };
-            thread::sleep(Duration::from_secs(delay));
+            let delay = rand::thread_rng().gen_range(120..300);
+            std::thread::sleep(Duration::from_secs(delay));
         }
     });