Suggest changes: Building a Real-Time B2B Lead Harvester and Dynamic Scoring Engine

Edit the text directly. Your proposal will remain pending until it is reviewed and approved. Only the content is displayed here — the title, date, and metadata cannot be changed from this form.

Content

Most B2B lead generation tools suffer from two major flaws: they rely on stale, expensive third-party databases that haven't been verified in months, or they depend on reckless scrapers and generative AI models that hallucinate fake businesses, non-existent phone numbers, and fabricated decision-makers.

When prospecting high-intent verticals—such as commercial real estate brokerages, property management firms, or community associations—you cannot afford to email imaginary companies or crawl domains that explicitly prohibit automated traffic. You need verified physical businesses, polite crawler behavior, deterministic signal extraction, dynamic multi-parameter scoring, and an AI outreach assistant strictly bounded by real facts.

I built **[LeadsPlus](https://github.com/PedroCaballero741/leadsplus)** to solve this end-to-end. It is a compliance-first Python web application powered by **FastAPI** with an embedded **Tailwind CSS** interface. It discovers real local businesses, respects `robots.txt`, crawls target websites with polite delays, extracts operational signals, computes custom opportunity scores, and streams verified leads in real time over Server-Sent Events (SSE).

Here is how the system is engineered from the ground up.

---

## Architecture Overview

The system is structured as an asynchronous pipeline with clean boundaries between discovery, compliance, extraction, scoring, and client-side streaming:

```mermaid
flowchart TB
    subgraph Client ["Client & Target Specification"]
        QUERY(["Target Location & Industry Criteria"])
    end

subgraph Discovery ["1. Pluggable Discovery Hierarchy (Zero Fabrication)"]
 direction TB
 GPLACES["Google Places API (New) Primary high-density local entity lookup"]
 4SQ["Foursquare Places API Secondary commercial directory fallback"]
 OSM["OpenStreetMap (Overpass QL) Spatial bounding-box fallback query"]
 NORM["Candidate Normalizer Schema standardization & deduplication"]

QUERY --> GPLACES
        GPLACES -.->|Quota exhausted or empty| 4SQ
        4SQ -.->|Fallback trigger| OSM
        GPLACES --> NORM
        4SQ --> NORM
        OSM --> NORM
    end

subgraph Compliance ["2. Compliance & Robots.txt Gate"]
 direction TB
 CACHE[("Domain Rules Cache Redis / In-memory TTL: 24h")]
 ROBOTS{"Robots.txt Permitted?"}
 QUEUE["Polite Delay Queue 0.5s min spacing per target host"]
 DISCARD(["Drop Candidate Log compliance audit record"])

subgraph Extraction ["3. Deep Website Extraction Pipeline"]
 direction TB
 BYPASS["Static Asset Bypass Filters out .png, .css, .js assets"]
 CRAWL["Subpage Crawler Inspects /about, /contact, /team"]
 REGEX["Contact & Signal Extractor Regex for emails, phones, LinkedIn profiles"]

QUEUE --> BYPASS
        BYPASS --> CRAWL
        CRAWL --> REGEX
    end

subgraph Scoring ["4. Dynamic Multi-Parameter Scoring Engine"]
 direction TB
 TECH["Tech Stack Signals Shopify, WordPress, custom portals"]
 VOLUME["Commercial Volume Signals Employee size & foot-traffic metrics"]
 REACH{"Reachability Gate Verified Direct Contact Found?"}
 TIER1["Tier 1 Qualified Lead High-intent profile + direct decision-maker contact"]
 TIER2["Tier 2 / 3 Lead Generic inbox or lower intent score"]

REGEX --> TECH
        REGEX --> VOLUME
        TECH --> REACH
        VOLUME --> REACH
        REACH -->|Yes| TIER1
        REACH -->|No| TIER2
    end

subgraph Delivery ["5. Real-Time Streaming & Outreach Synthesis"]
 direction TB
 SSE["FastAPI Event Stream (SSE) Live chunked progress events to frontend"]
 LLM["DeepSeek-V3 Outreach Engine Generates grounded cold email drafts"]
 DASHBOARD([("Browser Client Dashboard")])
 OUTREACH(["Personalized Email Drafts"])

TIER1 --> LLM
        TIER1 --> SSE
        TIER2 --> SSE
        SSE --> DASHBOARD
        LLM --> OUTREACH
    end
```

---

## 1. Zero-Fabrication Business Discovery

The foundation of any legitimate sales intelligence tool is that **it must never hallucinate a business**. If a query for property managers in a small municipality only yields 8 legitimate companies across all directory providers, the system must return exactly 8 leads—never 25 fabricated ones.

To accomplish this without locking the user into a single paid API, I built a pluggable discovery hierarchy with automated fallbacks across three providers:

1. **Google Places API (New) Text Search**: Offers the highest density and accuracy for local business entities, canonical websites, addresses, and phone numbers.
2. **Foursquare Places API**: Provides a secondary commercial venue directory.
3. **OpenStreetMap (Nominatim & Overpass)**: An open-source, zero-key fallback capable of querying commercial bounding boxes via spatial Overpass QL queries.

```python
# discovery.py - Provider selection and unified candidate normalization
def discover_businesses(
    city: str,
    query: str,
    limit: int = 25,
    logger_func: Optional[Callable[[str], None]] = None,
) -> List[dict]:
    providers = [
        p.strip().lower()
        for p in os.getenv("DISCOVERY_PROVIDERS", "google,foursquare,osm").split(",")
        if p.strip()
    ]
    seen_keys = set()
    leads = []

for provider in providers:
        if len(leads) >= limit:
            break
        candidates = []
        if provider == "google":
            candidates = search_google_places(city, query, limit - len(leads))
        elif provider == "foursquare":
            candidates = search_foursquare(city, query, limit - len(leads))
        elif provider == "osm":
            candidates = search_osm(city, query, limit - len(leads))

for cand in candidates:
            # Deduplicate by domain or normalized business name
            dedup_key = _domain_of(cand.get("website")) or cand["name"].lower()
            if dedup_key in seen_keys:
                continue
            seen_keys.add(dedup_key)
            leads.append(cand)
            if len(leads) >= limit:
                break

return leads
```

By normalizing all providers into a shared candidate dictionary, the downstream crawling and scoring pipeline remains completely agnostic to how the lead was initially found.

---

## 2. Compliance-First Crawling and robots.txt Caching

Unchecked crawlers cause IP bans, waste client bandwidth, and violate site terms. Every domain discovered by LeadsPlus is evaluated against its canonical `robots.txt` before any HTTP request touches the website.

Because parsing `robots.txt` over the network for every lead would introduce unacceptable latency, I implemented an in-memory robot parser cache:

```python
# pipeline.py - robots.txt evaluation with caching
def check_robots_txt(self, url: str) -> Tuple[bool, str]:
    try:
        parsed = urllib.parse.urlparse(url)
        domain = parsed.netloc.lower()
        if not domain:
            return True, "No domain to check"

if domain in self.robot_cache:
            rp, can_fetch = self.robot_cache[domain]
            return can_fetch, f"Cached robots.txt for {domain} -> {'ALLOWED' if can_fetch else 'DISALLOWED'}"

robots_url = f"{parsed.scheme}://{domain}/robots.txt"
        rp = urllib.robotparser.RobotFileParser()
        rp.set_url(robots_url)

try:
            resp = self.session.get(robots_url, timeout=(1.0, 1.5))
            if resp.status_code == 200:
                rp.parse(resp.text.splitlines())
            elif resp.status_code in (401, 403):
                # Standard crawler behavior: 401/403 robots.txt disallows all access
                self.robot_cache[domain] = (rp, False)
                return False, f"Robots.txt restricted (HTTP {resp.status_code}) -> Crawling Disallowed"
            else:
                rp.allow_all = True
        except Exception:
            rp.allow_all = True

can_fetch = rp.can_fetch(DEFAULT_HEADERS["User-Agent"], url)
        self.robot_cache[domain] = (rp, can_fetch)
        return can_fetch, f"Robots.txt check for {domain} -> {'ALLOWED' if can_fetch else 'DISALLOWED'}"
    except Exception as e:
        return True, f"Robots.txt check skipped due to error: {e}"
```

In addition to `robots.txt` verification, the crawler performs socket-level DNS pre-checks to immediately reject dead hostnames without waiting for HTTP socket timeouts, and enforces a mandatory 0.5-second polite crawl delay between requests.

---

## 3. Signal Extraction Without Scraping Junk

A notorious failure mode in naive web scrapers is email extraction. If you run a basic regex like `[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}` across raw HTML, you will end up scraping tracking pixels, image names (`logo.png@2x`), and error reporting tokens (`4837a@sentry.io` or `user@wixpress.com`).

To ensure contact integrity, LeadsPlus uses a two-stage filter:

```python
STATIC_EXT_RE = re.compile(
    r"\.(png|jpg|jpeg|gif|webp|svg|css|js|woff|woff2|ttf|eot|ico|pdf|zip|mp4|webm)($|\?)",
    re.IGNORECASE,
)

INVALID_EMAIL_DOMAINS = {
    "sentry.io", "wixpress.com", "example.com", "domain.com",
    "email.com", "yoursite.com", "placeholder.com",
}

def _clean_email(self, raw_email: Optional[str]) -> str:
 if not raw_email:
 return "Missing"
 email = raw_email.strip().lower()
 if STATIC_EXT_RE.search(email):
 return "Missing"
 parts = email.split("@")
 if len(parts) != 2 or parts[1] in INVALID_EMAIL_DOMAINS:
 return "Missing"
 if len(email) < 6 or len(email) > 80:
 return "Missing"
 return email
```

Beyond contact information, the crawler inspects the DOM for operational scale indicators (e.g. `(\d+)\s*(?:doors|units|listings|properties)`) and corporate social media links (LinkedIn company pages, Facebook business profiles, Instagram).

---

## 4. The Dynamic Scoring Engine and The Reachability Gate

Different B2B products require completely different qualification criteria. A software startup selling trust accounting automation to property managers cares about **AppFolio**, **Buildium**, or **Yardi** usage. A company selling deal room software to commercial brokerages looks for **CoStar**, **LoopNet**, and transaction volume metrics.

Rather than hardcoding evaluation heuristics, LeadsPlus implements a configurable multi-parameter scoring engine. Each parameter has a type (`keywords`, `numeric_threshold`, `regex`), an assigned weight, and an optional **mandatory knock-out** flag.

```python
# pipeline.py - Dynamic criteria evaluation and Reachability Gate
def _evaluate_lead_criteria(
    self,
    company_name: str,
    page_content: str,
    volume_metric: int,
    phone: str,
    email: str,
    is_unreachable: bool,
    parameters: List[DynamicCriterion],
    tier1_threshold: float = 0.70,
    tier2_threshold: float = 0.35,
) -> Tuple[float, str, str, List[str], Optional[str]]:
    matched_weights = []
    matched_descriptions = []

for param in parameters:
        # Match criteria across keywords, volume thresholds, or regex patterns
        is_matched, match_desc = self._test_parameter(param, page_content, volume_metric)
        if is_matched:
            matched_weights.append(param.weight)
            matched_descriptions.append(match_desc)
        elif param.is_mandatory:
            # Immediate knock-out: lead is disqualified regardless of other signals
            return 0.0, "Disqualified", f"[KNOCK-OUT: Failed {param.name}]", [], None

# Base score capped at 1.0
    base_score = min(1.0, sum(matched_weights))

# The Reachability Gate
    if is_unreachable:
        reachability_multiplier = 0.0
    else:
        has_phone = bool(phone and phone != "Missing")
        has_email = bool(email and email != "Missing")
        if has_phone and has_email:
            reachability_multiplier = 1.0
        else:
            reachability_multiplier = 0.7  # Single-channel penalty

opportunity_score = round(base_score * reachability_multiplier, 2)

if opportunity_score >= tier1_threshold:
        tier = "Tier 1 (High)"
    elif opportunity_score >= tier2_threshold:
        tier = "Tier 2 (Mid)"
    else:
        tier = "Tier 3 (Low)"

return opportunity_score, tier, "; ".join(matched_descriptions), [], None
```

### Why the Reachability Gate Matters

A prospect might match 100% of your ideal customer profile (ICP)—they run your competitor's software, have 1,000 units, and run an investor portal. But if the website lists neither an email address nor a phone number, their immediate outbound value is zero.

The **Reachability Gate** addresses this directly:
- **Both Phone and Email present:** Multiplier = `1.0` (Full score).
- **Only One Channel present:** Multiplier = `0.7` (30% penalty reflecting outreach friction).
- **Both Missing:** Multiplier = `0.0` (Knock-out into Tier 3 / Unreachable).

This prevents sales reps from spending time researching high-fit companies that cannot be contacted.

---

## 5. Streaming Live Progress with Server-Sent Events

Crawling 25 websites with polite delays takes between 15 and 45 seconds. Traditional REST endpoints that block until the entire harvest completes provide a terrible user experience: the client stares at a spinner with no feedback on which sites are being analyzed, which domains blocked the crawler, or which leads qualified.

FastAPI makes it straightforward to stream live pipeline execution directly to the browser using Server-Sent Events (`text/event-stream`):

```python
# app.py - SSE Streaming Endpoint
@app.post("/api/harvest")
async def harvest_leads(req: HarvestRequest):
    async def event_generator():
        harvester = LeadHarvester()
        async for event in harvester.harvest(
            city=req.city,
            search_objective=req.search_objective,
            limit=req.limit,
            custom_config=req.scoring_config,
        ):
            yield f"data: {json.dumps(event)}\n\n"

return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )
```

In the browser, a standard `fetch` call reads the response stream chunk by chunk, dispatching discovery logs, crawler decisions, and validated lead cards to the UI in real time.

---

## 6. Grounded AI Cold Outreach with DeepSeek

Once a lead is scored, drafting personalized outreach usually becomes a manual bottleneck. However, delegating outreach copy to an unconstrained LLM often produces generic fluff or, worse, made-up claims ("I saw you manage 5,000 units in Dallas!").

LeadsPlus integrates **DeepSeek** (`deepseek-chat`) via an OpenAI-compatible API interface with a strictly constrained prompt:

```python
SYSTEM_PROMPT = """You are a B2B sales development assistant who writes short, honest, highly personalized cold emails.

Rules:
- Use ONLY the facts provided about the lead. Never invent numbers, software, people, clients, or results.
- Never invent a person's name. If no contact first name is provided, open with "Hi there," or "Hi {Company} team,".
- 70-120 words in the body. Plain text, no markdown, no emojis, no spammy words.
- One clear, low-friction call to action (e.g. a short call or a reply).
- Sign off with the sender name exactly as given.

Return ONLY a JSON object:
{
  "subject": "short subject line, under 60 characters",
  "body": "full email body including greeting and signature",
  "summary": "1-2 sentence summary of why this lead is a good fit",
  "contact_role": "the most likely decision-maker job title (a role, not a name)",
  "personalization_hooks": ["up to 3 concrete facts from the data that the email uses"]
}"""
```

If the API key is not configured or an error occurs, the engine automatically falls back to an offline deterministic template engine, ensuring the application remains fully functional in any environment.

---

## Results and Codebase

By combining multi-source real directory discovery, polite crawling, dynamic multi-parameter scoring, and streaming Server-Sent Events, LeadsPlus demonstrates that lead generation doesn't need to depend on bloated corporate databases or hallucinating AI models.

The entire project is open-source and ready to run locally:

- **GitHub Repository:** [PedroCaballero741/leadsplus](https://github.com/PedroCaballero741/leadsplus)
- **Tech Stack:** Python 3.12+, FastAPI, BeautifulSoup4, Uvicorn, DeepSeek API, Google Places API, OpenStreetMap, Tailwind CSS.