"""Cross-load ETL: tradein.houses → gendesign.newbuilding_listings (#976 950-E5). Читает houses + house_sources из tradein-БД (gendesign_reader role), UPSERT-ит в gendesign newbuilding_listings по (source, ext_house_id). Соединение к tradein через psycopg v3 (прямой коннект, НЕ SQLAlchemy). Запускается: - ночью Celery beat (03:30 МСК, tasks/etl_newbuilding_crossload.py) - вручную через POST /api/v1/admin/scrape/newbuilding-crossload Если settings.tradein_database_url пусто — сразу возвращает {"disabled": True}. """ from __future__ import annotations import json import logging from typing import Any import psycopg from sqlalchemy import text from sqlalchemy.orm import Session from app.core.config import settings from app.core.db import SessionLocal logger = logging.getLogger(__name__) _BATCH_SIZE = 500 # ── SELECT из tradein ──────────────────────────────────────────────────────── _SOURCE_SQL = """ SELECT h.source, h.ext_house_id, h.url, h.slug, h.address, h.full_address, h.lat, h.lon, h.house_class, h.developer_name, h.developer_key, h.year_built, h.total_floors, h.total_units, h.rating, h.reviews_count, h.houses_by_turn, h.cian_internal_house_id, h.raw_payload, h.last_scraped_at, -- yandex_jk_id из house_sources (LEFT JOIN — может отсутствовать) hs_y.ext_id AS yandex_jk_id FROM houses h LEFT JOIN house_sources hs_y ON hs_y.house_id = h.id AND hs_y.ext_source = 'yandex' WHERE h.lat IS NOT NULL AND h.lon IS NOT NULL """ # ── UPSERT в gendesign ─────────────────────────────────────────────────────── _UPSERT_SQL = text(""" INSERT INTO newbuilding_listings ( source, ext_house_id, url, slug, address, full_address, lat, lon, geom, house_class, developer_name, developer_key, year_built, total_floors, total_units, rating, reviews_count, houses_by_turn, cian_internal_house_id, yandex_jk_id, raw_payload, source_last_scraped_at, crossloaded_at ) VALUES ( :source, :ext_house_id, :url, :slug, :address, :full_address, :lat, :lon, CASE WHEN :lat2 IS NOT NULL AND :lon2 IS NOT NULL THEN ST_SetSRID(ST_MakePoint(CAST(:lon2 AS double precision), CAST(:lat2 AS double precision)), 4326) ELSE NULL END, :house_class, :developer_name, :developer_key, CAST(:year_built AS integer), CAST(:total_floors AS integer), CAST(:total_units AS integer), CAST(:rating AS numeric), CAST(:reviews_count AS integer), CAST(:houses_by_turn AS jsonb), CAST(:cian_internal_house_id AS bigint), :yandex_jk_id, CAST(:raw_payload AS jsonb), CAST(:source_last_scraped_at AS timestamptz), now() ) ON CONFLICT (source, ext_house_id) DO UPDATE SET url = EXCLUDED.url, slug = EXCLUDED.slug, address = EXCLUDED.address, full_address = EXCLUDED.full_address, lat = EXCLUDED.lat, lon = EXCLUDED.lon, geom = EXCLUDED.geom, house_class = EXCLUDED.house_class, developer_name = EXCLUDED.developer_name, developer_key = EXCLUDED.developer_key, year_built = EXCLUDED.year_built, total_floors = EXCLUDED.total_floors, total_units = EXCLUDED.total_units, rating = EXCLUDED.rating, reviews_count = EXCLUDED.reviews_count, houses_by_turn = EXCLUDED.houses_by_turn, cian_internal_house_id = COALESCE( EXCLUDED.cian_internal_house_id, newbuilding_listings.cian_internal_house_id), yandex_jk_id = COALESCE( EXCLUDED.yandex_jk_id, newbuilding_listings.yandex_jk_id), raw_payload = EXCLUDED.raw_payload, source_last_scraped_at = EXCLUDED.source_last_scraped_at, crossloaded_at = now() """) def _row_to_params(row: Any) -> dict[str, Any]: """Преобразовать строку из tradein-курсора в dict параметров для UPSERT. Публичная функция — тестируется отдельно (без БД). row — объект с атрибутами (psycopg RealDictRow / dict / named row). """ if hasattr(row, "_asdict"): r: dict[str, Any] = row._asdict() elif hasattr(row, "keys"): r = dict(row) else: r = dict(row) houses_by_turn = r.get("houses_by_turn") raw_payload = r.get("raw_payload") return { "source": r["source"], "ext_house_id": r["ext_house_id"], "url": r.get("url"), "slug": r.get("slug"), "address": r.get("address"), "full_address": r.get("full_address"), "lat": r.get("lat"), "lon": r.get("lon"), # дублируем lat/lon для CASE-ветки geom (psycopg не позволяет один bind два раза) "lat2": r.get("lat"), "lon2": r.get("lon"), "house_class": r.get("house_class"), "developer_name": r.get("developer_name"), "developer_key": r.get("developer_key"), "year_built": r.get("year_built"), "total_floors": r.get("total_floors"), "total_units": r.get("total_units"), "rating": str(r["rating"]) if r.get("rating") is not None else None, "reviews_count": r.get("reviews_count"), "houses_by_turn": json.dumps(houses_by_turn, ensure_ascii=False) if houses_by_turn is not None else None, "cian_internal_house_id": r.get("cian_internal_house_id"), "yandex_jk_id": r.get("yandex_jk_id"), "raw_payload": json.dumps(raw_payload, ensure_ascii=False) if raw_payload is not None else None, "source_last_scraped_at": r.get("last_scraped_at"), } def run_crossload(db: Session | None = None) -> dict[str, Any]: """Выполнить cross-load ETL tradein→gendesign. Если settings.tradein_database_url пуст — возвращает {"disabled": True}, не коннектится. Returns: dict с ключами: source_rows, upserted, skipped — или {"disabled": True}. """ if not settings.tradein_database_url: logger.warning("etl_newbuilding_crossload: TRADEIN_DATABASE_URL не задан — ETL отключён") return {"disabled": True} close_db = db is None if db is None: db = SessionLocal() source_rows = 0 upserted = 0 skipped = 0 logger.info("etl_newbuilding_crossload: start (batch_size=%d)", _BATCH_SIZE) try: connstr = settings.tradein_database_url with psycopg.connect(connstr, row_factory=psycopg.rows.dict_row) as src: with src.cursor(name="crossload_cur") as cur: # server-side cursor cur.execute(_SOURCE_SQL) while True: batch = cur.fetchmany(_BATCH_SIZE) if not batch: break source_rows += len(batch) params_list = [_row_to_params(r) for r in batch] for params in params_list: try: with db.begin_nested(): db.execute(_UPSERT_SQL, params) upserted += 1 except Exception as exc: skipped += 1 logger.warning( "etl_newbuilding_crossload: upsert failed " "source=%s ext_id=%s: %s", params.get("source"), params.get("ext_house_id"), exc, ) db.commit() except Exception: if not close_db: db.rollback() raise finally: if close_db: db.close() logger.info( "etl_newbuilding_crossload: done — source_rows=%d upserted=%d skipped=%d", source_rows, upserted, skipped, ) return {"source_rows": source_rows, "upserted": upserted, "skipped": skipped}