gendesign/data/sql/71_etl_anton_sqlite_to_pg.py

"""ETL: SQLite Антона (sf_anton_snapshot.db) → наша PG.

Грузит 3 таблицы из его SQLite в нашу schema 68 v2:
  - objective_lots         (303 677 строк)
  - objective_corp_month   ( 19 738 строк)
  - jk_objective_match     (    229 строк)

Запуск:
    cd backend
    DATABASE_URL=postgresql://gendesign:PWD@localhost:15432/gendesign \\
      uv run python -X utf8 ../data/sql/71_etl_anton_sqlite_to_pg.py [--dry-run] [--limit N]

Идемпотентно через ON CONFLICT DO UPDATE по UNIQUE-ключам.
Для повторной заливки: данные в таблицах перезатрутся (snapshot_date обновится).

Источник: Антоновский /sf/api/* SQLite (~340 MB), скачанный в repo-root через scp:
    scp gendesign:/opt/gendesign/site-finder/analysis.db sf_anton_snapshot.db

Mapping полей задокументирован в _ROW_TO_LOT / _ROW_TO_CRM / _ROW_TO_MAPPING ниже.

После прогона ETL рекомендуется DISABLE наш Celery objective-sync beat task
(Антон уже еженедельно тянет — не дублируем платный API).
"""

from __future__ import annotations

import argparse
import logging
import os
import re
import sqlite3
import sys
from datetime import date, datetime
from pathlib import Path

import psycopg2
import psycopg2.extras

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("etl_anton")

REPO_ROOT = Path(__file__).resolve().parent.parent.parent
SQLITE_DEFAULT = REPO_ROOT / "sf_anton_snapshot.db"

GROUP_NAME = "Екатеринбург"  # У Антона нет колонки group_name — у нас всё под Екб

# ── helpers ─────────────────────────────────────────────────────────────────


def _parse_date(s: str | None) -> date | None:
    """SQLite хранит как ISO-строку или None."""
    if s is None or s == "":
        return None
    try:
        return date.fromisoformat(s[:10])
    except (ValueError, TypeError):
        return None


def _parse_month_to_first_day(s: str | None) -> date | None:
    """'2025-08' → date(2025, 8, 1)."""
    if not s:
        return None
    try:
        y, m = s.split("-")
        return date(int(y), int(m), 1)
    except (ValueError, TypeError):
        return None


def _normalize_rooms(s: str | None) -> int | None:
    """'студия'→0, '1'→1, …, '4+'→4 (как в наш парсер 70)."""
    if not s:
        return None
    sl = s.strip().lower()
    if "студ" in sl:
        return 0
    m = re.match(r"^(\d+)", sl)
    if m:
        return min(int(m.group(1)), 5)
    return None


def _bool_from_yes(v) -> bool | None:
    if v is None or v == "":
        return None
    s = str(v).strip().lower()
    if s in ("да", "true", "1", "yes"):
        return True
    if s in ("нет", "false", "0", "no"):
        return False
    return None


def _site_id_to_obj_id(s: str | None) -> int | None:
    """'jk:49854' → 49854. 'parcel:...' → None."""
    if not s:
        return None
    if s.startswith("jk:"):
        try:
            return int(s[3:])
        except ValueError:
            return None
    return None


def _to_int(v) -> int | None:
    if v is None or v == "":
        return None
    try:
        return int(v)
    except (TypeError, ValueError):
        return None


def _to_num(v) -> float | None:
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


# ── row mappings ────────────────────────────────────────────────────────────


def _row_to_lot(r: sqlite3.Row, snapshot_date: date) -> tuple:
    """SQLite objective_lots row → tuple для INSERT в наш objective_lots."""
    return (
        r["lot_id"],                            # objective_lot_id
        r["project_id"],                        # objective_project_id
        r["project"],                           # project_name (NOT NULL)
        r["developer"],
        r["city"],
        r["district"],
        r["corpus"],                            # corpus_name
        r["address"],
        r["obj_class"],                         # class
        r["section"],
        _to_int(r["floor"]),
        r["lot_num"],                           # lot_number
        r["room_kind"],                         # premise_kind
        _parse_date(r["sales_start"]),
        _parse_date(r["plan_date"]),
        _parse_date(r["fact_date"]),
        _to_int(r["readiness_pct"]),
        r["construction_stage"],
        r["finish_type"],                       # finishing
        r["status"],
        _bool_from_yes(r["sold"]),              # is_sold
        r["rooms_dev"],                         # rooms_dev_site
        r["rooms_pd"],
        r["rooms_obj"],                         # rooms_objective
        _normalize_rooms(r["rooms_obj"]),       # rooms_int (0..5)
        _to_num(r["area_dev"]),                 # area_dev_site
        _to_num(r["area_pd"]),
        _to_num(r["budget_rub"]),               # price_calculated_total_rub
        _to_num(r["price_per_m2"]),             # price_per_m2_rub (Р/м², не тыс)
        _to_num(r["offer_price"]),              # price_offer_total_rub
        _to_num(r["delta_price_rub"]),
        _to_num(r["delta_price_pct"]),
        r["price_method"],                      # pricing_method
        _parse_date(r["price_set_date"]),
        _parse_date(r["price_actual_date"]),
        _parse_date(r["contract_date"]),
        _parse_date(r["register_date"]),        # registration_date
        r["deal_type"],                         # contract_type
        r["buyer_type"],
        r["register_num"],                      # registration_number
        r["encumbrance"],                       # encumbrance_type
        r["bank"],                              # bank_name
        _parse_date(r["encumbrance_start"]),    # encumbrance_start_date
        _parse_date(r["egrn_actual_date"]),
        snapshot_date,                          # snapshot_date — день ETL
    )


_LOT_INSERT = """
INSERT INTO objective_lots (
    objective_lot_id, objective_project_id, project_name, developer, city,
    district, corpus_name, address, class, section, floor, lot_number,
    premise_kind, sales_start_date, plan_completion_date, actual_completion_date,
    readiness_pct, construction_stage, finishing, status, is_sold,
    rooms_dev_site, rooms_pd, rooms_objective, rooms_int,
    area_dev_site, area_pd,
    price_calculated_total_rub, price_per_m2_rub, price_offer_total_rub,
    price_delta_rub, price_delta_pct, pricing_method, price_set_date,
    price_actual_date, contract_date, registration_date, contract_type,
    buyer_type, registration_number, encumbrance_type, bank_name,
    encumbrance_start_date, egrn_actual_date, snapshot_date
) VALUES %s
ON CONFLICT (objective_lot_id) DO UPDATE SET
    project_name = EXCLUDED.project_name,
    developer = EXCLUDED.developer, district = EXCLUDED.district,
    corpus_name = EXCLUDED.corpus_name, address = EXCLUDED.address,
    class = EXCLUDED.class, section = EXCLUDED.section, floor = EXCLUDED.floor,
    lot_number = EXCLUDED.lot_number, premise_kind = EXCLUDED.premise_kind,
    sales_start_date = EXCLUDED.sales_start_date,
    plan_completion_date = EXCLUDED.plan_completion_date,
    actual_completion_date = EXCLUDED.actual_completion_date,
    readiness_pct = EXCLUDED.readiness_pct,
    construction_stage = EXCLUDED.construction_stage,
    finishing = EXCLUDED.finishing,
    status = EXCLUDED.status, is_sold = EXCLUDED.is_sold,
    rooms_dev_site = EXCLUDED.rooms_dev_site,
    rooms_pd = EXCLUDED.rooms_pd,
    rooms_objective = EXCLUDED.rooms_objective,
    rooms_int = EXCLUDED.rooms_int,
    area_dev_site = EXCLUDED.area_dev_site, area_pd = EXCLUDED.area_pd,
    price_calculated_total_rub = EXCLUDED.price_calculated_total_rub,
    price_per_m2_rub = EXCLUDED.price_per_m2_rub,
    price_offer_total_rub = EXCLUDED.price_offer_total_rub,
    price_delta_rub = EXCLUDED.price_delta_rub,
    price_delta_pct = EXCLUDED.price_delta_pct,
    pricing_method = EXCLUDED.pricing_method,
    price_set_date = EXCLUDED.price_set_date,
    price_actual_date = EXCLUDED.price_actual_date,
    contract_date = EXCLUDED.contract_date,
    registration_date = EXCLUDED.registration_date,
    contract_type = EXCLUDED.contract_type, buyer_type = EXCLUDED.buyer_type,
    registration_number = EXCLUDED.registration_number,
    encumbrance_type = EXCLUDED.encumbrance_type,
    bank_name = EXCLUDED.bank_name,
    encumbrance_start_date = EXCLUDED.encumbrance_start_date,
    egrn_actual_date = EXCLUDED.egrn_actual_date,
    snapshot_date = EXCLUDED.snapshot_date,
    updated_at = NOW()
"""


def _row_to_crm(r: sqlite3.Row) -> tuple:
    """objective_corp_month → objective_corpus_room_month.

    Антон НЕ разделяет ДДУ vs ДКП vs всего — у нас только totals из его данных.
    NULL'и в ddu/dkp колонках — Антон их не давал.
    """
    return (
        _parse_month_to_first_day(r["month"]),  # report_month
        GROUP_NAME,                              # group_name
        r["project"],                            # project_name
        r["developer"],
        r["district"],
        r["obj_class"],                          # class
        r["corpus"],                             # corpus_name
        r["rooms_bucket"],                       # room_bucket
        _normalize_rooms(r["rooms_bucket"]),     # rooms_int
        _parse_date(r["sales_start"]),
        _parse_date(r["plan_date"]),
        _parse_date(r["fact_date"]),
        _to_int(r["months_in_sales"]),           # months_in_realization
        _to_int(r["lots_pd"]),                   # lots_pd_count
        _to_num(r["area_pd"]),                   # lots_pd_area
        # ДДУ — у Антона нет, NULL'им
        None, None, None, None, None, None, None,
        # ДКП — то же
        None, None, None, None, None, None, None,
        # Всего (Антон даёт только это)
        _to_int(r["deals_total"]),               # deals_total_count
        _to_int(r["deals_priced"]),              # deals_total_count_priced
        _to_num(r["sold_volume_m2"]),            # deals_total_vol_m2
        _to_num(r["sold_volume_m2"]),            # deals_total_vol_m2_priced (= тот же)
        None,                                    # deals_total_sum_mln_rub — нет
        _to_num(r["avg_price_m2"]),              # deals_total_avg_price_thousand_rub_per_m2 (тыс.Р/м²)
        _to_num(r["avg_area_m2"]),               # deals_total_avg_area_m2
        # Offer
        _to_int(r["stock_lots"]),                # offer_count
        _to_num(r["stock_m2"]),                  # offer_area_m2
        None,                                    # offer_sum_mln_rub — нет
        _to_num(r["stock_avg_price_m2"]),        # offer_avg_price_thousand_rub_per_m2
    )


_CRM_INSERT = """
INSERT INTO objective_corpus_room_month (
    report_month, group_name, project_name, developer, district, class,
    corpus_name, room_bucket, rooms_int,
    sales_start_date, plan_completion_date, actual_completion_date,
    months_in_realization, lots_pd_count, lots_pd_area,
    deals_ddu_count, deals_ddu_count_priced, deals_ddu_vol_m2,
    deals_ddu_vol_m2_priced, deals_ddu_sum_mln_rub,
    deals_ddu_avg_price_thousand_rub_per_m2, deals_ddu_avg_area_m2,
    deals_dkp_count, deals_dkp_count_priced, deals_dkp_vol_m2,
    deals_dkp_vol_m2_priced, deals_dkp_sum_mln_rub,
    deals_dkp_avg_price_thousand_rub_per_m2, deals_dkp_avg_area_m2,
    deals_total_count, deals_total_count_priced, deals_total_vol_m2,
    deals_total_vol_m2_priced, deals_total_sum_mln_rub,
    deals_total_avg_price_thousand_rub_per_m2, deals_total_avg_area_m2,
    offer_count, offer_area_m2, offer_sum_mln_rub,
    offer_avg_price_thousand_rub_per_m2
) VALUES %s
ON CONFLICT (report_month, group_name, project_name, corpus_name, room_bucket)
DO UPDATE SET
    developer = EXCLUDED.developer, district = EXCLUDED.district,
    class = EXCLUDED.class, rooms_int = EXCLUDED.rooms_int,
    sales_start_date = EXCLUDED.sales_start_date,
    plan_completion_date = EXCLUDED.plan_completion_date,
    actual_completion_date = EXCLUDED.actual_completion_date,
    months_in_realization = EXCLUDED.months_in_realization,
    lots_pd_count = EXCLUDED.lots_pd_count,
    lots_pd_area = EXCLUDED.lots_pd_area,
    deals_total_count = EXCLUDED.deals_total_count,
    deals_total_count_priced = EXCLUDED.deals_total_count_priced,
    deals_total_vol_m2 = EXCLUDED.deals_total_vol_m2,
    deals_total_vol_m2_priced = EXCLUDED.deals_total_vol_m2_priced,
    deals_total_avg_price_thousand_rub_per_m2 = EXCLUDED.deals_total_avg_price_thousand_rub_per_m2,
    deals_total_avg_area_m2 = EXCLUDED.deals_total_avg_area_m2,
    offer_count = EXCLUDED.offer_count,
    offer_area_m2 = EXCLUDED.offer_area_m2,
    offer_avg_price_thousand_rub_per_m2 = EXCLUDED.offer_avg_price_thousand_rub_per_m2,
    fetched_at = NOW()
"""


def _row_to_mapping(r: sqlite3.Row) -> tuple | None:
    """jk_objective_match → objective_complex_mapping.
    Возвращает None если site_id не jk: (parcel пропускаем — у нас нет parcel-таблицы).
    """
    obj_id = _site_id_to_obj_id(r["site_id"])
    if obj_id is None:
        return None
    return (
        r["project"],                  # objective_complex_name
        None,                          # objective_project_id (Антон не хранит)
        GROUP_NAME,
        obj_id,                        # domrf_obj_id
        False,                         # is_reviewed
        r["method"],                   # match_method ('fuzzy', etc.)
        _to_num(r["score"]),           # match_score
        "imported_from_anton_sf",      # note
    )


_MAP_INSERT = """
INSERT INTO objective_complex_mapping (
    objective_complex_name, objective_project_id, objective_group,
    domrf_obj_id, is_reviewed, match_method, match_score, note
) VALUES %s
ON CONFLICT (objective_complex_name, objective_group) DO UPDATE SET
    domrf_obj_id = EXCLUDED.domrf_obj_id,
    match_method = EXCLUDED.match_method,
    match_score = EXCLUDED.match_score,
    note = EXCLUDED.note,
    updated_at = NOW()
"""


# ── orchestration ───────────────────────────────────────────────────────────


def etl_lots(sl: sqlite3.Connection, pg, dry_run: bool, limit: int | None,
             batch_size: int) -> int:
    snap = date.today()
    sql = "SELECT * FROM objective_lots"
    if limit:
        sql += f" LIMIT {limit}"
    cur = sl.execute(sql)
    n = 0
    batch: list[tuple] = []
    pg_cur = pg.cursor() if pg else None
    while True:
        rows = cur.fetchmany(batch_size)
        if not rows:
            break
        for r in rows:
            batch.append(_row_to_lot(r, snap))
            if len(batch) >= batch_size:
                if not dry_run:
                    psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=batch_size)
                    pg.commit()
                n += len(batch)
                batch.clear()
                if n % (batch_size * 5) == 0:
                    logger.info("  lots: %d upserted", n)
    if batch:
        if not dry_run:
            psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=len(batch))
            pg.commit()
        n += len(batch)
    if pg_cur is not None:
        pg_cur.close()
    return n


def etl_crm(sl: sqlite3.Connection, pg, dry_run: bool, limit: int | None,
            batch_size: int) -> int:
    sql = "SELECT * FROM objective_corp_month"
    if limit:
        sql += f" LIMIT {limit}"
    cur = sl.execute(sql)
    n = 0
    skipped = 0
    batch: list[tuple] = []
    pg_cur = pg.cursor() if pg else None
    while True:
        rows = cur.fetchmany(batch_size)
        if not rows:
            break
        for r in rows:
            t = _row_to_crm(r)
            # Skip строк где report_month/project/corpus/room_bucket NULL — они нарушат UNIQUE.
            if not t[0] or not t[2] or not t[6] or not t[7]:
                skipped += 1
                continue
            batch.append(t)
            if len(batch) >= batch_size:
                if not dry_run:
                    psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=batch_size)
                    pg.commit()
                n += len(batch)
                batch.clear()
    if batch:
        if not dry_run:
            psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=len(batch))
            pg.commit()
        n += len(batch)
    if pg_cur is not None:
        pg_cur.close()
    if skipped:
        logger.warning("  corp_month: пропущено %d строк с пустыми обязательными полями", skipped)
    return n


def etl_mapping(sl: sqlite3.Connection, pg, dry_run: bool) -> int:
    # ORDER BY score DESC — при дублях берём mapping с наибольшим match score.
    cur = sl.execute("SELECT * FROM jk_objective_match ORDER BY score DESC NULLS LAST")
    rows = cur.fetchall()
    # Дедупликация по (objective_complex_name, objective_group) — наш UNIQUE.
    # У Антона разные site_id (jk:N) могут матчиться в один и тот же Objective project
    # (fuzzy mapping для близких имён). Берём первый = с лучшим score.
    seen: set[tuple[str, str]] = set()
    batch: list[tuple] = []
    skipped_dups = 0
    for r in rows:
        t = _row_to_mapping(r)
        if t is None:
            continue
        key = (t[0], t[2])  # (complex_name, group)
        if key in seen:
            skipped_dups += 1
            continue
        seen.add(key)
        batch.append(t)
    if skipped_dups:
        logger.info("  mapping: дедуплицировано %d дублей (один Objective project ↔ несколько jk)", skipped_dups)
    if not batch:
        return 0
    if not dry_run:
        pg_cur = pg.cursor()
        psycopg2.extras.execute_values(pg_cur, _MAP_INSERT, batch, page_size=500)
        pg.commit()
        pg_cur.close()
    return len(batch)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--sqlite", default=str(SQLITE_DEFAULT),
                    help=f"Путь к SQLite (default: {SQLITE_DEFAULT})")
    ap.add_argument("--db-url", default=os.environ.get("DATABASE_URL"),
                    help="postgres URL (или env DATABASE_URL). Через SSH-туннель: "
                         "postgresql://gendesign:PWD@localhost:15432/gendesign")
    ap.add_argument("--dry-run", action="store_true",
                    help="Не писать в PG (только посчитать строки)")
    ap.add_argument("--limit", type=int, default=None,
                    help="Ограничить кол-во строк (для пробного прогона)")
    ap.add_argument("--batch-size", type=int, default=500,
                    help="execute_values batch size (default 500)")
    ap.add_argument("--only", choices=("lots", "crm", "mapping"), default=None,
                    help="Грузить только одну таблицу")
    args = ap.parse_args()

    sqlite_path = Path(args.sqlite)
    if not sqlite_path.exists():
        logger.error("SQLite не найден: %s", sqlite_path)
        return 1
    if not args.db_url and not args.dry_run:
        logger.error("Нужен --db-url или env DATABASE_URL")
        return 2

    logger.info("ETL: %s → %s", sqlite_path.name, args.db_url or "(dry-run)")

    sl = sqlite3.connect(sqlite_path)
    sl.row_factory = sqlite3.Row

    pg = None
    if not args.dry_run:
        pg = psycopg2.connect(args.db_url)

    t0 = datetime.now()
    try:
        if args.only in (None, "mapping"):
            logger.info("[1/3] objective_complex_mapping (Антоновский match)…")
            n_map = etl_mapping(sl, pg, args.dry_run)
            logger.info("  → upserted %d mappings", n_map)

        if args.only in (None, "crm"):
            logger.info("[2/3] objective_corpus_room_month (~19 738 строк)…")
            n_crm = etl_crm(sl, pg, args.dry_run, args.limit, args.batch_size)
            logger.info("  → upserted %d corp_month rows", n_crm)

        if args.only in (None, "lots"):
            logger.info("[3/3] objective_lots (~303 677 строк)…")
            n_lots = etl_lots(sl, pg, args.dry_run, args.limit, args.batch_size)
            logger.info("  → upserted %d lots", n_lots)
    finally:
        sl.close()
        if pg is not None:
            pg.close()

    elapsed = (datetime.now() - t0).total_seconds()
    logger.info("ETL DONE in %.1fs%s", elapsed, " (dry-run)" if args.dry_run else "")
    return 0


if __name__ == "__main__":
    sys.exit(main())