522ecca5c4
442-sweep instructions #16 Dockerfile: runner (lean, 1.6 GB) + runner-with-chromium (2.9 GB). Compose + deploy.yml pushes 2 GHCR images. Backend saves 1.3 GB. #15 5 backend object endpoints + /analytics/objects/[id] page (gallery, POI table, sale chart) + sparkline-driven PrinzipObjectsTable. #14 prinzip_leads/deals + 2 MVs. Import script with PII-hashing & --inspect-only mode. 3 funnel endpoints + Sankey/Monthly charts. #13 README section with UI/CLI/SQL for 442-object sweep.
301 lines
12 KiB
Python
301 lines
12 KiB
Python
"""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()
|