From cfe529fd1417498b7065b6e361a07ddfd8aee839 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 8 Sep 2026 22:45:55 +0300 Subject: [PATCH 1/5] =?UTF-8?q?feat(#3051):=20deals.doc=5Ftype=20+=20?= =?UTF-8?q?=D1=80=D0=B5=D0=B3=D0=B8=D0=BE=D0=BD=20=D0=B8=20=D1=82=D0=B8?= =?UTF-8?q?=D0=BF=20=D0=B4=D0=BE=D0=BA=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D0=B0?= =?UTF-8?q?=20=D0=BA=D0=B0=D0=BA=20=D0=BF=D0=B0=D1=80=D0=B0=D0=BC=D0=B5?= =?UTF-8?q?=D1=82=D1=80=D1=8B=20=D0=B8=D0=BC=D0=BF=D0=BE=D1=80=D1=82=D0=B0?= =?UTF-8?q?=20=D0=A0=D0=BE=D1=81=D1=80=D0=B5=D0=B5=D1=81=D1=82=D1=80=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ПОЧЕМУ: расширение на Москву упирается в два литерала. В источнике за 2024 по региону 77 лежат 30 627 ДДУ с медианой 112 743 против 107 005 ДКП с медианой 256 250 — это цены котлована, и без различимого признака в deals они развалят любую оценку. При этом тип сделки терялся при загрузке вовсе (в deals колонки не было), а фильтры region_code = 66 и doc_type = 'ДКП' стояли литералами в scheduler.import_rosreestr_dkp и в двойнике deploy/import-rosreestr.sh — сменить регион было нельзя, не правя код. ЧТО: - миграция 288: deals.doc_type text (idempotent) + бэкфилл 'ДКП' для source='rosreestr' (корректен, а не эвристика: всё загруженное прошло фильтр ДКП — и в импорте, и в 077) + явный region_code=66 в default_params расписания rosreestr_dkp_import вместо неявного дефолта в коде. Индекс НЕ добавлен: 2-3 значения, живые выборки идут по region_code/deal_date/geom — заведём частичный, когда появится режущий запрос; - import_rosreestr_dkp: region_code (default 66) и doc_types (default ['ДКП']) из params, фильтры через bind-параметры CAST(:region_code AS int) / ANY(CAST(:doc_types AS text[])), doc_type едет из SELECT в INSERT и в ON CONFLICT DO UPDATE. Дефолты сохраняют текущее прод-поведение байт-в-байт; - dedup_hash оставлен как 'ros:dkp:' || id: id уникален в источнике независимо от типа документа, а смена формы ключа осиротила бы уже загруженные строки (ровно то, что разгребала миграция 077); - deploy/import-rosreestr.sh: REGION_CODE / DOC_TYPE как env со старыми дефолтами, doc_type протащен через staging в deals; шапка про «ЕКБ квартиры» переписана честно — city-фильтр снят давно, скоуп = весь регион; - тесты: test_rosreestr_dedup_key переведён с ассертов на литералы на проверку «параметр + дефолт = скоуп 077»; новый test_3051_* проверяет bind-параметры реальным вызовом с моком Session, дефолты 66/['ДКП'], doc_type в колонках INSERT и текст 288. --- tradein-mvp/backend/app/services/scheduler.py | 52 ++++-- .../backend/data/sql/288_deals_doc_type.sql | 63 ++++++++ .../test_3051_deals_doc_type_region_param.py | 148 ++++++++++++++++++ .../backend/tests/test_rosreestr_dedup_key.py | 36 ++++- tradein-mvp/deploy/import-rosreestr.sh | 29 ++-- 5 files changed, 299 insertions(+), 29 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/288_deals_doc_type.sql create mode 100644 tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 0f67306d..0b2af764 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -328,26 +328,36 @@ def import_rosreestr_dkp( SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql. USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader). - Область покрытия: вся Свердловская область (region_code=66), не только Екатеринбург — - прежний ILIKE-фильтр по подстроке города (ограничивавший импорт одним Екатеринбургом) - снят (Mera trade-in расширяется на весь регион, unlocks +47183 сделок вне ЕКБ уже + Область покрытия задаётся параметром, а не литералом (#3051 п.6): region_code + приходит из params, дефолт 66 = вся Свердловская область (не только Екатеринбург — + прежний ILIKE-фильтр по подстроке города снят, unlocks +47183 сделок вне ЕКБ уже сидящих в source foreign table). address и deals.city строятся из реального city источника (не хардкод "Екатеринбург"), deals.region_code заполняется из строки - источника (= 66 при текущем фильтре). + источника (= запрошенному региону при текущем фильтре). + + Типы документов тоже параметр (#3051 п.3): doc_types, дефолт ['ДКП'] = прежнее + поведение (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24). Для Москвы + ДДУ идут по ценам котлована и медиану развалят, поэтому смешивать их с ДКП можно + только осознанно и с колонкой deals.doc_type (миграция 288), которая теперь + заполняется на импорте. Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24): - - region_code = 66 (вся Свердловская область, все города) + - region_code = :region_code (param, default 66 — вся Свердловская область) - city IS NOT NULL AND trim(city) != '' (непустой город → корректный address) - realestate_type_code = '002001003000' (квартира) - area BETWEEN 18 AND 200 - deal_price BETWEEN 1000000 AND 100000000 - street IS NOT NULL AND trim(street) != '' - - doc_type = 'ДКП' (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24) + - doc_type = ANY(:doc_types) (param, default ['ДКП']) - period_start_date >= since (default '2024-01-01') dedup_hash: 'ros:dkp:' || id — плоский натуральный ключ (инъективный, без коллизий, human-readable). До #576 здесь был md5('ros:dkp:' || id); миграция 077 конвертировала существующие строки. source_id хранит исходный rosreestr id (дедуп переустанавливаем). + Префикс ':dkp:' НАМЕРЕННО оставлен неизменным после параметризации doc_types: id + уникален в источнике сам по себе, независимо от типа документа, поэтому ключ и без + того не коллизирует; а вот смена формы ключа осиротила бы все уже загруженные строки + (их пришлось бы конвертировать ещё одной миграцией — ровно то, что делала 077). Rooms: выводятся из площади (Росреестр не отдаёт кол-во комнат). Batch-процессинг: читаем из FDW батчами по batch_size через cursor-based пагинацию @@ -364,6 +374,11 @@ def import_rosreestr_dkp( """ since: str = str(params.get("since", "2024-01-01")) batch_size: int = int(params.get("batch_size", 2000)) + # #3051 п.6: регион — параметр, дефолт 66 сохраняет текущее прод-поведение + # (расписание получает явный region_code в миграции 288). + region_code: int = int(params.get("region_code", 66)) + # #3051 п.3: типы документов — параметр, дефолт ['ДКП'] = прежний литерал. + doc_types: list[str] = [str(t) for t in params.get("doc_types") or ["ДКП"]] counters: dict[str, int] = { "rows_fetched": 0, @@ -472,21 +487,28 @@ def import_rosreestr_dkp( year_build AS year_built, round(deal_price)::bigint AS price_rub, 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 FROM gendesign_rosreestr_deals - WHERE region_code = 66 + WHERE region_code = CAST(:region_code AS int) AND city IS NOT NULL AND trim(city) <> '' AND realestate_type_code = '002001003000' AND area BETWEEN 18 AND 200 AND deal_price BETWEEN 1000000 AND 100000000 AND street IS NOT NULL AND trim(street) <> '' - AND doc_type = 'ДКП' + AND doc_type = ANY(CAST(:doc_types AS text[])) AND period_start_date >= CAST(:since AS date) AND id > CAST(:last_id AS bigint) ORDER BY id LIMIT CAST(:batch_size AS int) """), - {"since": since, "last_id": last_id, "batch_size": batch_size}, + { + "since": since, + "last_id": last_id, + "batch_size": batch_size, + "region_code": region_code, + "doc_types": doc_types, + }, ) .mappings() .all() @@ -524,7 +546,7 @@ def import_rosreestr_dkp( INSERT INTO deals ( source, dedup_hash, source_id, address, region_code, city, rooms, area_m2, floor, year_built, price_rub, price_per_m2, - deal_date + deal_date, doc_type ) VALUES ( 'rosreestr', @@ -539,7 +561,8 @@ def import_rosreestr_dkp( CAST(:year_built AS int), CAST(:price_rub AS bigint), CAST(:price_per_m2 AS int), - CAST(:deal_date AS date) + CAST(:deal_date AS date), + CAST(:doc_type AS text) ) ON CONFLICT (dedup_hash) DO UPDATE SET address = EXCLUDED.address, @@ -551,7 +574,8 @@ def import_rosreestr_dkp( year_built = EXCLUDED.year_built, price_rub = EXCLUDED.price_rub, price_per_m2 = EXCLUDED.price_per_m2, - deal_date = EXCLUDED.deal_date + deal_date = EXCLUDED.deal_date, + doc_type = EXCLUDED.doc_type WHERE deals.address IS DISTINCT FROM EXCLUDED.address OR deals.region_code IS DISTINCT FROM EXCLUDED.region_code OR deals.city IS DISTINCT FROM EXCLUDED.city @@ -562,6 +586,7 @@ def import_rosreestr_dkp( OR deals.price_rub IS DISTINCT FROM EXCLUDED.price_rub OR deals.price_per_m2 IS DISTINCT FROM EXCLUDED.price_per_m2 OR deals.deal_date IS DISTINCT FROM EXCLUDED.deal_date + OR deals.doc_type IS DISTINCT FROM EXCLUDED.doc_type RETURNING (xmax = 0) AS was_inserted """), { @@ -577,6 +602,7 @@ def import_rosreestr_dkp( "price_rub": row["price_rub"], "price_per_m2": row["price_per_m2"], "deal_date": row["deal_date"], + "doc_type": row["doc_type"], }, ).fetchone() if result is None: diff --git a/tradein-mvp/backend/data/sql/288_deals_doc_type.sql b/tradein-mvp/backend/data/sql/288_deals_doc_type.sql new file mode 100644 index 00000000..2e002720 --- /dev/null +++ b/tradein-mvp/backend/data/sql/288_deals_doc_type.sql @@ -0,0 +1,63 @@ +-- 288_deals_doc_type.sql +-- deals.doc_type — тип документа сделки Росреестра (#3051 п.3, п.6). +-- +-- Dependencies: 002_core_tables.sql (deals), 015_scrape_runs.sql + 072_scrape_schedules_seed_cian_rosreestr.sql +-- (scrape_schedules, строка source='rosreestr_dkp_import') +-- Apply after: 287_proxy_run_attribution.sql +-- +-- ЧАСТЬ A — WHY: +-- Импорт Росреестра до сих пор ронял тип документа на пол: фильтр doc_type = 'ДКП' +-- стоял литералом в WHERE, а в deals не приезжало ничего. Пока скоуп был один +-- (Свердловская обл., только вторичка) это было безобидно — все строки источника +-- 'rosreestr' по построению ДКП. С расширением на Москву безобидность кончается: +-- в источнике за 2024 по region_code=77 лежит 30 627 ДДУ с медианой 112 743 ₽/м² +-- против 107 005 ДКП с медианой 256 250 ₽/м² — это цены котлована, и смешать их +-- в одной таблице без различимого признака значит развалить любую оценку. +-- Колонка нужна ДО того, как импорт начнёт тянуть больше одного типа. +-- +-- ЧАСТЬ A — WHAT: +-- doc_type text NULLable — источник (foreign table gendesign_rosreestr_deals.doc_type) +-- тоже text и NULL допускает; строгий NOT NULL сломал бы не-росреестровые источники +-- (etazhi / domklik_history), у которых понятия «тип документа» нет вовсе. +-- Индекс НЕ добавляем: селективность низкая (2-3 значения), а все живые выборки +-- по deals идут по region_code/deal_date/geom — doc_type там в лучшем случае +-- довесок к уже отобранному диапазону. Появится запрос, который реально режет +-- по doc_type на большом наборе — заведём частичный индекс тогда, по EXPLAIN. +-- +-- ЧАСТЬ B — бэкфилл: +-- Всё, что лежит в deals с source='rosreestr', прошло через WHERE doc_type = 'ДКП' +-- (и в scheduler.import_rosreestr_dkp, и в deploy/import-rosreestr.sh, и в +-- backfill-миграции 077) — других типов там физически быть не может. Поэтому +-- проставить 'ДКП' задним числом корректно, а не эвристика. +-- Остальные источники остаются NULL осознанно. +-- +-- ЧАСТЬ C — расписание: +-- scrape_schedules.default_params для 'rosreestr_dkp_import' получает явный +-- region_code=66. Раньше регион был неявным дефолтом в коде; после параметризации +-- (params.get('region_code', 66)) неявность становится ловушкой — прод-строка должна +-- сама говорить, какой регион она тянет. doc_types в default_params НЕ пишем: +-- дефолт ['ДКП'] в коде и есть текущее поведение, а запись его в расписание +-- создала бы второе место, где надо не забыть поменять. +-- +-- Идемпотентна, безопасна к повторному запуску. + +BEGIN; + +-- A: колонка +ALTER TABLE deals ADD COLUMN IF NOT EXISTS doc_type text; + +COMMENT ON COLUMN deals.doc_type IS + 'Тип документа сделки Росреестра (ДКП / ДДУ). NULL для источников без этого понятия.'; + +-- B: бэкфилл — все rosreestr-строки прошли фильтр ДКП на импорте +UPDATE deals + SET doc_type = 'ДКП' + WHERE source = 'rosreestr' + AND doc_type IS NULL; + +-- C: явный регион в расписании импорта вместо неявного дефолта в коде +UPDATE scrape_schedules + SET default_params = default_params || '{"region_code": 66}'::jsonb + WHERE source = 'rosreestr_dkp_import'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py b/tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py new file mode 100644 index 00000000..f24ebaae --- /dev/null +++ b/tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py @@ -0,0 +1,148 @@ +"""#3051 п.3 + п.6: doc_type в deals и параметризация региона/типа документа на импорте. + +ЗАЧЕМ: для Москвы в источнике за 2024 лежат 30 627 ДДУ с медианой 112 743 ₽/м² против +107 005 ДКП с медианой 256 250 — это цены котлована. Тип сделки терялся при загрузке +(в deals не приезжал вовсе), а фильтры `region_code = 66` и `doc_type = 'ДКП'` стояли +литералами, поэтому регион 77 было не подключить, не правя код. Тесты фиксируют, что +параметры реально доезжают до bind-словаря, дефолты воспроизводят прежнее поведение, +а doc_type пишется в deals. + +Чисто-юнит: ассертим SQL-текст import_rosreestr_dkp (inspect.getsource) + текст миграции +288, плюс вызываем функцию с моком Session, чтобы увидеть фактические bind-параметры. +Живой FDW / БД не нужны. +""" + +import inspect +import os +import re +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +# Как в test_rosreestr_dedup_key.py: Settings требует DATABASE_URL до импорта. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import scheduler + +_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" +_MIGRATION_288 = _SQL_DIR / "288_deals_doc_type.sql" +_IMPORT_SRC = inspect.getsource(scheduler.import_rosreestr_dkp) + + +# ── SQL-форма импорта ──────────────────────────────────────────────────────── + + +def test_import_filters_region_and_doc_type_via_bind_params() -> None: + """Оба фильтра — bind-параметры, литералов 66 / 'ДКП' в SQL больше нет.""" + assert "region_code = CAST(:region_code AS int)" in _IMPORT_SRC + assert "doc_type = ANY(CAST(:doc_types AS text[]))" in _IMPORT_SRC + assert "WHERE region_code = 66" not in _IMPORT_SRC + assert "AND doc_type = 'ДКП'" not in _IMPORT_SRC + # psycopg v3: ни один bind-параметр не приводится через :name::type. + assert not re.search(r":\w+::", _IMPORT_SRC) + + +def test_import_writes_doc_type_into_deals() -> None: + """doc_type едет из SELECT в список колонок INSERT, в VALUES и в ON CONFLICT.""" + assert "doc_type AS doc_type" in _IMPORT_SRC + assert "deal_date, doc_type" in _IMPORT_SRC + assert "CAST(:doc_type AS text)" in _IMPORT_SRC + # Переопубликованный квартал может исправить тип документа — обновляем. + assert "doc_type = EXCLUDED.doc_type" in _IMPORT_SRC + assert "IS DISTINCT FROM EXCLUDED.doc_type" in _IMPORT_SRC + assert '"doc_type": row["doc_type"]' in _IMPORT_SRC + + +def test_import_keeps_dedup_key_shape_unchanged() -> None: + """Ключ дедупа НЕ зависит от типа документа — иначе осиротели бы старые строки. + + id уникален в источнике сам по себе; префикс 'ros:dkp:' остаётся историческим + именем формата (см. миграцию 077, которая уже конвертировала все строки). + """ + assert "'ros:dkp:' || CAST(id AS text) AS dedup_hash" in _IMPORT_SRC + assert "ON CONFLICT (dedup_hash) DO UPDATE SET" in _IMPORT_SRC + + +# ── фактические bind-параметры (вызов с моком) ─────────────────────────────── + + +def _capture_select_params(params: dict[str, Any]) -> dict[str, Any]: + """Прогоняет import_rosreestr_dkp до первого SELECT из FDW и возвращает его binds. + + Мок Session: cleanup-DELETE и _resume_dkp_cursor получают пустые результаты, + первый батч — пустой список, поэтому цикл завершается сразу после SELECT. + """ + captured: list[dict[str, Any]] = [] + + def _execute(stmt: Any, binds: Any = None, *a: Any, **kw: Any) -> MagicMock: + sql = str(stmt) + if "FROM gendesign_rosreestr_deals" in sql and isinstance(binds, dict): + captured.append(binds) + result = MagicMock() + result.fetchall.return_value = [] + result.fetchone.return_value = None + result.mappings.return_value.all.return_value = [] + result.scalar.return_value = None + return result + + db = MagicMock() + db.execute.side_effect = _execute + + monkeys = { + "is_cancelled": False, + } + orig_is_cancelled = scheduler.runs_mod.is_cancelled + orig_mark_done = scheduler.runs_mod.mark_done + orig_heartbeat = scheduler.kit_runs.update_heartbeat + try: + scheduler.runs_mod.is_cancelled = lambda *a, **kw: monkeys["is_cancelled"] # type: ignore[assignment] + scheduler.runs_mod.mark_done = lambda *a, **kw: None # type: ignore[assignment] + scheduler.kit_runs.update_heartbeat = lambda *a, **kw: None # type: ignore[assignment] + scheduler.import_rosreestr_dkp(db, 1, params) + finally: + scheduler.runs_mod.is_cancelled = orig_is_cancelled # type: ignore[assignment] + scheduler.runs_mod.mark_done = orig_mark_done # type: ignore[assignment] + scheduler.kit_runs.update_heartbeat = orig_heartbeat # type: ignore[assignment] + + assert captured, "SELECT из gendesign_rosreestr_deals не выполнился" + return captured[0] + + +def test_defaults_preserve_current_production_behaviour() -> None: + """Без параметров — ровно прежний скоуп: Свердловская обл., только ДКП.""" + binds = _capture_select_params({}) + assert binds["region_code"] == 66 + assert binds["doc_types"] == ["ДКП"] + + +def test_params_override_region_and_doc_types() -> None: + """Москва + оба типа документа приходят из params, а не из кода.""" + binds = _capture_select_params({"region_code": 77, "doc_types": ["ДКП", "ДДУ"]}) + assert binds["region_code"] == 77 + assert binds["doc_types"] == ["ДКП", "ДДУ"] + + +def test_region_code_accepts_string_from_jsonb_params() -> None: + """default_params прилетают из jsonb — строковый регион не должен ломать CAST.""" + binds = _capture_select_params({"region_code": "77"}) + assert binds["region_code"] == 77 + assert isinstance(binds["region_code"], int) + + +# ── миграция 288 ───────────────────────────────────────────────────────────── + + +def test_migration_288_adds_column_backfills_and_sets_schedule_region() -> None: + """288: idempotent-колонка, бэкфилл только rosreestr-строк, явный регион в расписании.""" + assert _MIGRATION_288.is_file(), f"missing migration: {_MIGRATION_288}" + sql = _MIGRATION_288.read_text("utf-8") + assert "ALTER TABLE deals ADD COLUMN IF NOT EXISTS doc_type text" in sql + # Бэкфилл корректен ровно потому, что всё загруженное прошло фильтр ДКП. + assert "SET doc_type = 'ДКП'" in sql + assert "WHERE source = 'rosreestr'" in sql + assert "AND doc_type IS NULL" in sql + # Явный region_code в расписании вместо неявного дефолта в коде. + assert "UPDATE scrape_schedules" in sql + assert "'{\"region_code\": 66}'::jsonb" in sql + assert "source = 'rosreestr_dkp_import'" in sql + assert "BEGIN;" in sql and "COMMIT;" in sql diff --git a/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py b/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py index 90176d3c..145101c0 100644 --- a/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py +++ b/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py @@ -151,23 +151,45 @@ def test_migration_077_converts_md5_to_plain_key() -> None: def test_migration_077_shared_filter_matches_live_import() -> None: """Дедуп-релевантные фильтры src CTE миграции 077 совпадают с живым импортом. - Исключение — city ILIKE (см. test_live_import_dropped_ekb_city_filter ниже): - 077 backfill'ил ЕКБ-строки под EKB-only scope того времени; живой импорт расширен - на всю Свердловскую область (region_code=66, все города), поэтому city-фильтр из - живого импорта СНЯТ намеренно. Остальные клозы обязаны совпадать байт-в-байт, - иначе 077 конвертировал бы не тот набор строк. + Два исключения: + - city ILIKE (см. test_live_import_dropped_ekb_city_filter ниже): 077 backfill'ил + ЕКБ-строки под EKB-only scope того времени; живой импорт расширен на всю + Свердловскую область, поэтому city-фильтр из живого импорта СНЯТ намеренно; + - region_code / doc_type: в живом импорте это ПАРАМЕТРЫ (#3051), их совпадение + с 077 проверяется через дефолты — см. тест ниже. + Остальные клозы обязаны совпадать байт-в-байт, иначе 077 конвертировал бы не тот + набор строк. """ sql = _MIGRATION_077.read_text("utf-8") for clause in ( - "region_code = 66", "realestate_type_code = '002001003000'", "area BETWEEN 18 AND 200", "deal_price BETWEEN 1000000 AND 100000000", "street IS NOT NULL AND trim(street) <> ''", - "doc_type = 'ДКП'", ): assert clause in sql, f"missing filter clause in migration: {clause!r}" assert clause in _IMPORT_SRC, f"missing filter clause in import: {clause!r}" + # Исторические литералы 077 остаются на месте (миграция применена, её не правят). + assert "region_code = 66" in sql + assert "doc_type = 'ДКП'" in sql + + +def test_live_import_parametrized_region_and_doc_type_keep_077_defaults() -> None: + """#3051: регион и типы документов — bind-параметры, но ДЕФОЛТЫ = скоуп 077. + + Литералы `region_code = 66` / `doc_type = 'ДКП'` ушли из SQL живого импорта + (иначе регион 77 не подключить), но дефолты обязаны воспроизводить ровно тот + набор строк, который 077 когда-то backfill'ил, — иначе штатный прогон молча + сменил бы скоуп. + """ + # В SQL — параметры, не литералы. + assert "region_code = CAST(:region_code AS int)" in _IMPORT_SRC + assert "doc_type = ANY(CAST(:doc_types AS text[]))" in _IMPORT_SRC + assert "region_code = 66" not in _IMPORT_BODY + assert "doc_type = 'ДКП'" not in _IMPORT_BODY.split("params.get", 1)[-1].split("SELECT", 1)[0] + # Дефолты — прежнее поведение. + assert 'params.get("region_code", 66)' in _IMPORT_SRC + assert 'params.get("doc_types") or ["ДКП"]' in _IMPORT_SRC def test_live_import_dropped_ekb_city_filter() -> None: diff --git a/tradein-mvp/deploy/import-rosreestr.sh b/tradein-mvp/deploy/import-rosreestr.sh index 460b076f..e78424c7 100755 --- a/tradein-mvp/deploy/import-rosreestr.sh +++ b/tradein-mvp/deploy/import-rosreestr.sh @@ -2,13 +2,19 @@ # Импорт реальных сделок Росреестра из gendesign-БД в tradein deals. # # Источник: gendesign-postgres-1 / rosreestr_deals (6.8М строк, партиц.). -# Берём ЕКБ квартиры (realestate_type_code=002001003000) за последние ~18 мес. +# Берём квартиры (realestate_type_code=002001003000) региона REGION_CODE — по умолчанию +# 66 = вся Свердловская область, все города (не только ЕКБ: city-фильтр снят давно, +# шапка про «ЕКБ квартиры» была неправдой). # rooms выводим из площади (Росреестр не отдаёт кол-во комнат). # Координаты — NULL, проставляются отдельно (geocode по street). # -# Только ДКП (вторичка) — ДДУ застройщиков скёюят median, отделены сознательно. См. PR-A. +# DOC_TYPE по умолчанию ДКП (вторичка): ДДУ идут по ценам котлована и скёюят median, +# отделены сознательно (см. PR-A). С #3051 это параметр, а не литерал — для Москвы +# (REGION_CODE=77) типы разделяются колонкой deals.doc_type (миграция 288), которую +# скрипт теперь заполняет. # # Запуск на прод-хосте: ./import-rosreestr.sh +# REGION_CODE=77 DOC_TYPE='ДДУ' ./import-rosreestr.sh # Москва, первичка # Повторяемо: дедуп по dedup_hash, новый запуск подтянет свежие кварталы. set -euo pipefail @@ -18,8 +24,11 @@ DST_PG="${DST_PG:-tradein-postgres}" SRC_DB="${SRC_DB:-gendesign}" SRC_USER="${SRC_USER:-gendesign}" SINCE="${SINCE:-2024-01-01}" +# #3051: регион и тип документа — параметры со старыми дефолтами (поведение не меняется). +REGION_CODE="${REGION_CODE:-66}" +DOC_TYPE="${DOC_TYPE:-ДКП}" -echo "[$(date -u +%H:%M:%S)] import-rosreestr: ЕКБ квартиры с $SINCE" +echo "[$(date -u +%H:%M:%S)] import-rosreestr: region=$REGION_CODE doc_type=$DOC_TYPE с $SINCE" # 1. Staging-таблица в tradein. docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c " @@ -29,7 +38,8 @@ docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c " CREATE TABLE deals_ros_staging ( dedup_hash text PRIMARY KEY, source_id text, address text, region_code int, city text, rooms int, area_m2 numeric, - floor int, year_built int, price_rub bigint, price_per_m2 int, deal_date date + floor int, year_built int, price_rub bigint, price_per_m2 int, deal_date date, + doc_type text ); " @@ -51,15 +61,16 @@ docker exec "$SRC_PG" psql -U "$SRC_USER" -d "$SRC_DB" -v ON_ERROR_STOP=on -c " year_build AS year_built, round(deal_price)::bigint AS price_rub, 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 FROM rosreestr_deals - WHERE region_code = 66 + WHERE region_code = $REGION_CODE AND city IS NOT NULL AND trim(city) <> '' AND realestate_type_code = '002001003000' AND area BETWEEN 18 AND 200 AND deal_price BETWEEN 1000000 AND 100000000 AND street IS NOT NULL AND trim(street) <> '' - AND doc_type = 'ДКП' + AND doc_type = '$DOC_TYPE' AND period_start_date >= '$SINCE' ) TO STDOUT WITH CSV " | docker exec -i "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c " @@ -75,10 +86,10 @@ docker exec "$DST_PG" psql -U tradein -d tradein -v ON_ERROR_STOP=on -c " INSERT INTO deals ( source, dedup_hash, source_id, address, region_code, city, rooms, area_m2, floor, - year_built, price_rub, price_per_m2, deal_date + year_built, price_rub, price_per_m2, deal_date, doc_type ) SELECT 'rosreestr', dedup_hash, source_id, address, region_code, city, rooms, area_m2, floor, - year_built, price_rub, price_per_m2, deal_date + year_built, price_rub, price_per_m2, deal_date, doc_type FROM deals_ros_staging ON CONFLICT (dedup_hash) DO NOTHING; From 7b33ed9a45cdc33881d0c475959a6a2d789a3e3c Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 8 Sep 2026 22:48:37 +0300 Subject: [PATCH 2/5] =?UTF-8?q?feat(#2989):=20=D0=BB=D0=BE=D0=BA=D0=B0?= =?UTF-8?q?=D0=BB=D1=8C=D0=BD=D1=8B=D0=B9=20=D1=80=D1=83=D1=87=D0=BD=D0=BE?= =?UTF-8?q?=D0=B9=20=D1=81=D0=B1=D0=BE=D1=80=D1=89=D0=B8=D0=BA=20SERP=20?= =?UTF-8?q?=D0=90=D0=B2=D0=B8=D1=82=D0=BE=20=D0=BF=D0=BE=20=D0=9C=D0=BE?= =?UTF-8?q?=D1=81=D0=BA=D0=B2=D0=B5=20=D0=B8=20=D0=9C=D0=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Трек 1 эпика: нужен разовый корпус вторички Москвы+МО в msk_raw, а прод-скрейпер для этого не подходит — его расписания, прокси-пул и сайдкар держат ЕКБ и трогать их ради ручного замера нельзя. Поэтому скрипт: браузер — уже открытый Chrome владельца с залогиненным техаккаунтом (connect_over_cdp, своя вкладка, чужие вкладки/контекст/браузер не трогаем и не закрываем; своего профиля не поднимаем); парсер — импорт _parse_html / _extract_total_count / _is_firewall_page из scraper-kit, а не копия (копия разъедется с прод-парсером на первом же DOM-drift); заливка — поток в psql через ssh, потому что прямого доступа к прод-Postgres с локалки нет, а одиночный psql -c ломается на квотинге. avito_serp_ekb_only=False обязателен: с True парсер выбрасывает всё, где в URL нет /ekaterinburg/ — из московской выдачи не осталось бы ни одной карточки. Потолок Авито 30x60=1800 на запрос, поэтому план ценовых коридоров с бисекцией по ГЕОМЕТРИЧЕСКОЙ середине: цены логнормальны, арифметическая середина 1млн..100млн даёт вырожденно-пустую верхнюю половину. Коридор, который не влезает в 1800 даже на минимальной ширине, помечается truncated и недобор пишется в batches.notes — молчаливое усечение читалось бы как полный охват. Стоп на первом признаке блока (403/439, 429, firewall, PoW, 0 карточек при ненулевом счётчике) без ретраев: ретрай по забаненному техаккаунту только углубляет бан. Дефолт — --measure 100, полный проход только по явному --full. --- tradein-mvp/scripts/local-avito-msk/README.md | 147 ++++ .../scripts/local-avito-msk/collect.py | 663 ++++++++++++++++++ 2 files changed, 810 insertions(+) create mode 100644 tradein-mvp/scripts/local-avito-msk/README.md create mode 100644 tradein-mvp/scripts/local-avito-msk/collect.py diff --git a/tradein-mvp/scripts/local-avito-msk/README.md b/tradein-mvp/scripts/local-avito-msk/README.md new file mode 100644 index 00000000..499a22b9 --- /dev/null +++ b/tradein-mvp/scripts/local-avito-msk/README.md @@ -0,0 +1,147 @@ +# Локальный сбор SERP Авито по Москве и МО (эпик #2989, трек 1) + +`collect.py` — ручной скрипт **с машины владельца**. Собирает карточки выдачи Авито +(вторичка, Москва + МО) и заливает их в прод-схему `msk_raw`. + +Прод-скрейпер, его расписания, прокси-пул и сайдкар **не задействованы вообще**. +Браузер — уже открытый Chrome владельца (подключение по CDP), парсер — импорт из +`packages/scraper-kit`, заливка — поток в `psql` через `ssh`. + +## Предусловия + +1. **Chrome владельца** запущен с залогиненным техническим аккаунтом Авито и + remote debugging: + + ```powershell + & "C:\Program Files\Google\Chrome\Application\chrome.exe" --remote-debugging-port=9222 + ``` + + Проверка: `curl http://localhost:9222/json/version` отдаёт JSON с `Browser: Chrome/...`. + Другой адрес — переменная `AVITO_CDP`. + + Скрипт **не поднимает свой профиль** (`launch_persistent_context` не используется): + он подключается к существующему браузеру, берёт `browser.contexts[0]`, открывает + **свою** вкладку и в конце закрывает **только её**. Чужие вкладки, контекст и сам + браузер не трогаются — это рабочий Chrome владельца. + +2. **ssh-доступ на прод** (`ssh selectel` без пароля) — для заливки в БД. + Прямого подключения к прод-Postgres с локалки нет: туннель `:35432` ведёт в мёртвую + копию на Beget. + +3. **Playwright в текущем интерпретаторе**: + + ```powershell + pip install playwright + python -m playwright install chromium + ``` + + В uv-зависимости проекта playwright **не добавлять**: прод-сайдкар пинит + playwright 1.60 под camoufox, и подъём версии сломает его. + + Больше ничего ставить не нужно: только stdlib + playwright + импорт `scraper_kit` + (путь `packages/scraper-kit/src` скрипт добавляет в `sys.path` сам, от `__file__`). + +## Запуск (PowerShell) + +```powershell +cd D:\prjct\gendesign\tradein-mvp\scripts\local-avito-msk + +# 0) сухой прогон: ничего не шлём на прод, карточки пишем в runs\cards-.csv +python .\collect.py --dry-run --measure 5 + +# 1) обязательный первый прогон — замер (дефолт, 100 загрузок страниц) +python .\collect.py + +# 2) полный проход — только явно +python .\collect.py --full --batch-id msk-serp-20260908 + +# 3) продолжить прерванный прогон по сохранённому плану коридоров +python .\collect.py --full --resume --batch-id msk-serp-20260908 +``` + +Без аргументов скрипт работает в режиме `--measure 100` и полный проход **не начинает**. + +Ключи: `--delay` (пауза между загрузками, дефолт 8.0 с ±20 % джиттера — сознательно +совпадает с прод-расписаниями `request_delay_sec` 7–10 с), `--batch-size` (карточек в +одной заливке, дефолт 1000), `--target-count` (целевой размер коридора, дефолт 1500), +`--base-url`, `--batch-id`, `--out-dir`, `--ssh-host/--container/--db-user/--db-name`. + +## Как режется выдача + +Потолок пагинации Авито — 30 страниц по 60 = **1800 объявлений на запрос**. Любой +запрос с `count > 1800` целиком не добирается, поэтому строится план ценовых коридоров: + +* читаем счётчик «N объявлений» со страницы 1 (`page-title/count`); +* `count > --target-count` → делим коридор пополам **по геометрической середине** + (`sqrt(lo*hi)`): цены логнормальны, арифметическая середина диапазона 1 млн … 100 млн + даёт вырожденно-пустую верхнюю половину; +* верхняя граница открытого коридора подбирается удвоением от 8 млн ₽; +* предохранители: глубина рекурсии ≤ 12 и минимальная ширина коридора (отношение + границ ≤ 1.05). Если коридор уже узкий, а `count` всё ещё > 1800 — он помечается + `truncated: true` в плане, а в лог и в `msk_raw.batches.notes` пишется, сколько + объявлений заведомо не добрано; +* гео-параметры `radius`/`geoCoords` не используются: сервером они не применяются (#3043). + +План лежит в `runs/plan-.json` и обновляется после каждой страницы — отсюда +работает `--resume`. + +## Стоп на первом признаке блока + +Проверки в фиксированном порядке, первое срабатывание = немедленный стоп +(никаких ретраев и никакого «продолжим со следующего коридора»): + +| # | Признак | Причина в `notes` | +|---|---|---| +| 1 | HTTP 403 / 439 | `platform` | +| 2 | HTTP 429 | `ratelimit` | +| 3 | `_is_firewall_page(html)` — «доступ ограничен», «проблема с ip», `firewall-container` | `firewall` | +| 4 | `startpow` / «доступ ограничен: проверка безопасности» в первых 4 КБ | `challenge` | +| 5 | 0 карточек при ненулевом счётчике (DOM-drift или тихий блок) | `empty_page` | + +При стопе: недоотправленный батч дозаливается, у батча проставляются `finished_at` и +`notes`, скрипт выходит с кодом **2**. + +**Что делать при стопе.** Не перезапускать сразу и не крутить ретраи. `platform` / +`firewall` / `challenge` — техаккаунт или IP помечены: пауза на несколько часов, +проверить вручную в браузере, что выдача открывается и аккаунт жив, при повторе +увеличить `--delay`. `ratelimit` — темп слишком высокий: `--delay 15` и выше. +`empty_page` — сначала посмотреть сохранённую страницу в браузере: если выдача +рисуется, значит уехал DOM и чинить надо парсер в `scraper-kit`, а не скрипт. +После разбора — `--resume` с тем же `--batch-id`, уже собранное не потеряется. + +## Куда пишем + +Схема `msk_raw` на проде (создана заранее): + +* `msk_raw.batches(batch_id PK, kind, query, started_at, finished_at, rows_sent, rows_new, notes, uploaded_at)` +* `msk_raw.avito_cards(id, source_id, observed_at, batch_id → batches, kind, url, price, payload, UNIQUE(source_id,batch_id,kind))` +* `msk_raw.avito_latest` — вью `DISTINCT ON (source_id) … ORDER BY source_id, observed_at DESC` + +Форма заливки: поток в +`ssh "docker exec -i tradein-postgres psql -U tradein -d tradein -v ON_ERROR_STOP=1 -f -"`. +Внутри одной транзакции: `INSERT` батча (`ON CONFLICT DO NOTHING` — строка обязана +существовать до карточек из-за FK), `CREATE TEMP TABLE _stg (LIKE msk_raw.avito_cards +INCLUDING DEFAULTS)`, `\copy _stg (...) FROM STDIN WITH (FORMAT csv)`, затем +`INSERT … SELECT` в `avito_cards` с `ON CONFLICT (source_id,batch_id,kind) DO NOTHING` +и `UPDATE batches SET rows_sent/rows_new` (`rows_new` = разница `count(*)` по batch_id +до и после вставки). + +Одиночный `psql -c` через ssh ломается на квотинге скобок и кавычек — поэтому только поток. + +`payload` = `lot.model_dump(mode="json")`. `source_id` в БД `bigint`, а у `ScrapedLot` — +строка: приводится к `int`, нечисловые пропускаются со счётчиком (он попадает в +`notes` как `skipped_non_numeric=N`). + +## Как проверить залитое + +```powershell +ssh selectel "docker exec -i tradein-postgres psql -U tradein -d tradein -c \"SELECT count(*) FROM msk_raw.avito_latest;\"" +ssh selectel "docker exec -i tradein-postgres psql -U tradein -d tradein -c \"SELECT batch_id, rows_sent, rows_new, started_at, finished_at, notes FROM msk_raw.batches ORDER BY uploaded_at DESC LIMIT 5;\"" +``` + +## Про фильтр городов + +Парсер конструируется как `AvitoScraper(SimpleNamespace(avito_serp_ekb_only=False), +target_city_slug="moskva")`. `avito_serp_ekb_only=False` **обязателен**: с `True` +`_parse_html` выбрасывает всё, у чего в URL нет `/ekaterinburg/`, включая все +подмосковные слаги — из московской выдачи не осталось бы ничего. diff --git a/tradein-mvp/scripts/local-avito-msk/collect.py b/tradein-mvp/scripts/local-avito-msk/collect.py new file mode 100644 index 00000000..2432cc86 --- /dev/null +++ b/tradein-mvp/scripts/local-avito-msk/collect.py @@ -0,0 +1,663 @@ +#!/usr/bin/env python3 +"""Локальный ручной сборщик SERP Авито по Москве и МО (эпик #2989, трек 1). + +Запускается ВРУЧНУЮ с машины владельца. Прод-скрейпер, его расписания и +прокси-пул не задействованы вообще: браузер — уже открытый Chrome владельца +(подключение по CDP), парсер — импорт из scraper-kit, заливка — поток в psql +через ssh. Скрипт ничего не устанавливает и своего профиля не поднимает. + +Дефолтный режим — --measure 100 (замер): полный проход только по явному --full. +""" + +from __future__ import annotations + +import argparse +import asyncio +import csv +import io +import json +import math +import os +import random +import re +import subprocess +import sys +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Iterable +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +# --- импорт парсера из scraper-kit без установки backend ------------------- +_KIT_SRC = Path(__file__).resolve().parents[2] / "packages" / "scraper-kit" / "src" +if str(_KIT_SRC) not in sys.path: + sys.path.insert(0, str(_KIT_SRC)) + +from scraper_kit.providers.avito.serp import ( # noqa: E402 + AvitoScraper, + _is_firewall_page, +) + +# Вкладка, открытая у владельца: вторичка, Москва + МО. +DEFAULT_BASE_URL = ( + "https://www.avito.ru/moskva_i_mo/kvartiry/prodam/vtorichka-ASgBAgICAkSSA8YQ5geMUg" + "?f=ASgBAgICA0SSA8YQ5geMUuDC3c0CgOkP" +) + +PAGE_SIZE = 60 # карточек на странице выдачи +MAX_PAGES = 30 # потолок пагинации Авито → 30*60 = 1800 на один запрос +HARD_CAP = PAGE_SIZE * MAX_PAGES + +PRICE_FLOOR = 500_000 # нижняя граница осмысленного коридора, ₽ +PRICE_PROBE_START = 8_000_000 # старт удвоения при поиске верхней границы +PRICE_CEIL = 2_000_000_000 +MIN_WIDTH_RATIO = 1.05 # уже этого коридор не делим (геометрическая ширина) +MAX_DEPTH = 12 + +_BATCH_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$") +_POW_MARKERS = ("startpow", "доступ ограничен: проверка безопасности") + + +class Blocked(Exception): + """Первый признак блока. Ретраев нет — только немедленный стоп.""" + + def __init__(self, reason: str, detail: str = "") -> None: + super().__init__(f"{reason}: {detail}" if detail else reason) + self.reason = reason + self.detail = detail + + +class BudgetExhausted(Exception): + """Потолок --measure выбран: штатный выход, не ошибка.""" + + +# --- план коридоров -------------------------------------------------------- + + +@dataclass +class Corridor: + lo: int | None + hi: int | None + count: int | None = None + truncated: bool = False + pages_done: int = 0 + status: str = "pending" # pending | done + missed: int = 0 # заведомо недобрано (count - HARD_CAP), если truncated + + def label(self) -> str: + lo = "-" if self.lo is None else f"{self.lo:_}" + hi = "-" if self.hi is None else f"{self.hi:_}" + return f"[{lo} .. {hi}]" + + def to_json(self) -> dict[str, Any]: + return { + "lo": self.lo, "hi": self.hi, "count": self.count, + "truncated": self.truncated, "pages_done": self.pages_done, + "status": self.status, "missed": self.missed, + } + + @staticmethod + def from_json(d: dict[str, Any]) -> "Corridor": + return Corridor( + lo=d.get("lo"), hi=d.get("hi"), count=d.get("count"), + truncated=bool(d.get("truncated")), + pages_done=int(d.get("pages_done") or 0), + status=d.get("status") or "pending", missed=int(d.get("missed") or 0), + ) + + def planned_pages(self) -> int: + if not self.count: + return 1 + return max(1, min(MAX_PAGES, math.ceil(self.count / PAGE_SIZE))) + + +@dataclass +class Plan: + base_url: str + target: int + batch_id: str + corridors: list[Corridor] = field(default_factory=list) + created_at: str = "" + + def save(self, path: Path) -> None: + path.write_text( + json.dumps( + { + "version": 1, "base_url": self.base_url, "target": self.target, + "batch_id": self.batch_id, "created_at": self.created_at, + "corridors": [c.to_json() for c in self.corridors], + }, + ensure_ascii=False, indent=1, + ), + encoding="utf-8", + ) + + @staticmethod + def load(path: Path) -> "Plan": + d = json.loads(path.read_text(encoding="utf-8")) + return Plan( + base_url=d["base_url"], target=int(d["target"]), batch_id=d["batch_id"], + created_at=d.get("created_at", ""), + corridors=[Corridor.from_json(c) for c in d.get("corridors", [])], + ) + + +def build_url(base_url: str, page: int, lo: int | None, hi: int | None) -> str: + """URL коридора: pmin/pmax + пагинация. Гео-параметры не добавляем — #3043.""" + parts = urlsplit(base_url) + q = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True) + if k not in {"p", "pmin", "pmax"}] + if lo is not None: + q.append(("pmin", str(int(lo)))) + if hi is not None: + q.append(("pmax", str(int(hi)))) + if page > 1: + q.append(("p", str(page))) + return urlunsplit( + (parts.scheme, parts.netloc, parts.path, urlencode(q), parts.fragment) + ) + + +def geometric_mid(lo: int | None, hi: int) -> int: + """Геометрическая середина коридора. + + Цены логнормальны: арифметическая середина 1 млн..100 млн (≈50 млн) + отрезает вырожденно-пустую верхнюю половину. sqrt(lo*hi) делит выборку + заметно ровнее. + """ + low = max(int(lo or PRICE_FLOOR), 1) + mid = int(math.sqrt(low * float(hi))) + return max(low + 1, min(hi - 1, mid)) + + +def width_ratio(lo: int | None, hi: int | None) -> float: + if hi is None: + return float("inf") + return float(hi) / max(float(lo or PRICE_FLOOR), 1.0) + + +# --- загрузка страницы ----------------------------------------------------- + + +def _guard(html: str, status: int | None) -> None: + """Порядок проверок фиксирован заданием; первое срабатывание = стоп.""" + if status in (403, 439): + raise Blocked("platform", f"HTTP {status}") + if status == 429: + raise Blocked("ratelimit", "HTTP 429") + if _is_firewall_page(html): + raise Blocked("firewall", "firewall-страница на HTTP 200") + head = html[:4096].lower() + if any(m in head for m in _POW_MARKERS): + raise Blocked("challenge", "PoW / проверка безопасности") + + +class Loader: + """Одна СВОЯ вкладка в уже открытом Chrome владельца (CDP). + + Ни браузер, ни контекст, ни чужие вкладки не закрываются и не трогаются: + это рабочий Chrome с залогиненным техаккаунтом. + """ + + def __init__(self, delay: float, page_budget: int | None) -> None: + self._delay = delay + self._budget = page_budget + self.loads = 0 + self._page: Any = None + self._pw: Any = None + self._browser: Any = None + self._last_load = 0.0 + + async def __aenter__(self) -> "Loader": + from playwright.async_api import async_playwright + + endpoint = os.environ.get("AVITO_CDP", "http://localhost:9222") + self._pw = await async_playwright().start() + try: + self._browser = await self._pw.chromium.connect_over_cdp(endpoint) + except Exception as exc: # noqa: BLE001 — подсказка важнее типа + await self._pw.stop() + raise SystemExit( + f"Не удалось подключиться по CDP к {endpoint}: {exc}\n" + "Запусти Chrome с залогиненным техаккаунтом Авито и ключом " + "--remote-debugging-port=9222, либо укажи адрес в AVITO_CDP." + ) from exc + if not self._browser.contexts: + await self._pw.stop() + raise SystemExit( + "В подключённом Chrome нет ни одного контекста. Открой обычное окно " + "Chrome, запущенное с --remote-debugging-port=9222." + ) + ctx = self._browser.contexts[0] + self._page = await ctx.new_page() + return self + + async def __aexit__(self, *exc: object) -> None: + if self._page is not None: + try: + await self._page.close() # ТОЛЬКО своя вкладка + except Exception: # noqa: BLE001 + pass + if self._pw is not None: + try: + await self._pw.stop() + except Exception: # noqa: BLE001 + pass + + def budget_left(self) -> bool: + return self._budget is None or self.loads < self._budget + + async def _pause(self) -> None: + if self._last_load == 0.0: + return + jitter = self._delay * random.uniform(-0.2, 0.2) + wait = max(0.0, self._delay + jitter - (time.monotonic() - self._last_load)) + if wait > 0: + print(f" пауза {wait:.1f} с", flush=True) + await asyncio.sleep(wait) + + async def fetch(self, url: str) -> tuple[str, int | None]: + if not self.budget_left(): + raise BudgetExhausted() + await self._pause() + resp = await self._page.goto(url, wait_until="domcontentloaded", timeout=90_000) + self.loads += 1 + self._last_load = time.monotonic() + status = resp.status if resp is not None else None + try: + await self._page.wait_for_selector('[data-marker="item"]', timeout=7_000) + except Exception: # noqa: BLE001 — пустая/блочная страница разбирается ниже + pass + html = await self._page.content() + _guard(html, status) + return html, status + + +# --- заливка в msk_raw ----------------------------------------------------- + + +def _sql_str(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def _csv_rows(rows: Iterable[dict[str, Any]]) -> str: + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + for r in rows: + writer.writerow([ + r["source_id"], r["observed_at"], r["batch_id"], r["kind"], + r["url"], r["price"], r["payload"], + ]) + return buf.getvalue() + + +def build_sql(batch_id: str, query: str, rows: list[dict[str, Any]], + started_at: str, kind: str = "serp") -> str: + """Один поток на `psql -f -`: batch (FK!) → TEMP staging → \\copy → INSERT. + + Одиночный `psql -c` через ssh ломается на квотинге скобок и кавычек, поэтому + только поток. rows_new = разница count(*) по batch_id до и после вставки. + """ + bid = _sql_str(batch_id) + return ( + "BEGIN;\n" + "INSERT INTO msk_raw.batches (batch_id, kind, query, started_at)\n" + f"VALUES ({bid}, {_sql_str(kind)}, {_sql_str(query)}, " + f"CAST({_sql_str(started_at)} AS timestamptz))\n" + "ON CONFLICT (batch_id) DO NOTHING;\n" + "CREATE TEMP TABLE _stg (LIKE msk_raw.avito_cards INCLUDING DEFAULTS) " + "ON COMMIT DROP;\n" + "CREATE TEMP TABLE _before ON COMMIT DROP AS\n" + f" SELECT count(*) AS n FROM msk_raw.avito_cards WHERE batch_id = {bid};\n" + "\\copy _stg (source_id,observed_at,batch_id,kind,url,price,payload) " + "FROM STDIN WITH (FORMAT csv)\n" + + _csv_rows(rows) + + "\\.\n" + "INSERT INTO msk_raw.avito_cards " + "(source_id,observed_at,batch_id,kind,url,price,payload)\n" + "SELECT source_id,observed_at,batch_id,kind,url,price,payload FROM _stg\n" + "ON CONFLICT (source_id,batch_id,kind) DO NOTHING;\n" + "UPDATE msk_raw.batches b SET\n" + " rows_sent = coalesce(b.rows_sent,0) + (SELECT count(*) FROM _stg),\n" + " rows_new = coalesce(b.rows_new,0) +\n" + f" ((SELECT count(*) FROM msk_raw.avito_cards WHERE batch_id = {bid})\n" + " - (SELECT n FROM _before))\n" + f"WHERE b.batch_id = {bid};\n" + "COMMIT;\n" + ) + + +def build_finalize_sql(batch_id: str, query: str, notes: str) -> str: + bid = _sql_str(batch_id) + return ( + "INSERT INTO msk_raw.batches (batch_id, kind, query, started_at)\n" + f"VALUES ({bid}, 'serp', {_sql_str(query)}, now())\n" + "ON CONFLICT (batch_id) DO NOTHING;\n" + f"UPDATE msk_raw.batches SET finished_at = now(), notes = {_sql_str(notes)}\n" + f"WHERE batch_id = {bid};\n" + ) + + +def run_psql(sql: str, ssh_host: str, container: str, db_user: str, db_name: str) -> None: + cmd = [ + "ssh", ssh_host, + f"docker exec -i {container} psql -U {db_user} -d {db_name} " + "-v ON_ERROR_STOP=1 -f -", + ] + proc = subprocess.run(cmd, input=sql.encode("utf-8"), capture_output=True) + out = (proc.stdout + proc.stderr).decode("utf-8", "replace").strip() + if proc.returncode != 0: + raise RuntimeError(f"psql через ssh вернул {proc.returncode}:\n{out}") + if out: + print(f" psql: {out}", flush=True) + + +# --- накопитель карточек --------------------------------------------------- + + +@dataclass +class Sink: + """Батчами на прод (ssh+psql) или в локальный CSV при --dry-run.""" + + batch_id: str + started_at: str + query: str + batch_size: int + dry_run: bool + csv_path: Path + ssh_host: str + container: str + db_user: str + db_name: str + buffer: list[dict[str, Any]] = field(default_factory=list) + sent: int = 0 + skipped_non_numeric: int = 0 + + def add(self, lot: Any) -> None: + raw_id = str(getattr(lot, "source_id", "") or "") + try: + source_id = int(raw_id) # в БД bigint, у ScrapedLot — строка + except (TypeError, ValueError): + self.skipped_non_numeric += 1 + return + payload = lot.model_dump(mode="json") + self.buffer.append({ + "source_id": source_id, + "observed_at": datetime.now(timezone.utc).isoformat(), + "batch_id": self.batch_id, + "kind": "serp", + "url": payload.get("source_url"), + "price": payload.get("price_rub"), + "payload": json.dumps(payload, ensure_ascii=False), + }) + + def maybe_flush(self) -> None: + if len(self.buffer) >= self.batch_size: + self.flush() + + def flush(self) -> None: + if not self.buffer: + return + rows, self.buffer = self.buffer, [] + if self.dry_run: + fresh = not self.csv_path.exists() + with self.csv_path.open("a", encoding="utf-8", newline="") as fh: + if fresh: + fh.write("source_id,observed_at,batch_id,kind,url,price,payload\n") + fh.write(_csv_rows(rows)) + print(f" [dry-run] {len(rows)} строк → {self.csv_path}", flush=True) + else: + run_psql( + build_sql(self.batch_id, self.query, rows, self.started_at), + self.ssh_host, self.container, self.db_user, self.db_name, + ) + print(f" залито {len(rows)} строк в msk_raw.avito_cards", flush=True) + self.sent += len(rows) + + def finalize(self, notes: str) -> None: + self.flush() + if self.dry_run: + print(f" [dry-run] finalize: {notes}", flush=True) + return + run_psql( + build_finalize_sql(self.batch_id, self.query, notes), + self.ssh_host, self.container, self.db_user, self.db_name, + ) + + +# --- сбор ------------------------------------------------------------------ + + +def parse_page(scraper: AvitoScraper, html: str, url: str) -> tuple[int | None, list[Any]]: + count = scraper._extract_total_count(html) + lots = scraper._parse_html(html, "https://www.avito.ru") + if not lots and count: + raise Blocked("empty_page", f"0 карточек при счётчике {count}: {url}") + return count, lots + + +async def probe(loader: Loader, scraper: AvitoScraper, base_url: str, + lo: int | None, hi: int | None) -> tuple[int | None, list[Any]]: + url = build_url(base_url, 1, lo, hi) + html, _ = await loader.fetch(url) + return parse_page(scraper, html, url) + + +async def build_plan(loader: Loader, scraper: AvitoScraper, base_url: str, target: int, + cache: dict[tuple[int | None, int | None], list[Any]] + ) -> list[Corridor]: + """Адаптивная бисекция по цене; страница 1 каждого коридора кэшируется.""" + corridors: list[Corridor] = [] + + def emit(lo: int | None, hi: int | None, count: int | None, + truncated: bool, lots: list[Any]) -> None: + missed = max(0, (count or 0) - HARD_CAP) if truncated else 0 + c = Corridor(lo=lo, hi=hi, count=count, truncated=truncated, missed=missed) + corridors.append(c) + cache[(lo, hi)] = lots + flag = " TRUNCATED" if truncated else "" + print(f" коридор {c.label()} count={count} " + f"страниц={c.planned_pages()}{flag}", flush=True) + if truncated: + print(f" ВНИМАНИЕ: коридор {c.label()} не влезает в потолок " + f"{HARD_CAP}; заведомо не добрано ~{missed} объявлений", flush=True) + + async def find_upper(lo: int | None) -> int: + """Верхнюю границу открытого коридора ищем удвоением от разумного старта.""" + cand = max(int(lo or PRICE_FLOOR) * 2, PRICE_PROBE_START) + while cand < PRICE_CEIL: + cnt, _ = await probe(loader, scraper, base_url, cand, None) + print(f" проба хвоста pmin={cand:_} count={cnt}", flush=True) + if cnt is not None and cnt <= target: + return cand + cand *= 2 + return cand + + async def split(lo: int | None, hi: int | None, depth: int, + count: int | None, lots: list[Any]) -> None: + if count is None: + url = build_url(base_url, 1, lo, hi) + raise Blocked("empty_page", f"счётчик не прочитался: {url}") + if count <= target: + emit(lo, hi, count, False, lots) + return + if depth >= MAX_DEPTH or width_ratio(lo, hi) <= MIN_WIDTH_RATIO: + # Предохранитель: не молчим — помечаем truncated и считаем недобор. + emit(lo, hi, count, count > HARD_CAP, lots) + return + upper = hi if hi is not None else await find_upper(lo) + if hi is None: + tail_cnt, tail_lots = await probe(loader, scraper, base_url, upper, None) + emit(upper, None, tail_cnt, + bool(tail_cnt and tail_cnt > HARD_CAP), tail_lots) + mid = geometric_mid(lo, upper) + for sub_lo, sub_hi in ((lo, mid), (mid, upper)): + sub_cnt, sub_lots = await probe(loader, scraper, base_url, sub_lo, sub_hi) + print(f" проба {sub_lo or '-'}..{sub_hi} count={sub_cnt}", flush=True) + await split(sub_lo, sub_hi, depth + 1, sub_cnt, sub_lots) + + root_cnt, root_lots = await probe(loader, scraper, base_url, None, None) + print(f"Всего по базовому запросу: {root_cnt}", flush=True) + await split(None, None, 0, root_cnt, root_lots) + return corridors + + +async def collect(args: argparse.Namespace) -> int: + # avito_serp_ekb_only=False обязателен: с True парсер выбрасывает всё, где в + # URL нет /ekaterinburg/ — то есть все подмосковные слаги (serp.py:2154). + scraper = AvitoScraper( + SimpleNamespace(avito_serp_ekb_only=False), # type: ignore[arg-type] + target_city_slug="moskva", + ) + out_dir = Path(args.out_dir).resolve() + out_dir.mkdir(parents=True, exist_ok=True) + plan_path = out_dir / f"plan-{args.batch_id}.json" + csv_path = out_dir / f"cards-{args.batch_id}.csv" + started_at = datetime.now(timezone.utc).isoformat() + + plan: Plan | None = None + if args.resume: + if not plan_path.exists(): + print(f"--resume: плана нет — {plan_path}", file=sys.stderr) + return 1 + plan = Plan.load(plan_path) + done = sum(1 for c in plan.corridors if c.status == "done") + print(f"Resume по {plan_path}: коридоров {len(plan.corridors)}, " + f"готово {done}", flush=True) + + page_budget = None if args.full else args.measure + mode = "FULL" if args.full else f"MEASURE<={page_budget}" + print(f"Режим: {mode}; batch_id={args.batch_id}; delay={args.delay}s; " + f"target={args.target_count}; dry_run={args.dry_run}", flush=True) + + sink = Sink( + batch_id=args.batch_id, started_at=started_at, query=args.base_url, + batch_size=args.batch_size, dry_run=args.dry_run, csv_path=csv_path, + ssh_host=args.ssh_host, container=args.container, + db_user=args.db_user, db_name=args.db_name, + ) + cache: dict[tuple[int | None, int | None], list[Any]] = {} + total = 0 + stop_reason = "" + rc = 0 + loads = 0 + + async with Loader(args.delay, page_budget) as loader: + try: + if plan is None: + print("Строю план коридоров...", flush=True) + corridors = await build_plan(loader, scraper, args.base_url, + args.target_count, cache) + plan = Plan(base_url=args.base_url, target=args.target_count, + batch_id=args.batch_id, corridors=corridors, + created_at=started_at) + plan.save(plan_path) + print(f"План сохранён: {plan_path} ({len(corridors)} коридоров)", + flush=True) + + for corridor in plan.corridors: + if corridor.status == "done": + continue + pages = corridor.planned_pages() + print(f"Коридор {corridor.label()} count={corridor.count} " + f"страниц={pages} (с {corridor.pages_done + 1})", flush=True) + for page in range(corridor.pages_done + 1, pages + 1): + key = (corridor.lo, corridor.hi) + if page == 1 and key in cache: + lots = cache.pop(key) # страница 1 уже скачана при планировании + else: + url = build_url(args.base_url, page, corridor.lo, corridor.hi) + html, _ = await loader.fetch(url) + _, lots = parse_page(scraper, html, url) + for lot in lots: + sink.add(lot) + total += len(lots) + corridor.pages_done = page + print(f" стр.{page}/{pages}: карточек {len(lots)}, " + f"итого {total}", flush=True) + sink.maybe_flush() + plan.save(plan_path) + if not lots: + print(" пустая страница — конец коридора", flush=True) + break + corridor.status = "done" + plan.save(plan_path) + except BudgetExhausted: + stop_reason = "потолок --measure исчерпан" + print(f"Стоп: {stop_reason}", flush=True) + except Blocked as exc: + stop_reason = f"BLOCKED/{exc.reason}: {exc.detail}" + print(f"СТОП: {stop_reason}", file=sys.stderr, flush=True) + rc = 2 + finally: + loads = loader.loads + if plan is not None: + plan.save(plan_path) + + truncated = [c for c in (plan.corridors if plan else []) if c.truncated] + missed = sum(c.missed for c in truncated) + notes = "; ".join(x for x in [ + f"mode={mode}", f"loads={loads}", f"cards={total}", + f"skipped_non_numeric={sink.skipped_non_numeric}", + (f"truncated_corridors={len(truncated)} missed~{missed}" if truncated else ""), + stop_reason, + ] if x) + try: + sink.finalize(notes) + except Exception as exc: # noqa: BLE001 — не прятать исходную причину стопа + print(f"finalize провалился: {exc}", file=sys.stderr) + rc = rc or 1 + print(f"Готово. Загрузок: {loads}; карточек: {total}; отправлено: {sink.sent}; " + f"пропущено нечисловых source_id: {sink.skipped_non_numeric}; " + f"notes: {notes}", flush=True) + return rc + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + p = argparse.ArgumentParser( + prog="collect.py", + description="Ручной сбор SERP Авито (вторичка, Москва+МО) в прод-схему msk_raw.", + ) + p.add_argument("--base-url", default=DEFAULT_BASE_URL, + help="базовый URL выдачи (дефолт — вкладка владельца)") + p.add_argument("--measure", type=int, default=100, metavar="N", + help="режим замера: не больше N загрузок страниц (дефолт 100)") + p.add_argument("--full", action="store_true", + help="полный проход без потолка страниц (включается только явно)") + p.add_argument("--dry-run", action="store_true", + help="ничего не слать на прод, писать CSV локально") + p.add_argument("--resume", action="store_true", + help="продолжить по сохранённому плану коридоров") + p.add_argument("--delay", type=float, default=8.0, + help="пауза между загрузками, с (±20%% джиттер, дефолт 8.0)") + p.add_argument("--batch-size", type=int, default=1000, + help="карточек в одной заливке (дефолт 1000)") + p.add_argument("--target-count", type=int, default=1500, + help="целевой размер коридора; больше — делим (дефолт 1500)") + p.add_argument("--batch-id", default=None, + help="batch_id в msk_raw.batches (дефолт msk-serp-)") + p.add_argument("--out-dir", default=str(Path(__file__).resolve().parent / "runs"), + help="каталог плана/CSV") + p.add_argument("--ssh-host", default="selectel", help="ssh-хост прода") + p.add_argument("--container", default="tradein-postgres", + help="имя контейнера Postgres на проде") + p.add_argument("--db-user", default="tradein") + p.add_argument("--db-name", default="tradein") + args = p.parse_args(argv) + if args.batch_id is None: + args.batch_id = "msk-serp-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + if not _BATCH_ID_RE.match(args.batch_id): + p.error("--batch-id: допустимы только символы [A-Za-z0-9._-]") + if args.measure < 1: + p.error("--measure должен быть >= 1") + return args + + +def main(argv: list[str] | None = None) -> int: + return asyncio.run(collect(parse_args(argv))) + + +if __name__ == "__main__": + raise SystemExit(main()) From 476fb9b77a0148a80fd4de54de6d062610d31e81 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 8 Sep 2026 22:53:20 +0300 Subject: [PATCH 3/5] =?UTF-8?q?chore(#3051):=20=D1=81=D0=BD=D1=8F=D1=82?= =?UTF-8?q?=D1=8C=20=D0=BD=D0=BE=D0=B2=D1=8B=D0=B5=20=D1=82=D0=B5=D1=81?= =?UTF-8?q?=D1=82=D1=8B=20=E2=80=94=20=D0=B4=D0=BE=20=D0=B6=D0=B8=D0=B2?= =?UTF-8?q?=D0=BE=D0=B9=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5=D1=80=D0=BA=D0=B8?= =?UTF-8?q?=20=D1=84=D1=83=D0=BD=D0=BA=D1=86=D0=B8=D0=BE=D0=BD=D0=B0=D0=BB?= =?UTF-8?q?=D0=B0=20=D1=82=D0=B5=D1=81=D1=82=D1=8B=20=D0=BD=D0=B5=20=D0=BF?= =?UTF-8?q?=D0=B8=D1=88=D0=B5=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit В существующем test_rosreestr_dedup_key оставлена только правка, без которой он падает после параметризации (литералы region_code=66 / doc_type='ДКП' ушли из SQL живого импорта). Новый тест-файл и добавленная функция сняты: поведение ещё не проверено на живом импорте, тест зафиксировал бы догадку. --- .../test_3051_deals_doc_type_region_param.py | 148 ------------------ .../backend/tests/test_rosreestr_dedup_key.py | 20 +-- 2 files changed, 1 insertion(+), 167 deletions(-) delete mode 100644 tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py diff --git a/tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py b/tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py deleted file mode 100644 index f24ebaae..00000000 --- a/tradein-mvp/backend/tests/test_3051_deals_doc_type_region_param.py +++ /dev/null @@ -1,148 +0,0 @@ -"""#3051 п.3 + п.6: doc_type в deals и параметризация региона/типа документа на импорте. - -ЗАЧЕМ: для Москвы в источнике за 2024 лежат 30 627 ДДУ с медианой 112 743 ₽/м² против -107 005 ДКП с медианой 256 250 — это цены котлована. Тип сделки терялся при загрузке -(в deals не приезжал вовсе), а фильтры `region_code = 66` и `doc_type = 'ДКП'` стояли -литералами, поэтому регион 77 было не подключить, не правя код. Тесты фиксируют, что -параметры реально доезжают до bind-словаря, дефолты воспроизводят прежнее поведение, -а doc_type пишется в deals. - -Чисто-юнит: ассертим SQL-текст import_rosreestr_dkp (inspect.getsource) + текст миграции -288, плюс вызываем функцию с моком Session, чтобы увидеть фактические bind-параметры. -Живой FDW / БД не нужны. -""" - -import inspect -import os -import re -from pathlib import Path -from typing import Any -from unittest.mock import MagicMock - -# Как в test_rosreestr_dedup_key.py: Settings требует DATABASE_URL до импорта. -os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") - -from app.services import scheduler - -_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql" -_MIGRATION_288 = _SQL_DIR / "288_deals_doc_type.sql" -_IMPORT_SRC = inspect.getsource(scheduler.import_rosreestr_dkp) - - -# ── SQL-форма импорта ──────────────────────────────────────────────────────── - - -def test_import_filters_region_and_doc_type_via_bind_params() -> None: - """Оба фильтра — bind-параметры, литералов 66 / 'ДКП' в SQL больше нет.""" - assert "region_code = CAST(:region_code AS int)" in _IMPORT_SRC - assert "doc_type = ANY(CAST(:doc_types AS text[]))" in _IMPORT_SRC - assert "WHERE region_code = 66" not in _IMPORT_SRC - assert "AND doc_type = 'ДКП'" not in _IMPORT_SRC - # psycopg v3: ни один bind-параметр не приводится через :name::type. - assert not re.search(r":\w+::", _IMPORT_SRC) - - -def test_import_writes_doc_type_into_deals() -> None: - """doc_type едет из SELECT в список колонок INSERT, в VALUES и в ON CONFLICT.""" - assert "doc_type AS doc_type" in _IMPORT_SRC - assert "deal_date, doc_type" in _IMPORT_SRC - assert "CAST(:doc_type AS text)" in _IMPORT_SRC - # Переопубликованный квартал может исправить тип документа — обновляем. - assert "doc_type = EXCLUDED.doc_type" in _IMPORT_SRC - assert "IS DISTINCT FROM EXCLUDED.doc_type" in _IMPORT_SRC - assert '"doc_type": row["doc_type"]' in _IMPORT_SRC - - -def test_import_keeps_dedup_key_shape_unchanged() -> None: - """Ключ дедупа НЕ зависит от типа документа — иначе осиротели бы старые строки. - - id уникален в источнике сам по себе; префикс 'ros:dkp:' остаётся историческим - именем формата (см. миграцию 077, которая уже конвертировала все строки). - """ - assert "'ros:dkp:' || CAST(id AS text) AS dedup_hash" in _IMPORT_SRC - assert "ON CONFLICT (dedup_hash) DO UPDATE SET" in _IMPORT_SRC - - -# ── фактические bind-параметры (вызов с моком) ─────────────────────────────── - - -def _capture_select_params(params: dict[str, Any]) -> dict[str, Any]: - """Прогоняет import_rosreestr_dkp до первого SELECT из FDW и возвращает его binds. - - Мок Session: cleanup-DELETE и _resume_dkp_cursor получают пустые результаты, - первый батч — пустой список, поэтому цикл завершается сразу после SELECT. - """ - captured: list[dict[str, Any]] = [] - - def _execute(stmt: Any, binds: Any = None, *a: Any, **kw: Any) -> MagicMock: - sql = str(stmt) - if "FROM gendesign_rosreestr_deals" in sql and isinstance(binds, dict): - captured.append(binds) - result = MagicMock() - result.fetchall.return_value = [] - result.fetchone.return_value = None - result.mappings.return_value.all.return_value = [] - result.scalar.return_value = None - return result - - db = MagicMock() - db.execute.side_effect = _execute - - monkeys = { - "is_cancelled": False, - } - orig_is_cancelled = scheduler.runs_mod.is_cancelled - orig_mark_done = scheduler.runs_mod.mark_done - orig_heartbeat = scheduler.kit_runs.update_heartbeat - try: - scheduler.runs_mod.is_cancelled = lambda *a, **kw: monkeys["is_cancelled"] # type: ignore[assignment] - scheduler.runs_mod.mark_done = lambda *a, **kw: None # type: ignore[assignment] - scheduler.kit_runs.update_heartbeat = lambda *a, **kw: None # type: ignore[assignment] - scheduler.import_rosreestr_dkp(db, 1, params) - finally: - scheduler.runs_mod.is_cancelled = orig_is_cancelled # type: ignore[assignment] - scheduler.runs_mod.mark_done = orig_mark_done # type: ignore[assignment] - scheduler.kit_runs.update_heartbeat = orig_heartbeat # type: ignore[assignment] - - assert captured, "SELECT из gendesign_rosreestr_deals не выполнился" - return captured[0] - - -def test_defaults_preserve_current_production_behaviour() -> None: - """Без параметров — ровно прежний скоуп: Свердловская обл., только ДКП.""" - binds = _capture_select_params({}) - assert binds["region_code"] == 66 - assert binds["doc_types"] == ["ДКП"] - - -def test_params_override_region_and_doc_types() -> None: - """Москва + оба типа документа приходят из params, а не из кода.""" - binds = _capture_select_params({"region_code": 77, "doc_types": ["ДКП", "ДДУ"]}) - assert binds["region_code"] == 77 - assert binds["doc_types"] == ["ДКП", "ДДУ"] - - -def test_region_code_accepts_string_from_jsonb_params() -> None: - """default_params прилетают из jsonb — строковый регион не должен ломать CAST.""" - binds = _capture_select_params({"region_code": "77"}) - assert binds["region_code"] == 77 - assert isinstance(binds["region_code"], int) - - -# ── миграция 288 ───────────────────────────────────────────────────────────── - - -def test_migration_288_adds_column_backfills_and_sets_schedule_region() -> None: - """288: idempotent-колонка, бэкфилл только rosreestr-строк, явный регион в расписании.""" - assert _MIGRATION_288.is_file(), f"missing migration: {_MIGRATION_288}" - sql = _MIGRATION_288.read_text("utf-8") - assert "ALTER TABLE deals ADD COLUMN IF NOT EXISTS doc_type text" in sql - # Бэкфилл корректен ровно потому, что всё загруженное прошло фильтр ДКП. - assert "SET doc_type = 'ДКП'" in sql - assert "WHERE source = 'rosreestr'" in sql - assert "AND doc_type IS NULL" in sql - # Явный region_code в расписании вместо неявного дефолта в коде. - assert "UPDATE scrape_schedules" in sql - assert "'{\"region_code\": 66}'::jsonb" in sql - assert "source = 'rosreestr_dkp_import'" in sql - assert "BEGIN;" in sql and "COMMIT;" in sql diff --git a/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py b/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py index 145101c0..275a8a47 100644 --- a/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py +++ b/tradein-mvp/backend/tests/test_rosreestr_dedup_key.py @@ -156,7 +156,7 @@ def test_migration_077_shared_filter_matches_live_import() -> None: ЕКБ-строки под EKB-only scope того времени; живой импорт расширен на всю Свердловскую область, поэтому city-фильтр из живого импорта СНЯТ намеренно; - region_code / doc_type: в живом импорте это ПАРАМЕТРЫ (#3051), их совпадение - с 077 проверяется через дефолты — см. тест ниже. + с 077 проверяется дефолтами params.get(..., 66) / ["ДКП"] в самом импорте. Остальные клозы обязаны совпадать байт-в-байт, иначе 077 конвертировал бы не тот набор строк. """ @@ -174,24 +174,6 @@ def test_migration_077_shared_filter_matches_live_import() -> None: assert "doc_type = 'ДКП'" in sql -def test_live_import_parametrized_region_and_doc_type_keep_077_defaults() -> None: - """#3051: регион и типы документов — bind-параметры, но ДЕФОЛТЫ = скоуп 077. - - Литералы `region_code = 66` / `doc_type = 'ДКП'` ушли из SQL живого импорта - (иначе регион 77 не подключить), но дефолты обязаны воспроизводить ровно тот - набор строк, который 077 когда-то backfill'ил, — иначе штатный прогон молча - сменил бы скоуп. - """ - # В SQL — параметры, не литералы. - assert "region_code = CAST(:region_code AS int)" in _IMPORT_SRC - assert "doc_type = ANY(CAST(:doc_types AS text[]))" in _IMPORT_SRC - assert "region_code = 66" not in _IMPORT_BODY - assert "doc_type = 'ДКП'" not in _IMPORT_BODY.split("params.get", 1)[-1].split("SELECT", 1)[0] - # Дефолты — прежнее поведение. - assert 'params.get("region_code", 66)' in _IMPORT_SRC - assert 'params.get("doc_types") or ["ДКП"]' in _IMPORT_SRC - - def test_live_import_dropped_ekb_city_filter() -> None: """Живой импорт БОЛЬШЕ не фильтрует по городу — Mera расширена на всю обл. 66. From bde38cdce750e9f6d15502ba0a300195b60c4aef Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 8 Sep 2026 22:54:44 +0300 Subject: [PATCH 4/5] =?UTF-8?q?fix(msk-collector):=20--resume=20=D0=B1?= =?UTF-8?q?=D0=B5=D1=80=D1=91=D1=82=20base=5Furl=20=D0=B8=D0=B7=20=D1=81?= =?UTF-8?q?=D0=BE=D1=85=D1=80=D0=B0=D0=BD=D1=91=D0=BD=D0=BD=D0=BE=D0=B3?= =?UTF-8?q?=D0=BE=20=D0=BF=D0=BB=D0=B0=D0=BD=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Коридоры считаются под конкретный URL выдачи. При --resume код брал свежий args.base_url, поэтому запуск без повтора --base-url молча качал другую выдачу под тем же batch_id. Теперь URL из плана, расхождение с аргументом — явная ошибка вместо тихого выбора одного из двух. --- .../scripts/local-avito-msk/collect.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/tradein-mvp/scripts/local-avito-msk/collect.py b/tradein-mvp/scripts/local-avito-msk/collect.py index 2432cc86..8df91f48 100644 --- a/tradein-mvp/scripts/local-avito-msk/collect.py +++ b/tradein-mvp/scripts/local-avito-msk/collect.py @@ -527,13 +527,24 @@ async def collect(args: argparse.Namespace) -> int: print(f"Resume по {plan_path}: коридоров {len(plan.corridors)}, " f"готово {done}", flush=True) + # Resume: URL берём из сохранённого плана, а не из CLI — коридоры посчитаны + # именно под него. Расхождение = молчаливая заливка чужой выдачи под тем же + # batch_id, поэтому это ошибка, а не тихий приоритет одного из двух. + if plan is not None and plan.base_url != args.base_url: + raise SystemExit( + "--resume: план построен для другого URL." + f" В плане {plan.base_url}, в аргументах {args.base_url}." + " Убери --base-url (возьмётся из плана) либо начни новый batch_id." + ) + base_url = plan.base_url if plan is not None else args.base_url + page_budget = None if args.full else args.measure mode = "FULL" if args.full else f"MEASURE<={page_budget}" print(f"Режим: {mode}; batch_id={args.batch_id}; delay={args.delay}s; " f"target={args.target_count}; dry_run={args.dry_run}", flush=True) sink = Sink( - batch_id=args.batch_id, started_at=started_at, query=args.base_url, + batch_id=args.batch_id, started_at=started_at, query=base_url, batch_size=args.batch_size, dry_run=args.dry_run, csv_path=csv_path, ssh_host=args.ssh_host, container=args.container, db_user=args.db_user, db_name=args.db_name, @@ -548,9 +559,9 @@ async def collect(args: argparse.Namespace) -> int: try: if plan is None: print("Строю план коридоров...", flush=True) - corridors = await build_plan(loader, scraper, args.base_url, + corridors = await build_plan(loader, scraper, base_url, args.target_count, cache) - plan = Plan(base_url=args.base_url, target=args.target_count, + plan = Plan(base_url=base_url, target=args.target_count, batch_id=args.batch_id, corridors=corridors, created_at=started_at) plan.save(plan_path) @@ -568,7 +579,7 @@ async def collect(args: argparse.Namespace) -> int: if page == 1 and key in cache: lots = cache.pop(key) # страница 1 уже скачана при планировании else: - url = build_url(args.base_url, page, corridor.lo, corridor.hi) + url = build_url(base_url, page, corridor.lo, corridor.hi) html, _ = await loader.fetch(url) _, lots = parse_page(scraper, html, url) for lot in lots: From e9370818ea231b1bd9b5b018ea036edef5e40705 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 8 Sep 2026 22:58:38 +0300 Subject: [PATCH 5/5] =?UTF-8?q?fix(#3051):=20lock=5Ftimeout=20=D0=B2=20?= =?UTF-8?q?=D0=BC=D0=B8=D0=B3=D1=80=D0=B0=D1=86=D0=B8=D0=B8=20288=20?= =?UTF-8?q?=E2=80=94=20=D0=B3=D0=B5=D0=B9=D1=82=20CI=20#2752?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ALTER TABLE deals ADD COLUMN без SET LOCAL lock_timeout встаёт в очередь за чужой сессией и уводит за собой запросы приложения. Лучше упасть по таймауту и повторить деплой. --- tradein-mvp/backend/data/sql/288_deals_doc_type.sql | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tradein-mvp/backend/data/sql/288_deals_doc_type.sql b/tradein-mvp/backend/data/sql/288_deals_doc_type.sql index 2e002720..113caa48 100644 --- a/tradein-mvp/backend/data/sql/288_deals_doc_type.sql +++ b/tradein-mvp/backend/data/sql/288_deals_doc_type.sql @@ -43,6 +43,10 @@ BEGIN; +-- #2752: блокирующий DDL не должен вставать в очередь за чужой сессией и уводить +-- за собой запросы приложения — лучше упасть по таймауту и повторить деплой. +SET LOCAL lock_timeout = '5s'; + -- A: колонка ALTER TABLE deals ADD COLUMN IF NOT EXISTS doc_type text;