fix(tradein/houses): вернуть координаты объявлений в дом, когда объявления согласны (#2771)

Дом берёт координаты один раз — при создании, из породившего его объявления.
Объявление геокодируется ПОЗЖЕ (geoportal_coords_backfill, geocode_missing_listings)
и пишет в listings; ON CONFLICT у вставки дома обновляет только address. Обратно
координаты не возвращались: на проде 1945 домов из 9625 (20.2%) без координат, при
этом у 1490 из них объявления с координатами уже есть.

Новая ежедневная задача house_coords_from_listings (окно 03:00-04:00 UTC, перед
house_dedup_merge) считает центроид объявлений дома и их разброс — max
ST_DistanceSphere до центроида. Разброс <= 125 м → центроид пишется в дом.

Порог 125 м взят не из данных (гистограмма обрыва не даёт: ядро 0-125 м, дальше
ровное плато до 400 м), а от потребителя координат: это половина 250-метрового
гео-ограждения house_dedup_merge (#2187). Записанный центроид — оценка с
погрешностью ровно в разброс; отдать ей весь бюджет ограждения значит позволить
двум домам разъехаться на 500 м. Цена решения мала: 125 м пропускает 1382 дома,
250 м — 1403.

Дома с бо́льшим разбросом не трогаются и считаются отдельным счётчиком: это дефект
сопоставителя (#2777), а не задача переноса — в одну запись дома сшиты объявления
разных населённых пунктов, худший разброс 289.9 км. Пропуск виден по каждой
причине отдельно (no_listings / no_listing_coords / scattered), худшие дома
называются в логе поимённо.

Идемпотентно: UPDATE только по lat IS NULL, повторный прогон — no-op.

Замер тем же SQL, что мержится (прод, read-only, ROLLBACK):
candidates=1945 eligible=1382 no_listings=114 no_listing_coords=341
scattered=108 (18 дальше 5 км, худший 289.9 км).

tradein-mvp/backend/app/services/product_handlers.py

@@ -350,6 +350,19 @@ async def _job_domclick_detail_backfill(
     await run_domclick_detail_backfill(db, run_id=run_id, params=params)
 
 
+# ── house_coords_from_listings — sync set-based UPDATE в executor (#2771) ─────
+async def _job_house_coords_from_listings(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.backfill_house_coords_from_listings import run_house_coords_from_listings
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(
+        None,
+        lambda: run_house_coords_from_listings(db, run_id=run_id, params=params),
+    )
+
+
 # ── geoportal_coords_backfill — sync local exact match в executor (#1967) ─────
 async def _job_geoportal_coords_backfill(
     db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
@@ -586,6 +599,9 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
         "geoportal_coords_backfill": Handler(
             _job_geoportal_coords_backfill, "geoportal_coords_backfill"
         ),
+        "house_coords_from_listings": Handler(
+            _job_house_coords_from_listings, "house_coords_from_listings"
+        ),
         "geocode_missing_listings": Handler(
             _job_geocode_missing_listings, "geocode_missing_listings"
         ),

tradein-mvp/backend/app/tasks/backfill_house_coords_from_listings.py

@@ -0,0 +1,323 @@
+"""Перенос координат из объявлений в дом, когда объявления между собой согласны (#2771).
+
+ЗАЧЕМ. Дом получает координаты РОВНО ОДИН РАЗ — в момент создания, из породившего его
+объявления (`matching/houses.py`). Объявление без координат создаёт дом без координат;
+позже объявление геокодируется (`geoportal_coords_backfill`, `geocode_missing_listings` —
+обе живые ежедневные задачи), но пишут они в `listings`, а `ON CONFLICT` у вставки дома
+обновляет только `address`. Обратно в дом координаты не возвращались никогда: из 52
+источников расписания `houses.lat`/`houses.lon` не писал ни один. Замер на проде 2026-08-07:
+1945 домов из 9625 (20.2%) без координат, при этом у 1490 из них объявления с координатами
+УЖЕ ЕСТЬ. Эта задача — недостающее звено проводки, и именно ПОВТОРЯЮЩЕЕСЯ: однократный
+скрипт закрыл бы дыру на день, а приток новых домов без координат никуда не девается.
+
+ЧТО ДЕЛАЕТ. Для каждого дома с lat IS NULL берёт geom всех связанных объявлений, считает
+центроид и РАЗБРОС — максимальное расстояние от центроида до объявления (ST_DistanceSphere,
+метры на сфере). Разброс <= max_spread_m → центроид пишется в дом; иначе дом НЕ трогается
+и попадает в счётчик skipped_scattered.
+
+ПОРОГ 125 м — почему столько (замер на проде 2026-08-07, ST_DistanceSphere, не приближение
+«градусы × 111320»). Гистограмма разброса по 258 домам с >1 различной точкой не даёт обрыва:
+плотное ядро 0-125 м (150 домов), затем ровное плато ~4-5 домов на каждые 25 м вплоть до
+400 м, и масса из 67 домов дальше 400 м. Раз в данных обрыва нет, порог берётся не из них,
+а из ПОТРЕБИТЕЛЯ: гео-ограждение `house_dedup_merge` (#2187) считает два дома одним зданием
+при ST_DistanceSphere <= 250 м. Записанный сюда центроид — оценка со своей погрешностью
+ровно в этот разброс; если разрешить каждому дому ошибаться на все 250 м, два дома с
+максимальной ошибкой разъедутся на 500 м, и ограждение отвергнет верную пару (или сведёт
+неверную). Отдаём своей погрешности ПОЛОВИНУ бюджета ограждения — 125 м. Цена решения
+мала и потому оно не несущее: 125 м пропускает 1382 дома, 250 м — 1403, разница 21 дом,
+и это не потеря, а отсрочка (дом остаётся NULL и попадёт в следующий прогон, если его
+объявления сойдутся).
+
+ЧЕГО НЕ ДЕЛАЕТ. Не обещает роста числа слияний дублей. Сейчас гео-ограждение отбраковывает
+проигравших с NULL geom, потому что проверить тождество нечем; после переноса у него
+появляется ВОЗМОЖНОСТЬ оценить пару — часть сольётся, часть будет отвергнута правильно
+(дома в разных городах). Ни то ни другое здесь не предрешается.
+
+ДОМА С БОЛЬШИМ РАЗБРОСОМ — не задача переноса, а дефект сопоставителя: в одну запись дома
+сшиты объявления из разных населённых пунктов (худший на проде — «ул. Кирова,4», разброс
+290 км). Они не переносятся, считаются отдельным счётчиком и логируются поимённо (топ-5),
+чтобы пропуск был виден, а не читался как «покрыто всё».
+
+ИДЕМПОТЕНТНОСТЬ. UPDATE только по `h.lat IS NULL` (тот же приём, что в
+`backfill_listings_coords_geoportal`). Повторный прогон не трогает уже проставленные
+координаты и на чистой базе — no-op с нулевыми счётчиками.
+
+geom СТРОИТСЯ ЯВНО. Триггер `houses_set_geom_trg` на проде — BEFORE INSERT OR UPDATE OF
+lat, lon (проверено 2026-08-07), то есть geom построился бы и сам; UPDATE всё равно пишет
+geom явно — как это делает `_update_listing_coords` в geoportal-задаче — чтобы правка не
+зависела от живости триггера и не оказалась беззвучной, если его когда-нибудь снимут.
+
+Запуск:
+  python -m app.tasks.backfill_house_coords_from_listings
+  python -m app.tasks.backfill_house_coords_from_listings --max-spread-m 125
+Ежедневно через in-app scheduler (source='house_coords_from_listings', migration 239).
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import SessionLocal
+from app.services import scrape_runs as runs_mod
+
+logger = logging.getLogger(__name__)
+
+# Половина 250-метрового гео-ограждения house_dedup_merge (#2187) — обоснование в docstring.
+DEFAULT_MAX_SPREAD_M = 125.0
+
+# «Разброс в километрах» — порог отдельного счётчика для дефектов сопоставителя (#2777).
+SCATTER_DEFECT_M = 5000.0
+
+# Сколько худших домов с разбросом назвать поимённо в логе.
+_SCATTER_LOG_TOP = 5
+
+# Кандидаты: КАЖДЫЙ дом без координат + агрегат по его объявлениям. LEFT JOIN, чтобы
+# «объявлений нет вовсе» и «объявления есть, но без координат» различались в счётчиках.
+# ST_Collect/count(geom) пропускают NULL — дом без геокодированных объявлений получает
+# cen IS NULL → lat/lon NULL → в UPDATE не попадает (n_geocoded = 0).
+_BUILD_CANDIDATES_SQL = """
+CREATE TEMP TABLE _2771_house_coords ON COMMIT DROP AS
+WITH coordless AS (
+    SELECT id FROM houses WHERE lat IS NULL OR lon IS NULL
+),
+linked AS (
+    SELECT c.id AS house_id, l.id AS listing_id, l.geom
+    FROM coordless c
+    LEFT JOIN listings l ON l.house_id_fk = c.id
+),
+agg AS (
+    SELECT house_id,
+           count(listing_id) AS n_listings,
+           count(geom) AS n_geocoded,
+           ST_Centroid(ST_Collect(geom)) AS cen
+    FROM linked
+    GROUP BY house_id
+),
+spread AS (
+    SELECT l.house_id,
+           max(ST_DistanceSphere(l.geom, a.cen)) AS spread_m
+    FROM linked l
+    JOIN agg a ON a.house_id = l.house_id
+    WHERE l.geom IS NOT NULL
+    GROUP BY l.house_id
+)
+SELECT a.house_id,
+       a.n_listings,
+       a.n_geocoded,
+       ST_Y(a.cen) AS lat,
+       ST_X(a.cen) AS lon,
+       s.spread_m
+FROM agg a
+LEFT JOIN spread s ON s.house_id = a.house_id
+"""
+
+_COUNTERS_SQL = """
+SELECT count(*)                                                        AS candidates,
+       count(*) FILTER (WHERE n_listings = 0)                          AS skipped_no_listings,
+       count(*) FILTER (WHERE n_listings > 0 AND n_geocoded = 0)       AS skipped_no_listing_coords,
+       count(*) FILTER (WHERE n_geocoded > 0
+                          AND spread_m > CAST(:max_spread_m AS double precision))
+                                                                       AS skipped_scattered,
+       count(*) FILTER (WHERE n_geocoded > 0
+                          AND spread_m > CAST(:defect_m AS double precision))
+                                                                       AS scattered_over_5km,
+       COALESCE(max(spread_m) FILTER (WHERE n_geocoded > 0
+                          AND spread_m > CAST(:max_spread_m AS double precision)), 0)
+                                                                       AS worst_spread_m,
+       count(*) FILTER (WHERE n_geocoded > 0
+                          AND spread_m <= CAST(:max_spread_m AS double precision))
+                                                                       AS eligible
+FROM _2771_house_coords
+"""
+
+# Идемпотентность: h.lat IS NULL повторяется в UPDATE (а не только в отборе кандидатов) —
+# уже проставленные координаты не перезаписываются ни при каком стечении обстоятельств.
+_APPLY_SQL = """
+UPDATE houses h
+SET lat  = c.lat,
+    lon  = c.lon,
+    geom = ST_SetSRID(ST_MakePoint(c.lon, c.lat), 4326)
+FROM _2771_house_coords c
+WHERE h.id = c.house_id
+  AND h.lat IS NULL
+  AND c.n_geocoded > 0
+  AND c.spread_m <= CAST(:max_spread_m AS double precision)
+"""
+
+_SCATTER_TOP_SQL = """
+SELECT c.house_id, h.address, c.n_geocoded, c.spread_m
+FROM _2771_house_coords c
+JOIN houses h ON h.id = c.house_id
+WHERE c.n_geocoded > 0
+  AND c.spread_m > CAST(:max_spread_m AS double precision)
+ORDER BY c.spread_m DESC
+LIMIT CAST(:top AS integer)
+"""
+
+
+@dataclass
+class HouseCoordsResult:
+    """Счётчики прогона. Пропуск виден отдельно по КАЖДОЙ причине — молчаливое усечение
+    читается как «покрыто всё»."""
+
+    candidates: int = 0  # домов без координат всего
+    eligible: int = 0  # объявления согласны (разброс <= порога)
+    updated: int = 0  # реально обновлено (UPDATE rowcount)
+    skipped_no_listings: int = 0  # объявлений у дома нет вовсе
+    skipped_no_listing_coords: int = 0  # объявления есть, но сами без координат
+    skipped_scattered: int = 0  # разброс больше порога — дефект сопоставителя (#2777)
+    scattered_over_5km: int = 0  # из них разъехавшиеся на километры
+    worst_spread_m: int = 0  # худший пропущенный разброс, метры
+    max_spread_m: int = 0  # применённый порог, метры (чтобы счётчики читались без кода)
+    duration_sec: float = field(default=0.0)
+
+    def to_counters(self) -> dict[str, int]:
+        return {
+            "candidates": self.candidates,
+            "eligible": self.eligible,
+            "updated": self.updated,
+            "skipped_no_listings": self.skipped_no_listings,
+            "skipped_no_listing_coords": self.skipped_no_listing_coords,
+            "skipped_scattered": self.skipped_scattered,
+            "scattered_over_5km": self.scattered_over_5km,
+            "worst_spread_m": self.worst_spread_m,
+            "max_spread_m": self.max_spread_m,
+            "duration_sec": int(self.duration_sec),
+        }
+
+
+def backfill_house_coords(
+    db: Session,
+    *,
+    max_spread_m: float = DEFAULT_MAX_SPREAD_M,
+) -> HouseCoordsResult:
+    """Перенести координаты объявлений в дома, где объявления согласны.
+
+    Одна транзакция: TEMP-таблица кандидатов (ON COMMIT DROP) → счётчики → UPDATE.
+    Счётчики считаются по ТОЙ ЖЕ таблице, из которой идёт UPDATE, поэтому «сколько
+    пропущено» относится ровно к этому прогону, а не к соседнему запросу.
+
+    Args:
+        db: сессия tradein-БД.
+        max_spread_m: предел разброса объявлений, метры (см. DEFAULT_MAX_SPREAD_M).
+
+    Returns:
+        HouseCoordsResult со счётчиками по каждой причине пропуска.
+    """
+    start = time.monotonic()
+    res = HouseCoordsResult(max_spread_m=int(max_spread_m))
+    params = {"max_spread_m": max_spread_m, "defect_m": SCATTER_DEFECT_M}
+
+    db.execute(text(_BUILD_CANDIDATES_SQL))
+
+    row = db.execute(text(_COUNTERS_SQL), params).mappings().one()
+    res.candidates = int(row["candidates"])
+    res.eligible = int(row["eligible"])
+    res.skipped_no_listings = int(row["skipped_no_listings"])
+    res.skipped_no_listing_coords = int(row["skipped_no_listing_coords"])
+    res.skipped_scattered = int(row["skipped_scattered"])
+    res.scattered_over_5km = int(row["scattered_over_5km"])
+    res.worst_spread_m = int(row["worst_spread_m"])
+
+    if res.skipped_scattered:
+        worst = (
+            db.execute(text(_SCATTER_TOP_SQL), {**params, "top": _SCATTER_LOG_TOP}).mappings().all()
+        )
+        logger.warning(
+            "house_coords: ПРОПУЩЕНО %d домов с разбросом объявлений > %.0f м "
+            "(из них %d дальше %.0f км) — это дефект сопоставителя (#2777), не переноса. "
+            "Худшие: %s",
+            res.skipped_scattered,
+            max_spread_m,
+            res.scattered_over_5km,
+            SCATTER_DEFECT_M / 1000,
+            "; ".join(
+                f"id={w['house_id']} {w['address']!r} n={w['n_geocoded']} "
+                f"разброс={w['spread_m'] / 1000:.1f} км"
+                for w in worst
+            ),
+        )
+
+    res.updated = int(db.execute(text(_APPLY_SQL), params).rowcount or 0)
+    db.commit()
+
+    res.duration_sec = time.monotonic() - start
+    logger.info(
+        "house_coords: DONE — кандидатов=%d годных=%d обновлено=%d | пропущено: "
+        "без_объявлений=%d объявления_без_координат=%d разброс>%.0fм=%d (худший %d м) "
+        "время=%.1fs",
+        res.candidates,
+        res.eligible,
+        res.updated,
+        res.skipped_no_listings,
+        res.skipped_no_listing_coords,
+        max_spread_m,
+        res.skipped_scattered,
+        res.worst_spread_m,
+        res.duration_sec,
+    )
+    return res
+
+
+# ── Run lifecycle wrapper (scheduler entrypoint) ─────────────────────────────
+def run_house_coords_from_listings(
+    db: Session,
+    *,
+    run_id: int,
+    params: dict,
+) -> HouseCoordsResult:
+    """Run-lifecycle обёртка (kit scheduler, source='house_coords_from_listings').
+
+    Params (default_params jsonb):
+        max_spread_m: предел разброса в метрах (default 125).
+    """
+    max_spread_m = float(params.get("max_spread_m", DEFAULT_MAX_SPREAD_M))
+    counters: dict[str, int] = HouseCoordsResult(max_spread_m=int(max_spread_m)).to_counters()
+
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+        res = backfill_house_coords(db, max_spread_m=max_spread_m)
+        counters = res.to_counters()
+        runs_mod.mark_done(db, run_id, counters)
+        return res
+    except Exception as exc:
+        logger.exception("run_house_coords_from_listings: run_id=%d FAILED", run_id)
+        try:
+            db.rollback()
+        except Exception:
+            logger.warning("run_house_coords_from_listings: rollback failed", exc_info=True)
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    parser = argparse.ArgumentParser(
+        description="Перенос координат объявлений в дома, где объявления согласны (#2771)"
+    )
+    parser.add_argument(
+        "--max-spread-m",
+        type=float,
+        default=DEFAULT_MAX_SPREAD_M,
+        help=f"Предел разброса объявлений, метры (default: {DEFAULT_MAX_SPREAD_M:.0f})",
+    )
+    args = parser.parse_args()
+
+    db = SessionLocal()
+    try:
+        backfill_house_coords(db, max_spread_m=args.max_spread_m)
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    main()

tradein-mvp/backend/data/sql/239_scrape_schedules_seed_house_coords_from_listings.sql

@@ -0,0 +1,72 @@
+-- 239_scrape_schedules_seed_house_coords_from_listings.sql
+-- Ежедневный перенос координат из объявлений в дом, когда объявления согласны (#2771).
+--
+-- ЧТО (source='house_coords_from_listings'):
+--   product_handlers._job_house_coords_from_listings → run_house_coords_from_listings
+--   (app/tasks/backfill_house_coords_from_listings.py). Для каждого дома с lat IS NULL:
+--   центроид geom его объявлений + разброс (max ST_DistanceSphere до центроида).
+--   Разброс <= max_spread_m → центроид пишется в дом; больше → дом не трогается и
+--   попадает в отдельный счётчик skipped_scattered.
+--
+-- ЗАЧЕМ ПОВТОРЯЮЩАЯСЯ, а не однократный скрипт:
+--   Дом берёт координаты один раз при создании, из породившего его объявления. Объявление
+--   геокодируется ПОЗЖЕ (geoportal_coords_backfill, geocode_missing_listings) и пишет в
+--   listings; ON CONFLICT у вставки дома обновляет только address. Обратно координаты не
+--   возвращались — на проде 2026-08-07 накопилось 1945 домов из 9625 (20.2%) без координат,
+--   1490 из них уже имеют объявления С координатами. Приток новых таких домов не
+--   прекращается, поэтому однократный прогон закрыл бы дыру на день.
+--
+-- ПОРОГ 125 м: половина 250-метрового гео-ограждения house_dedup_merge (#2187). Записанный
+--   центроид — оценка с погрешностью ровно в разброс; отдать ей весь бюджет ограждения
+--   значит позволить двум домам разъехаться на 500 м и сломать сравнение. Гистограмма
+--   разброса обрыва не даёт (ядро 0-125 м, затем ровное плато до 400 м), поэтому порог
+--   взят от потребителя, а не подогнан к данным. Цена: 125 м пропускает 1382 дома,
+--   250 м — 1403; 21 дом разницы не теряется, а откладывается до следующего прогона.
+--
+-- ОКНО 03:00-04:00 UTC, ежедневно (interval_days по умолчанию = 1):
+--   ПЕРЕД house_dedup_merge (04:00-05:00, migration 135) — недельный дедуп застаёт
+--   координаты, перенесённые этой ночью из объявлений, геокодированных накануне
+--   (geoportal_coords_backfill 05:00-06:00, geocode_missing_listings 06:00-09:00).
+--   Чисто внутренняя операция БД: ни HTTP, ни прокси, ни anti-bot — конкуренции нет.
+--
+-- enabled=true: операция НЕ деструктивная — пишет только туда, где lat IS NULL, и не
+--   перезаписывает ни одной существующей координаты (тот же приём, что у
+--   geoportal_coords_backfill, migration 171, тоже засеянного enabled=true).
+--
+-- default_params:
+--   max_spread_m -- 125: предел разброса объявлений в метрах (см. выше).
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)).
+-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно применять повторно.
+-- Runner применяет миграции БЕЗ --single-transaction, поэтому явные BEGIN/COMMIT.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'house_coords_from_listings',
+    true,
+    3,
+    4,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 3)) AT TIME ZONE 'UTC',
+    '{"max_spread_m": 125}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет cron-script setup). Источники перечислены в '
+    'tests/test_scraper_kit_scheduler_parity.py::_PRODUCT_SOURCES и в сид-миграциях '
+    'data/sql/*scrape_schedules*seed*.sql. Последний добавленный: '
+    'house_coords_from_listings (#2771: ежедневный перенос координат из объявлений в дом '
+    'при согласии объявлений, разброс <= 125 м; дома с большим разбросом пропускаются '
+    'и считаются отдельно — это дефект сопоставителя #2777, не переноса).';
+
+COMMIT;

tradein-mvp/backend/tests/skip_allowlist.txt

@@ -28,6 +28,7 @@ tests/test_pdf_real_render.py
 #   docker run -d -p 5432:5432 -e POSTGRES_DB=tradein -e POSTGRES_USER=tradein \
 #     -e POSTGRES_PASSWORD=tradein postgis/postgis:16-3.4
 #   psql ... -f backend/data/sql/*.sql   # порядок как в deploy-tradein.yml
+tests/tasks/test_backfill_house_coords_from_listings.py::test_real_transfers_agreeing_coords_skips_scattered_and_is_idempotent
 tests/tasks/test_cadastral_geo_match.py::test_real_knn_nearest_within_threshold_picked
 tests/test_audit_api.py::test_real_accounts_and_analytics_aggregate_inserted_rows
 tests/test_gar_flats_loader.py::test_upsert_and_canon_match_populates_gar_flat_count

tradein-mvp/backend/tests/tasks/test_backfill_house_coords_from_listings.py

@@ -0,0 +1,218 @@
+"""Тесты переноса координат из объявлений в дом (#2771).
+
+Форма — как у соседей (test_house_dedup_merge.py, tests/tasks/test_cadastral_geo_match.py):
+статические проверки эмитируемого SQL + миграции (бегут везде) и поведенческий прогон на
+ЖИВОМ Postgres/PostGIS (в CI он есть — postgis/postgis:16-3.4; на ноутбуке self-skip,
+запись в tests/skip_allowlist.txt).
+
+Поведенческий тест обязан покрывать НЕ ТОЛЬКО перенос: главный риск задачи — молча усечь
+дома с разбросом и отчитаться «покрыто всё». Поэтому в фикстуре есть дом с объявлениями
+за 11 км друг от друга, и на нём проверяется именно ПРОПУСК (lat остался NULL), а не запись.
+
+Регистрация handler'а в scheduler-реестре проверяется НЕ здесь, а в каноническом месте —
+test_scraper_kit_scheduler_parity.py::_PRODUCT_SOURCES (туда source добавлен).
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.tasks import backfill_house_coords_from_listings as mod
+
+_MIGRATION = (
+    Path(__file__).resolve().parents[2]
+    / "data"
+    / "sql"
+    / "239_scrape_schedules_seed_house_coords_from_listings.sql"
+)
+
+
+# ── статика: инварианты, которые нельзя потерять при рефакторе ────────────────
+
+
+def test_update_touches_only_coordless_houses() -> None:
+    """Идемпотентность: UPDATE обязан нести h.lat IS NULL — иначе прогон затрёт уже
+    проставленные (возможно более точные) координаты."""
+    assert "h.lat IS NULL" in mod._APPLY_SQL
+
+
+def test_update_is_gated_by_spread_threshold() -> None:
+    """Без этого условия перенос применился бы и к домам с объявлениями в разных городах."""
+    assert "c.spread_m <= CAST(:max_spread_m AS double precision)" in mod._APPLY_SQL
+    assert "c.n_geocoded > 0" in mod._APPLY_SQL
+
+
+def test_spread_measured_in_metres_not_degrees() -> None:
+    """Разброс считается ST_DistanceSphere (метры), а не приближением «градусы × 111320»."""
+    assert "ST_DistanceSphere" in mod._BUILD_CANDIDATES_SQL
+
+
+def test_threshold_is_half_of_dedup_geo_guard() -> None:
+    """125 м = половина 250-метрового гео-ограждения house_dedup_merge (#2187).
+
+    Если ограждение когда-нибудь поменяют, этот тест покажет, что порог переноса больше
+    не согласован с потребителем координат.
+    """
+    from app.services import house_dedup_merge as hdm
+
+    assert "<= 250" in str(hdm._BUILD_MAPPING_SQL.text)
+    assert mod.DEFAULT_MAX_SPREAD_M == 125.0
+
+
+def test_counters_name_every_skip_reason_separately() -> None:
+    """Пропуск должен быть виден по причинам, иначе счётчики читаются как «покрыто всё»."""
+    counters = mod.HouseCoordsResult().to_counters()
+    for key in (
+        "skipped_no_listings",
+        "skipped_no_listing_coords",
+        "skipped_scattered",
+        "scattered_over_5km",
+        "worst_spread_m",
+    ):
+        assert key in counters
+
+
+def test_migration_239_seeds_daily_schedule_before_dedup() -> None:
+    sql = _MIGRATION.read_text(encoding="utf-8")
+    assert "'house_coords_from_listings'" in sql
+    assert '"max_spread_m": 125' in sql
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+    # окно 03:00-04:00 UTC — до house_dedup_merge (04:00-05:00, migration 135)
+    assert re.search(r"^\s+3,\s*$", sql, re.M), "window_start_hour must be 3"
+
+
+def test_migration_239_no_psycopg_v3_colon_colon_cast() -> None:
+    assert not re.search(r":\w+::", _MIGRATION.read_text(encoding="utf-8"))
+
+
+# ── поведение на живом Postgres/PostGIS (self-skip без БД) ────────────────────
+
+
+def _live_session() -> Any | None:
+    """Session на достижимом не-заглушечном Postgres, иначе None (копия приёма из
+    tests/test_house_dedup_merge.py)."""
+    try:
+        from sqlalchemy import create_engine, text
+        from sqlalchemy.orm import sessionmaker
+
+        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+        if not dsn or "localhost:5432/test" in dsn:
+            return None
+        engine = create_engine(dsn, future=True)
+        conn = engine.connect()
+        conn.execute(text("SELECT 1"))
+        conn.close()
+        return sessionmaker(bind=engine, future=True)()
+    except Exception:
+        return None
+
+
+# list, не tuple: psycopg3 адаптирует в массив только list (tuple уходит как composite).
+_HOUSE_IDS = [902771, 902772, 902773, 902774, 902775]
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_transfers_agreeing_coords_skips_scattered_and_is_idempotent() -> None:
+    """Пять домов, пять исходов:
+
+    902771 — 3 объявления в ~30 м  → координаты перенесены (центроид)
+    902772 — 2 объявления в 11 км  → НЕ тронут, посчитан как scattered
+    902773 — объявления без geom   → НЕ тронут, посчитан отдельно
+    902774 — объявлений нет вовсе  → НЕ тронут, посчитан отдельно
+    902775 — координаты уже есть   → не кандидат, значение не перезаписано
+    """
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    try:
+        db.execute(_t("DELETE FROM listings WHERE house_id_fk = ANY(:ids)"), {"ids": _HOUSE_IDS})
+        db.execute(_t("DELETE FROM houses WHERE id = ANY(:ids)"), {"ids": _HOUSE_IDS})
+        # url NOT NULL (002_core_tables); 902775 приходит с уже проставленной координатой.
+        db.execute(
+            _t(
+                "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon) VALUES "
+                "(902771,'avito','E-2771-A','u','тестдом 2771 A', NULL, NULL),"
+                "(902772,'avito','E-2771-B','u','тестдом 2771 B', NULL, NULL),"
+                "(902773,'avito','E-2771-C','u','тестдом 2771 C', NULL, NULL),"
+                "(902774,'avito','E-2771-D','u','тестдом 2771 D', NULL, NULL),"
+                "(902775,'avito','E-2771-E','u','тестдом 2771 E', 56.80000, 60.50000)"
+            )
+        )
+        # listings: source_url/dedup_hash/price_rub NOT NULL, dedup_hash UNIQUE.
+        # 0.0002° по широте ≈ 22 м; 0.1° по широте ≈ 11.1 км (разброс = радиус от
+        # центроида, то есть ≈ 5.6 км — половина расстояния между точками).
+        db.execute(
+            _t("""
+                INSERT INTO listings
+                  (id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk, lat, lon,
+                   geom)
+                VALUES
+                  (912771,'avito','http://t/2771/1','L1','dh-2771-1',5000000,902771,
+                   56.840000,60.600000, ST_SetSRID(ST_MakePoint(60.600000,56.840000),4326)),
+                  (912772,'avito','http://t/2771/2','L2','dh-2771-2',5000000,902771,
+                   56.840200,60.600000, ST_SetSRID(ST_MakePoint(60.600000,56.840200),4326)),
+                  (912773,'avito','http://t/2771/3','L3','dh-2771-3',5000000,902771,
+                   56.840100,60.600300, ST_SetSRID(ST_MakePoint(60.600300,56.840100),4326)),
+                  (912774,'avito','http://t/2771/4','L4','dh-2771-4',5000000,902772,
+                   56.840000,60.600000, ST_SetSRID(ST_MakePoint(60.600000,56.840000),4326)),
+                  (912775,'avito','http://t/2771/5','L5','dh-2771-5',5000000,902772,
+                   56.940000,60.600000, ST_SetSRID(ST_MakePoint(60.600000,56.940000),4326)),
+                  (912776,'avito','http://t/2771/6','L6','dh-2771-6',5000000,902773,
+                   NULL,NULL,NULL)
+            """)
+        )
+        db.commit()
+
+        res = mod.backfill_house_coords(db, max_spread_m=mod.DEFAULT_MAX_SPREAD_M)
+
+        rows = dict(
+            db.execute(
+                _t("SELECT id, lat FROM houses WHERE id = ANY(:ids)"), {"ids": _HOUSE_IDS}
+            ).all()
+        )
+        # A: перенесено, и это ЦЕНТРОИД трёх точек, а не первая попавшаяся.
+        assert rows[902771] is not None
+        assert rows[902771] == pytest.approx(56.8401, abs=1e-4)
+        # geom построен (триггер houses_set_geom_trg + явный ST_SetSRID в UPDATE).
+        geom_ok = db.execute(_t("SELECT ST_Y(geom) FROM houses WHERE id = 902771")).scalar()
+        assert geom_ok == pytest.approx(56.8401, abs=1e-4)
+        # B: объявления в 11 км друг от друга — дом НЕ тронут (главная проверка задачи).
+        assert rows[902772] is None
+        # C/D: пропущены по своим причинам.
+        assert rows[902773] is None and rows[902774] is None
+        # E: чужая уже проставленная координата не перезаписана.
+        assert rows[902775] == pytest.approx(56.80000, abs=1e-6)
+
+        # Счётчики: таблица общая с другими тестами, поэтому проверяем вклад «не меньше».
+        assert res.updated >= 1
+        assert res.skipped_scattered >= 1
+        assert res.skipped_no_listings >= 1
+        assert res.skipped_no_listing_coords >= 1
+        assert res.worst_spread_m >= 5000
+        assert res.scattered_over_5km >= 1
+
+        # Второй прогон — no-op для наших домов (идемпотентность).
+        before = res.updated
+        res2 = mod.backfill_house_coords(db, max_spread_m=mod.DEFAULT_MAX_SPREAD_M)
+        assert res2.updated < before or res2.updated == 0
+        again = dict(
+            db.execute(
+                _t("SELECT id, lat FROM houses WHERE id = ANY(:ids)"), {"ids": _HOUSE_IDS}
+            ).all()
+        )
+        assert again[902771] == rows[902771]
+        assert again[902772] is None
+    finally:
+        db.rollback()
+        db.execute(_t("DELETE FROM listings WHERE house_id_fk = ANY(:ids)"), {"ids": _HOUSE_IDS})
+        db.execute(_t("DELETE FROM houses WHERE id = ANY(:ids)"), {"ids": _HOUSE_IDS})
+        db.commit()
+        db.close()

tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py

@@ -71,6 +71,7 @@ _PRODUCT_SOURCES: set[str] = {
     "newbuilding_enrich",
     "yandex_newbuilding_sweep",
     "geoportal_coords_backfill",
+    "house_coords_from_listings",
     "geocode_missing_listings",
     "avito_detail_backfill",
     "yandex_detail_backfill",