sales/lead_manager.py

import logging
from typing import Any, Dict, List, Optional
from sales.models import Lead, LeadStatus
from sqlalchemy.orm import Session

from core.automation_settings import get_automation_settings
from core.websockets import manager

logger = logging.getLogger(__name__)

try:
    from integrations.atom_communication_ingestion_pipeline import (
        CommunicationAppType,
        ingestion_pipeline,
    )
    INGESTION_AVAILABLE = True
except ImportError:
    INGESTION_AVAILABLE = False
    logger.warning("Ingestion pipeline not available. Sales memory will be disabled.")

class LeadManager:
    """
    Handles lead ingestion, qualification, and AI scoring.
    """
    def __init__(self, db: Session):
        self.db = db
        self.settings = get_automation_settings()

    async def ingest_lead(self, workspace_id: str, lead_data: Dict[str, Any]) -> Lead:
        """
        Normalize and ingest a lead from any source.
        """
        if not self.settings.is_sales_enabled():
            logger.info("Sales automations are disabled. Skipping lead ingestion.")
            return None

        # Check for duplicate
        existing = self.db.query(Lead).filter(
            Lead.workspace_id == workspace_id,
            Lead.email == lead_data["email"]
        ).first()

        if existing:
            logger.info(f"Lead {lead_data['email']} already exists. Updating...")
            for key, value in lead_data.items():
                if hasattr(existing, key):
                    setattr(existing, key, value)
            return existing

        lead = Lead(
            workspace_id=workspace_id,
            email=lead_data["email"],
            first_name=lead_data.get("first_name"),
            last_name=lead_data.get("last_name"),
            company=lead_data.get("company"),
            source=lead_data.get("source"),
            status=LeadStatus.NEW,
            metadata_json=lead_data.get("metadata", {})
        )

        self.db.add(lead)
        self.db.flush()

        # Trigger AI Scoring
        await self.score_lead(lead)

        return lead

    async def score_lead(self, lead: Lead):
        """
        Use AI to score the lead and detect spam/competitors.
        """
        # In a real implementation, this would call an LLM with lead context
        # For now, we'll implement a rule-based mock that simulates AI behavior
        
        email_domain = lead.email.split("@")[-1].lower()
        competitor_domains = ["competitor.com", "rival.io"]
        disposable_domains = ["mailinator.com", "tempmail.com"]

        score = 50.0 # Base score
        
        if email_domain in competitor_domains:
            lead.is_spam = True
            lead.status = LeadStatus.SPAM
            lead.ai_qualification_summary = "Detected as competitor research."
            score = 0.0
        elif email_domain in disposable_domains:
            lead.is_spam = True
            lead.ai_qualification_summary = "Disposable email address used."
            score = 10.0
        else:
            # Simulate positive signals
            if lead.company:
                score += 20.0
            if lead.source == "request_demo":
                score += 20.0
            
            lead.ai_qualification_summary = f"High intent lead from {lead.source}."

        lead.ai_score = score
        self.db.commit()
        logger.info(f"Scored lead {lead.email}: {score}")

        # Broadcast update
        try:
            await manager.broadcast(f"workspace:{lead.workspace_id}", {
                "type": "new_lead",
                "workspace_id": lead.workspace_id,
                "data": {
                    "id": lead.id,
                    "first_name": lead.first_name,
                    "last_name": lead.last_name,
                    "company": lead.company,
                    "ai_score": lead.ai_score,
                    "status": lead.status.value,
                    "summary": lead.ai_qualification_summary
                }
            })
        except Exception as e:
            logger.error(f"Failed to broadcast lead update: {e}")

        # Ingest into LanceDB Memory
        if INGESTION_AVAILABLE:
            try:
                from datetime import datetime
                ingestion_pipeline.ingest_message(
                    CommunicationAppType.CRM_LEAD.value,
                    {
                        "id": lead.id,
                        "timestamp": datetime.now().isoformat(),
                        "sender": lead.email,
                        "content": f"New Lead: {lead.first_name or ''} {lead.last_name or ''} from {lead.company or 'Unknown'}. Source: {lead.source}. AI Score: {lead.ai_score}. Summary: {lead.ai_qualification_summary}",
                        "metadata": {
                            "lead_id": lead.id,
                            "workspace_id": lead.workspace_id,
                            "company": lead.company,
                            "ai_score": lead.ai_score,
                            "status": lead.status.value
                        }
                    }
                )
            except Exception as e:
                logger.error(f"Failed to ingest lead into LanceDB: {e}")