Merge pull request 'feat(tradein/rosreestr): импорт ДКП по Москве (77) — canonical_city, raw_payload, wildcard-расписание rosreestr_dkp_import_*, per-source чекпоинт (#3051)' (#3422) from feat/3051-rosreestr-import-region-param into main
All checks were successful
Deploy Trade-In / changes (push) Successful in 13s
Deploy Trade-In / build-browser (push) Successful in 39s
Deploy Trade-In / build-frontend (push) Successful in 2m13s
Deploy Trade-In / test (push) Successful in 4m22s
Deploy Trade-In / build-backend (push) Successful in 1m1s
Deploy Trade-In / deploy (push) Successful in 1m23s
Deploy Trade-In / deploy-status (push) Successful in 1s
Deploy Trade-In / perimeter-smoke (push) Successful in 12s
All checks were successful
Deploy Trade-In / changes (push) Successful in 13s
Deploy Trade-In / build-browser (push) Successful in 39s
Deploy Trade-In / build-frontend (push) Successful in 2m13s
Deploy Trade-In / test (push) Successful in 4m22s
Deploy Trade-In / build-backend (push) Successful in 1m1s
Deploy Trade-In / deploy (push) Successful in 1m23s
Deploy Trade-In / deploy-status (push) Successful in 1s
Deploy Trade-In / perimeter-smoke (push) Successful in 12s
This commit is contained in:
commit
e2045582ab
7 changed files with 228 additions and 20 deletions
|
|
@ -750,6 +750,13 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
|
||||||
post_claim=reschedule_after_minutes(param="interval_minutes", default=360),
|
post_claim=reschedule_after_minutes(param="interval_minutes", default=360),
|
||||||
),
|
),
|
||||||
"rosreestr_dkp_import": Handler(_job_rosreestr_dkp, "rosreestr_dkp_import"),
|
"rosreestr_dkp_import": Handler(_job_rosreestr_dkp, "rosreestr_dkp_import"),
|
||||||
|
# Wildcard (#3051 п.3): rosreestr_dkp_import_77 (Москва, миграция 288) и любой
|
||||||
|
# будущий region_code-суффикс из той же семьи резолвятся сюда через
|
||||||
|
# resolve_handler по префиксу (тот же механизм, что deactivate_stale_* /
|
||||||
|
# avito_city_sweep_* — см. scraper_kit.orchestration.scheduler.resolve_handler).
|
||||||
|
# region_code берётся из default_params строки расписания (import_rosreestr_dkp
|
||||||
|
# сам валидирует его через app.services.regions.REGIONS), Handler-тело общее.
|
||||||
|
"rosreestr_dkp_import_*": Handler(_job_rosreestr_dkp, "rosreestr_dkp_import_*"),
|
||||||
"listing_source_snapshot": Handler(_job_listing_source_snapshot, "listing_source_snapshot"),
|
"listing_source_snapshot": Handler(_job_listing_source_snapshot, "listing_source_snapshot"),
|
||||||
"asking_to_sold_ratio_refresh": Handler(
|
"asking_to_sold_ratio_refresh": Handler(
|
||||||
_job_asking_to_sold_ratio, "asking_to_sold_ratio_refresh"
|
_job_asking_to_sold_ratio, "asking_to_sold_ratio_refresh"
|
||||||
|
|
|
||||||
|
|
@ -47,6 +47,15 @@ class Region:
|
||||||
Регион без тира должен деградировать ЯВНО (потребитель
|
Регион без тира должен деградировать ЯВНО (потребитель
|
||||||
спрашивает unsupported_tier_reason и логирует/маркирует),
|
спрашивает unsupported_tier_reason и логирует/маркирует),
|
||||||
а не молча считать дальше без источника.
|
а не молча считать дальше без источника.
|
||||||
|
canonical_city — #3051: имя города, которым ПЕРЕЗАПИСЫВАЕТСЯ `city`
|
||||||
|
строк, приходящих из источника без надёжного city-поля
|
||||||
|
(Росреестр по Москве отдаёт муниципальный округ/поселение
|
||||||
|
вместо города — «Раменки», «Сосенское» — а не «Москва»).
|
||||||
|
None — источник несёт свой city как есть, без override
|
||||||
|
(регион 66: byte-for-byte прежнее поведение). Not-None —
|
||||||
|
потребитель (import_rosreestr_dkp) подставляет это имя
|
||||||
|
вместо city источника и НЕ фильтрует по city IS NOT NULL
|
||||||
|
(иначе на 77 теряется ~10% строк с пустым city).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
code: int
|
code: int
|
||||||
|
|
@ -58,6 +67,7 @@ class Region:
|
||||||
city_token: str
|
city_token: str
|
||||||
cities: frozenset[str]
|
cities: frozenset[str]
|
||||||
enrichment_tiers: frozenset[str]
|
enrichment_tiers: frozenset[str]
|
||||||
|
canonical_city: str | None = None
|
||||||
|
|
||||||
|
|
||||||
def is_within_bbox(lat: float, lon: float, bbox: BBox) -> bool:
|
def is_within_bbox(lat: float, lon: float, bbox: BBox) -> bool:
|
||||||
|
|
@ -144,6 +154,10 @@ REGIONS: dict[int, Region] = {
|
||||||
# sber_index покрывают регион 66. Пустое множество здесь — не заглушка,
|
# sber_index покрывают регион 66. Пустое множество здесь — не заглушка,
|
||||||
# а ФАКТ, который потребители обязаны озвучивать (см. класс-докстринг).
|
# а ФАКТ, который потребители обязаны озвучивать (см. класс-докстринг).
|
||||||
enrichment_tiers=frozenset(),
|
enrichment_tiers=frozenset(),
|
||||||
|
# #3051: Росреестр по Москве отдаёт в city муниципальный округ/поселение
|
||||||
|
# ("муниципальный округ Раменки", "поселение Сосенское"), не сам город —
|
||||||
|
# import_rosreestr_dkp подставляет каноничное имя вместо city источника.
|
||||||
|
canonical_city="Москва",
|
||||||
),
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -25,6 +25,7 @@ Zombie-reap, advisory-lock claim и tick-loop теперь целиком в
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
import logging
|
import logging
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
|
@ -50,6 +51,7 @@ from sqlalchemy.orm import Session
|
||||||
|
|
||||||
from app.core.shutdown import shutdown_requested
|
from app.core.shutdown import shutdown_requested
|
||||||
from app.services import scrape_runs as runs_mod
|
from app.services import scrape_runs as runs_mod
|
||||||
|
from app.services.regions import REGIONS
|
||||||
|
|
||||||
__all__ = ["compute_next_run_at", "has_running_run"]
|
__all__ = ["compute_next_run_at", "has_running_run"]
|
||||||
|
|
||||||
|
|
@ -238,6 +240,24 @@ async def _execute_cian_backfill(
|
||||||
|
|
||||||
_DKP_SOURCE = "rosreestr_dkp_import"
|
_DKP_SOURCE = "rosreestr_dkp_import"
|
||||||
|
|
||||||
|
|
||||||
|
def _dkp_source_for_region(region_code: int) -> str:
|
||||||
|
"""Имя scrape_runs.source для чекпоинта данного региона (#3051 п.3).
|
||||||
|
|
||||||
|
66 — байт-в-байт прежнее имя ('rosreestr_dkp_import'), под которым годами
|
||||||
|
писались scrape_runs. Остальные регионы получают суффикс кода — тот же
|
||||||
|
формат, что и у строки scrape_schedules ('rosreestr_dkp_import_77',
|
||||||
|
seed — миграция 289), которую резолвит wildcard 'rosreestr_dkp_import_*'
|
||||||
|
в product_handlers.py. Изоляция чекпоинтов между регионами держится именно
|
||||||
|
на разных source: _resume_dkp_cursor ищет ПРЕДЫДУЩИЙ прогон с ТЕМ ЖЕ source,
|
||||||
|
поэтому курсор региона 77 никогда не подхватит last_id региона 66 (и
|
||||||
|
наоборот) — они просто разные строки в scrape_runs.source.
|
||||||
|
"""
|
||||||
|
if region_code == 66:
|
||||||
|
return _DKP_SOURCE
|
||||||
|
return f"{_DKP_SOURCE}_{region_code}"
|
||||||
|
|
||||||
|
|
||||||
# Потолок возраста чекпоинта: старше — last_id прошлого прогона не подхватываем, прогон
|
# Потолок возраста чекпоинта: старше — last_id прошлого прогона не подхватываем, прогон
|
||||||
# стартует с id=0 (issue #3168). У предиката `id > last_id` нет протухания в смысле
|
# стартует с id=0 (issue #3168). У предиката `id > last_id` нет протухания в смысле
|
||||||
# свипов (он остаётся корректным сколь угодно долго), но апстрим
|
# свипов (он остаётся корректным сколь угодно долго), но апстрим
|
||||||
|
|
@ -263,7 +283,9 @@ _DKP_RESUME_CANDIDATE_SQL = text("""
|
||||||
""")
|
""")
|
||||||
|
|
||||||
|
|
||||||
def _resume_dkp_cursor(db: Session, run_id: int) -> tuple[int, dict[str, Any]]:
|
def _resume_dkp_cursor(
|
||||||
|
db: Session, run_id: int, source: str = _DKP_SOURCE
|
||||||
|
) -> tuple[int, dict[str, Any]]:
|
||||||
"""Продолжить last_id прошлого прогона или начать с 0 — решение + explain (issue #3168).
|
"""Продолжить last_id прошлого прогона или начать с 0 — решение + explain (issue #3168).
|
||||||
|
|
||||||
last_id раньше жил только в памяти процесса (init на 0 при каждом запуске): heartbeat
|
last_id раньше жил только в памяти процесса (init на 0 при каждом запуске): heartbeat
|
||||||
|
|
@ -271,7 +293,14 @@ def _resume_dkp_cursor(db: Session, run_id: int) -> tuple[int, dict[str, Any]]:
|
||||||
обратно. Обрыв (деплой/OOM/рестарт хоста) откатывал прогресс на 0 и заставлял
|
обратно. Обрыв (деплой/OOM/рестарт хоста) откатывал прогресс на 0 и заставлял
|
||||||
пере-сканировать источник с начала.
|
пере-сканировать источник с начала.
|
||||||
|
|
||||||
Кандидат — ПОСЛЕДНИЙ прогон этого source (тот же принцип, что и
|
`source` (#3051 п.3) — per-region ключ чекпоинта (см. _dkp_source_for_region):
|
||||||
|
дефолт _DKP_SOURCE сохраняет прежнее поведение вызовов без явного аргумента
|
||||||
|
(регион 66). Кандидат ищется СТРОГО по этому source — прогон региона 77
|
||||||
|
(source='rosreestr_dkp_import_77') никогда не видит last_id региона 66
|
||||||
|
(source='rosreestr_dkp_import') и наоборот: разные регионы физически не
|
||||||
|
матчат друг друга в WHERE source = :source ниже.
|
||||||
|
|
||||||
|
Кандидат — ПОСЛЕДНИЙ прогон ЭТОГО source (тот же принцип, что и
|
||||||
scraper_kit.orchestration.scheduler._pick_resume, локальная копия ладдера — контракт
|
scraper_kit.orchestration.scheduler._pick_resume, локальная копия ладдера — контракт
|
||||||
другой: нет params/interval_days, курсор числовой, а не bucket-set):
|
другой: нет params/interval_days, курсор числовой, а не bucket-set):
|
||||||
- 'running' / 'zombie' — прогон, которого не завершили штатно.
|
- 'running' / 'zombie' — прогон, которого не завершили штатно.
|
||||||
|
|
@ -285,7 +314,7 @@ def _resume_dkp_cursor(db: Session, run_id: int) -> tuple[int, dict[str, Any]]:
|
||||||
кодом (kit_runs.update_heartbeat — merge, не замена), чтобы решение было видно в
|
кодом (kit_runs.update_heartbeat — merge, не замена), чтобы решение было видно в
|
||||||
scrape_runs, а не только в логе.
|
scrape_runs, а не только в логе.
|
||||||
"""
|
"""
|
||||||
row = db.execute(_DKP_RESUME_CANDIDATE_SQL, {"source": _DKP_SOURCE, "rid": run_id}).fetchone()
|
row = db.execute(_DKP_RESUME_CANDIDATE_SQL, {"source": source, "rid": run_id}).fetchone()
|
||||||
verdict: dict[str, Any] = {"resume_from": None}
|
verdict: dict[str, Any] = {"resume_from": None}
|
||||||
|
|
||||||
if row is None:
|
if row is None:
|
||||||
|
|
@ -322,18 +351,33 @@ def import_rosreestr_dkp(
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Import ДКП-сделок из gendesign rosreestr_deals через postgres_fdw.
|
"""Import ДКП-сделок из gendesign rosreestr_deals через postgres_fdw.
|
||||||
|
|
||||||
Python-порт import-rosreestr.sh (Variant C из #563).
|
Python-порт import-rosreestr.sh (Variant C из #563). #3051 п.3: параметризовано
|
||||||
|
по региону (params["region_code"], реестр — app.services.regions.REGIONS) —
|
||||||
|
было хардкод region_code=66.
|
||||||
|
|
||||||
Источник: foreign table gendesign_rosreestr_deals (создана в migration 072).
|
Источник: foreign table gendesign_rosreestr_deals (создана в migration 072,
|
||||||
|
okato/quarter_cad_number/district добавлены миграцией 289).
|
||||||
SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql.
|
SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql.
|
||||||
USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader).
|
USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader).
|
||||||
|
|
||||||
Область покрытия задаётся параметром, а не литералом (#3051 п.6): region_code
|
Область покрытия задаётся параметром, а не литералом (#3051 п.6): region_code
|
||||||
приходит из params, дефолт 66 = вся Свердловская область (не только Екатеринбург —
|
приходит из params, дефолт 66 = вся Свердловская область (не только Екатеринбург —
|
||||||
прежний ILIKE-фильтр по подстроке города снят, unlocks +47183 сделок вне ЕКБ уже
|
прежний ILIKE-фильтр по подстроке города снят, unlocks +47183 сделок вне ЕКБ уже
|
||||||
сидящих в source foreign table). address и deals.city строятся из реального city
|
сидящих в source foreign table). Неизвестный код региона (нет в REGIONS) — ValueError,
|
||||||
источника (не хардкод "Екатеринбург"), deals.region_code заполняется из строки
|
прогон падает явно, а не молча импортирует мусор с чужим region_code.
|
||||||
источника (= запрошенному региону при текущем фильтре).
|
|
||||||
|
region_code=66 (регион БЕЗ canonical_city в реестре) — поведение байт-в-байт
|
||||||
|
прежнее: city/address строятся из city источника, обязателен фильтр
|
||||||
|
city IS NOT NULL AND trim(city) != ''.
|
||||||
|
|
||||||
|
Регион С canonical_city (77 — Москва): Росреестр отдаёт в city муниципальный
|
||||||
|
округ/поселение ("муниципальный округ Раменки", "поселение Сосенское"), НЕ
|
||||||
|
город — city/address подставляют region.canonical_city, а не city источника;
|
||||||
|
фильтр city IS NOT NULL НЕ применяется (иначе теряется ~10% строк с пустым
|
||||||
|
city источника). Исходные city/okato/quarter_cad_number/district уходят в
|
||||||
|
raw_payload (jsonb) — единственная ветка SQL решает это через bind-параметр
|
||||||
|
:canonical_city (CASE WHEN ... IS NOT NULL), а не отдельный Python if/else на
|
||||||
|
конкретный код региона.
|
||||||
|
|
||||||
Типы документов тоже параметр (#3051 п.3): doc_types, дефолт ['ДКП'] = прежнее
|
Типы документов тоже параметр (#3051 п.3): doc_types, дефолт ['ДКП'] = прежнее
|
||||||
поведение (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24). Для Москвы
|
поведение (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24). Для Москвы
|
||||||
|
|
@ -342,8 +386,8 @@ def import_rosreestr_dkp(
|
||||||
заполняется на импорте.
|
заполняется на импорте.
|
||||||
|
|
||||||
Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
|
Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
|
||||||
- region_code = :region_code (param, default 66 — вся Свердловская область)
|
- region_code = :region_code (параметризовано, было хардкод 66)
|
||||||
- city IS NOT NULL AND trim(city) != '' (непустой город → корректный address)
|
- city IS NOT NULL AND trim(city) != '' — ТОЛЬКО если у региона нет canonical_city
|
||||||
- realestate_type_code = '002001003000' (квартира)
|
- realestate_type_code = '002001003000' (квартира)
|
||||||
- area BETWEEN 18 AND 200
|
- area BETWEEN 18 AND 200
|
||||||
- deal_price BETWEEN 1000000 AND 100000000
|
- deal_price BETWEEN 1000000 AND 100000000
|
||||||
|
|
@ -364,13 +408,18 @@ def import_rosreestr_dkp(
|
||||||
(WHERE id > last_id ORDER BY id). Heartbeat обновляется каждый батч (= checkpoint),
|
(WHERE id > last_id ORDER BY id). Heartbeat обновляется каждый батч (= checkpoint),
|
||||||
мержем (kit_runs.update_heartbeat), а не заменой. На старте _resume_dkp_cursor решает
|
мержем (kit_runs.update_heartbeat), а не заменой. На старте _resume_dkp_cursor решает
|
||||||
продолжить с last_id прошлого прогона или начать с 0 — чекпоинт переживает рестарт
|
продолжить с last_id прошлого прогона или начать с 0 — чекпоинт переживает рестарт
|
||||||
процесса (деплой/OOM/SIGTERM), пока не старше суток (issue #3168).
|
процесса (деплой/OOM/SIGTERM), пока не старше суток (issue #3168). Курсор — ПЕР
|
||||||
|
РЕГИОН (#3051 п.3): _resume_dkp_cursor вызывается с source=_dkp_source_for_region
|
||||||
|
(region_code), поэтому last_id региона 77 никогда не подхватывает last_id региона
|
||||||
|
66 — они разные scrape_runs.source ('rosreestr_dkp_import' vs
|
||||||
|
'rosreestr_dkp_import_77'), см. докстринг _dkp_source_for_region.
|
||||||
SAVEPOINT per row — один сбойный row не откатывает батч.
|
SAVEPOINT per row — один сбойный row не откатывает батч.
|
||||||
|
|
||||||
Координаты: NULL после импорта — геокодинг остаётся follow-up (geocode-deals).
|
Координаты: NULL после импорта — геокодинг остаётся follow-up (geocode-deals).
|
||||||
TODO (follow-up): запустить geocode backfill после import.
|
TODO (follow-up): запустить geocode backfill после import.
|
||||||
|
|
||||||
Cleanup: удаляет legacy строки address='Екатеринбург, реальная сделка' (pre-#549).
|
Cleanup: удаляет legacy строки address='Екатеринбург, реальная сделка' (pre-#549,
|
||||||
|
region-agnostic — синтетические строки существовали только для ЕКБ).
|
||||||
"""
|
"""
|
||||||
since: str = str(params.get("since", "2024-01-01"))
|
since: str = str(params.get("since", "2024-01-01"))
|
||||||
batch_size: int = int(params.get("batch_size", 2000))
|
batch_size: int = int(params.get("batch_size", 2000))
|
||||||
|
|
@ -379,6 +428,15 @@ def import_rosreestr_dkp(
|
||||||
region_code: int = int(params.get("region_code", 66))
|
region_code: int = int(params.get("region_code", 66))
|
||||||
# #3051 п.3: типы документов — параметр, дефолт ['ДКП'] = прежний литерал.
|
# #3051 п.3: типы документов — параметр, дефолт ['ДКП'] = прежний литерал.
|
||||||
doc_types: list[str] = [str(t) for t in params.get("doc_types") or ["ДКП"]]
|
doc_types: list[str] = [str(t) for t in params.get("doc_types") or ["ДКП"]]
|
||||||
|
region = REGIONS.get(region_code)
|
||||||
|
if region is None:
|
||||||
|
raise ValueError(
|
||||||
|
f"rosreestr_dkp_import: region_code={region_code} не найден в "
|
||||||
|
f"app.services.regions.REGIONS (известны: {sorted(REGIONS)}) — "
|
||||||
|
"прогон остановлен, чтобы не импортировать сделки с неизвестным "
|
||||||
|
"региональным контекстом (city/address-правила для него не определены)"
|
||||||
|
)
|
||||||
|
dkp_source = _dkp_source_for_region(region_code)
|
||||||
|
|
||||||
counters: dict[str, int] = {
|
counters: dict[str, int] = {
|
||||||
"rows_fetched": 0,
|
"rows_fetched": 0,
|
||||||
|
|
@ -415,7 +473,7 @@ def import_rosreestr_dkp(
|
||||||
)
|
)
|
||||||
db.rollback()
|
db.rollback()
|
||||||
|
|
||||||
last_id, resume_verdict = _resume_dkp_cursor(db, run_id)
|
last_id, resume_verdict = _resume_dkp_cursor(db, run_id, source=dkp_source)
|
||||||
total_batches = 0
|
total_batches = 0
|
||||||
kit_runs.update_heartbeat(db, run_id, resume_verdict)
|
kit_runs.update_heartbeat(db, run_id, resume_verdict)
|
||||||
logger.info(
|
logger.info(
|
||||||
|
|
@ -465,9 +523,20 @@ def import_rosreestr_dkp(
|
||||||
id,
|
id,
|
||||||
id AS source_id_src,
|
id AS source_id_src,
|
||||||
'ros:dkp:' || CAST(id AS text) AS dedup_hash,
|
'ros:dkp:' || CAST(id AS text) AS dedup_hash,
|
||||||
trim(city) || ', ' || trim(street) AS address,
|
-- #3051: регион с canonical_city (Москва) подставляет его вместо
|
||||||
|
-- city источника (округ/поселение, не город) — CASE на bind-параметре,
|
||||||
|
-- не Python if/else на код региона.
|
||||||
|
CASE
|
||||||
|
WHEN CAST(:canonical_city AS text) IS NOT NULL
|
||||||
|
THEN CAST(:canonical_city AS text) || ', ' || trim(street)
|
||||||
|
ELSE trim(city) || ', ' || trim(street)
|
||||||
|
END AS address,
|
||||||
region_code,
|
region_code,
|
||||||
trim(city) AS city,
|
CASE
|
||||||
|
WHEN CAST(:canonical_city AS text) IS NOT NULL
|
||||||
|
THEN CAST(:canonical_city AS text)
|
||||||
|
ELSE trim(city)
|
||||||
|
END AS city,
|
||||||
CASE
|
CASE
|
||||||
WHEN area < 30 THEN 0
|
WHEN area < 30 THEN 0
|
||||||
WHEN area < 44 THEN 1
|
WHEN area < 44 THEN 1
|
||||||
|
|
@ -488,10 +557,26 @@ def import_rosreestr_dkp(
|
||||||
round(deal_price)::bigint AS price_rub,
|
round(deal_price)::bigint AS price_rub,
|
||||||
round(price_per_sqm)::int AS price_per_m2,
|
round(price_per_sqm)::int AS price_per_m2,
|
||||||
period_start_date AS deal_date,
|
period_start_date AS deal_date,
|
||||||
doc_type AS doc_type
|
doc_type,
|
||||||
|
-- Исходный city/okato/quarter_cad_number/district — ТОЛЬКО когда
|
||||||
|
-- city перезаписан canonical_city выше (иначе NULL, регион 66
|
||||||
|
-- byte-for-byte прежний: raw_payload не заполнялся и не заполняется).
|
||||||
|
CASE
|
||||||
|
WHEN CAST(:canonical_city AS text) IS NOT NULL THEN
|
||||||
|
jsonb_build_object(
|
||||||
|
'src_city', city,
|
||||||
|
'okato', okato,
|
||||||
|
'quarter_cad_number', quarter_cad_number,
|
||||||
|
'district', district
|
||||||
|
)
|
||||||
|
ELSE NULL
|
||||||
|
END AS raw_payload
|
||||||
FROM gendesign_rosreestr_deals
|
FROM gendesign_rosreestr_deals
|
||||||
WHERE region_code = CAST(:region_code AS int)
|
WHERE region_code = CAST(:region_code AS int)
|
||||||
AND city IS NOT NULL AND trim(city) <> ''
|
AND (
|
||||||
|
CAST(:canonical_city AS text) IS NOT NULL
|
||||||
|
OR (city IS NOT NULL AND trim(city) <> '')
|
||||||
|
)
|
||||||
AND realestate_type_code = '002001003000'
|
AND realestate_type_code = '002001003000'
|
||||||
AND area BETWEEN 18 AND 200
|
AND area BETWEEN 18 AND 200
|
||||||
AND deal_price BETWEEN 1000000 AND 100000000
|
AND deal_price BETWEEN 1000000 AND 100000000
|
||||||
|
|
@ -507,6 +592,7 @@ def import_rosreestr_dkp(
|
||||||
"last_id": last_id,
|
"last_id": last_id,
|
||||||
"batch_size": batch_size,
|
"batch_size": batch_size,
|
||||||
"region_code": region_code,
|
"region_code": region_code,
|
||||||
|
"canonical_city": region.canonical_city,
|
||||||
"doc_types": doc_types,
|
"doc_types": doc_types,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
@ -546,7 +632,7 @@ def import_rosreestr_dkp(
|
||||||
INSERT INTO deals (
|
INSERT INTO deals (
|
||||||
source, dedup_hash, source_id, address, region_code, city,
|
source, dedup_hash, source_id, address, region_code, city,
|
||||||
rooms, area_m2, floor, year_built, price_rub, price_per_m2,
|
rooms, area_m2, floor, year_built, price_rub, price_per_m2,
|
||||||
deal_date, doc_type
|
deal_date, doc_type, raw_payload
|
||||||
)
|
)
|
||||||
VALUES (
|
VALUES (
|
||||||
'rosreestr',
|
'rosreestr',
|
||||||
|
|
@ -562,7 +648,8 @@ def import_rosreestr_dkp(
|
||||||
CAST(:price_rub AS bigint),
|
CAST(:price_rub AS bigint),
|
||||||
CAST(:price_per_m2 AS int),
|
CAST(:price_per_m2 AS int),
|
||||||
CAST(:deal_date AS date),
|
CAST(:deal_date AS date),
|
||||||
CAST(:doc_type AS text)
|
CAST(:doc_type AS text),
|
||||||
|
CAST(:raw_payload AS jsonb)
|
||||||
)
|
)
|
||||||
ON CONFLICT (dedup_hash) DO UPDATE SET
|
ON CONFLICT (dedup_hash) DO UPDATE SET
|
||||||
address = EXCLUDED.address,
|
address = EXCLUDED.address,
|
||||||
|
|
@ -575,7 +662,8 @@ def import_rosreestr_dkp(
|
||||||
price_rub = EXCLUDED.price_rub,
|
price_rub = EXCLUDED.price_rub,
|
||||||
price_per_m2 = EXCLUDED.price_per_m2,
|
price_per_m2 = EXCLUDED.price_per_m2,
|
||||||
deal_date = EXCLUDED.deal_date,
|
deal_date = EXCLUDED.deal_date,
|
||||||
doc_type = EXCLUDED.doc_type
|
doc_type = EXCLUDED.doc_type,
|
||||||
|
raw_payload = EXCLUDED.raw_payload
|
||||||
WHERE deals.address IS DISTINCT FROM EXCLUDED.address
|
WHERE deals.address IS DISTINCT FROM EXCLUDED.address
|
||||||
OR deals.region_code IS DISTINCT FROM EXCLUDED.region_code
|
OR deals.region_code IS DISTINCT FROM EXCLUDED.region_code
|
||||||
OR deals.city IS DISTINCT FROM EXCLUDED.city
|
OR deals.city IS DISTINCT FROM EXCLUDED.city
|
||||||
|
|
@ -587,6 +675,7 @@ def import_rosreestr_dkp(
|
||||||
OR deals.price_per_m2 IS DISTINCT FROM EXCLUDED.price_per_m2
|
OR deals.price_per_m2 IS DISTINCT FROM EXCLUDED.price_per_m2
|
||||||
OR deals.deal_date IS DISTINCT FROM EXCLUDED.deal_date
|
OR deals.deal_date IS DISTINCT FROM EXCLUDED.deal_date
|
||||||
OR deals.doc_type IS DISTINCT FROM EXCLUDED.doc_type
|
OR deals.doc_type IS DISTINCT FROM EXCLUDED.doc_type
|
||||||
|
OR deals.raw_payload IS DISTINCT FROM EXCLUDED.raw_payload
|
||||||
RETURNING (xmax = 0) AS was_inserted
|
RETURNING (xmax = 0) AS was_inserted
|
||||||
"""),
|
"""),
|
||||||
{
|
{
|
||||||
|
|
@ -603,6 +692,11 @@ def import_rosreestr_dkp(
|
||||||
"price_per_m2": row["price_per_m2"],
|
"price_per_m2": row["price_per_m2"],
|
||||||
"deal_date": row["deal_date"],
|
"deal_date": row["deal_date"],
|
||||||
"doc_type": row["doc_type"],
|
"doc_type": row["doc_type"],
|
||||||
|
"raw_payload": (
|
||||||
|
json.dumps(row["raw_payload"], ensure_ascii=False)
|
||||||
|
if row["raw_payload"] is not None
|
||||||
|
else None
|
||||||
|
),
|
||||||
},
|
},
|
||||||
).fetchone()
|
).fetchone()
|
||||||
if result is None:
|
if result is None:
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,60 @@
|
||||||
|
-- 289_rosreestr_fdw_msk_columns_seed77.sql
|
||||||
|
-- gendesign_rosreestr_deals: колонки okato/quarter_cad_number/district + disabled
|
||||||
|
-- seed-строка scrape_schedules для региона 77 (Москва) — #3051 п.3.
|
||||||
|
--
|
||||||
|
-- Dependencies: 072_scrape_schedules_seed_cian_rosreestr.sql (gendesign_rosreestr_deals,
|
||||||
|
-- scrape_schedules).
|
||||||
|
-- Apply after: 288_deals_doc_type.sql
|
||||||
|
--
|
||||||
|
-- WHY:
|
||||||
|
-- Трек 2 подготовки Mera к Москве — импорт сделок Росреестра параметризуется по
|
||||||
|
-- региону (66 Свердловская обл. / 77 Москва, code-часть в scheduler.py). Для
|
||||||
|
-- региона С canonical_city (Москва) исходные city/okato/quarter_cad_number/district
|
||||||
|
-- уходят в deals.raw_payload (jsonb), т.к. city источника там — муниципальный
|
||||||
|
-- округ/поселение, не город, и обычный city/address его не покрывают. Foreign
|
||||||
|
-- table расширена тремя колонками источника; существование их в
|
||||||
|
-- public.rosreestr_deals на gendesign-стороне проверено live (2026-09-08, prod psql).
|
||||||
|
--
|
||||||
|
-- Seed-строка rosreestr_dkp_import_77 — ВЫКЛЮЧЕНА (enabled=false): миграция
|
||||||
|
-- только заводит расписание, включение и первый прогон по Москве — отдельное
|
||||||
|
-- решение main-сессии после ревью кода-части.
|
||||||
|
--
|
||||||
|
-- ИДЕМПОТЕНТНОСТЬ: ADD COLUMN IF NOT EXISTS × 3, ON CONFLICT (source) DO NOTHING
|
||||||
|
-- для seed — повторный прогон no-op.
|
||||||
|
|
||||||
|
BEGIN;
|
||||||
|
|
||||||
|
SET LOCAL lock_timeout = '5s';
|
||||||
|
|
||||||
|
-- ── gendesign_rosreestr_deals: колонки под региональный raw_payload (#3051) ────
|
||||||
|
-- ALTER FOREIGN TABLE ADD COLUMN — только локальные метаданные (не трогает
|
||||||
|
-- реальную remote-таблицу), безопасно как ALTER TABLE ADD COLUMN без DEFAULT.
|
||||||
|
|
||||||
|
ALTER FOREIGN TABLE gendesign_rosreestr_deals
|
||||||
|
ADD COLUMN IF NOT EXISTS okato text;
|
||||||
|
|
||||||
|
ALTER FOREIGN TABLE gendesign_rosreestr_deals
|
||||||
|
ADD COLUMN IF NOT EXISTS quarter_cad_number text;
|
||||||
|
|
||||||
|
ALTER FOREIGN TABLE gendesign_rosreestr_deals
|
||||||
|
ADD COLUMN IF NOT EXISTS district text;
|
||||||
|
|
||||||
|
-- ── Seed: rosreestr_dkp_import_77 (выключено) ──────────────────────────────────
|
||||||
|
|
||||||
|
INSERT INTO scrape_schedules (
|
||||||
|
source,
|
||||||
|
enabled,
|
||||||
|
window_start_hour,
|
||||||
|
window_end_hour,
|
||||||
|
default_params
|
||||||
|
)
|
||||||
|
VALUES (
|
||||||
|
'rosreestr_dkp_import_77',
|
||||||
|
false,
|
||||||
|
4,
|
||||||
|
6,
|
||||||
|
'{"region_code": 77, "since": "2024-01-01", "batch_size": 2000}'::jsonb
|
||||||
|
)
|
||||||
|
ON CONFLICT (source) DO NOTHING;
|
||||||
|
|
||||||
|
COMMIT;
|
||||||
|
|
@ -161,6 +161,7 @@ def test_migration_077_shared_filter_matches_live_import() -> None:
|
||||||
набор строк.
|
набор строк.
|
||||||
"""
|
"""
|
||||||
sql = _MIGRATION_077.read_text("utf-8")
|
sql = _MIGRATION_077.read_text("utf-8")
|
||||||
|
assert "region_code = 66" in sql, "migration 077 must keep its historical literal"
|
||||||
for clause in (
|
for clause in (
|
||||||
"realestate_type_code = '002001003000'",
|
"realestate_type_code = '002001003000'",
|
||||||
"area BETWEEN 18 AND 200",
|
"area BETWEEN 18 AND 200",
|
||||||
|
|
@ -174,6 +175,18 @@ def test_migration_077_shared_filter_matches_live_import() -> None:
|
||||||
assert "doc_type = 'ДКП'" in sql
|
assert "doc_type = 'ДКП'" in sql
|
||||||
|
|
||||||
|
|
||||||
|
def test_live_import_region_code_is_bind_param() -> None:
|
||||||
|
"""#3051 п.3: живой импорт параметризован по региону, литерала 66 в SQL нет.
|
||||||
|
|
||||||
|
До #3051 живой импорт хардкодил `WHERE region_code = 66` — единственный регион
|
||||||
|
покрытия. С параметризацией region_code приходит из params (default 66 — обратная
|
||||||
|
совместимость), в SQL идёт bind-параметром через CAST, не литералом.
|
||||||
|
"""
|
||||||
|
assert "region_code = CAST(:region_code AS int)" in _IMPORT_SRC
|
||||||
|
assert "region_code = 66" not in _IMPORT_BODY
|
||||||
|
assert 'params.get("region_code", 66)' in _IMPORT_SRC
|
||||||
|
|
||||||
|
|
||||||
def test_live_import_dropped_ekb_city_filter() -> None:
|
def test_live_import_dropped_ekb_city_filter() -> None:
|
||||||
"""Живой импорт БОЛЬШЕ не фильтрует по городу — Mera расширена на всю обл. 66.
|
"""Живой импорт БОЛЬШЕ не фильтрует по городу — Mera расширена на всю обл. 66.
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -57,6 +57,10 @@ from scraper_kit.orchestration.scheduler import (
|
||||||
_PRODUCT_SOURCES: set[str] = {
|
_PRODUCT_SOURCES: set[str] = {
|
||||||
"cian_history_backfill",
|
"cian_history_backfill",
|
||||||
"rosreestr_dkp_import",
|
"rosreestr_dkp_import",
|
||||||
|
# #3051 п.3: member-source семейства "rosreestr_dkp_import_*" (Москва, миграция 288
|
||||||
|
# — то же раскрытие wildcard в конкретный member, что deactivate_stale_avito/yandex/
|
||||||
|
# cian ниже, а не сам wildcard "rosreestr_dkp_import_*").
|
||||||
|
"rosreestr_dkp_import_77",
|
||||||
"listing_source_snapshot",
|
"listing_source_snapshot",
|
||||||
"asking_to_sold_ratio_refresh",
|
"asking_to_sold_ratio_refresh",
|
||||||
"deal_city_price_bands_refresh",
|
"deal_city_price_bands_refresh",
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,16 @@
|
||||||
# (REGION_CODE=77) типы разделяются колонкой deals.doc_type (миграция 288), которую
|
# (REGION_CODE=77) типы разделяются колонкой deals.doc_type (миграция 288), которую
|
||||||
# скрипт теперь заполняет.
|
# скрипт теперь заполняет.
|
||||||
#
|
#
|
||||||
|
# #3051 п.3: REGION_CODE параметризован (default 66 — Свердловская обл., byte-for-byte
|
||||||
|
# прежнее поведение). Этот bash-путь — НЕ region-generic: для region_code=77 (Москва) он
|
||||||
|
# НЕ подставляет canonical_city вместо city источника (Росреестр по Москве отдаёт
|
||||||
|
# муниципальный округ/поселение, не сам город) и не пишет raw_payload с
|
||||||
|
# okato/quarter_cad_number/district — эту логику несёт только Python-путь
|
||||||
|
# (app/services/scheduler.py::import_rosreestr_dkp, боевой планировщик). Если этот
|
||||||
|
# скрипт когда-нибудь запустят вручную с REGION_CODE=77 — city/address будут
|
||||||
|
# "муниципальный округ Раменки, ..." как есть из источника, НЕ "Москва, ...". Держать
|
||||||
|
# паритет фильтров (area/price/doc_type) обязательно, паритет city-override — нет.
|
||||||
|
#
|
||||||
# Запуск на прод-хосте: ./import-rosreestr.sh
|
# Запуск на прод-хосте: ./import-rosreestr.sh
|
||||||
# REGION_CODE=77 DOC_TYPE='ДДУ' ./import-rosreestr.sh # Москва, первичка
|
# REGION_CODE=77 DOC_TYPE='ДДУ' ./import-rosreestr.sh # Москва, первичка
|
||||||
# Повторяемо: дедуп по dedup_hash, новый запуск подтянет свежие кварталы.
|
# Повторяемо: дедуп по dedup_hash, новый запуск подтянет свежие кварталы.
|
||||||
|
|
@ -27,6 +37,12 @@ SINCE="${SINCE:-2024-01-01}"
|
||||||
# #3051: регион и тип документа — параметры со старыми дефолтами (поведение не меняется).
|
# #3051: регион и тип документа — параметры со старыми дефолтами (поведение не меняется).
|
||||||
REGION_CODE="${REGION_CODE:-66}"
|
REGION_CODE="${REGION_CODE:-66}"
|
||||||
DOC_TYPE="${DOC_TYPE:-ДКП}"
|
DOC_TYPE="${DOC_TYPE:-ДКП}"
|
||||||
|
# Оба значения интерполируются в SQL текстом (не bind-параметром) — валидация здесь
|
||||||
|
# и есть единственная защита от SQL-инъекции. Паттерн — в переменной (не инлайн в
|
||||||
|
# [[ =~ ]]): голая кавычка в regex-операнде ломает bash-парсинг команды.
|
||||||
|
DOC_TYPE_RE="^[^;']+\$"
|
||||||
|
[[ "$REGION_CODE" =~ ^[0-9]+$ ]] || { echo "REGION_CODE должен быть целым числом, получено: '$REGION_CODE'" >&2; exit 1; }
|
||||||
|
[[ "$DOC_TYPE" =~ $DOC_TYPE_RE ]] || { echo "DOC_TYPE не должен содержать ' или ;, получено: '$DOC_TYPE'" >&2; exit 1; }
|
||||||
|
|
||||||
echo "[$(date -u +%H:%M:%S)] import-rosreestr: region=$REGION_CODE doc_type=$DOC_TYPE с $SINCE"
|
echo "[$(date -u +%H:%M:%S)] import-rosreestr: region=$REGION_CODE doc_type=$DOC_TYPE с $SINCE"
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue