lead_gen/utils/database.py

"""
utils/database.py — SQLite persistence layer for leads.

Provides CRUD operations and cross-run deduplication so leads
survive server restarts and are never scraped twice.

Uses WAL journal mode for concurrent read access from Flask threads.
"""

import os
import re
import shutil
import sqlite3
import threading
from datetime import datetime
from typing import Dict, List, Optional, Tuple

import config
from models import Lead
from utils.logger import get_logger

logger = get_logger(__name__)

# Thread-local storage for connections
_local = threading.local()

# ═══════════════════════════════════════════════════════════════════════════════
# Schema
# ═══════════════════════════════════════════════════════════════════════════════

_SCHEMA = """
CREATE TABLE IF NOT EXISTS leads (
    id                  INTEGER PRIMARY KEY AUTOINCREMENT,
    business_name       TEXT NOT NULL,
    service_category    TEXT DEFAULT '',
    phone               TEXT DEFAULT '',
    email               TEXT DEFAULT '',
    address             TEXT DEFAULT '',
    website             TEXT DEFAULT '',
    contact_person      TEXT DEFAULT '',
    source              TEXT DEFAULT '',
    lead_source_scraper TEXT DEFAULT '',
    notes               TEXT DEFAULT '',
    user_notes          TEXT DEFAULT '',
    score               TEXT DEFAULT 'LOW',
    score_points        INTEGER DEFAULT 0,
    industry            TEXT DEFAULT '',
    product_category    TEXT DEFAULT '',
    status              TEXT DEFAULT 'New',
    fingerprint         TEXT DEFAULT '',

    -- Website analysis flags
    has_https           INTEGER DEFAULT 0,
    is_mobile_friendly  INTEGER DEFAULT 1,
    is_outdated         INTEGER DEFAULT 0,
    has_chatbot         INTEGER DEFAULT 0,
    has_online_booking  INTEGER DEFAULT 0,
    social_facebook     TEXT DEFAULT '',
    social_instagram    TEXT DEFAULT '',
    has_contact_form    INTEGER DEFAULT 0,
    ssl_expiry_days     INTEGER,
    page_speed_score    REAL,
    load_time_sec       REAL,
    site_status         TEXT DEFAULT '',
    gmb_exists          INTEGER DEFAULT 0,
    review_count        INTEGER DEFAULT 0,
    opportunity_pitch   TEXT DEFAULT '',

    -- Timestamps
    scraped_at          TEXT DEFAULT '',
    updated_at          TEXT DEFAULT '',

    -- Dedup
    UNIQUE(fingerprint)
);

CREATE INDEX IF NOT EXISTS idx_leads_score ON leads(score);
CREATE INDEX IF NOT EXISTS idx_leads_status ON leads(status);
CREATE INDEX IF NOT EXISTS idx_leads_service ON leads(service_category);
CREATE INDEX IF NOT EXISTS idx_leads_fingerprint ON leads(fingerprint);
CREATE INDEX IF NOT EXISTS idx_leads_name ON leads(business_name COLLATE NOCASE);
"""


# ═══════════════════════════════════════════════════════════════════════════════
# Connection management
# ═══════════════════════════════════════════════════════════════════════════════

def _get_conn() -> sqlite3.Connection:
    """Get a thread-local SQLite connection with WAL mode enabled."""
    if not hasattr(_local, "conn") or _local.conn is None:
        _local.conn = sqlite3.connect(
            config.DB_PATH,
            check_same_thread=False,
            timeout=10,
        )
        _local.conn.execute("PRAGMA journal_mode=WAL")
        _local.conn.execute("PRAGMA busy_timeout=5000")
        _local.conn.execute("PRAGMA foreign_keys=ON")
        _local.conn.row_factory = sqlite3.Row
    return _local.conn


def init_db() -> None:
    """Create tables and indexes if they don't exist. Backup if configured."""
    # Backup existing DB
    if config.DB_BACKUP_ON_STARTUP and os.path.isfile(config.DB_PATH):
        bak = config.DB_PATH + ".bak"
        try:
            shutil.copy2(config.DB_PATH, bak)
            logger.debug(f"Database backed up to {bak}")
        except Exception as e:
            logger.warning(f"DB backup failed: {e}")

    conn = _get_conn()
    conn.executescript(_SCHEMA)
    conn.commit()
    _maybe_migrate(conn)
    logger.info(f"Database initialised at {config.DB_PATH}")


def _maybe_migrate(conn: sqlite3.Connection) -> None:
    """Add any missing columns from newer schema versions."""
    cursor = conn.execute("PRAGMA table_info(leads)")
    existing_cols = {row["name"] for row in cursor.fetchall()}

    # Columns added in v2 — add if missing
    v2_columns = {
        "contact_person":      "TEXT DEFAULT ''",
        "lead_source_scraper": "TEXT DEFAULT ''",
        "user_notes":          "TEXT DEFAULT ''",
        "industry":            "TEXT DEFAULT ''",
        "product_category":    "TEXT DEFAULT ''",
        "status":              "TEXT DEFAULT 'New'",
        "social_facebook":     "TEXT DEFAULT ''",
        "social_instagram":    "TEXT DEFAULT ''",
        "has_contact_form":    "INTEGER DEFAULT 0",
        "ssl_expiry_days":     "INTEGER",
        "page_speed_score":    "REAL",
        "load_time_sec":       "REAL",
        "site_status":         "TEXT DEFAULT ''",
        "gmb_exists":          "INTEGER DEFAULT 0",
        "review_count":        "INTEGER DEFAULT 0",
        "opportunity_pitch":   "TEXT DEFAULT ''",
    }

    for col_name, col_def in v2_columns.items():
        if col_name not in existing_cols:
            try:
                conn.execute(f"ALTER TABLE leads ADD COLUMN {col_name} {col_def}")
                logger.info(f"Migration: added column '{col_name}'")
            except sqlite3.OperationalError:
                pass  # Column already exists

    conn.commit()


# ═══════════════════════════════════════════════════════════════════════════════
# CRUD Operations
# ═══════════════════════════════════════════════════════════════════════════════

def insert_lead(lead: Lead) -> Optional[int]:
    """
    Insert a Lead into the database.

    Returns the row ID on success, or None if the lead already exists
    (fingerprint conflict).
    """
    conn = _get_conn()
    try:
        lead.stamp_updated()
        if not lead.scraped_at:
            lead.stamp_scraped()

        cursor = conn.execute("""
            INSERT OR IGNORE INTO leads (
                business_name, service_category, phone, email, address, website,
                contact_person, source, lead_source_scraper, notes, user_notes,
                score, score_points, industry, product_category, status,
                fingerprint, has_https, is_mobile_friendly, is_outdated,
                has_chatbot, has_online_booking, social_facebook,
                social_instagram, has_contact_form, ssl_expiry_days,
                page_speed_score, load_time_sec, site_status, gmb_exists,
                review_count, opportunity_pitch, scraped_at, updated_at
            ) VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
        """, (
            lead.business_name, lead.service_category, lead.phone, lead.email,
            lead.address, lead.website, lead.contact_person, lead.source,
            lead.lead_source_scraper, lead.notes, lead.user_notes,
            lead.score, lead.score_points, lead.industry, lead.product_category,
            lead.status, lead.fingerprint,
            int(lead.has_https), int(lead.is_mobile_friendly),
            int(lead.is_outdated), int(lead.has_chatbot),
            int(lead.has_online_booking), lead.social_facebook,
            lead.social_instagram, int(lead.has_contact_form),
            lead.ssl_expiry_days, lead.page_speed_score, lead.load_time_sec,
            lead.site_status, int(lead.gmb_exists), lead.review_count,
            lead.opportunity_pitch, lead.scraped_at, lead.updated_at,
        ))
        conn.commit()

        if cursor.rowcount > 0:
            lead.db_id = cursor.lastrowid
            return cursor.lastrowid
        return None  # Already existed

    except sqlite3.Error as e:
        logger.error(f"DB insert error: {e}")
        conn.rollback()
        return None


def insert_leads(leads: List[Lead]) -> Tuple[int, int]:
    """Insert multiple leads. Returns (inserted_count, skipped_count)."""
    inserted = 0
    skipped = 0
    for lead in leads:
        row_id = insert_lead(lead)
        if row_id:
            inserted += 1
        else:
            skipped += 1
    logger.info(f"Bulk insert: {inserted} new, {skipped} skipped (duplicates)")
    return inserted, skipped


def lead_exists(fingerprint: str) -> bool:
    """Check if a lead with this fingerprint already exists in the DB."""
    conn = _get_conn()
    row = conn.execute(
        "SELECT 1 FROM leads WHERE fingerprint = ? LIMIT 1",
        (fingerprint,)
    ).fetchone()
    return row is not None


def get_all_leads(
    page: int = 1,
    per_page: int = 25,
    score_filter: str = "",
    service_filter: str = "",
    status_filter: str = "",
    search: str = "",
    sort_by: str = "score_points",
    sort_dir: str = "DESC",
) -> Tuple[List[Lead], int]:
    """
    Retrieve leads from the database with filtering, sorting, and pagination.
    Returns (list_of_leads, total_count).
    """
    conn = _get_conn()

    where_clauses = []
    params = []

    if score_filter and score_filter.upper() != "ALL":
        where_clauses.append("score = ?")
        params.append(score_filter.upper())

    if service_filter and service_filter.lower() != "all":
        where_clauses.append("service_category = ?")
        params.append(service_filter.lower())

    if status_filter and status_filter != "all":
        where_clauses.append("status = ?")
        params.append(status_filter)

    if search:
        where_clauses.append(
            "(business_name LIKE ? OR email LIKE ? OR phone LIKE ? OR address LIKE ?)"
        )
        like = f"%{search}%"
        params.extend([like, like, like, like])

    where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""

    # Validate sort column to prevent SQL injection
    allowed_sorts = {
        "score_points", "business_name", "service_category",
        "score", "status", "scraped_at", "updated_at",
    }
    if sort_by not in allowed_sorts:
        sort_by = "score_points"

    sort_dir = "DESC" if sort_dir.upper() == "DESC" else "ASC"

    # Total count
    count_sql = f"SELECT COUNT(*) FROM leads{where_sql}"
    total = conn.execute(count_sql, params).fetchone()[0]

    # Paginated query
    offset = (page - 1) * per_page
    data_sql = (
        f"SELECT * FROM leads{where_sql} "
        f"ORDER BY {sort_by} {sort_dir} "
        f"LIMIT ? OFFSET ?"
    )
    rows = conn.execute(data_sql, params + [per_page, offset]).fetchall()

    leads = [_row_to_lead(row) for row in rows]
    return leads, total


def get_lead_by_id(db_id: int) -> Optional[Lead]:
    """Retrieve a single lead by its database ID."""
    conn = _get_conn()
    row = conn.execute("SELECT * FROM leads WHERE id = ?", (db_id,)).fetchone()
    return _row_to_lead(row) if row else None


def update_lead_status(db_id: int, status: str) -> bool:
    """Update the status of a lead (New/Contacted/Converted/Rejected)."""
    valid = {"New", "Contacted", "Converted", "Rejected"}
    if status not in valid:
        logger.warning(f"Invalid status: {status}")
        return False

    conn = _get_conn()
    conn.execute(
        "UPDATE leads SET status = ?, updated_at = ? WHERE id = ?",
        (status, datetime.now().isoformat(), db_id),
    )
    conn.commit()
    return True


def bulk_update_status(db_ids: List[int], status: str) -> int:
    """Update status for multiple leads. Returns count updated."""
    valid = {"New", "Contacted", "Converted", "Rejected"}
    if status not in valid:
        return 0

    conn = _get_conn()
    now = datetime.now().isoformat()
    placeholders = ",".join("?" * len(db_ids))
    conn.execute(
        f"UPDATE leads SET status = ?, updated_at = ? WHERE id IN ({placeholders})",
        [status, now] + db_ids,
    )
    conn.commit()
    return len(db_ids)


def update_lead_notes(db_id: int, notes: str) -> bool:
    """Update user notes for a lead."""
    conn = _get_conn()
    conn.execute(
        "UPDATE leads SET user_notes = ?, updated_at = ? WHERE id = ?",
        (notes, datetime.now().isoformat(), db_id),
    )
    conn.commit()
    return True


def update_lead_full(lead: Lead) -> bool:
    """Update all fields of an existing lead."""
    if lead.db_id is None:
        return False
    conn = _get_conn()
    lead.stamp_updated()
    conn.execute("""
        UPDATE leads SET
            business_name=?, service_category=?, phone=?, email=?, address=?,
            website=?, contact_person=?, source=?, lead_source_scraper=?,
            notes=?, score=?, score_points=?, industry=?, product_category=?,
            has_https=?, is_mobile_friendly=?, is_outdated=?, has_chatbot=?,
            has_online_booking=?, social_facebook=?, social_instagram=?,
            has_contact_form=?, ssl_expiry_days=?, page_speed_score=?,
            load_time_sec=?, site_status=?, gmb_exists=?, review_count=?,
            opportunity_pitch=?, updated_at=?
        WHERE id=?
    """, (
        lead.business_name, lead.service_category, lead.phone, lead.email,
        lead.address, lead.website, lead.contact_person, lead.source,
        lead.lead_source_scraper, lead.notes, lead.score, lead.score_points,
        lead.industry, lead.product_category,
        int(lead.has_https), int(lead.is_mobile_friendly),
        int(lead.is_outdated), int(lead.has_chatbot),
        int(lead.has_online_booking), lead.social_facebook,
        lead.social_instagram, int(lead.has_contact_form),
        lead.ssl_expiry_days, lead.page_speed_score, lead.load_time_sec,
        lead.site_status, int(lead.gmb_exists), lead.review_count,
        lead.opportunity_pitch, lead.updated_at, lead.db_id,
    ))
    conn.commit()
    return True


def get_stats() -> Dict:
    """Aggregate statistics for the dashboard."""
    conn = _get_conn()

    total = conn.execute("SELECT COUNT(*) FROM leads").fetchone()[0]

    scores = {}
    for row in conn.execute("SELECT score, COUNT(*) as c FROM leads GROUP BY score"):
        scores[row["score"]] = row["c"]

    services = {}
    for row in conn.execute("SELECT service_category, COUNT(*) as c FROM leads GROUP BY service_category"):
        services[row["service_category"]] = row["c"]

    statuses = {}
    for row in conn.execute("SELECT status, COUNT(*) as c FROM leads GROUP BY status"):
        statuses[row["status"]] = row["c"]

    avg_score = conn.execute("SELECT AVG(score_points) FROM leads").fetchone()[0] or 0

    analysis = conn.execute("""
        SELECT
            ROUND(100.0 * SUM(CASE WHEN website != '' THEN 1 ELSE 0 END) / MAX(COUNT(*), 1), 1) as has_website_pct,
            ROUND(100.0 * SUM(CASE WHEN website != '' AND has_https = 0 THEN 1 ELSE 0 END) / MAX(COUNT(*), 1), 1) as no_https_pct,
            ROUND(100.0 * SUM(CASE WHEN is_outdated = 1 THEN 1 ELSE 0 END) / MAX(COUNT(*), 1), 1) as outdated_pct,
            ROUND(100.0 * SUM(CASE WHEN has_chatbot = 0 AND has_online_booking = 0 THEN 1 ELSE 0 END) / MAX(COUNT(*), 1), 1) as no_automation_pct
        FROM leads
    """).fetchone()

    return {
        "total": total,
        "by_score": {
            "HIGH": scores.get("HIGH", 0),
            "MEDIUM": scores.get("MEDIUM", 0),
            "LOW": scores.get("LOW", 0),
        },
        "by_service": {
            "web": services.get("web", 0),
            "app": services.get("app", 0),
            "ai": services.get("ai", 0),
        },
        "by_status": {
            "New": statuses.get("New", 0),
            "Contacted": statuses.get("Contacted", 0),
            "Converted": statuses.get("Converted", 0),
            "Rejected": statuses.get("Rejected", 0),
        },
        "avg_score": round(avg_score, 1),
        "has_website_pct": analysis["has_website_pct"] or 0,
        "no_https_pct": analysis["no_https_pct"] or 0,
        "outdated_pct": analysis["outdated_pct"] or 0,
        "no_automation_pct": analysis["no_automation_pct"] or 0,
    }


def get_top_leads(limit: int = 5) -> List[Lead]:
    """Get top N leads by score points."""
    conn = _get_conn()
    rows = conn.execute(
        "SELECT * FROM leads ORDER BY score_points DESC LIMIT ?", (limit,)
    ).fetchall()
    return [_row_to_lead(row) for row in rows]


def delete_all_leads() -> int:
    """Delete ALL leads from the database. Returns the number of rows deleted."""
    conn = _get_conn()
    cursor = conn.execute("DELETE FROM leads")
    deleted = cursor.rowcount
    conn.commit()
    logger.info(f"Deleted all leads: {deleted} rows removed")
    return deleted


def search_by_name(name: str, threshold: float = 0.85) -> List[Lead]:
    """Search for leads with similar names (for cross-run dedup)."""
    conn = _get_conn()
    rows = conn.execute(
        "SELECT * FROM leads WHERE business_name LIKE ?",
        (f"%{name}%",),
    ).fetchall()
    return [_row_to_lead(row) for row in rows]


# ═══════════════════════════════════════════════════════════════════════════════
# Row ↔ Lead conversion
# ═══════════════════════════════════════════════════════════════════════════════

def _row_to_lead(row: sqlite3.Row) -> Lead:
    """Convert a SQLite Row to a Lead object."""
    return Lead(
        business_name=row["business_name"] or "",
        service_category=row["service_category"] or "",
        phone=row["phone"] or "",
        email=row["email"] or "",
        address=row["address"] or "",
        website=row["website"] or "",
        contact_person=row["contact_person"] or "",
        source=row["source"] or "",
        lead_source_scraper=row["lead_source_scraper"] or "",
        notes=row["notes"] or "",
        user_notes=row["user_notes"] or "",
        score=row["score"] or "LOW",
        score_points=row["score_points"] or 0,
        industry=row["industry"] or "",
        product_category=row["product_category"] or "",
        status=row["status"] or "New",
        db_id=row["id"],
        scraped_at=row["scraped_at"] or "",
        updated_at=row["updated_at"] or "",
        has_https=bool(row["has_https"]),
        is_mobile_friendly=bool(row["is_mobile_friendly"]),
        is_outdated=bool(row["is_outdated"]),
        has_chatbot=bool(row["has_chatbot"]),
        has_online_booking=bool(row["has_online_booking"]),
        social_facebook=row["social_facebook"] or "",
        social_instagram=row["social_instagram"] or "",
        has_contact_form=bool(row["has_contact_form"]),
        ssl_expiry_days=row["ssl_expiry_days"],
        page_speed_score=row["page_speed_score"],
        load_time_sec=row["load_time_sec"],
        site_status=row["site_status"] or "",
        gmb_exists=bool(row["gmb_exists"]),
        review_count=row["review_count"] or 0,
        opportunity_pitch=row["opportunity_pitch"] or "",
        fingerprint=row["fingerprint"] or "",
    )