Merge pull request 'chore(tradein/scraper): cut wasteful scraping load - cian_full_load tempo + zero-yield jobs' (#2633) from chore/tradein-cut-wasteful-scraping into main

tradein-mvp/backend/data/sql/206_scrape_schedules_cut_wasteful_load.sql

@@ -0,0 +1,136 @@
+-- 206_scrape_schedules_cut_wasteful_load.sql
+-- Срезать бесполезную нагрузку на источники (=нагрузку на единственный живой общий
+-- прокси: scrape_proxies enabled=true AND provider_affinity='any' → ровно 1 узел
+-- asocks-mobile-2 на момент этой миграции; asocks-residential-1 закреплён отдельно
+-- за domclick). Только UPDATE scrape_schedules.default_params / .enabled — код
+-- скраперов НЕ меняется. Все цифры ниже — прод, scrape_runs.counters, 30 дней
+-- (2026-08-02), проверено read-only перед написанием файла.
+--
+-- 1) cian_full_load — 110.0 ч из 30-дневного окна (29 runs), доминирующий потребитель
+--    прокси-времени в системе. Текущий default_params подтверждён на проде:
+--    concurrency=5, request_delay_sec=4.0 → эффективный интервал 4.0/5=0.8с между
+--    запросами. 30-дневные counters: unique_fetched=44011, saved_inserted=1682,
+--    saved_updated=32862 — сигнал реальный (НЕ нулевой выхлоп), но темп избыточен
+--    относительно ценности. concurrency 5→2, request_delay_sec 4.0→6.0 даёт
+--    эффективный интервал 6.0/2=3.0с (в 3.75 раза медленнее); interval_days 1→3
+--    (daily → раз в 3 дня) сокращает число прогонов в 3 раза. Совместно — падение
+--    запросов к Циану на порядок, в духе оценки задачи (~11-16 тыс./сутки → ~1-1.5 тыс.).
+--    Потеря свежести: ~1095 price-update/сутки в среднем откладываются на срок до
+--    2 суток между прогонами — не исчезают, детектируются позже; оценщик использует
+--    LISTINGS_FRESH_DAYS=14, лаг в 1-2 дня внутри этого окна некритичен.
+BEGIN;
+
+UPDATE scrape_schedules
+SET default_params = COALESCE(default_params, '{}'::jsonb)
+    || '{"concurrency": 2, "request_delay_sec": 6.0, "interval_days": 3}'::jsonb
+WHERE source = 'cian_full_load';
+
+-- 2) avito_full_load — request_delay_sec=1.0, самая агрессивная настройка в конфиге
+--    (подтверждено). 30-дневные scrape_runs.status: 20/30 banned мгновенно
+--    (done_buckets=[], 0 fetched), 9/30 failed (0 fetched), 1/30 done (357 inserted /
+--    3397 updated). avito_full_load_exhaustive — тот же traversal, но УЖЕ на
+--    request_delay_sec=7.0 + interval_days=7 — тем не менее 4/4 runs banned за 30д:
+--    сама скорость запроса не единственная причина бана (вероятно паттерн полного
+--    обхода всех room×price buckets), но замедление всё равно валидно снижает
+--    бесполезную нагрузку на прокси при каждой попытке. request_delay_sec 1.0→7.0
+--    (уравнено с городскими развёртками) и interval_days 1→7 (недельный такт, решение
+--    по данным — см. § "рассмотри и перевод на недельный такт, но реши по данным":
+--    ежедневный прогон 29 из последних 30 раз не даёт НИ ОДНОЙ новой/обновлённой
+--    строки, недельный такт не теряет свежести, которой и так нет).
+UPDATE scrape_schedules
+SET default_params = COALESCE(default_params, '{}'::jsonb)
+    || '{"request_delay_sec": 7.0, "interval_days": 7}'::jsonb
+WHERE source = 'avito_full_load';
+
+-- 3a) yandex_address_backfill (отдельная джоба) — 30-дневные counters: checked=6000,
+--     saved=4 (0.07%), errors=406. Ненулевой поток (4 записи/мес) — по границам задачи
+--     НЕ выключаем полностью, переводим на недельный такт (1→7).
+UPDATE scrape_schedules
+SET default_params = COALESCE(default_params, '{}'::jsonb)
+    || '{"interval_days": 7}'::jsonb
+WHERE source = 'yandex_address_backfill';
+
+-- 3b) address-enrich ФАЗА ВНУТРИ yandex_city_sweep (ЕКБ) — run_yandex_city_sweep()
+--     принимает enrich_address: bool (orchestration/pipeline.py:1870), scheduler.py:545
+--     читает его ИМЕННО из default_params.get("enrich_address", True) — управляется
+--     параметром, правка кода НЕ требуется (в отличие от того, если бы флаг был
+--     захардкожен — этого на проверке НЕТ, поэтому трогаем только данные).
+--     30-дневные counters ТОЛЬКО для source='yandex_city_sweep' (ЕКБ, без city-суффикса):
+--     address_attempted=5829, address_enriched=0, address_failed=40 — фаза полностью
+--     впустую. ВАЖНО: 5 областных yandex_city_sweep_<city> за те же 30 дней показывают
+--     address_attempted=0 (фаза там и так не тратит запросы — не из-за enrich_address,
+--     а потому что WHERE-условие backfill'а — address IS NOT NULL AND NOT ~ ',\s*\d+' —
+--     там просто ничего не находит) — их НЕ трогаем, нечего чинить по данным.
+UPDATE scrape_schedules
+SET default_params = COALESCE(default_params, '{}'::jsonb)
+    || '{"enrich_address": false}'::jsonb
+WHERE source = 'yandex_city_sweep';
+
+-- 4) house_imv_backfill — 30-дневные counters: checked=1500, saved=44 (2.9%),
+--    errors=1301 (87%), skipped=155. Ненулевой поток — НЕ выключаем (граница задачи),
+--    втрое снижаем частоту (1→3 дня) до отдельного разбора причины 87%-ошибок —
+--    сокращает объём бесполезных попыток пропорционально при сохранении прогресса
+--    по валидным 13%.
+UPDATE scrape_schedules
+SET default_params = COALESCE(default_params, '{}'::jsonb)
+    || '{"interval_days": 3}'::jsonb
+WHERE source = 'house_imv_backfill';
+
+-- 5) domclick_detail_backfill — 30-дневные counters: attempted=491, enriched=0,
+--    failed=431, blocked=60 — 100% впустую (0 обогащений вообще), включая на
+--    ДЕДИКЕЙТЕД прокси (scrape_proxies.provider_affinity='domclick',
+--    asocks-residential-1) — тот прокси тоже палится в никуда. Полностью выключаем
+--    до починки (единственный пункт этой миграции, где нулевой выход подтверждён
+--    буквально — enabled=false оправдан границей задачи).
+UPDATE scrape_schedules
+SET enabled = false
+WHERE source = 'domclick_detail_backfill';
+
+-- 6) yandex_newbuilding_sweep — 30/30 runs status=done, но rows_inserted=0 во ВСЕХ
+--    (failed_resolve стабильно ~4-5/run, backlog pending растёт 351→367 за 30д —
+--    джоба не успевает и не разбирает очередь). interval_days 1→7.
+UPDATE scrape_schedules
+SET default_params = COALESCE(default_params, '{}'::jsonb)
+    || '{"interval_days": 7}'::jsonb
+WHERE source = 'yandex_newbuilding_sweep';
+
+-- 7) Областные развёртки (15 job'ов = 5 городов × {avito,cian,yandex}_city_sweep_<city>,
+--    см. миграцию 179) — ежедневно → раз в 3 дня. Независимая проверка (НЕ те же цифры,
+--    что в задаче — посчитано отдельно по listings_snapshots за последние 14 дней для
+--    ~1245 активных объявлений в 5 областных городах): 4 события изменения цены на
+--    2282 снапшот-строки = ~0.023%/сутки — НИЖЕ заявленных в задаче 0.15%/сутки,
+--    подтверждает избыточность daily-такта. cian_city_sweep (ЕКБ, БЕЗ суффикса города,
+--    id=128) и его newbuilding_only-логику НЕ трогаем (недавно правились, вне периметра
+--    этой миграции). Потеря свежести: при трёхдневном такте цена/новый лот в областном
+--    городе детектируется с лагом до 2 суток — при ~0.02-0.15%/сутки волатильности и
+--    LISTINGS_FRESH_DAYS=14 эффект на оценку пренебрежим.
+UPDATE scrape_schedules
+SET default_params = COALESCE(default_params, '{}'::jsonb)
+    || '{"interval_days": 3}'::jsonb
+WHERE source IN (
+    'avito_city_sweep_nizhniy_tagil',
+    'avito_city_sweep_kamensk_uralskiy',
+    'avito_city_sweep_pervouralsk',
+    'avito_city_sweep_verkhnyaya_pyshma',
+    'avito_city_sweep_serov',
+    'cian_city_sweep_nizhniy_tagil',
+    'cian_city_sweep_kamensk_uralskiy',
+    'cian_city_sweep_pervouralsk',
+    'cian_city_sweep_verkhnyaya_pyshma',
+    'cian_city_sweep_serov',
+    'yandex_city_sweep_nizhniy_tagil',
+    'yandex_city_sweep_kamensk_uralskiy',
+    'yandex_city_sweep_pervouralsk',
+    'yandex_city_sweep_verkhnyaya_pyshma',
+    'yandex_city_sweep_serov'
+);
+
+-- НЕ тронуто (сознательно, данные не подтвердили действие):
+--   avito_detail_backfill (2295 attempted / 494 enriched = 78% брака, но 494
+--     обогащения/мес — реальный, не близкий к нулю поток; вне "Предлагаемого набора"
+--     задачи, полноценно вне периметра этой миграции).
+--   cian_city_sweep (ЕКБ) / newbuilding_only — явный запрет задачи.
+--   yandex_city_sweep_<city> (5 областных) enrich_address — address_attempted=0 там,
+--     нечего выключать.
+
+COMMIT;

tradein-mvp/backend/tests/test_migration_206_scrape_schedules_cut_wasteful_load.py

@@ -0,0 +1,246 @@
+"""Static guards for migration 206 (срезать бесполезную нагрузку на источники —
+scrape_schedules.default_params / .enabled, единственный живой общий прокси).
+
+Прод применяет data/sql построчно строго (ON_ERROR_STOP). Полный DB-прогон требует
+живой БД; здесь фиксируем структурные инварианты: транзакционность, отсутствие DDL,
+отсутствие psycopg CAST-ловушки, jsonb-мердж (не перезапись — соседние ключи
+default_params должны сохраниться), точные target-source списки и, главное,
+regression-guard'ы на то, что миграция НЕ должна трогать:
+  1. cian_city_sweep (ЕКБ, БЕЗ суффикса города) и его newbuilding_only-логику
+     (недавно правились — явный запрет задачи).
+  2. avito_full_load_exhaustive (соседняя джоба с похожим именем — легко зацепить
+     случайным LIKE/подстрокой).
+  3. 5 областных yandex_city_sweep_<city> — address_attempted=0 там (нечего чинить),
+     enrich_address:false ставим ТОЛЬКО на 'yandex_city_sweep' (ЕКБ).
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_206 = _SQL_DIR / "206_scrape_schedules_cut_wasteful_load.sql"
+
+_OBLAST_SOURCES = (
+    "avito_city_sweep_nizhniy_tagil",
+    "avito_city_sweep_kamensk_uralskiy",
+    "avito_city_sweep_pervouralsk",
+    "avito_city_sweep_verkhnyaya_pyshma",
+    "avito_city_sweep_serov",
+    "cian_city_sweep_nizhniy_tagil",
+    "cian_city_sweep_kamensk_uralskiy",
+    "cian_city_sweep_pervouralsk",
+    "cian_city_sweep_verkhnyaya_pyshma",
+    "cian_city_sweep_serov",
+    "yandex_city_sweep_nizhniy_tagil",
+    "yandex_city_sweep_kamensk_uralskiy",
+    "yandex_city_sweep_pervouralsk",
+    "yandex_city_sweep_verkhnyaya_pyshma",
+    "yandex_city_sweep_serov",
+)
+
+
+def _sql() -> str:
+    return _MIGRATION_206.read_text(encoding="utf-8")
+
+
+def _executable_sql() -> str:
+    """SQL без построчных `--`-комментариев — только исполняемый код."""
+    lines = []
+    for raw in _sql().splitlines():
+        code = raw.split("--", 1)[0]
+        if code.strip():
+            lines.append(code)
+    return "\n".join(lines)
+
+
+def _flat(text: str) -> str:
+    return re.sub(r"\s+", " ", text).strip().lower()
+
+
+def test_migration_206_exists() -> None:
+    assert _MIGRATION_206.exists(), f"missing migration: {_MIGRATION_206}"
+
+
+def test_migration_206_is_transactional() -> None:
+    sql = _sql()
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+    # BEGIN должен предшествовать первому UPDATE, COMMIT — идти последним.
+    # (сверяем по исполняемому SQL — header-комментарий упоминает "UPDATE
+    # scrape_schedules.default_params" текстом раньше самого BEGIN;)
+    executable = _executable_sql()
+    assert executable.index("BEGIN;") < executable.index("UPDATE scrape_schedules")
+    assert executable.rindex("COMMIT;") > executable.rindex("UPDATE scrape_schedules")
+
+
+def test_migration_206_no_ddl() -> None:
+    """Только UPDATE данных (default_params/enabled) — никакого ALTER/CREATE/DROP
+    TABLE/TRUNCATE. Код скраперов не меняется в этой миграции."""
+    flat = _flat(_executable_sql())
+    assert "alter table" not in flat
+    assert "create table" not in flat
+    assert "drop table" not in flat
+    assert "truncate" not in flat
+    assert "drop function" not in flat
+    assert "create or replace function" not in flat
+
+
+def test_migration_206_no_psycopg_cast_trap() -> None:
+    assert not re.search(r":\w+::", _sql())
+
+
+def test_migration_206_uses_jsonb_merge_not_overwrite() -> None:
+    """Все default_params-обновления идут через COALESCE(default_params, '{}'::jsonb)
+    || '{...}'::jsonb — мердж, НЕ прямая перезапись (иначе стёрлись бы соседние ключи
+    вроде concurrency/detail_top_n/price_cap_per_bucket/batch_size/radius_m/city)."""
+    flat = _flat(_executable_sql())
+    merge_count = flat.count("set default_params = coalesce(default_params, '{}'::jsonb) ||")
+    # 7 UPDATE трогают default_params через merge (cian_full_load, avito_full_load,
+    # yandex_address_backfill, yandex_city_sweep, house_imv_backfill,
+    # yandex_newbuilding_sweep, oblast-15) — domclick_detail_backfill трогает
+    # только enabled (boolean column), не default_params.
+    assert merge_count == 7
+    # Regression guard: нигде нет прямой перезаписи default_params = '{...}'::jsonb
+    # (без COALESCE/||) — это стёрло бы соседние ключи.
+    assert "set default_params = '{" not in flat
+
+
+def test_migration_206_cian_full_load_params() -> None:
+    flat = _flat(_executable_sql())
+    assert '\'{"concurrency": 2, "request_delay_sec": 6.0, "interval_days": 3}\'::jsonb' in flat
+    assert "where source = 'cian_full_load'" in flat
+
+
+def test_migration_206_avito_full_load_params() -> None:
+    flat = _flat(_executable_sql())
+    assert '\'{"request_delay_sec": 7.0, "interval_days": 7}\'::jsonb' in flat
+    assert "where source = 'avito_full_load'" in flat
+
+
+def test_migration_206_does_not_touch_avito_full_load_exhaustive() -> None:
+    """Regression-guard: 'avito_full_load' — точное совпадение source в WHERE,
+    НЕ подстрока 'avito_full_load_exhaustive' (соседняя джоба с похожим именем,
+    уже на request_delay_sec=7.0/interval_days=7 — трогать её эта миграция не должна,
+    иначе повторное указание тех же ключей маскирует regression, если имена
+    когда-нибудь разъедутся). Header-комментарий ССЫЛАЕТСЯ на avito_full_load_exhaustive
+    как на контекст (её собственные 30-дневные counters), но НЕ в исполняемом SQL."""
+    assert "avito_full_load_exhaustive" not in _executable_sql()
+
+
+def test_migration_206_yandex_address_backfill_weekly_not_disabled() -> None:
+    """4 записи/мес — ненулевой поток, граница задачи запрещает полное выключение:
+    interval_days:7, БЕЗ enabled=false для этого source."""
+    flat = _flat(_executable_sql())
+    assert (
+        "set default_params = coalesce(default_params, '{}'::jsonb) "
+        "|| '{\"interval_days\": 7}'::jsonb where source = 'yandex_address_backfill'" in flat
+    )
+    assert "enabled = false where source = 'yandex_address_backfill'" not in flat
+
+
+def test_migration_206_yandex_city_sweep_disables_address_enrich_ekb_only() -> None:
+    """enrich_address:false — ТОЛЬКО на 'yandex_city_sweep' (ЕКБ, address_attempted=5829/
+    address_enriched=0 за 30д). Параметр читается scheduler.py:545
+    bool(params.get("enrich_address", True)) — код НЕ меняется."""
+    flat = _flat(_executable_sql())
+    assert "'{\"enrich_address\": false}'::jsonb" in flat
+    assert "where source = 'yandex_city_sweep'" in flat
+    # Ровно один литерал 'yandex_city_sweep' без city-суффикса среди WHERE-таргетов
+    # (bare source, не 'yandex_city_sweep_nizhniy_tagil' и т.д.).
+    bare_matches = re.findall(r"where source = 'yandex_city_sweep'", flat)
+    assert len(bare_matches) == 1
+
+
+def test_migration_206_does_not_touch_oblast_yandex_address_enrich() -> None:
+    """Regression-guard: 5 областных yandex_city_sweep_<city> НЕ фигурируют рядом с
+    enrich_address — address_attempted=0 там за 30д (нечего чинить по данным)."""
+    sql = _sql()
+    enrich_idx = sql.index('"enrich_address": false')
+    # В statement с enrich_address нет city-суффиксов областных yandex sweep'ов.
+    stmt_start = sql.rindex("UPDATE scrape_schedules", 0, enrich_idx)
+    stmt_end = sql.index(";", enrich_idx)
+    stmt = sql[stmt_start:stmt_end]
+    for city in ("nizhniy_tagil", "kamensk_uralskiy", "pervouralsk", "verkhnyaya_pyshma", "serov"):
+        assert f"yandex_city_sweep_{city}" not in stmt
+
+
+def test_migration_206_house_imv_backfill_throttled_not_disabled() -> None:
+    """87% ошибок, но 44 saved/мес (ненулевой поток) — снижаем частоту (interval_days:3),
+    НЕ выключаем полностью (граница задачи)."""
+    flat = _flat(_executable_sql())
+    assert (
+        "set default_params = coalesce(default_params, '{}'::jsonb) "
+        "|| '{\"interval_days\": 3}'::jsonb where source = 'house_imv_backfill'" in flat
+    )
+    assert "enabled = false where source = 'house_imv_backfill'" not in flat
+
+
+def test_migration_206_domclick_detail_backfill_fully_disabled() -> None:
+    """0 обогащений из 491 попытки за 30д (буквально нулевой выход) — единственный
+    пункт, где enabled=false оправдан границей задачи."""
+    flat = _flat(_executable_sql())
+    assert "set enabled = false" in flat
+    assert "where source = 'domclick_detail_backfill'" in flat
+
+
+def test_migration_206_yandex_newbuilding_sweep_weekly() -> None:
+    flat = _flat(_executable_sql())
+    assert (
+        "set default_params = coalesce(default_params, '{}'::jsonb) "
+        "|| '{\"interval_days\": 7}'::jsonb where source = 'yandex_newbuilding_sweep'" in flat
+    )
+
+
+def test_migration_206_targets_exactly_fifteen_oblast_sources() -> None:
+    flat = _flat(_executable_sql())
+    for source in _OBLAST_SOURCES:
+        assert f"'{source}'" in flat, f"missing oblast target: {source}"
+    quoted = re.findall(
+        r"'((?:avito|cian|yandex)_city_sweep_[a-z_]+)'",
+        flat,
+    )
+    assert sorted(set(quoted)) == sorted(_OBLAST_SOURCES)
+    assert len(quoted) == len(_OBLAST_SOURCES) == 15
+
+
+def test_migration_206_oblast_update_uses_interval_days_3() -> None:
+    sql = _sql()
+    where_idx = sql.index("WHERE source IN (")
+    stmt_start = sql.rindex("UPDATE scrape_schedules", 0, where_idx)
+    stmt = sql[stmt_start:where_idx]
+    flat_stmt = _flat(stmt)
+    assert "|| '{\"interval_days\": 3}'::jsonb" in flat_stmt
+
+
+def test_migration_206_does_not_touch_ekaterinburg_cian_city_sweep() -> None:
+    """Regression-guard: 'cian_city_sweep' (ЕКБ, БЕЗ суффикса города) и
+    newbuilding_only НЕ должны фигурировать в исполняемом SQL — явный запрет задачи,
+    недавно правились отдельно (footer-комментарий упоминает newbuilding_only как
+    объяснение "НЕ тронуто" — это ожидаемо и не проверяется здесь)."""
+    executable = _executable_sql()
+    assert "newbuilding_only" not in executable
+    for match in re.finditer(r"'cian_city_sweep([a-z_]*)'", executable):
+        suffix = match.group(1)
+        assert suffix.startswith("_"), (
+            "нашли bare 'cian_city_sweep' (ЕКБ-расписание) среди таргетов миграции — "
+            "запрещено границами задачи"
+        )
+
+
+def test_migration_206_exactly_eight_update_statements() -> None:
+    """8 отдельных UPDATE: cian_full_load, avito_full_load, yandex_address_backfill,
+    yandex_city_sweep(enrich_address), house_imv_backfill,
+    domclick_detail_backfill(enabled), yandex_newbuilding_sweep, oblast-15 (одним IN)."""
+    flat = _flat(_executable_sql())
+    assert flat.count("update scrape_schedules") == 8
+
+
+def test_migration_206_idempotent_by_construction() -> None:
+    """Каждый UPDATE — SET x = merge(x, const) / SET enabled = false (константа) —
+    повторный прогон ставит те же значения, без отдельного guard-условия."""
+    flat = _flat(_executable_sql())
+    # Ни одного ON CONFLICT/INSERT — чистые UPDATE, все идемпотентны по конструкции.
+    assert "insert into" not in flat
+    assert "on conflict" not in flat