parser-v1/database.py

"""SQLite БД для лидов.

Главное: upsert_lead() — единственная точка входа для записи лида.
Сама занимается дедупликацией: ИНН > телефон > домен.
"""
import json
import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Optional

import config
from normalization import phone_dedup_key, normalize_domain

# ───────────────────────────────────────────────────────────────────────
# Схема таблиц
# ───────────────────────────────────────────────────────────────────────
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS leads (
    id                  INTEGER PRIMARY KEY AUTOINCREMENT,

    -- Идентичность
    name                TEXT NOT NULL,
    inn                 TEXT,
    ogrn                TEXT,
    director_name       TEXT,

    -- Контакты (JSON-массивы для множественных значений)
    phones              TEXT,              -- ["+74952580888", ...] — доверенные (Я.Карты)
    phones_extra        TEXT,              -- доп. телефоны с САЙТА (к проверке, D19)
    emails              TEXT,              -- ["info@x.ru", ...]
    phone_primary       TEXT,              -- E.164 формат
    email_primary       TEXT,

    -- Онлайн-присутствие
    website             TEXT,
    vk_url              TEXT,
    telegram_url        TEXT,
    instagram_url       TEXT,
    youtube_url         TEXT,

    -- Гео
    address             TEXT,
    city                TEXT DEFAULT 'Москва',
    region              TEXT,                  -- агрегатор (Московская область / Москва и МО)
    district            TEXT,                  -- район внутри города (Митино, Бутово, ...)

    -- Бизнес
    category            TEXT,

    -- Сигналы
    reviews_count       INTEGER DEFAULT 0,
    reviews_avg         REAL DEFAULT 0.0,
    has_website         INTEGER DEFAULT 0,
    has_vk              INTEGER DEFAULT 0,
    has_telegram        INTEGER DEFAULT 0,

    -- Скоринг (v5 — «решаемая боль» × ICP)
    score               INTEGER DEFAULT 0,
    score_breakdown     TEXT,              -- JSON: band, themes, pain_products, reasons, coverage
    pain_products       TEXT,              -- JSON {"P4":3.0,...} «с чем заходить»
    diagnostic_coverage REAL,              -- 0..1 полнота диагностики боли
    band                TEXT,              -- hot / warm / cold (по score)

    -- Email валидация
    email_valid         INTEGER,
    email_checked_at    TEXT,

    -- Tier 2 enrichment (анализ сайта компании)
    site_alive          INTEGER,           -- NULL=не проверяли, 1=200-399, 0=мёртвый
    site_status_code    INTEGER,           -- HTTP статус
    cms_type            TEXT,              -- tilda/wix/wordpress/bitrix/modx/custom/none
    has_live_chat       INTEGER,           -- 0/1 (jivo, talk-me, ...)
    has_online_booking  INTEGER,           -- 0/1 (yclients, dikidi, ...)
    has_analytics       INTEGER,           -- 0/1 (Я.Метрика / GA)
    email_domain_type   TEXT,              -- corporate / free
    site_checked_at     TEXT,              -- когда последний раз enrich'или

    -- ЕГРЮЛ enrichment
    registration_date   TEXT,              -- дата регистрации компании (ISO)
    egrul_checked_at    TEXT,              -- когда искали в ЕГРЮЛ
    egrul_status        TEXT,              -- found / not_found / error

    -- Финансы (ФНС через DaData, D20)
    employee_count      INTEGER,           -- среднесписочная численность
    revenue             INTEGER,           -- доходы/оборот за год, руб
    expense             INTEGER,           -- расходы за год, руб
    finance_year        INTEGER,           -- год отчётности
    finance_checked_at  TEXT,              -- когда тянули финансы

    -- Outreach (auto-pipeline Phase 3 + ручной CRM-режим)
    outreach_status     TEXT DEFAULT 'new',  -- new / inbox / triaged / in_work / done / skip / queued / sent / replied / converted
    outreach_channel    TEXT,                -- последний канал auto-pipeline: email / vk / telegram
    outreach_sent_at    TEXT,
    outreach_replied_at TEXT,

    -- CRM (ручной режим, пишется из Streamlit-UI или CLI --touch)
    comments            TEXT,                -- свободные заметки о лиде (append)
    last_action         TEXT,                -- последнее ручное действие: call / email / vk / telegram / whatsapp / sms
    last_reaction       TEXT,                -- последняя реакция: no_answer / refused / agreed / moved_to_tg / callback / spam / not_target
    last_touched_at     TEXT,                -- ISO datetime последнего ручного касания

    -- Системные
    source              TEXT NOT NULL,     -- yandex_maps,2gis,vk,...
    source_id           TEXT,
    source_url          TEXT,
    parsed_at           TEXT NOT NULL,
    updated_at          TEXT,

    -- Дедуп-ключи
    phone_dedup_key     TEXT,
    domain_dedup_key    TEXT,

    UNIQUE(phone_dedup_key),
    UNIQUE(inn)
);

CREATE INDEX IF NOT EXISTS idx_leads_score ON leads(score DESC);
CREATE INDEX IF NOT EXISTS idx_leads_source ON leads(source);
CREATE INDEX IF NOT EXISTS idx_leads_outreach ON leads(outreach_status);
CREATE INDEX IF NOT EXISTS idx_leads_has_website ON leads(has_website);

CREATE TABLE IF NOT EXISTS sources_log (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    source          TEXT NOT NULL,
    query           TEXT,
    city            TEXT,
    started_at      TEXT,
    finished_at     TEXT,
    records_found   INTEGER DEFAULT 0,
    records_saved   INTEGER DEFAULT 0,
    duplicates      INTEGER DEFAULT 0,
    errors          INTEGER DEFAULT 0,
    status          TEXT,
    error_msg       TEXT
);

-- История касаний с лидами.
-- Пишется из двух мест: Streamlit-UI / CLI --touch (ручной режим) и Phase 3 n8n auto-pipeline.
-- Один лид = много строк (история всех взаимодействий).
CREATE TABLE IF NOT EXISTS outreach_events (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    lead_id         INTEGER NOT NULL REFERENCES leads(id),
    channel         TEXT NOT NULL,         -- call / email / vk / telegram / whatsapp / sms
    template_id     TEXT,                  -- для auto-pipeline (Phase 3)
    message_text    TEXT,                  -- текст отправленного сообщения (auto-pipeline)
    sent_at         TEXT,                  -- ISO datetime отправки/действия
    delivery_status TEXT,                  -- sent / failed / bounced — для auto. NULL для ручных.
    reply_at        TEXT,                  -- ISO datetime ответа клиента
    reply_text      TEXT,                  -- текст ответа клиента
    converted_at    TEXT,                  -- ISO datetime конверсии (стал клиентом)
    reaction        TEXT,                  -- no_answer / refused / agreed / moved_to_tg / callback / spam / not_target
    notes           TEXT                   -- свободный комментарий от звонящего
);

CREATE INDEX IF NOT EXISTS idx_outreach_events_lead ON outreach_events(lead_id);
CREATE INDEX IF NOT EXISTS idx_outreach_events_sent_at ON outreach_events(sent_at DESC);

-- Связка прогон ↔ лиды (junction).
-- Один прогон даёт N лидов; один лид может попасть в M прогонов (если разные категории его поймали).
-- Под ТЗ «один CSV = один прогон» — JOIN sources_log → lead_in_run → leads вернёт ровно лидов прогона.
-- role: 'inserted' (этот лид впервые появился в этом прогоне) / 'merged' (уже был, обновлён).
CREATE TABLE IF NOT EXISTS lead_in_run (
    lead_id  INTEGER NOT NULL REFERENCES leads(id),
    run_id   INTEGER NOT NULL REFERENCES sources_log(id),
    role     TEXT NOT NULL,
    PRIMARY KEY (lead_id, run_id)
);
CREATE INDEX IF NOT EXISTS idx_lead_in_run_run ON lead_in_run(run_id);
"""

# Tier 2 enrichment поля (заполняются enricher/website_analyzer.py)
ENRICHMENT_FIELDS = [
    "site_alive", "site_status_code", "cms_type",
    "has_live_chat", "has_online_booking", "has_analytics",
    "email_domain_type", "site_checked_at",
]

# ЕГРЮЛ enrichment поля (заполняются enricher/egrul_enricher.py).
# inn / ogrn / director_name / address — уже есть в базовой схеме.
EGRUL_FIELDS = [
    "inn", "ogrn", "director_name",
    "registration_date", "egrul_checked_at", "egrul_status",
    # 2026-05-19: добавлены — Rusprofile нередко публикует сайт и телефон
    # (особенно для крупных ООО). Перезаписывают существующее ТОЛЬКО если
    # вызывающий код явно передал эти ключи (см. update_egrul).
    "website", "phone_primary", "address",
    # Финансы (D20) — пишутся когда DaData их отдала
    "employee_count", "revenue", "expense", "finance_year",
]

# Поля, которые мы вставляем/обновляем (без id, parsed_at и системных вычисляемых)
WRITABLE_FIELDS = [
    "name", "inn", "ogrn", "director_name",
    "phones", "emails", "phone_primary", "email_primary",
    "website", "vk_url", "telegram_url", "instagram_url", "youtube_url",
    "address", "city", "region", "district", "category",
    "reviews_count", "reviews_avg",
    "has_website", "has_vk", "has_telegram",
    "score", "score_breakdown", "pain_products", "diagnostic_coverage", "band",
    *ENRICHMENT_FIELDS,
    "source", "source_id", "source_url",
    "phone_dedup_key", "domain_dedup_key",
]


def get_connection(db_path: str = "leads.db") -> sqlite3.Connection:
    """Открыть соединение с включённым row_factory (sqlite3.Row для dict-доступа)."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    return conn


def init_db(db_path: str = "leads.db") -> None:
    """Создать БД и все таблицы (если ещё нет) + автоматически мигрировать."""
    Path(db_path).parent.mkdir(parents=True, exist_ok=True)
    conn = get_connection(db_path)
    conn.executescript(SCHEMA_SQL)
    conn.commit()
    migrate_db(conn)  # автоматически довести старую БД до текущей схемы
    conn.close()


# Колонки которых может не быть в старой БД (схема росла итеративно).
# Формат: (column_name, sql_type_with_default)
NEW_COLUMNS = [
    # Контакты (D19 — доп. телефоны с сайта отдельно от доверенных)
    ("phones_extra",       "TEXT"),
    # Гео
    ("district",           "TEXT"),
    # Tier 2 — анализ сайта
    ("site_alive",         "INTEGER"),
    ("site_status_code",   "INTEGER"),
    ("cms_type",           "TEXT"),
    ("has_live_chat",      "INTEGER"),
    ("has_online_booking", "INTEGER"),
    ("has_analytics",      "INTEGER"),
    ("email_domain_type",  "TEXT"),
    ("site_checked_at",    "TEXT"),
    # ЕГРЮЛ обогащение
    ("registration_date",  "TEXT"),
    ("egrul_checked_at",   "TEXT"),
    ("egrul_status",       "TEXT"),
    # Финансы (D20)
    ("employee_count",     "INTEGER"),
    ("revenue",            "INTEGER"),
    ("expense",            "INTEGER"),
    ("finance_year",       "INTEGER"),
    ("finance_checked_at", "TEXT"),
    # Скоринг v5 — производные поля из score_breakdown (для CRM-фильтров)
    ("pain_products",       "TEXT"),
    ("diagnostic_coverage", "REAL"),
    ("band",                "TEXT"),
    # CRM (ручные касания — добавлено в шаге 0.1 CRM-блока)
    ("comments",           "TEXT"),
    ("last_action",        "TEXT"),
    ("last_reaction",      "TEXT"),
    ("last_touched_at",    "TEXT"),
]


def migrate_db(conn: sqlite3.Connection) -> None:
    """Достроить отсутствующие колонки в существующей БД (idempotent)."""
    existing_cols = {row["name"] for row in conn.execute("PRAGMA table_info(leads)")}
    added = []
    for col_name, col_type in NEW_COLUMNS:
        if col_name not in existing_cols:
            conn.execute(f"ALTER TABLE leads ADD COLUMN {col_name} {col_type}")
            added.append(col_name)
    if added:
        conn.commit()
        print(f"📦 Миграция БД: добавлены колонки {added}")


def _prepare_lead(lead: dict) -> dict:
    """Дополнить лид техническими полями: phones/emails в JSON, dedup-ключи, флаги, parsed_at."""
    prepared = dict(lead)  # копия чтобы не портить входной

    # phones: list → JSON, phone_primary
    phones = prepared.get("phones") or []
    if isinstance(phones, list):
        prepared["phones"] = json.dumps(phones, ensure_ascii=False)
        if not prepared.get("phone_primary") and phones:
            prepared["phone_primary"] = phones[0]

    # emails: аналогично
    emails = prepared.get("emails") or []
    if isinstance(emails, list):
        prepared["emails"] = json.dumps(emails, ensure_ascii=False)
        if not prepared.get("email_primary") and emails:
            prepared["email_primary"] = emails[0]

    # Дедуп-ключи
    prepared["phone_dedup_key"] = phone_dedup_key(prepared.get("phone_primary"))
    prepared["domain_dedup_key"] = normalize_domain(prepared.get("website"))

    # Булевые флаги
    prepared["has_website"] = 1 if prepared.get("website") else 0
    prepared["has_vk"] = 1 if prepared.get("vk_url") else 0
    prepared["has_telegram"] = 1 if prepared.get("telegram_url") else 0

    # score_breakdown: dict → JSON + производные поля v5 (band/coverage/pain_products)
    # выносим в отдельные колонки, чтобы CRM могла фильтровать через SQL.
    sb = prepared.get("score_breakdown")
    if isinstance(sb, dict):
        prepared["band"] = sb.get("band")
        prepared["diagnostic_coverage"] = sb.get("coverage")
        pp = sb.get("pain_products")
        prepared["pain_products"] = json.dumps(pp, ensure_ascii=False) if pp is not None else None
        prepared["score_breakdown"] = json.dumps(sb, ensure_ascii=False)

    # parsed_at
    if not prepared.get("parsed_at"):
        prepared["parsed_at"] = datetime.now().isoformat(timespec="seconds")

    return prepared


def _find_existing(conn: sqlite3.Connection, lead: dict) -> Optional[sqlite3.Row]:
    """Поиск дубля по приоритету: ИНН → phone_dedup_key → domain_dedup_key."""
    inn = lead.get("inn")
    pkey = lead.get("phone_dedup_key")
    dkey = lead.get("domain_dedup_key")

    if inn:
        row = conn.execute("SELECT * FROM leads WHERE inn = ?", (inn,)).fetchone()
        if row:
            return row
    if pkey:
        row = conn.execute("SELECT * FROM leads WHERE phone_dedup_key = ?", (pkey,)).fetchone()
        if row:
            return row
    if dkey:
        row = conn.execute("SELECT * FROM leads WHERE domain_dedup_key = ?", (dkey,)).fetchone()
        if row:
            return row
    return None


def _merge_lead(existing: dict, new: dict) -> dict:
    """Слить два лида: заполнить пустые поля у existing, объединить телефоны/email/source."""
    merged = {**existing}

    # Объединить списки телефонов и email
    for list_field in ("phones", "emails"):
        ex_list = json.loads(existing.get(list_field) or "[]")
        new_raw = new.get(list_field)
        if isinstance(new_raw, str):
            new_list = json.loads(new_raw or "[]")
        elif isinstance(new_raw, list):
            new_list = new_raw
        else:
            new_list = []
        # Сохраняем порядок, без дублей
        combined = list(dict.fromkeys(ex_list + new_list))
        merged[list_field] = json.dumps(combined, ensure_ascii=False)

    # Заполнить пустые поля из нового
    fillable = [
        "inn", "ogrn", "director_name", "phone_primary", "email_primary",
        "website", "vk_url", "telegram_url", "instagram_url", "youtube_url",
        "address", "region", "district", "category",
        "reviews_count", "reviews_avg",
        "score", "score_breakdown",
        "pain_products", "diagnostic_coverage", "band",
    ]
    for f in fillable:
        if not merged.get(f) and new.get(f):
            merged[f] = new[f]

    # Source: добавить если новый
    sources = (existing.get("source") or "").split(",")
    if new.get("source") and new["source"] not in sources:
        sources = [s for s in sources if s] + [new["source"]]
        merged["source"] = ",".join(sources)

    # Пересчитать дедуп-ключи и флаги по итоговым значениям
    merged["phone_dedup_key"] = phone_dedup_key(merged.get("phone_primary")) or existing.get("phone_dedup_key")
    merged["domain_dedup_key"] = normalize_domain(merged.get("website")) or existing.get("domain_dedup_key")
    merged["has_website"] = 1 if merged.get("website") else 0
    merged["has_vk"] = 1 if merged.get("vk_url") else 0
    merged["has_telegram"] = 1 if merged.get("telegram_url") else 0
    merged["updated_at"] = datetime.now().isoformat(timespec="seconds")

    return merged


def upsert_lead(conn: sqlite3.Connection, lead: dict, run_id: int | None = None) -> str:
    """Вставить либо обновить лид. Возвращает 'inserted' / 'merged' / 'skipped'.

    Дедупликация: ищем сначала по ИНН, потом по 10-значному телефону, потом по домену.
    Если найден — мержим (заполняем пустые поля, объединяем списки телефонов/email).

    Если задан run_id — пишет в lead_in_run строку (lead_id, run_id, role)
    где role = 'inserted' или 'merged'. Это связка для CSV прогонов.
    Если run_id=None — поведение как раньше (backward compat).
    """
    prepared = _prepare_lead(lead)

    if not prepared.get("name"):
        return "skipped"

    existing_row = _find_existing(conn, prepared)
    lead_id: int | None = None
    result: str

    if existing_row:
        existing_dict = dict(existing_row)
        merged = _merge_lead(existing_dict, prepared)
        lead_id = existing_dict["id"]

        # UPDATE по id
        fields_to_update = [f for f in WRITABLE_FIELDS + ["updated_at"] if f in merged]
        set_clause = ", ".join(f"{f} = ?" for f in fields_to_update)
        values = [merged.get(f) for f in fields_to_update]
        values.append(lead_id)
        conn.execute(f"UPDATE leads SET {set_clause} WHERE id = ?", values)
        result = "merged"
    else:
        # INSERT
        fields = WRITABLE_FIELDS + ["parsed_at"]
        placeholders = ", ".join("?" for _ in fields)
        cols = ", ".join(fields)
        values = [prepared.get(f) for f in fields]
        cursor = conn.execute(f"INSERT INTO leads ({cols}) VALUES ({placeholders})", values)
        lead_id = cursor.lastrowid
        result = "inserted"

    # Связка с прогоном (если задан run_id)
    if run_id is not None and lead_id is not None:
        # INSERT OR IGNORE — guard на случай если лид встретился в прогоне дважды
        conn.execute(
            "INSERT OR IGNORE INTO lead_in_run (lead_id, run_id, role) VALUES (?, ?, ?)",
            (lead_id, run_id, result),
        )

    conn.commit()
    return result


def log_source_run(
    conn: sqlite3.Connection,
    source: str,
    query: str,
    city: str,
    started_at: str,
    finished_at: str,
    records_found: int,
    records_saved: int,
    duplicates: int,
    errors: int,
    status: str,
    error_msg: str | None = None,
) -> None:
    """[DEPRECATED] Запись в sources_log одной транзакцией после прогона.

    Оставлено для backward compat. Новый код должен использовать пару
    start_source_run() → finish_source_run() — она открывает run заранее,
    возвращает run_id для связки лидов через lead_in_run.
    """
    conn.execute("""
        INSERT INTO sources_log
        (source, query, city, started_at, finished_at, records_found,
         records_saved, duplicates, errors, status, error_msg)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (source, query, city, started_at, finished_at, records_found,
          records_saved, duplicates, errors, status, error_msg))
    conn.commit()


def start_source_run(
    conn: sqlite3.Connection,
    source: str,
    query: str | None,
    city: str | None,
) -> int:
    """Открыть запись прогона до начала парсинга. Возвращает run_id.

    После завершения парсинга закрывается через finish_source_run().
    run_id передаётся в upsert_lead для связи с лидами через lead_in_run.
    """
    started_at = datetime.now().isoformat(timespec="seconds")
    cursor = conn.execute(
        "INSERT INTO sources_log (source, query, city, started_at, status) "
        "VALUES (?, ?, ?, ?, 'running')",
        (source, query, city, started_at),
    )
    conn.commit()
    return cursor.lastrowid


def finish_source_run(
    conn: sqlite3.Connection,
    run_id: int,
    records_found: int,
    records_saved: int,
    duplicates: int,
    errors: int,
    status: str,
    error_msg: str | None = None,
) -> None:
    """Закрыть запись прогона: finished_at + счётчики + финальный статус.

    status: 'ok' / 'partial' / 'error' (исключая 'running' который был при start).
    """
    finished_at = datetime.now().isoformat(timespec="seconds")
    conn.execute("""
        UPDATE sources_log
        SET finished_at = ?, records_found = ?, records_saved = ?,
            duplicates = ?, errors = ?, status = ?, error_msg = ?
        WHERE id = ?
    """, (finished_at, records_found, records_saved, duplicates, errors,
          status, error_msg, run_id))
    conn.commit()


def get_leads_for_enrichment(
    conn: sqlite3.Connection,
    limit: int | None = None,
    only_unchecked: bool = True,
) -> list[sqlite3.Row]:
    """Лиды у которых есть website и которые ещё не проходили enrichment."""
    where = "website IS NOT NULL AND website != ''"
    if only_unchecked:
        where += " AND site_checked_at IS NULL"
    query = f"SELECT id, name, website FROM leads WHERE {where} ORDER BY id"
    if limit:
        query += f" LIMIT {int(limit)}"
    return conn.execute(query).fetchall()


def update_enrichment(conn: sqlite3.Connection, lead_id: int, enrichment: dict) -> None:
    """Обновить Tier 2 поля у лида."""
    fields = [k for k in ENRICHMENT_FIELDS if k in enrichment]
    if not fields:
        return
    set_clause = ", ".join(f"{f} = ?" for f in fields)
    values = [enrichment.get(f) for f in fields]
    values.append(lead_id)
    conn.execute(f"UPDATE leads SET {set_clause} WHERE id = ?", values)
    conn.commit()


def update_lead_contacts(
    conn: sqlite3.Connection,
    lead_id: int,
    emails_found: list[str] | None = None,
    phones_found: list[str] | None = None,
) -> tuple[int, int]:
    """Слить найденные email/phones в существующего лида (без дублей).

    Используется enricher'ами (Tier 2 — сайт) для добавления контактов,
    которые источник парсера не отдал явно.

    Returns: (added_emails, added_phones)
    """
    emails_found = emails_found or []
    phones_found = phones_found or []
    if not emails_found and not phones_found:
        return 0, 0

    row = conn.execute(
        "SELECT phones, phones_extra, emails, email_primary FROM leads WHERE id = ?",
        (lead_id,),
    ).fetchone()
    if not row:
        return 0, 0

    def _load(v):
        try:
            return json.loads(v) if v else []
        except (json.JSONDecodeError, TypeError):
            return []

    trusted = set(_load(row["phones"]))      # с Я.Карт — доверенные, НЕ трогаем
    extra = _load(row["phones_extra"])
    current_emails = _load(row["emails"])

    # D19: телефоны с САЙТА недостоверны (шаблоны конструкторов, партнёры, другие
    # города) → пишем в phones_extra «к проверке», а НЕ в доверенный phones /
    # phone_primary (тот формируется из Я.Карт). Исключаем уже-доверенные и дубли.
    added_phones = 0
    extra_set = set(extra)
    for p in phones_found:
        if p and p not in trusted and p not in extra_set:
            extra.append(p)
            extra_set.add(p)
            added_phones += 1

    # Email — мерж как раньше (analyze_website уже фильтрует по домену сайта).
    existing_emails_lc = {e.lower() for e in current_emails if e}
    added_emails = 0
    for e in emails_found:
        if not e:
            continue
        el = e.lower()
        if el not in existing_emails_lc:
            current_emails.append(el)
            existing_emails_lc.add(el)
            added_emails += 1

    if added_phones == 0 and added_emails == 0:
        return 0, 0

    email_primary_new = row["email_primary"] or (current_emails[0] if current_emails else None)
    now = datetime.now().isoformat(timespec="seconds")

    conn.execute("""
        UPDATE leads
        SET phones_extra = ?, emails = ?,
            email_primary = COALESCE(email_primary, ?),
            updated_at = ?
        WHERE id = ?
    """, (
        json.dumps(extra, ensure_ascii=False),
        json.dumps(current_emails, ensure_ascii=False),
        email_primary_new, now, lead_id,
    ))
    conn.commit()

    return added_emails, added_phones


def get_hh_leads_without_website(
    conn: sqlite3.Connection,
    limit: int | None = None,
) -> list[sqlite3.Row]:
    """HH-лиды без заполненного website — для запуска hh_employers enricher.

    Берём по source LIKE '%hh%' (включая мерж-источники типа 'yandex_maps,hh')
    и пустому website. source_id формата 'hh_12345' содержит employer_id —
    нужен для построения URL hh.ru/employer/{id}.
    """
    where = (
        "source LIKE '%hh%' "
        "AND (website IS NULL OR website = '') "
        "AND source_id IS NOT NULL AND source_id LIKE 'hh_%'"
    )
    query = f"SELECT id, name, source_id FROM leads WHERE {where} ORDER BY id"
    if limit:
        query += f" LIMIT {int(limit)}"
    return conn.execute(query).fetchall()


def update_lead_website(
    conn: sqlite3.Connection,
    lead_id: int,
    website: str,
) -> bool:
    """Дозаполнить website у лида, если оно было пустым.

    Не перезаписывает уже заполненный website (защита от затирания валидного
    сайта мусорным URL). Возвращает True если запись произошла.
    Обновляет domain_dedup_key и has_website автоматически.
    """
    if not website:
        return False
    domain_key = normalize_domain(website)
    now = datetime.now().isoformat(timespec="seconds")
    cursor = conn.execute(
        """
        UPDATE leads
        SET website = ?,
            has_website = 1,
            domain_dedup_key = COALESCE(domain_dedup_key, ?),
            updated_at = ?
        WHERE id = ? AND (website IS NULL OR website = '')
        """,
        (website, domain_key, now, lead_id),
    )
    conn.commit()
    return cursor.rowcount > 0


def get_leads_for_egrul(
    conn: sqlite3.Connection,
    limit: int | None = None,
    only_unchecked: bool = True,
) -> list[sqlite3.Row]:
    """Лиды у которых есть name и которые ещё не проходили ЕГРЮЛ-обогащение.

    Возвращает дополнительные колонки (inn, director_name, website, phone_primary)
    чтобы вызывающий код мог решить:
       • если все эти 4 поля заполнены → skip полностью (нет смысла дёргать Rusprofile)
       • если есть только inn → искать ПО ИНН (быстрее и точнее чем по имени)
       • если ничего нет → искать по имени (старое поведение)
    """
    where = "name IS NOT NULL AND name != ''"
    if only_unchecked:
        where += " AND egrul_checked_at IS NULL"
    query = (
        "SELECT id, name, address, city, inn, director_name, website, phone_primary "
        f"FROM leads WHERE {where} ORDER BY id"
    )
    if limit:
        query += f" LIMIT {int(limit)}"
    return conn.execute(query).fetchall()


def fix_categories_from_runs(conn: sqlite3.Connection) -> int:
    """Одноразовая миграция: установить lead.category = sources_log.query.

    До этой миграции парсеры писали в category неконсистентно:
    - Я.Карты — то что показала карточка («Ресторан, бар»)
    - HH — название вакансии («Оператор колл-центра»)

    Теперь правило для всех источников: category = поисковый запрос
    («кафе», «стоматология», ...). Эта функция переписывает существующих
    лидов на основе истории прогонов из lead_in_run + sources_log.

    Если лид был в нескольких прогонах — берём query из ПЕРВОГО прогона
    где он был вставлен (role='inserted'), иначе из самого раннего.

    Returns: количество обновлённых лидов.
    """
    sql = """
        UPDATE leads
        SET category = (
            SELECT sl.query
            FROM lead_in_run lir
            JOIN sources_log sl ON sl.id = lir.run_id
            WHERE lir.lead_id = leads.id
              AND sl.query IS NOT NULL
              AND sl.query != ''
            ORDER BY
              CASE WHEN lir.role = 'inserted' THEN 0 ELSE 1 END,
              lir.run_id ASC
            LIMIT 1
        )
        WHERE id IN (
            SELECT DISTINCT lead_id FROM lead_in_run
            WHERE lead_id IN (SELECT id FROM leads)
        )
    """
    cursor = conn.execute(sql)
    conn.commit()
    return cursor.rowcount


def cleanup_bad_director_names(conn: sqlite3.Connection) -> int:
    """Очистить director_name у лидов где туда попала должность вместо ФИО.

    Сбрасывает egrul_checked_at = NULL чтобы --enrich-egrul их перепрогнал.
    """
    bad_markers = [
        "Генеральный директор", "Директор", "Руководитель",
        "Председатель", "Конкурсный", "Учредитель", "Управляющий",
        "Производство", "Услуги", "Работ",
    ]
    # Совпадение если первое слово в director_name — должность
    where_parts = [f"director_name LIKE '{m}%'" for m in bad_markers]
    where_clause = " OR ".join(where_parts)
    sql = f"""
        UPDATE leads
        SET director_name = NULL,
            egrul_checked_at = NULL,
            egrul_status = NULL
        WHERE {where_clause}
    """
    cursor = conn.execute(sql)
    conn.commit()
    return cursor.rowcount


def update_egrul(conn: sqlite3.Connection, lead_id: int, egrul: dict) -> str:
    """Обновить ЕГРЮЛ-поля у лида.

    Возвращает:
        'updated'    — поля записаны
        'duplicate'  — ИНН уже у другого лида в БД (дубль), записали без ИНН
        'noop'       — нечего записывать
    """
    # Enrichment только ДОПОЛНЯЕТ — не перезаписываем существующие значения
    # пустотой. Rusprofile/DaData при not_found возвращают website/phone/address
    # = None; без этого фильтра update обнулял уже собранный сайт/телефон
    # (баг: website терялся у лидов со статусом not_found).
    fields = [k for k in EGRUL_FIELDS if k in egrul and egrul.get(k) not in (None, "")]
    if not fields:
        return "noop"

    set_clause = ", ".join(f"{f} = ?" for f in fields)
    values = [egrul.get(f) for f in fields]
    values.append(lead_id)

    try:
        conn.execute(f"UPDATE leads SET {set_clause} WHERE id = ?", values)
        conn.commit()
        return "updated"
    except sqlite3.IntegrityError as e:
        # UNIQUE(inn) — этот ИНН уже у другого лида (дубль одного юр.лица).
        # Записываем остальные поля БЕЗ inn, чтобы не потерять директора и дату.
        if "leads.inn" in str(e).lower() or "unique constraint failed: leads.inn" in str(e).lower():
            fields_no_inn = [f for f in fields if f != "inn"]
            if fields_no_inn:
                set_clause = ", ".join(f"{f} = ?" for f in fields_no_inn)
                values = [egrul.get(f) for f in fields_no_inn]
                values.append(lead_id)
                conn.execute(f"UPDATE leads SET {set_clause} WHERE id = ?", values)
                conn.commit()
            return "duplicate"
        raise


def get_leads_for_finance(
    conn: sqlite3.Connection,
    limit: int | None = None,
    only_unchecked: bool = True,
) -> list[sqlite3.Row]:
    """Лиды-ООО (ИНН = 10 цифр) для добора финансов через DaData findById.

    ИП (ИНН 12 цифр) исключены — они не сдают финотчётность в ФНС.
    """
    where = "inn IS NOT NULL AND length(inn) = 10"
    if only_unchecked:
        where += " AND finance_checked_at IS NULL"
    query = f"SELECT id, name, inn FROM leads WHERE {where} ORDER BY id"
    if limit:
        query += f" LIMIT {int(limit)}"
    return conn.execute(query).fetchall()


def update_finance(conn: sqlite3.Connection, lead_id: int, info: dict) -> bool:
    """Записать финансы (employee_count / revenue / expense / finance_year).

    COALESCE — не затираем уже заполненное пустым. `finance_checked_at`
    проставляется всегда (чтобы повторно не дёргать). Возвращает True если
    хоть одно финансовое поле непустое.
    """
    now = datetime.now().isoformat(timespec="seconds")
    has = any(info.get(f) is not None for f in ("employee_count", "revenue", "expense", "finance_year"))
    conn.execute(
        "UPDATE leads SET "
        "employee_count = COALESCE(?, employee_count), "
        "revenue = COALESCE(?, revenue), "
        "expense = COALESCE(?, expense), "
        "finance_year = COALESCE(?, finance_year), "
        "finance_checked_at = ? WHERE id = ?",
        (info.get("employee_count"), info.get("revenue"), info.get("expense"),
         info.get("finance_year"), now, lead_id),
    )
    conn.commit()
    return has


def get_all_leads(conn: sqlite3.Connection) -> list[sqlite3.Row]:
    """Все лиды (для пересчёта score)."""
    return conn.execute("SELECT * FROM leads ORDER BY id").fetchall()


def update_score(conn: sqlite3.Connection, lead_id: int, score: int, breakdown: dict) -> None:
    """Обновить score + breakdown + производные поля (band/pain_products/coverage).

    Производные поля дублируются в отдельные колонки из breakdown, чтобы CRM
    могла сортировать/фильтровать через SQL без парсинга JSON.
    """
    band = pain_products = coverage = None
    if isinstance(breakdown, dict):
        band = breakdown.get("band")
        coverage = breakdown.get("coverage")
        pp = breakdown.get("pain_products")
        pain_products = json.dumps(pp, ensure_ascii=False) if pp is not None else None
    conn.execute(
        "UPDATE leads SET score = ?, score_breakdown = ?, pain_products = ?, "
        "diagnostic_coverage = ?, band = ? WHERE id = ?",
        (score, json.dumps(breakdown, ensure_ascii=False), pain_products, coverage, band, lead_id),
    )
    conn.commit()


def get_stats(conn: sqlite3.Connection) -> dict:
    """Сводная статистика: всего, по source, по score-bucket, по outreach_status."""
    total = conn.execute("SELECT COUNT(*) FROM leads").fetchone()[0]
    by_source = dict(conn.execute(
        "SELECT source, COUNT(*) FROM leads GROUP BY source"
    ).fetchall())
    by_outreach = dict(conn.execute(
        "SELECT outreach_status, COUNT(*) FROM leads GROUP BY outreach_status"
    ).fetchall())
    threshold = config.HOT_LEAD_THRESHOLD
    hot = conn.execute(
        f"SELECT COUNT(*) FROM leads WHERE score >= {int(threshold)}"
    ).fetchone()[0]
    with_phone = conn.execute(
        "SELECT COUNT(*) FROM leads WHERE phone_primary IS NOT NULL"
    ).fetchone()[0]
    with_email = conn.execute(
        "SELECT COUNT(*) FROM leads WHERE email_primary IS NOT NULL"
    ).fetchone()[0]
    return {
        "total": total,
        "by_source": by_source,
        "by_outreach": by_outreach,
        f"hot_leads (score >= {int(threshold)})": hot,
        "with_phone": with_phone,
        "with_email": with_email,
    }


if __name__ == "__main__":
    # Smoke-test: создать тестовую БД, вставить 2 одинаковых лида, проверить мерж
    import os
    test_db = "test_leads.db"
    if os.path.exists(test_db):
        os.remove(test_db)

    init_db(test_db)
    conn = get_connection(test_db)

    lead1 = {
        "name": "Кафе Тест",
        "phones": ["+74951234567"],
        "phone_primary": "+74951234567",
        "website": "https://example.ru",
        "city": "Москва",
        "category": "кафе",
        "source": "yandex_maps",
    }
    r1 = upsert_lead(conn, lead1)
    assert r1 == "inserted", f"expected inserted, got {r1}"

    # Тот же телефон → должно смержиться
    lead2 = {
        "name": "Кафе Тест (другой источник)",
        "phones": ["+74951234567"],
        "phone_primary": "+74951234567",
        "website": "https://example.ru",
        "vk_url": "https://vk.com/cafetest",  # новое поле
        "city": "Москва",
        "source": "2gis",
    }
    r2 = upsert_lead(conn, lead2)
    assert r2 == "merged", f"expected merged, got {r2}"

    stats = get_stats(conn)
    assert stats["total"] == 1, f"должна быть 1 запись после мержа, есть {stats['total']}"

    # Source должен содержать оба
    row = conn.execute("SELECT source, vk_url FROM leads").fetchone()
    assert "yandex_maps" in row["source"] and "2gis" in row["source"]
    assert row["vk_url"] == "https://vk.com/cafetest"

    conn.close()
    os.remove(test_db)
    print("✅ database.py — smoke-тест дедупликации пройден")