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

# 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

# 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}"

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

# 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

# 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",
        },
    )

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"]
}"""