Spaces:

Shurem
/

mz_chatbot

Running

mz_chatbot / app /services /lead_engine.py

added verification

35f3660 21 days ago

23.8 kB

"""
lead_engine.py — Sourcing, Phone Scrubbing & High-Throughput Enrichment Engine for Elipse Studio.
Features:
1. Apollo.io API integration for querying verified decision makers with direct mobile/desk phones.
2. Bulk ICP CSV Importer: Auto-maps columns from Apollo, ZoomInfo, Clay, or Sales Navigator exports.
3. Phone Scrub & Normalization: Formats phone numbers, flags switchboards (1-800/888), and eliminates bad dials.
4. Battlecard Synthesis: Pairs each enriched lead with Gemini Flash cold-call scripts and objection rebuttals.
"""

import os
import re
import csv
import io
import requests
from datetime import datetime

from app.services import agent_core, countries, scraper, validation
from app.services.secrets import get_apollo_key
from app.services.signalhire import role_bucket as _role_bucket

# Title-seniority tiers for picking which contact on a multi-contact CSV
# company becomes the lead's own primary contact_* fields -- same scoring
# scripts/import_september_leads.py used one-off; now built into the real
# CSV Import feature (see parse_and_enrich_csv's per-company grouping
# below), per user request 2026-09-18.
_SENIORITY_TIERS = [
    (("ceo", "chief executive", "chairman", "owner", "founder", "president"), 100),
    (("managing director", "group managing director"), 80),
    (("coo", "cfo", "cto", "cio", "chief operating", "chief financial",
      "chief technology", "chief information", "group director"), 60),
    (("director",), 40),
    (("manager", "head of"), 20),
]


def _seniority(title: str) -> int:
    t = (title or "").lower()
    for keywords, score in _SENIORITY_TIERS:
        if any(k in t for k in keywords):
            return score
    return 10


def scrub_and_format_phone(raw_phone: str, region: str = "", source: str = "") -> dict:
    """
    Parse a phone number and say honestly what is known about it.
    Delegates to services.validation, which uses libphonenumber. The version this
    replaced assumed any 10-digit number was North American and prefixed "+1", so
    a real London number "020 7946 0958" came out as "+1 (207) 946-0958"; and it
    reported status "verified_direct" for anything that merely was not toll-free.
    `region` is an ISO alpha-2 hint (derive it with validation.region_from_url on
    the company website). `source` decides whether a valid number may be called
    verified -- digits alone never prove that.
    Return shape is unchanged: {"phone", "status", "type"}.
    """
    got = validation.normalize_phone(raw_phone, region)
    status = validation.phone_status_for(got["kind"], source)
    phone_type = {
        "missing": "none",
        "invalid": "none",
        "toll_free": "switchboard",
        "direct": "direct_line",
    }[got["kind"]]
    return {
        "phone": got["phone"],
        "status": status,
        "type": phone_type,
        "country": got["country"],
    }

def query_apollo_leads(icp_prompt: str, limit: int = 50, api_key: str = None) -> dict:
    key = api_key or get_apollo_key()
    if not key:
        return {
            "success": False,
            "error": "APOLLO_API_KEY not configured. Enter your key in the settings drawer or upload an Apollo CSV export.",
            "leads": [],
        }

    url = "https://api.apollo.io/v1/mixed_people/search"
    headers = {
        "Content-Type": "application/json",
        "Cache-Control": "no-cache",
        "X-Api-Key": key,
    }

    seniorities = ["owner", "founder", "c_suite", "vp", "director"]
    payload = {
        "q_keywords": icp_prompt,
        "person_titles": ["Owner", "Founder", "President", "Chief Executive Officer", "VP Sales", "Director of Marketing"],
        "seniorities": seniorities,
        "page": 1,
        "per_page": min(limit, 100),
    }

    try:
        r = requests.post(url, json=payload, headers=headers, timeout=15)
        if r.status_code == 403:
            return {
                "success": False,
                "error": "Apollo Free Plan Notice: Apollo.io reserves their direct REST Search API for paid plans ($49/mo). Free options: 1) Use the 'Free AI Discovery' tab to generate leads via Gemini + Hunter.io, OR 2) Do your search on the free Apollo website, click 'Export CSV', and drop it into the 'Bulk CSV' tab!",
                "leads": [],
            }
        if r.status_code != 200:
            return {
                "success": False,
                "error": f"Apollo API Error ({r.status_code}): {r.text[:200]}",
                "leads": [],
            }

        data = r.json()
        people = data.get("people", [])
        parsed_leads = []

        for p in people:
            org = p.get("organization") or {}
            comp_name = org.get("name") or p.get("headline", "").split(" at ")[-1].strip()
            if not comp_name:
                continue

            phone_numbers = p.get("phone_numbers") or []
            raw_phone = phone_numbers[0].get("sanitized_number") if phone_numbers else ""
            if not raw_phone:
                raw_phone = org.get("primary_phone", {}).get("number") or ""

            name = f"{p.get('first_name', '')} {p.get('last_name', '')}".strip() or "Executive"
            role = p.get("title") or "Decision Maker"
            # Apollo often withholds the email on the free plan. Do NOT invent
            # one -- the old fallback produced contact@company.com for any org
            # with no domain, which reads as a real address downstream.
            email = p.get("email") or "unknown"
            _domain = (org.get("primary_domain") or "").strip()
            website = org.get("website_url") or (f"https://{_domain}" if _domain else "")

            # Resolved above so the number is parsed against the company's own
            # region rather than defaulting to +1.
            scrubbed_phone = scrub_and_format_phone(
                raw_phone, region=validation.region_from_url(website), source="apollo"
            )
            linkedin = p.get("linkedin_url") or org.get("linkedin_url") or ""
            industry = org.get("industry") or "⚡ Tech & Commercial Products"

            clean_cat = agent_core.map_to_standard_category(industry, comp_name)

            parsed_leads.append({
                "company_name": comp_name,
                "company_website": website,
                "contact_name": name,
                "contact_role": role,
                "contact_email": email,
                "contact_phone": scrubbed_phone["phone"],
                "contact_linkedin": linkedin,
                "industry_tag": clean_cat,
                # structure-plan.md Phase 8: default deal value is 10000 everywhere (was 25000 here).
                "deal_value": 10000.0,
                "phone_status": scrubbed_phone["status"],
                "source_prompt": f"Apollo ICP: {icp_prompt}",
            })

        return {
            "success": True,
            "leads": parsed_leads,
            "total_found": data.get("pagination", {}).get("total_entries", len(parsed_leads)),
        }

    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "leads": [],
        }


def parse_and_enrich_csv(
    csv_content: str, max_records: int = 300, is_duplicate=None
) -> dict:
    """
    Parses an Apollo-style CSV export into lead dicts, one per COMPANY (not
    per row) -- multiple rows sharing a company name are the exact shape a
    real Apollo/ZoomInfo/Sales Navigator export uses for several people at
    the same company (company fields repeated on every row), and the earlier
    version treated every row after the first as a plain duplicate and threw
    it away. Confirmed live 2026-09-18 importing a hand-built sheet the same
    shape: 135 of 192 contact rows were silently dropped this way.
    Each returned lead dict now also carries a "contacts" list -- every
    person parsed for that company, each with a title-seniority score. The
    caller (jobs/runners.py run_import) writes these through
    crud_contacts.upsert_contacts()/promote_primary(), same plumbing the
    AI-qualify funnel and SignalHire reveal already use, so a company
    imported here looks identical (multiple contacts, most senior one
    auto-picked as primary) to one discovered normally. A single-contact
    company (the common case) behaves exactly as before -- one contact in a
    list of one.
    is_duplicate: callable(company_name, company_website="") -> bool, checked ONCE per company (not per
    row) -- an already-known company skips its whole row group. The Streamlit version called
    db.is_duplicate() once per row; over remote MySQL that is up to 300 round-trips, so the caller passes a
    set-backed closure resolved in a single query. The website arg (2026-10-06) catches the same real
    company imported under a different name string -- see crud.is_duplicate()'s docstring; this is the one
    path that used to skip that check entirely (AI Discovery/Google Maps/manual Add a Lead already had it).
    """
    if is_duplicate is None:
        is_duplicate = lambda _name, _website="": False  # noqa: E731

    company_names = ["company", "company_name", "organization", "account_name", "business_name"]

    try:
        # A real header row isn't always line 1 -- an exported report can lead
        # with a title row and a blank row (e.g. "Qualified global yacht
        # manufacturer CRM pipeline" then an empty line) before the actual
        # "#,Priority,...,Company,..." header. Scan the first few lines for one
        # that contains a recognisable Company column and start DictReader
        # there instead of silently failing on the title row.
        raw_lines = csv_content.strip().splitlines()
        header_idx = 0
        for i, line in enumerate(raw_lines[:20]):
            cells = next(csv.reader([line]), [])
            normalized = {c.strip().lower().replace(" ", "_").replace("-", "_") for c in cells}
            if normalized & set(company_names):
                header_idx = i
                break
        reader = csv.DictReader(io.StringIO("\n".join(raw_lines[header_idx:])))
        headers = [h.strip() for h in reader.fieldnames] if reader.fieldnames else []
    except Exception as e:
        return {"success": False, "error": f"Invalid CSV file: {str(e)}", "leads": []}

    def match_col(possible_names):
        for h in headers:
            normalized = h.lower().replace(" ", "_").replace("-", "_")
            if normalized in possible_names:
                return h
        return None

    c_company = match_col(company_names)
    c_website = match_col(["website", "company_website", "domain", "website_url", "url", "official_website"])
    c_name = match_col(["name", "contact_name", "full_name", "person_name", "champion_name"])
    c_first = match_col(["first_name", "firstname"])
    c_last = match_col(["last_name", "lastname"])
    c_role = match_col(["title", "role", "job_title", "position", "champion_title"])
    c_email = match_col(["email", "contact_email", "work_email", "corporate_email", "primary_email"])
    c_phone = match_col(["phone", "direct_phone", "mobile_phone", "corporate_phone", "phone_number", "mobile", "cell", "work_phone"])
    c_linkedin = match_col(["linkedin", "linkedin_url", "person_linkedin_url", "profile_url", "contact_linkedin_url", "person_linkedin", "contact_linkedin", "company_linkedin_url", "company_linkedin"])
    c_industry = match_col(["industry", "category", "keywords", "tags"])
    c_deal = match_col(["revenue", "annual_revenue", "deal_value", "estimated_value"])
    c_country = match_col(["country", "region", "nation"])

    if not c_company:
        return {
            "success": False,
            "error": "Could not identify Company column in CSV. Ensure header contains 'Company', 'Company Name', or 'Organization'.",
            "leads": [],
        }

    # Pass 1: parse every row, grouped by normalized company name, in
    # first-seen order. No dedup/skip here yet -- a repeat company name is
    # exactly how a second contact at the same company shows up.
    groups: dict[str, dict] = {}
    order: list[str] = []
    for row in reader:
        comp = row.get(c_company, "").strip()
        if not comp:
            continue
        key = comp.lower()
        if key not in groups:
            groups[key] = {"company_name": comp, "rows": []}
            order.append(key)
        groups[key]["rows"].append(row)

    parsed_leads = []
    skipped_duplicates = 0

    for key in order:
        if len(parsed_leads) >= max_records:
            break

        comp = groups[key]["company_name"]
        rows = groups[key]["rows"]

        # Best-effort website hint for the dedupe check (2026-10-06) -- same "first non-empty value across
        # the group" the real website extraction below does, just done early enough to pass to
        # is_duplicate() before deciding whether to skip this group entirely.
        website_hint = ""
        if c_website:
            for row in rows:
                if row.get(c_website, "").strip():
                    website_hint = row[c_website].strip()
                    break

        if is_duplicate(comp, website_hint):
            skipped_duplicates += 1
            continue

        # Company-level fields: first non-empty value across the group --
        # covers both a "company info on the first row only" sheet and an
        # Apollo-style export that repeats them on every row.
        website = ""
        industry_val = ""
        country_val = ""
        # structure-plan.md Phase 8: default deal value is 10000 everywhere.
        deal_val = 10000.0
        deal_found = False
        for row in rows:
            if not website and c_website and row.get(c_website, "").strip():
                website = row[c_website].strip()
            if not industry_val and c_industry and row.get(c_industry, "").strip():
                industry_val = row[c_industry].strip()
            if not country_val and c_country and row.get(c_country, "").strip():
                country_val = row[c_country].strip()
            if not deal_found and c_deal and row.get(c_deal):
                try:
                    v = float(re.sub(r"[^\d.]", "", str(row[c_deal])))
                    deal_val = v if v >= 5000 else 10000.0
                    deal_found = True
                except Exception:
                    pass
        if website and not website.startswith("http"):
            website = f"https://{website}"
        if not industry_val:
            industry_val = "⚡ Tech & Commercial Products"
        # A CSV Country/Region column may already be a 2-letter code or a
        # full name ("USA", "United Arab Emirates") -- try both before
        # falling back to the website-TLD guess.
        if len(country_val) == 2:
            country_val = country_val.upper()
        elif country_val:
            country_val = countries.name_to_iso2(country_val)
        if not country_val:
            country_val = validation.region_from_url(website)
        clean_category = agent_core.map_to_standard_category(industry_val, comp)

        # Every person on this company's rows becomes a contact -- confidence
        # is title-seniority, so the most senior one is what
        # crud_contacts.promote_primary() picks as primary downstream.
        contacts = []
        for row in rows:
            if c_name and row.get(c_name):
                name = row.get(c_name, "").strip()
            elif c_first and row.get(c_first):
                name = f"{row.get(c_first, '').strip()} {row.get(c_last, '').strip()}".strip()
            else:
                name = "Decision Maker"
            role = row.get(c_role, "").strip() if c_role else "Executive"
            email = row.get(c_email, "").strip() if c_email else ""
            raw_phone = row.get(c_phone, "").strip() if c_phone else ""
            # A single cell often carries several numbers for one person,
            # comma-separated (confirmed live 2026-09-18) -- feeding that
            # whole string to phonenumbers as ONE number reliably fails to
            # parse at all, silently losing every number in the cell. Split
            # first, validate each candidate loosely (looks_like_phone), and
            # keep every one that passes -- rejoined for storage. A CSV the
            # user exported themselves counts as observed data, so a valid
            # direct number may be reported as such -- but the region still
            # comes from the company website, never a hardcoded +1.
            region = validation.region_from_url(website)
            phone_candidates = [
                p.strip() for p in raw_phone.split(",") if p.strip() and validation.looks_like_phone(p.strip())
            ]
            formatted_phones = []
            for cand in phone_candidates:
                fmt = scrub_and_format_phone(cand, region=region, source="manual")
                formatted_phones.append(fmt["phone"] or cand)
            # Status reflects only the first number's classification --
            # "is there at least one good number", not every one of them.
            first_status = (
                scrub_and_format_phone(phone_candidates[0], region=region, source="manual")["status"]
                if phone_candidates else "missing"
            )
            linkedin = row.get(c_linkedin, "").strip() if c_linkedin else ""
            contacts.append({
                "name": name,
                "role": role,
                "role_bucket": _role_bucket(role),
                "email": email,
                "phone": ", ".join(formatted_phones),
                "phone_status": first_status,
                "linkedin": linkedin,
                "source": "csv_import",
                "confidence": _seniority(role),
            })

        best = max(contacts, key=lambda c: (c["confidence"], bool(c["email"]) or bool(c["phone"])))

        parsed_leads.append({
            "company_name": comp,
            "company_website": website,
            "contact_name": best["name"],
            "contact_role": best["role"],
            "contact_email": best["email"] or "unknown",
            "contact_phone": best["phone"],
            "contact_linkedin": best["linkedin"],
            "industry_tag": clean_category,
            "country": country_val,
            "deal_value": deal_val,
            "phone_status": best["phone_status"],
            "source_prompt": "CSV Batch Import",
            "contacts": contacts,
        })

    return {
        "success": True,
        "leads": parsed_leads,
        "total_parsed": len(parsed_leads),
        "skipped_duplicates": skipped_duplicates,
    }


def _slugify(text: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", (text or "lead").lower()).strip("-")[:60] or "lead"


def synthesize_lead_battlecards(
    leads_list: list, progress_callback=None, enrich: bool = False, scrape: bool = False, save_callback=None
) -> list:
    """
    Despite the name, this no longer calls Gemini for a battlecard per lead --
    see the comment above the removed call below. It still, per lead:
    - scrape=True : run the Playwright site scan (scraper.check_configurator) to detect
      existing 3D/visualization tech, product title/description and a screenshot.
    - enrich=True : run SignalHire enrichment (agent_core.enrich_lead), but only for leads
      still missing an email or phone, to bound API-credit spend on large batches.
    save_callback: optional callable(lead_dict) -> None, invoked right after each lead
    is processed. When given, the caller is expected to persist that one row itself (see
    runners.run_import) instead of waiting on the full return list -- a large CSV used to
    do one all-or-nothing DB write at the very end, so a slow/failed run (e.g. an
    exhausted LLM quota during the scrape verdict step) lost every row already processed.
    With save_callback, rows land in the DB as they finish. When omitted, the old
    return-the-whole-list behavior is unchanged (used by the Apollo import path too).
    """
    total = len(leads_list)
    enriched = []

    for idx, lead in enumerate(leads_list):
        if progress_callback:
            progress_callback(idx + 1, total, lead["company_name"])

        site_note = ""  # must reset per lead — otherwise it leaks to the next one

        if scrape and lead.get("company_website"):
            try:
                sc = scraper.scan_site(
                    lead["company_website"],
                    screenshot_path=f"screenshots/{_slugify(lead['company_name'])}.png",
                )
            except Exception:
                sc = {}
            lead["has_3d"] = 1 if sc.get("has_3d") else 0
            lead["matched_signals"] = ", ".join(sc.get("matched_signals", []) or [])
            lead["product_title"] = sc.get("product_title", "")
            lead["product_description"] = sc.get("product_description", "")
            lead["screenshot_path"] = sc.get("screenshot_path", "")
            if not sc.get("error"):
                verdict, why = agent_core.site_matches_lead(
                    lead.get("company_name", ""),
                    lead.get("industry_tag", ""),
                    f"{sc.get('product_title','')} {sc.get('product_description','')}",
                    blocked=bool(sc.get("blocked")),
                )
                lead["site_verdict"] = verdict
                # Held aside and prepended to the battlecard reason below, so the
                # AI-written "how we help" text is never lost.
                if verdict == "blocked":
                    pass  # site refused the scanner — proves nothing, leave the lead alone
                elif verdict == "mismatch":
                    lead["lead_score"] = 20
                    site_note = (
                        f"⚠️ UNVERIFIED — {lead.get('company_website','the listed site')} does not match this company "
                        f"({why}). Confirm the business exists before outreach."
                    )
                elif not sc.get("has_3d"):
                    site_note = "No interactive 3D on their site yet — strong fit for an Elipse configurator / 3D visualization."

        if enrich:
            cur_email = (lead.get("contact_email") or "").strip().lower()
            if cur_email in ("", "unknown") or not (lead.get("contact_phone") or "").strip():
                try:
                    lead = agent_core.enrich_lead(lead)
                except Exception:
                    pass

        # Phone script / objection matrix / email body are generated on demand
        # from the lead card ("Generate battlecard" button -> POST
        # /leads/{id}/battlecard, see app/api/leads.py), not here -- a
        # Gemini/Groq call per imported row exhausted the free Gemini quota
        # (20 requests/day/model) and made a 300-row CSV import crawl through
        # the whole model cascade + Groq fallback for every single lead. Same
        # move already made for the AI Search / Google Maps discovery paths;
        # the lead card UI falls back to a generic script/objections template
        # until a real one is generated.
        base_reason = lead.get("reason") or (
            "High customization product catalog that converts higher with interactive real-time 3D web builder."
        )
        lead["reason"] = f"{site_note} {base_reason}".strip() if site_note else base_reason
        lead.setdefault("phone_script", "")
        lead.setdefault("objection_notes", "")
        lead["subject"] = lead.get("subject") or f"3D configurator concept for {lead['company_name']}"
        lead.setdefault("body", "")

        if save_callback:
            save_callback(lead)
        else:
            enriched.append(lead)

    return enriched

	"""
	lead_engine.py — Sourcing, Phone Scrubbing & High-Throughput Enrichment Engine for Elipse Studio.

	Features:
	1. Apollo.io API integration for querying verified decision makers with direct mobile/desk phones.
	2. Bulk ICP CSV Importer: Auto-maps columns from Apollo, ZoomInfo, Clay, or Sales Navigator exports.
	3. Phone Scrub & Normalization: Formats phone numbers, flags switchboards (1-800/888), and eliminates bad dials.
	4. Battlecard Synthesis: Pairs each enriched lead with Gemini Flash cold-call scripts and objection rebuttals.
	"""

	import os
	import re
	import csv
	import io
	import requests
	from datetime import datetime

	from app.services import agent_core, countries, scraper, validation
	from app.services.secrets import get_apollo_key
	from app.services.signalhire import role_bucket as _role_bucket

	# Title-seniority tiers for picking which contact on a multi-contact CSV
	# company becomes the lead's own primary contact_* fields -- same scoring
	# scripts/import_september_leads.py used one-off; now built into the real
	# CSV Import feature (see parse_and_enrich_csv's per-company grouping
	# below), per user request 2026-09-18.
	_SENIORITY_TIERS = [
	(("ceo", "chief executive", "chairman", "owner", "founder", "president"), 100),
	(("managing director", "group managing director"), 80),
	(("coo", "cfo", "cto", "cio", "chief operating", "chief financial",
	"chief technology", "chief information", "group director"), 60),
	(("director",), 40),
	(("manager", "head of"), 20),
	]


	def _seniority(title: str) -> int:
	t = (title or "").lower()
	for keywords, score in _SENIORITY_TIERS:
	if any(k in t for k in keywords):
	return score
	return 10


	def scrub_and_format_phone(raw_phone: str, region: str = "", source: str = "") -> dict:
	"""
	Parse a phone number and say honestly what is known about it.

	Delegates to services.validation, which uses libphonenumber. The version this
	replaced assumed any 10-digit number was North American and prefixed "+1", so
	a real London number "020 7946 0958" came out as "+1 (207) 946-0958"; and it
	reported status "verified_direct" for anything that merely was not toll-free.

	`region` is an ISO alpha-2 hint (derive it with validation.region_from_url on
	the company website). `source` decides whether a valid number may be called
	verified -- digits alone never prove that.

	Return shape is unchanged: {"phone", "status", "type"}.
	"""
	got = validation.normalize_phone(raw_phone, region)
	status = validation.phone_status_for(got["kind"], source)
	phone_type = {
	"missing": "none",
	"invalid": "none",
	"toll_free": "switchboard",
	"direct": "direct_line",
	}[got["kind"]]
	return {
	"phone": got["phone"],
	"status": status,
	"type": phone_type,
	"country": got["country"],
	}

	def query_apollo_leads(icp_prompt: str, limit: int = 50, api_key: str = None) -> dict:
	key = api_key or get_apollo_key()
	if not key:
	return {
	"success": False,
	"error": "APOLLO_API_KEY not configured. Enter your key in the settings drawer or upload an Apollo CSV export.",
	"leads": [],
	}

	url = "https://api.apollo.io/v1/mixed_people/search"
	headers = {
	"Content-Type": "application/json",
	"Cache-Control": "no-cache",
	"X-Api-Key": key,
	}

	seniorities = ["owner", "founder", "c_suite", "vp", "director"]
	payload = {
	"q_keywords": icp_prompt,
	"person_titles": ["Owner", "Founder", "President", "Chief Executive Officer", "VP Sales", "Director of Marketing"],
	"seniorities": seniorities,
	"page": 1,
	"per_page": min(limit, 100),
	}

	try:
	r = requests.post(url, json=payload, headers=headers, timeout=15)
	if r.status_code == 403:
	return {
	"success": False,
	"error": "Apollo Free Plan Notice: Apollo.io reserves their direct REST Search API for paid plans ($49/mo). Free options: 1) Use the 'Free AI Discovery' tab to generate leads via Gemini + Hunter.io, OR 2) Do your search on the free Apollo website, click 'Export CSV', and drop it into the 'Bulk CSV' tab!",
	"leads": [],
	}
	if r.status_code != 200:
	return {
	"success": False,
	"error": f"Apollo API Error ({r.status_code}): {r.text[:200]}",
	"leads": [],
	}

	data = r.json()
	people = data.get("people", [])
	parsed_leads = []

	for p in people:
	org = p.get("organization") or {}
	comp_name = org.get("name") or p.get("headline", "").split(" at ")[-1].strip()
	if not comp_name:
	continue

	phone_numbers = p.get("phone_numbers") or []
	raw_phone = phone_numbers[0].get("sanitized_number") if phone_numbers else ""
	if not raw_phone:
	raw_phone = org.get("primary_phone", {}).get("number") or ""

	name = f"{p.get('first_name', '')} {p.get('last_name', '')}".strip() or "Executive"
	role = p.get("title") or "Decision Maker"
	# Apollo often withholds the email on the free plan. Do NOT invent
	# one -- the old fallback produced contact@company.com for any org
	# with no domain, which reads as a real address downstream.
	email = p.get("email") or "unknown"
	_domain = (org.get("primary_domain") or "").strip()
	website = org.get("website_url") or (f"https://{_domain}" if _domain else "")

	# Resolved above so the number is parsed against the company's own
	# region rather than defaulting to +1.
	scrubbed_phone = scrub_and_format_phone(
	raw_phone, region=validation.region_from_url(website), source="apollo"
	)
	linkedin = p.get("linkedin_url") or org.get("linkedin_url") or ""
	industry = org.get("industry") or "⚡ Tech & Commercial Products"

	clean_cat = agent_core.map_to_standard_category(industry, comp_name)

	parsed_leads.append({
	"company_name": comp_name,
	"company_website": website,
	"contact_name": name,
	"contact_role": role,
	"contact_email": email,
	"contact_phone": scrubbed_phone["phone"],
	"contact_linkedin": linkedin,
	"industry_tag": clean_cat,
	# structure-plan.md Phase 8: default deal value is 10000 everywhere (was 25000 here).
	"deal_value": 10000.0,
	"phone_status": scrubbed_phone["status"],
	"source_prompt": f"Apollo ICP: {icp_prompt}",
	})

	return {
	"success": True,
	"leads": parsed_leads,
	"total_found": data.get("pagination", {}).get("total_entries", len(parsed_leads)),
	}

	except Exception as e:
	return {
	"success": False,
	"error": str(e),
	"leads": [],
	}


	def parse_and_enrich_csv(
	csv_content: str, max_records: int = 300, is_duplicate=None
	) -> dict:
	"""
	Parses an Apollo-style CSV export into lead dicts, one per COMPANY (not
	per row) -- multiple rows sharing a company name are the exact shape a
	real Apollo/ZoomInfo/Sales Navigator export uses for several people at
	the same company (company fields repeated on every row), and the earlier
	version treated every row after the first as a plain duplicate and threw
	it away. Confirmed live 2026-09-18 importing a hand-built sheet the same
	shape: 135 of 192 contact rows were silently dropped this way.

	Each returned lead dict now also carries a "contacts" list -- every
	person parsed for that company, each with a title-seniority score. The
	caller (jobs/runners.py run_import) writes these through
	crud_contacts.upsert_contacts()/promote_primary(), same plumbing the
	AI-qualify funnel and SignalHire reveal already use, so a company
	imported here looks identical (multiple contacts, most senior one
	auto-picked as primary) to one discovered normally. A single-contact
	company (the common case) behaves exactly as before -- one contact in a
	list of one.

	is_duplicate: callable(company_name, company_website="") -> bool, checked ONCE per company (not per
	row) -- an already-known company skips its whole row group. The Streamlit version called
	db.is_duplicate() once per row; over remote MySQL that is up to 300 round-trips, so the caller passes a
	set-backed closure resolved in a single query. The website arg (2026-10-06) catches the same real
	company imported under a different name string -- see crud.is_duplicate()'s docstring; this is the one
	path that used to skip that check entirely (AI Discovery/Google Maps/manual Add a Lead already had it).
	"""
	if is_duplicate is None:
	is_duplicate = lambda _name, _website="": False # noqa: E731

	company_names = ["company", "company_name", "organization", "account_name", "business_name"]

	try:
	# A real header row isn't always line 1 -- an exported report can lead
	# with a title row and a blank row (e.g. "Qualified global yacht
	# manufacturer CRM pipeline" then an empty line) before the actual
	# "#,Priority,...,Company,..." header. Scan the first few lines for one
	# that contains a recognisable Company column and start DictReader
	# there instead of silently failing on the title row.
	raw_lines = csv_content.strip().splitlines()
	header_idx = 0
	for i, line in enumerate(raw_lines[:20]):
	cells = next(csv.reader([line]), [])
	normalized = {c.strip().lower().replace(" ", "_").replace("-", "_") for c in cells}
	if normalized & set(company_names):
	header_idx = i
	break
	reader = csv.DictReader(io.StringIO("\n".join(raw_lines[header_idx:])))
	headers = [h.strip() for h in reader.fieldnames] if reader.fieldnames else []
	except Exception as e:
	return {"success": False, "error": f"Invalid CSV file: {str(e)}", "leads": []}

	def match_col(possible_names):
	for h in headers:
	normalized = h.lower().replace(" ", "_").replace("-", "_")
	if normalized in possible_names:
	return h
	return None

	c_company = match_col(company_names)
	c_website = match_col(["website", "company_website", "domain", "website_url", "url", "official_website"])
	c_name = match_col(["name", "contact_name", "full_name", "person_name", "champion_name"])
	c_first = match_col(["first_name", "firstname"])
	c_last = match_col(["last_name", "lastname"])
	c_role = match_col(["title", "role", "job_title", "position", "champion_title"])
	c_email = match_col(["email", "contact_email", "work_email", "corporate_email", "primary_email"])
	c_phone = match_col(["phone", "direct_phone", "mobile_phone", "corporate_phone", "phone_number", "mobile", "cell", "work_phone"])
	c_linkedin = match_col(["linkedin", "linkedin_url", "person_linkedin_url", "profile_url", "contact_linkedin_url", "person_linkedin", "contact_linkedin", "company_linkedin_url", "company_linkedin"])
	c_industry = match_col(["industry", "category", "keywords", "tags"])
	c_deal = match_col(["revenue", "annual_revenue", "deal_value", "estimated_value"])
	c_country = match_col(["country", "region", "nation"])

	if not c_company:
	return {
	"success": False,
	"error": "Could not identify Company column in CSV. Ensure header contains 'Company', 'Company Name', or 'Organization'.",
	"leads": [],
	}

	# Pass 1: parse every row, grouped by normalized company name, in
	# first-seen order. No dedup/skip here yet -- a repeat company name is
	# exactly how a second contact at the same company shows up.
	groups: dict[str, dict] = {}
	order: list[str] = []
	for row in reader:
	comp = row.get(c_company, "").strip()
	if not comp:
	continue
	key = comp.lower()
	if key not in groups:
	groups[key] = {"company_name": comp, "rows": []}
	order.append(key)
	groups[key]["rows"].append(row)

	parsed_leads = []
	skipped_duplicates = 0

	for key in order:
	if len(parsed_leads) >= max_records:
	break

	comp = groups[key]["company_name"]
	rows = groups[key]["rows"]

	# Best-effort website hint for the dedupe check (2026-10-06) -- same "first non-empty value across
	# the group" the real website extraction below does, just done early enough to pass to
	# is_duplicate() before deciding whether to skip this group entirely.
	website_hint = ""
	if c_website:
	for row in rows:
	if row.get(c_website, "").strip():
	website_hint = row[c_website].strip()
	break

	if is_duplicate(comp, website_hint):
	skipped_duplicates += 1
	continue

	# Company-level fields: first non-empty value across the group --
	# covers both a "company info on the first row only" sheet and an
	# Apollo-style export that repeats them on every row.
	website = ""
	industry_val = ""
	country_val = ""
	# structure-plan.md Phase 8: default deal value is 10000 everywhere.
	deal_val = 10000.0
	deal_found = False
	for row in rows:
	if not website and c_website and row.get(c_website, "").strip():
	website = row[c_website].strip()
	if not industry_val and c_industry and row.get(c_industry, "").strip():
	industry_val = row[c_industry].strip()
	if not country_val and c_country and row.get(c_country, "").strip():
	country_val = row[c_country].strip()
	if not deal_found and c_deal and row.get(c_deal):
	try:
	v = float(re.sub(r"[^\d.]", "", str(row[c_deal])))
	deal_val = v if v >= 5000 else 10000.0
	deal_found = True
	except Exception:
	pass
	if website and not website.startswith("http"):
	website = f"https://{website}"
	if not industry_val:
	industry_val = "⚡ Tech & Commercial Products"
	# A CSV Country/Region column may already be a 2-letter code or a
	# full name ("USA", "United Arab Emirates") -- try both before
	# falling back to the website-TLD guess.
	if len(country_val) == 2:
	country_val = country_val.upper()
	elif country_val:
	country_val = countries.name_to_iso2(country_val)
	if not country_val:
	country_val = validation.region_from_url(website)
	clean_category = agent_core.map_to_standard_category(industry_val, comp)

	# Every person on this company's rows becomes a contact -- confidence
	# is title-seniority, so the most senior one is what
	# crud_contacts.promote_primary() picks as primary downstream.
	contacts = []
	for row in rows:
	if c_name and row.get(c_name):
	name = row.get(c_name, "").strip()
	elif c_first and row.get(c_first):
	name = f"{row.get(c_first, '').strip()} {row.get(c_last, '').strip()}".strip()
	else:
	name = "Decision Maker"
	role = row.get(c_role, "").strip() if c_role else "Executive"
	email = row.get(c_email, "").strip() if c_email else ""
	raw_phone = row.get(c_phone, "").strip() if c_phone else ""
	# A single cell often carries several numbers for one person,
	# comma-separated (confirmed live 2026-09-18) -- feeding that
	# whole string to phonenumbers as ONE number reliably fails to
	# parse at all, silently losing every number in the cell. Split
	# first, validate each candidate loosely (looks_like_phone), and
	# keep every one that passes -- rejoined for storage. A CSV the
	# user exported themselves counts as observed data, so a valid
	# direct number may be reported as such -- but the region still
	# comes from the company website, never a hardcoded +1.
	region = validation.region_from_url(website)
	phone_candidates = [
	p.strip() for p in raw_phone.split(",") if p.strip() and validation.looks_like_phone(p.strip())
	]
	formatted_phones = []
	for cand in phone_candidates:
	fmt = scrub_and_format_phone(cand, region=region, source="manual")
	formatted_phones.append(fmt["phone"] or cand)
	# Status reflects only the first number's classification --
	# "is there at least one good number", not every one of them.
	first_status = (
	scrub_and_format_phone(phone_candidates[0], region=region, source="manual")["status"]
	if phone_candidates else "missing"
	)
	linkedin = row.get(c_linkedin, "").strip() if c_linkedin else ""
	contacts.append({
	"name": name,
	"role": role,
	"role_bucket": _role_bucket(role),
	"email": email,
	"phone": ", ".join(formatted_phones),
	"phone_status": first_status,
	"linkedin": linkedin,
	"source": "csv_import",
	"confidence": _seniority(role),
	})

	best = max(contacts, key=lambda c: (c["confidence"], bool(c["email"]) or bool(c["phone"])))

	parsed_leads.append({
	"company_name": comp,
	"company_website": website,
	"contact_name": best["name"],
	"contact_role": best["role"],
	"contact_email": best["email"] or "unknown",
	"contact_phone": best["phone"],
	"contact_linkedin": best["linkedin"],
	"industry_tag": clean_category,
	"country": country_val,
	"deal_value": deal_val,
	"phone_status": best["phone_status"],
	"source_prompt": "CSV Batch Import",
	"contacts": contacts,
	})

	return {
	"success": True,
	"leads": parsed_leads,
	"total_parsed": len(parsed_leads),
	"skipped_duplicates": skipped_duplicates,
	}


	def _slugify(text: str) -> str:
	return re.sub(r"[^a-z0-9]+", "-", (text or "lead").lower()).strip("-")[:60] or "lead"


	def synthesize_lead_battlecards(
	leads_list: list, progress_callback=None, enrich: bool = False, scrape: bool = False, save_callback=None
	) -> list:
	"""
	Despite the name, this no longer calls Gemini for a battlecard per lead --
	see the comment above the removed call below. It still, per lead:
	- scrape=True : run the Playwright site scan (scraper.check_configurator) to detect
	existing 3D/visualization tech, product title/description and a screenshot.
	- enrich=True : run SignalHire enrichment (agent_core.enrich_lead), but only for leads
	still missing an email or phone, to bound API-credit spend on large batches.

	save_callback: optional callable(lead_dict) -> None, invoked right after each lead
	is processed. When given, the caller is expected to persist that one row itself (see
	runners.run_import) instead of waiting on the full return list -- a large CSV used to
	do one all-or-nothing DB write at the very end, so a slow/failed run (e.g. an
	exhausted LLM quota during the scrape verdict step) lost every row already processed.
	With save_callback, rows land in the DB as they finish. When omitted, the old
	return-the-whole-list behavior is unchanged (used by the Apollo import path too).
	"""
	total = len(leads_list)
	enriched = []

	for idx, lead in enumerate(leads_list):
	if progress_callback:
	progress_callback(idx + 1, total, lead["company_name"])

	site_note = "" # must reset per lead — otherwise it leaks to the next one

	if scrape and lead.get("company_website"):
	try:
	sc = scraper.scan_site(
	lead["company_website"],
	screenshot_path=f"screenshots/{_slugify(lead['company_name'])}.png",
	)
	except Exception:
	sc = {}
	lead["has_3d"] = 1 if sc.get("has_3d") else 0
	lead["matched_signals"] = ", ".join(sc.get("matched_signals", []) or [])
	lead["product_title"] = sc.get("product_title", "")
	lead["product_description"] = sc.get("product_description", "")
	lead["screenshot_path"] = sc.get("screenshot_path", "")
	if not sc.get("error"):
	verdict, why = agent_core.site_matches_lead(
	lead.get("company_name", ""),
	lead.get("industry_tag", ""),
	f"{sc.get('product_title','')} {sc.get('product_description','')}",
	blocked=bool(sc.get("blocked")),
	)
	lead["site_verdict"] = verdict
	# Held aside and prepended to the battlecard reason below, so the
	# AI-written "how we help" text is never lost.
	if verdict == "blocked":
	pass # site refused the scanner — proves nothing, leave the lead alone
	elif verdict == "mismatch":
	lead["lead_score"] = 20
	site_note = (
	f"⚠️ UNVERIFIED — {lead.get('company_website','the listed site')} does not match this company "
	f"({why}). Confirm the business exists before outreach."
	)
	elif not sc.get("has_3d"):
	site_note = "No interactive 3D on their site yet — strong fit for an Elipse configurator / 3D visualization."

	if enrich:
	cur_email = (lead.get("contact_email") or "").strip().lower()
	if cur_email in ("", "unknown") or not (lead.get("contact_phone") or "").strip():
	try:
	lead = agent_core.enrich_lead(lead)
	except Exception:
	pass

	# Phone script / objection matrix / email body are generated on demand
	# from the lead card ("Generate battlecard" button -> POST
	# /leads/{id}/battlecard, see app/api/leads.py), not here -- a
	# Gemini/Groq call per imported row exhausted the free Gemini quota
	# (20 requests/day/model) and made a 300-row CSV import crawl through
	# the whole model cascade + Groq fallback for every single lead. Same
	# move already made for the AI Search / Google Maps discovery paths;
	# the lead card UI falls back to a generic script/objections template
	# until a real one is generated.
	base_reason = lead.get("reason") or (
	"High customization product catalog that converts higher with interactive real-time 3D web builder."
	)
	lead["reason"] = f"{site_note} {base_reason}".strip() if site_note else base_reason
	lead.setdefault("phone_script", "")
	lead.setdefault("objection_notes", "")
	lead["subject"] = lead.get("subject") or f"3D configurator concept for {lead['company_name']}"
	lead.setdefault("body", "")

	if save_callback:
	save_callback(lead)
	else:
	enriched.append(lead)

	return enriched