gendesign/backend/app/services/objective_etl.py

"""ETL: SQLite Антона (/sf/) → наша PG (objective_* таблицы).

Reusable библиотека вокруг логики `data/sql/71_etl_anton_sqlite_to_pg.py`.
Используется Celery-task `tasks.objective_etl.import_anton_objective` и
admin-endpoint `POST /api/v1/admin/scrape/objective`.

Файл-источник по умолчанию: settings.objective_anton_sqlite_path
(на проде смонтирован bind-mount-ом docker-compose из /opt/gendesign/site-finder).

Стратегия:
  - lots:    INSERT … ON CONFLICT (objective_lot_id) DO UPDATE
  - crm:     INSERT … ON CONFLICT (report_month, group_name, project_name,
                                   corpus_name, room_bucket) DO UPDATE
  - mapping: дедупликация по (objective_complex_name, objective_group),
             потом INSERT … ON CONFLICT DO UPDATE

Heartbeat / прогресс (для UI) выставляется через callback `progress_cb`,
который Celery-task передаёт со UPDATE objective_scrape_runs.
"""

from __future__ import annotations

import logging
import re
import sqlite3
from collections.abc import Callable
from datetime import date
from pathlib import Path
from typing import Any

import psycopg2
import psycopg2.extras

logger = logging.getLogger(__name__)

GROUP_NAME = "Екатеринбург"  # У Антона в SQLite нет колонки group_name


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


def _parse_date(s) -> date | None:
    if s is None or s == "":
        return None
    try:
        return date.fromisoformat(str(s)[:10])
    except (ValueError, TypeError):
        return None


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


def _normalize_rooms(s) -> int | None:
    if not s:
        return None
    sl = str(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) -> int | None:
    """'jk:49854' → 49854. 'parcel:...' → None."""
    if not s:
        return None
    s = str(s)
    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:
    return (
        r["lot_id"],
        r["project_id"],
        r["project"],
        r["developer"],
        r["city"],
        r["district"],
        r["corpus"],
        r["address"],
        r["obj_class"],
        r["section"],
        _to_int(r["floor"]),
        r["lot_num"],
        r["room_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"],
        r["status"],
        _bool_from_yes(r["sold"]),
        r["rooms_dev"],
        r["rooms_pd"],
        r["rooms_obj"],
        _normalize_rooms(r["rooms_obj"]),
        _to_num(r["area_dev"]),
        _to_num(r["area_pd"]),
        _to_num(r["budget_rub"]),
        _to_num(r["price_per_m2"]),
        _to_num(r["offer_price"]),
        _to_num(r["delta_price_rub"]),
        _to_num(r["delta_price_pct"]),
        r["price_method"],
        _parse_date(r["price_set_date"]),
        _parse_date(r["price_actual_date"]),
        _parse_date(r["contract_date"]),
        _parse_date(r["register_date"]),
        r["deal_type"],
        r["buyer_type"],
        r["register_num"],
        r["encumbrance"],
        r["bank"],
        _parse_date(r["encumbrance_start"]),
        _parse_date(r["egrn_actual_date"]),
        snapshot_date,
    )


_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:
    return (
        _parse_month_to_first_day(r["month"]),
        GROUP_NAME,
        r["project"],
        r["developer"],
        r["district"],
        r["obj_class"],
        r["corpus"],
        r["rooms_bucket"],
        _normalize_rooms(r["rooms_bucket"]),
        _parse_date(r["sales_start"]),
        _parse_date(r["plan_date"]),
        _parse_date(r["fact_date"]),
        _to_int(r["months_in_sales"]),
        _to_int(r["lots_pd"]),
        _to_num(r["area_pd"]),
        # ДДУ — у Антона нет
        None,
        None,
        None,
        None,
        None,
        None,
        None,
        # ДКП
        None,
        None,
        None,
        None,
        None,
        None,
        None,
        # Всего
        _to_int(r["deals_total"]),
        _to_int(r["deals_priced"]),
        _to_num(r["sold_volume_m2"]),
        _to_num(r["sold_volume_m2"]),
        None,
        _to_num(r["avg_price_m2"]),
        _to_num(r["avg_area_m2"]),
        # Offer
        _to_int(r["stock_lots"]),
        _to_num(r["stock_m2"]),
        None,
        _to_num(r["stock_avg_price_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:
    obj_id = _site_id_to_obj_id(r["site_id"])
    if obj_id is None:
        return None
    return (
        r["project"],
        None,
        GROUP_NAME,
        obj_id,
        False,
        r["method"],
        _to_num(r["score"]),
        "imported_from_anton_sf",
    )


_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, batch_size: int, progress_cb: Callable[[int], None] | None
) -> int:
    snap = date.today()
    cur = sl.execute("SELECT * FROM objective_lots")
    n = 0
    batch: list[tuple] = []
    pg_cur = pg.cursor()
    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:
                psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=batch_size)
                pg.commit()
                n += len(batch)
                batch.clear()
                if progress_cb and n % (batch_size * 5) == 0:
                    progress_cb(n)
    if batch:
        psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=len(batch))
        pg.commit()
        n += len(batch)
    pg_cur.close()
    if progress_cb:
        progress_cb(n)
    return n


def _etl_crm(
    sl: sqlite3.Connection, pg, batch_size: int, progress_cb: Callable[[int], None] | None
) -> tuple[int, int]:
    cur = sl.execute("SELECT * FROM objective_corp_month")
    n = 0
    skipped = 0
    batch: list[tuple] = []
    pg_cur = pg.cursor()
    while True:
        rows = cur.fetchmany(batch_size)
        if not rows:
            break
        for r in rows:
            t = _row_to_crm(r)
            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:
                psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=batch_size)
                pg.commit()
                n += len(batch)
                batch.clear()
    if batch:
        psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=len(batch))
        pg.commit()
        n += len(batch)
    pg_cur.close()
    if progress_cb:
        progress_cb(n)
    return n, skipped


def _etl_mapping(sl: sqlite3.Connection, pg) -> tuple[int, int]:
    cur = sl.execute("SELECT * FROM jk_objective_match ORDER BY score DESC NULLS LAST")
    rows = cur.fetchall()
    seen: set[tuple] = 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 not batch:
        return 0, skipped_dups
    pg_cur = pg.cursor()
    psycopg2.extras.execute_values(pg_cur, _MAP_INSERT, batch, page_size=500)
    pg.commit()
    pg_cur.close()
    return len(batch), skipped_dups


# ── public API ──────────────────────────────────────────────────────────────


def get_sqlite_info(sqlite_path: str | Path) -> dict[str, Any]:
    """Метаданные SQLite-файла без полного запуска ETL.

    Используется в admin-coverage endpoint чтобы UI знал свежесть данных,
    размер файла и сколько потенциально rows будет залито.
    """
    p = Path(sqlite_path)
    info: dict[str, Any] = {
        "path": str(p),
        "exists": p.exists(),
    }
    if not p.exists():
        return info
    st = p.stat()
    info["size_bytes"] = st.st_size
    info["modified_at"] = st.st_mtime  # epoch seconds
    try:
        c = sqlite3.connect(p)
        info["lots"] = c.execute("SELECT COUNT(*) FROM objective_lots").fetchone()[0]
        info["corp_room_month"] = c.execute("SELECT COUNT(*) FROM objective_corp_month").fetchone()[
            0
        ]
        info["mappings"] = c.execute("SELECT COUNT(*) FROM jk_objective_match").fetchone()[0]
        c.close()
    except sqlite3.Error as e:
        info["error"] = str(e)
    return info


def run_etl(
    sqlite_path: str | Path,
    db_url: str,
    *,
    batch_size: int = 2000,
    progress_cb: Callable[[str, int], None] | None = None,
) -> dict[str, int]:
    """Полный ETL: SQLite → PG.

    Args:
        sqlite_path: путь к analysis.db (на проде — bind-mount /data/anton-sqlite/...)
        db_url: postgresql://… connection string
        batch_size: размер batch для execute_values (default 2000)
        progress_cb: optional callback(stage: str, n_rows: int) для UI прогресса

    Returns:
        dict с counts: {'lots', 'corp_room_month', 'mappings', 'crm_skipped',
                        'mapping_dups_skipped'}

    Raises:
        FileNotFoundError если SQLite не найден
        psycopg2.Error при ошибках БД
    """
    p = Path(sqlite_path)
    if not p.exists():
        raise FileNotFoundError(f"SQLite не найден: {p}")

    logger.info("ETL Anton SQLite → PG: %s", p)
    sl = sqlite3.connect(p)
    sl.row_factory = sqlite3.Row
    pg = psycopg2.connect(db_url)
    try:
        if progress_cb:
            progress_cb("mapping_start", 0)
        n_map, dups = _etl_mapping(sl, pg)
        logger.info("  mapping: %d upserted, %d dups skipped", n_map, dups)
        if progress_cb:
            progress_cb("mapping_done", n_map)

        if progress_cb:
            progress_cb("crm_start", 0)
        n_crm, crm_skipped = _etl_crm(
            sl,
            pg,
            batch_size,
            lambda n: progress_cb("crm_progress", n) if progress_cb else None,
        )
        logger.info("  crm: %d upserted, %d skipped", n_crm, crm_skipped)
        if progress_cb:
            progress_cb("crm_done", n_crm)

        if progress_cb:
            progress_cb("lots_start", 0)
        n_lots = _etl_lots(
            sl,
            pg,
            batch_size,
            lambda n: progress_cb("lots_progress", n) if progress_cb else None,
        )
        logger.info("  lots: %d upserted", n_lots)
        if progress_cb:
            progress_cb("lots_done", n_lots)
    finally:
        sl.close()
        pg.close()

    return {
        "lots": n_lots,
        "corp_room_month": n_crm,
        "mappings": n_map,
        "crm_skipped": crm_skipped,
        "mapping_dups_skipped": dups,
    }