fix(tradein/houses): вернуть координаты объявлений в дом, когда объявления согласны (#2771) #2780
6 changed files with 631 additions and 0 deletions
|
|
@ -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"
|
||||
),
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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;
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue