lead_parser.py

import asyncio
import typer
import uuid
import os
from datetime import datetime
from dotenv import load_dotenv

from database import init_db, get_db_connection
from adapters.discovery import get_discovery_adapter
from core.dedupe import normalize_domain, is_duplicate
from core.crawler import Crawler
from core.extractor import get_email_confidence
from core.scoring import score_lead
from core.exporter import export_leads
from schema import Lead

load_dotenv()
app = typer.Typer()

@app.command()
def init():
    """Initialize the local SQLite database."""
    init_db()
    typer.echo("Database initialized.")

@app.command()
def discover(country: str, city: str, category: str, limit: int = 50):
    """Discover leads using search engines/directories."""
    typer.echo(f"Discovering {category} in {city}, {country} (limit: {limit})")
    
    adapter = get_discovery_adapter()
    results = asyncio.run(adapter.discover(city, category, limit))
    
    conn = get_db_connection()
    cursor = conn.cursor()
    
    added = 0
    for res in results:
        lead = Lead(
            id=str(uuid.uuid4()),
            business_name=res.business_name,
            category=res.category,
            country=country,
            city=res.city,
            website_url=res.website_url,
            normalized_domain=normalize_domain(res.website_url),
            phone=res.phone,
            source_type=res.source_type,
            source_url=res.source_url,
            created_at=datetime.utcnow().isoformat(),
            updated_at=datetime.utcnow().isoformat(),
        )
        
        if not is_duplicate(lead, conn):
            cursor.execute("""
                INSERT INTO leads (id, business_name, category, country, city, website_url, 
                                 normalized_domain, phone, source_type, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (lead.id, lead.business_name, lead.category, lead.country, lead.city, 
                  lead.website_url, lead.normalized_domain, lead.phone, lead.source_type, 
                  lead.created_at, lead.updated_at))
            added += 1
            
    conn.commit()
    conn.close()
    typer.echo(f"Added {added} new leads to database.")

@app.command()
def crawl(limit: int = 100):
    """Crawl discovered leads to extract emails and score them."""
    conn = get_db_connection()
    cursor = conn.cursor()
    
    # Get leads that haven't been crawled yet
    cursor.execute("SELECT * FROM leads WHERE status = 'discovered' LIMIT ?", (limit,))
    rows = cursor.fetchall()
    
    if not rows:
        typer.echo("No leads to crawl.")
        return

    crawler = Crawler()
    
    async def process_leads():
        for row in rows:
            row_dict = dict(row)
            for k, v in row_dict.items():
                if v is None and (k.startswith('has_') or k.endswith('_estimate') or k.endswith('_score')):
                    row_dict[k] = 0
                    
            lead = Lead(**row_dict)
            typer.echo(f"Crawling {lead.website_url} ...")
            
            crawl_data = await crawler.crawl_site(lead.website_url)
            
            emails = crawl_data['emails']
            if emails:
                # Naive: pick first high/medium confidence email
                best_email = None
                best_conf = 'low'
                
                for email in emails:
                    conf = get_email_confidence(email, lead.normalized_domain, 'crawl')
                    if conf == 'high':
                        best_email = email
                        best_conf = conf
                        break
                    elif conf == 'medium' and best_conf == 'low':
                        best_email = email
                        best_conf = conf
                
                if best_email:
                    lead.email = best_email
                    lead.email_confidence = best_conf
                    lead.status = "email_found"
                else:
                    lead.email = emails[0]
                    lead.email_confidence = get_email_confidence(emails[0], lead.normalized_domain, 'crawl')
                    lead.status = "email_found"
            else:
                lead.status = "no_email"
                
            lead.contact_page_url = crawl_data['contact_page_url']
            lead.last_checked_at = datetime.utcnow().isoformat()
            
            # Score Lead
            lead = score_lead(lead)
            
            # Update DB
            cursor.execute("""
                UPDATE leads 
                SET email = ?, email_confidence = ?, contact_page_url = ?, status = ?, 
                    website_quality_score = ?, redesign_opportunity_score = ?, lead_priority = ?,
                    suggested_offer = ?, last_checked_at = ?, updated_at = ?
                WHERE id = ?
            """, (lead.email, lead.email_confidence, lead.contact_page_url, lead.status,
                  lead.website_quality_score, lead.redesign_opportunity_score, lead.lead_priority,
                  lead.suggested_offer, lead.last_checked_at, lead.last_checked_at, lead.id))
            conn.commit()

        # Close crawler inside async loop
        await crawler.close()

    asyncio.run(process_leads())
    conn.close()
    typer.echo("Crawl completed.")

@app.command()
def export(output: str = "leads.csv", format: str = "csv"):
    """Export scored leads."""
    count = export_leads(output, format)
    typer.echo(f"Exported {count} leads to {output}")

@app.command()
def stats():
    """Print database stats."""
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT status, count(*) as count FROM leads GROUP BY status")
    rows = cursor.fetchall()
    
    typer.echo("Lead Stats:")
    for row in rows:
        typer.echo(f"  {row['status']}: {row['count']}")
    conn.close()

@app.command()
def enrich(limit: int = 50):
    """Run AI analysis to generate outreach messages and extract extra details."""
    from core.ai_analyzer import generate_outreach_message
    
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM leads WHERE status = 'email_found' AND outreach_message IS NULL LIMIT ?", (limit,))
    rows = cursor.fetchall()
    
    if not rows:
        typer.echo("No leads ready for enrichment.")
        return
        
    enriched = 0
    for row in rows:
        row_dict = dict(row)
        for k, v in row_dict.items():
            if v is None and (k.startswith('has_') or k.endswith('_estimate') or k.endswith('_score')):
                row_dict[k] = 0
                
        lead = Lead(**row_dict)
        typer.echo(f"Enriching {lead.business_name}...")
        message = generate_outreach_message(lead)
        
        if message:
            cursor.execute("UPDATE leads SET outreach_message = ? WHERE id = ?", (message, lead.id))
            conn.commit()
            enriched += 1
            
    conn.close()
    typer.echo(f"Enriched {enriched} leads.")

@app.command()
def screenshot(limit: int = 10):
    """Capture screenshots for leads."""
    from core.screenshot import capture_screenshot
    
    conn = get_db_connection()
    cursor = conn.cursor()
    # Ideally we add a has_screenshot column, but we can just screenshot top priority leads for now
    cursor.execute("SELECT id, website_url FROM leads WHERE lead_priority IN ('A', 'B') LIMIT ?", (limit,))
    rows = cursor.fetchall()
    
    if not rows:
        typer.echo("No high priority leads found to screenshot.")
        return
        
    os.makedirs("screenshots", exist_ok=True)
    
    async def run_screenshots():
        for row in rows:
            lead_id, url = row["id"], row["website_url"]
            output = f"screenshots/{lead_id}.png"
            if not os.path.exists(output):
                typer.echo(f"Capturing {url} ...")
                await capture_screenshot(url, output)
                
    asyncio.run(run_screenshots())
    typer.echo("Screenshots complete.")

@app.command()
def ui():
    """Launch the local Streamlit dashboard."""
    typer.echo("Starting local Streamlit dashboard...")
    os.system("streamlit run dashboard.py")

if __name__ == "__main__":
    app()