gendesign/data/sql/52_import_prinzip_crm.py

"""Import PRINZIP CRM SQLite dump → Postgres prinzip_leads / prinzip_deals.

Источник: 49 MB SQLite-дамп с analytics.prinzipevent.ru. user скачивает руками
(доступ подтверждён) и кладёт в data/raw/prinzip_crm.sqlite (gitignored).

Запуск:
    PYTHONIOENCODING=utf-8 python data/sql/52_import_prinzip_crm.py \\
      --sqlite data/raw/prinzip_crm.sqlite

Скрипт идемпотентный — UPSERT по lead_id / deal_id. Можно перезапускать
после очередного экспорта SQLite.

⚠ Точная схема таблиц в SQLite неизвестна заранее. Скрипт сначала логирует
sqlite_master + table_info по каждой таблице, чтобы можно было поправить
mapping (поле LEAD_FIELD_MAP внизу).

PII compliance (152-ФЗ):
- phone → SHA256(phone)[:16] в phone_hash + last 4 digits в phone_last4
- email → MD5(email) в email_hash
- name остаётся (legal: client может попросить удалить)
"""

from __future__ import annotations

import argparse
import hashlib
import logging
import os
import sqlite3
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "backend"))

from sqlalchemy import create_engine, text  # noqa: E402

logger = logging.getLogger(__name__)


# ── PII helpers ──────────────────────────────────────────────────────────────


def hash_phone(phone: str | None) -> tuple[str | None, str | None]:
    """Return (sha256_short, last4_digits)."""
    if not phone:
        return None, None
    digits = "".join(c for c in str(phone) if c.isdigit())
    if not digits:
        return None, None
    h = hashlib.sha256(digits.encode("utf-8")).hexdigest()[:16]
    last4 = digits[-4:] if len(digits) >= 4 else digits
    return h, last4


def hash_email(email: str | None) -> str | None:
    if not email:
        return None
    return hashlib.md5(str(email).strip().lower().encode("utf-8")).hexdigest()


# ── SQLite → Postgres column mapping (правится после первого dry-run) ───────
#
# Скрипт ВПЕРВЫЕ логирует table_info — посмотри логи и поправь LEAD_FIELD_MAP
# и DEAL_FIELD_MAP под реальные имена в SQLite.

LEAD_FIELD_MAP = {
    "lead_id": ["id", "lead_id", "request_id"],
    "created_at": ["created_at", "createdAt", "datetime", "date"],
    "source": ["source", "utm_source", "channel"],
    "channel": ["channel", "utm_campaign"],
    "name": ["name", "client_name", "fio"],
    "phone": ["phone", "tel", "telephone"],
    "email": ["email", "mail"],
    "obj_interest": ["obj_id", "object_id", "complex_id", "jk_id"],
    "rooms": ["rooms", "room_count"],
    "budget_min": ["budget_min", "price_min"],
    "budget_max": ["budget_max", "price_max"],
    "status": ["status", "state"],
}

DEAL_FIELD_MAP = {
    "deal_id": ["id", "deal_id"],
    "lead_id": ["lead_id", "request_id"],
    "closed_at": ["closed_at", "deal_date", "date", "datetime"],
    "obj_id": ["obj_id", "object_id", "complex_id"],
    "flat_ods_id": ["flat_id", "flat_ods_id", "ods_id"],
    "rooms": ["rooms", "room_count"],
    "total_area": ["area", "total_area"],
    "deal_price": ["price", "deal_price", "amount"],
    "payment_type": ["payment_type", "payment_method"],
    "mortgage_bank": ["bank", "mortgage_bank"],
    "mortgage_rate": ["rate", "mortgage_rate"],
    "deal_stage": ["stage", "deal_stage"],
}


def find_field(row: sqlite3.Row, candidates: list[str]):
    """Pick first existing column from candidates in a sqlite Row."""
    keys = row.keys() if hasattr(row, "keys") else []
    for c in candidates:
        if c in keys:
            return row[c]
    return None


# ── Main import ──────────────────────────────────────────────────────────────


UPSERT_LEAD = text(
    """
    INSERT INTO prinzip_leads (
        lead_id, created_at, source, channel, name, phone_hash, phone_last4,
        email_hash, obj_interest, rooms, budget_min, budget_max, status, converted, deal_id
    ) VALUES (
        :lead_id, :created_at, :source, :channel, :name, :phone_hash, :phone_last4,
        :email_hash, :obj_interest, :rooms, :budget_min, :budget_max, :status,
        :converted, :deal_id
    )
    ON CONFLICT (lead_id) DO UPDATE SET
        status = EXCLUDED.status,
        converted = EXCLUDED.converted,
        deal_id = EXCLUDED.deal_id
    """
)

UPSERT_DEAL = text(
    """
    INSERT INTO prinzip_deals (
        deal_id, lead_id, closed_at, obj_id, flat_ods_id, rooms, total_area,
        deal_price, price_per_m2, payment_type, mortgage_bank, mortgage_rate, deal_stage
    ) VALUES (
        :deal_id, :lead_id, :closed_at, :obj_id, :flat_ods_id, :rooms, :total_area,
        :deal_price, :price_per_m2, :payment_type, :mortgage_bank, :mortgage_rate, :deal_stage
    )
    ON CONFLICT (deal_id) DO UPDATE SET
        deal_stage = EXCLUDED.deal_stage,
        deal_price = EXCLUDED.deal_price,
        mortgage_bank = EXCLUDED.mortgage_bank
    """
)


def inspect_sqlite(sqlite_path: Path) -> None:
    """Log all tables + columns for first-time schema discovery."""
    conn = sqlite3.connect(sqlite_path)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    tables = [
        r[0]
        for r in cur.execute(
            "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
        ).fetchall()
    ]
    logger.info("SQLite contains %d tables: %s", len(tables), tables)
    for t in tables:
        cols = cur.execute(f"PRAGMA table_info({t})").fetchall()
        col_names = [c[1] for c in cols]
        cnt = cur.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
        logger.info("  %s (%d rows): %s", t, cnt, col_names)
    conn.close()


def import_leads(sqlite_path: Path, pg_engine, table_name: str) -> int:
    conn = sqlite3.connect(sqlite_path)
    conn.row_factory = sqlite3.Row
    cur = conn.execute(f"SELECT * FROM {table_name}")
    n = 0
    with pg_engine.begin() as pg_conn:
        for row in cur:
            phone_h, phone_last4 = hash_phone(find_field(row, LEAD_FIELD_MAP["phone"]))
            try:
                pg_conn.execute(
                    UPSERT_LEAD,
                    {
                        "lead_id": find_field(row, LEAD_FIELD_MAP["lead_id"]),
                        "created_at": find_field(row, LEAD_FIELD_MAP["created_at"]),
                        "source": find_field(row, LEAD_FIELD_MAP["source"]),
                        "channel": find_field(row, LEAD_FIELD_MAP["channel"]),
                        "name": find_field(row, LEAD_FIELD_MAP["name"]),
                        "phone_hash": phone_h,
                        "phone_last4": phone_last4,
                        "email_hash": hash_email(find_field(row, LEAD_FIELD_MAP["email"])),
                        "obj_interest": find_field(row, LEAD_FIELD_MAP["obj_interest"]),
                        "rooms": find_field(row, LEAD_FIELD_MAP["rooms"]),
                        "budget_min": find_field(row, LEAD_FIELD_MAP["budget_min"]),
                        "budget_max": find_field(row, LEAD_FIELD_MAP["budget_max"]),
                        "status": find_field(row, LEAD_FIELD_MAP["status"]),
                        "converted": (find_field(row, ["converted", "is_converted"]) or False),
                        "deal_id": find_field(row, ["deal_id"]),
                    },
                )
                n += 1
            except Exception as e:
                logger.warning("lead %s skipped: %s", find_field(row, LEAD_FIELD_MAP["lead_id"]), e)
    conn.close()
    return n


def import_deals(sqlite_path: Path, pg_engine, table_name: str) -> int:
    conn = sqlite3.connect(sqlite_path)
    conn.row_factory = sqlite3.Row
    cur = conn.execute(f"SELECT * FROM {table_name}")
    n = 0
    with pg_engine.begin() as pg_conn:
        for row in cur:
            area = find_field(row, DEAL_FIELD_MAP["total_area"])
            price = find_field(row, DEAL_FIELD_MAP["deal_price"])
            ppm = (price / area) if area and price else None
            try:
                pg_conn.execute(
                    UPSERT_DEAL,
                    {
                        "deal_id": find_field(row, DEAL_FIELD_MAP["deal_id"]),
                        "lead_id": find_field(row, DEAL_FIELD_MAP["lead_id"]),
                        "closed_at": find_field(row, DEAL_FIELD_MAP["closed_at"]),
                        "obj_id": find_field(row, DEAL_FIELD_MAP["obj_id"]),
                        "flat_ods_id": find_field(row, DEAL_FIELD_MAP["flat_ods_id"]),
                        "rooms": find_field(row, DEAL_FIELD_MAP["rooms"]),
                        "total_area": area,
                        "deal_price": price,
                        "price_per_m2": ppm,
                        "payment_type": find_field(row, DEAL_FIELD_MAP["payment_type"]),
                        "mortgage_bank": find_field(row, DEAL_FIELD_MAP["mortgage_bank"]),
                        "mortgage_rate": find_field(row, DEAL_FIELD_MAP["mortgage_rate"]),
                        "deal_stage": find_field(row, DEAL_FIELD_MAP["deal_stage"]),
                    },
                )
                n += 1
            except Exception as e:
                logger.warning("deal %s skipped: %s", find_field(row, DEAL_FIELD_MAP["deal_id"]), e)
    conn.close()
    return n


def refresh_views(pg_engine) -> None:
    with pg_engine.begin() as pg_conn:
        pg_conn.execute(text("REFRESH MATERIALIZED VIEW prinzip_funnel_monthly"))
        pg_conn.execute(text("REFRESH MATERIALIZED VIEW prinzip_funnel_by_object"))


def main() -> None:
    parser = argparse.ArgumentParser(description="PRINZIP CRM SQLite → Postgres importer")
    parser.add_argument(
        "--sqlite",
        type=Path,
        default=Path("data/raw/prinzip_crm.sqlite"),
        help="Путь к SQLite-дампу",
    )
    parser.add_argument(
        "--inspect-only",
        action="store_true",
        help="Только перечислить таблицы и колонки SQLite (для подбора field-mapping)",
    )
    parser.add_argument(
        "--leads-table", default="leads", help="Имя таблицы leads в SQLite"
    )
    parser.add_argument(
        "--deals-table", default="deals", help="Имя таблицы deals в SQLite"
    )
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )

    if not args.sqlite.exists():
        logger.error(
            "SQLite file not found: %s. Скачай файл с analytics.prinzipevent.ru "
            "и положи туда.",
            args.sqlite,
        )
        sys.exit(1)

    inspect_sqlite(args.sqlite)
    if args.inspect_only:
        return

    db_url = os.environ.get(
        "DATABASE_URL",
        "postgresql+psycopg://gendesign:gendesign@localhost:5432/gendesign",
    )
    engine = create_engine(db_url)

    logger.info("Importing leads from table %s …", args.leads_table)
    n_leads = import_leads(args.sqlite, engine, args.leads_table)
    logger.info("Imported %d leads.", n_leads)

    logger.info("Importing deals from table %s …", args.deals_table)
    n_deals = import_deals(args.sqlite, engine, args.deals_table)
    logger.info("Imported %d deals.", n_deals)

    logger.info("Refreshing materialized views…")
    refresh_views(engine)
    logger.info("Done.")


if __name__ == "__main__":
    main()