Merge pull request 'fix(tradein/avito): обогащать деталями листинги всей области, не только ЕКБ (#2576)' (#2578) from fix/tradein-avito-detail-oblast into main

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

@@ -30,6 +30,7 @@ from scraper_kit.avito_exceptions import (
     AvitoRateLimitedError,
 )
 from scraper_kit.browser_fetcher import BrowserFetcher
+from scraper_kit.orchestration.pipeline import CITY_LOCATIONS
 
 # #2397 slice B (эпик #2277 decommission scrape_pipeline.py, Part E): раньше
 # _CHROME_HEADERS/_avito_proxies() импортировались из app.services.scrape_pipeline.
@@ -70,6 +71,31 @@ __all__ = [
     "run_avito_detail_backfill",
 ]
 
+# #2576 этап B: oblast-города (region 66, вне ЕКБ) уже дают листинги (Каменск-
+# Уральский), но snapshot-SELECT ниже раньше фильтровал ЖЁСТКО '%/ekaterinburg/%' —
+# у всех остальных detail_enriched_at оставался NULL навсегда (без detail-страницы
+# нет lat/lon -> листинг молча выпадает из подбора аналогов по радиусу).
+# CITY_LOCATIONS.avito_slug — единственный источник правды для avito URL-слага
+# города (может отличаться от нашего city_slug: kamensk-uralskiy через дефис,
+# verhnyaya_pyshma без "kh") -- дублировать список тут вместо импорта было бы
+# risk дрейфа при добавлении новых oblast-городов.
+#
+# #2578 review: Postgres LIKE трактует '_' как wildcard "один любой символ" (не
+# литерал) и '%' как wildcard "любая последовательность" -- два слага из пяти
+# (nizhniy_tagil, verhnyaya_pyshma) содержат '_', без экранирования это латентная
+# дыра: город с похожим слагом (напр. nizhniyXtagil) молча совпал бы. Сегодня
+# коллизий нет (проверено на проде: raw vs escaped паттерны дают одинаковые 776
+# совпадений), но экранируем сейчас, а не когда появится реальная коллизия.
+# LIKE по умолчанию использует '\' как escape-символ (без явного ESCAPE) —
+# подтверждено на живом Postgres 16.4 (см. коммит #2578-fixup): 'nizhniyXtagil'
+# матчит неэкранированный '%/nizhniy_tagil/%' (LIKE default '_'=wildcard) и НЕ
+# матчит экранированный '%/nizhniy\_tagil/%' (LIKE '\_' = литерал '_'); точный
+# слаг 'nizhniy_tagil' матчит оба варианта -- позитивный кейс не сломан.
+_OBLAST_AVITO_URL_PATTERNS = tuple(
+    "%/" + loc.avito_slug.replace("\\", "\\\\").replace("_", "\\_").replace("%", "\\%") + "/%"
+    for loc in CITY_LOCATIONS.values()
+)
+
 
 @dataclass
 class AvitoDetailBackfillResult:
@@ -102,7 +128,12 @@ async def run_avito_detail_backfill(
     """Backfill detail_enriched_at for legacy avito listings via mobile proxy.
 
     Params (from default_params jsonb in scrape_schedules):
-        batch_size: int -- snapshot size (SELECT LIMIT), default 800.
+        batch_size: int -- ЕКБ snapshot size (SELECT LIMIT), default 800 (unchanged,
+            #2576 -- volume/order for ЕКБ stay byte-identical to pre-oblast behaviour).
+        oblast_batch_size: int -- ДОПОЛНИТЕЛЬНАЯ reserved-квота для листингов
+            области (#2576), default 100. Отдельный LIMIT, НЕ отъедает от batch_size
+            ЕКБ -- гарантирует области честную обработку и одновременно не даёт
+            всплеску свежих oblast-листингов вытеснить ЕКБ из top-N по scraped_at.
         budget_sec: float -- wall-clock budget per run, default 3600s.
         request_delay_sec: float -- delay between listings, default 6.0s.
         max_consecutive_blocks: int -- abort threshold, default 5.
@@ -111,6 +142,7 @@ async def run_avito_detail_backfill(
     mark_done (incl. partial/block-abort) / mark_failed (exception only).
     """
     batch_size = int(params.get("batch_size", 800))
+    oblast_batch_size = int(params.get("oblast_batch_size", 100))
     budget_sec = float(params.get("budget_sec", 3600))
     request_delay_sec = float(params.get("request_delay_sec", 6.0))
     max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))
@@ -189,30 +221,59 @@ async def run_avito_detail_backfill(
         runs_mod.update_heartbeat(db, run_id, current_counters)
 
         # SNAPSHOT: single SELECT at start -- NOT re-selected in loop.
-        # Scope (#1814): только активные ЕКБ-листинги. region_code на insert
+        # Scope (#1814, расширено #2576): активные листинги ЕКБ + известных oblast-
-        # хардкодится в 66 (base.py) → НЕ дискриминирует legacy не-ЕКБ; реальный
+        # городов (region 66). region_code на insert хардкодится в 66 (base.py) →
-        # признак региона у Avito — путь URL (/ekaterinburg/ для ЕКБ; legacy
+        # НЕ дискриминирует город; реальный признак города у Avito — путь URL
-        # Москва/СПб/Тюмень — /moskva//sankt-peterburg//tyumen/). browser-fetch
+        # (/ekaterinburg/ для ЕКБ; legacy Москва/СПб/Тюмень — /moskva//sankt-
-        # на legacy не-ЕКБ спотыкается → curl-fallback → 429-бан curl-фингерпринта.
+        # peterburg//tyumen/ — те по-прежнему вне scope, НЕ входят ни в ekb, ни в
-        # Не тратим фетчи на мёртвые (is_active) и не-ЕКБ.
+        # oblast CTE). browser-fetch на legacy не-ЕКБ/не-oblast спотыкается →
+        # curl-fallback → 429-бан curl-фингерпринта. Не тратим фетчи на мёртвые
+        # (is_active) и на регионы вне scope.
+        #
+        # Два CTE вместо одного WHERE ... OR ...: ekb сохраняет ТОЧНО прежний
+        # LIMIT/ORDER (#2576 требование "ЕКБ не деградирует") -- oblast НЕ может
+        # вытеснить ЕКБ из batch_size ни при каком всплеске свежих oblast-строк
+        # (ORDER BY ... scraped_at DESC в общем WHERE отдал бы приоритет самым
+        # свежим независимо от города). oblast получает отдельную честную квоту
+        # oblast_batch_size, добавленную ПОСЛЕ ekb-квоты (не вычтенную из неё).
         snapshot = (
             db.execute(
                 text(
                     """
-                    SELECT id, source_url
+                    WITH ekb AS (
-                    FROM listings
+                        SELECT id, source_url, 'ekb' AS city_scope
-                    WHERE source = 'avito'
+                        FROM listings
-                      AND detail_enriched_at IS NULL
+                        WHERE source = 'avito'
-                      AND source_url IS NOT NULL
+                          AND detail_enriched_at IS NULL
-                      AND is_active = TRUE
+                          AND source_url IS NOT NULL
-                      AND source_url LIKE '%/ekaterinburg/%'
+                          AND is_active = TRUE
-                    -- сперва листинги без координат (#1967 — detail-страница даёт
+                          AND source_url LIKE '%/ekaterinburg/%'
-                    -- координаты здания), затем по свежести
+                        -- сперва листинги без координат (#1967 — detail-страница
-                    ORDER BY (lat IS NULL) DESC, scraped_at DESC NULLS LAST
+                        -- даёт координаты здания), затем по свежести
-                    LIMIT CAST(:batch_size AS int)
+                        ORDER BY (lat IS NULL) DESC, scraped_at DESC NULLS LAST
+                        LIMIT CAST(:batch_size AS int)
+                    ),
+                    oblast AS (
+                        SELECT id, source_url, 'oblast' AS city_scope
+                        FROM listings
+                        WHERE source = 'avito'
+                          AND detail_enriched_at IS NULL
+                          AND source_url IS NOT NULL
+                          AND is_active = TRUE
+                          AND source_url LIKE ANY(CAST(:oblast_patterns AS text[]))
+                        ORDER BY (lat IS NULL) DESC, scraped_at DESC NULLS LAST
+                        LIMIT CAST(:oblast_batch_size AS int)
+                    )
+                    SELECT id, source_url, city_scope FROM ekb
+                    UNION ALL
+                    SELECT id, source_url, city_scope FROM oblast
                     """
                 ),
-                {"batch_size": batch_size},
+                {
+                    "batch_size": batch_size,
+                    "oblast_patterns": list(_OBLAST_AVITO_URL_PATTERNS),
+                    "oblast_batch_size": oblast_batch_size,
+                },
             )
             .mappings()
             .all()
@@ -227,11 +288,16 @@ async def run_avito_detail_backfill(
             runs_mod.mark_done(db, run_id, current_counters)
             return counters
 
+        # #2576: разбивка ekb/oblast только для наблюдаемости -- .get() консервативен
+        # (city_scope нет в mock-снапшотах старых тестов, дефолт "ekb" их не ломает).
+        oblast_count = sum(1 for row in snapshot if row.get("city_scope") == "oblast")
         logger.info(
-            "avito_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
+            "avito_detail_backfill: run_id=%d snapshot=%d (ekb=%d oblast=%d, "
-            "delay=%.1fs max_blocks=%d mode=%s)",
+            "budget=%.0fs delay=%.1fs max_blocks=%d mode=%s)",
             run_id,
             len(snapshot),
+            len(snapshot) - oblast_count,
+            oblast_count,
             budget_sec,
             request_delay_sec,
             max_consecutive_blocks,

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

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import fnmatch
 import os
 import sys
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -14,6 +15,7 @@ import pytest  # noqa: E402
 
 from app.core import shutdown as _sd  # noqa: E402
 from app.tasks.avito_detail_backfill import (  # noqa: E402
+    _OBLAST_AVITO_URL_PATTERNS,
     AvitoDetailBackfillResult,
     run_avito_detail_backfill,
 )
@@ -353,11 +355,14 @@ async def test_backfill_rotate_ip_called_on_each_block() -> None:
 
 @pytest.mark.asyncio
 async def test_backfill_snapshot_filters_ekb_active_only() -> None:
-    """Снапшот-SELECT (#1814) фильтрует только активные ЕКБ-листинги.
+    """Снапшот-SELECT (#1814, расширено #2576) фильтрует активные ЕКБ- И
+    известные oblast-листинги (region 66), НЕ всё подряд.
 
-    Проверяем, что текст запроса содержит `is_active = TRUE` и
+    Проверяем, что текст запроса содержит `is_active = TRUE`, `LIKE
-    `LIKE '%/ekaterinburg/%'` — legacy не-ЕКБ (moskva/spb/tyumen) и мёртвые
+    '%/ekaterinburg/%'` (ekb CTE, LIMIT batch_size НЕ сокращён) и `LIKE ANY(...)`
-    листинги не попадают в фетч, иначе browser спотыкается → curl-бан 429.
+    по oblast-паттернам (oblast CTE, отдельный LIMIT oblast_batch_size) — legacy
+    не-ЕКБ/не-область (moskva/spb/tyumen) и мёртвые листинги не попадают в фетч,
+    иначе browser спотыкается → curl-бан 429.
     """
     db = _mock_db([])
     runs = MagicMock()
@@ -376,8 +381,127 @@ async def test_backfill_snapshot_filters_ekb_active_only() -> None:
     sql_text = str(snapshot_call.args[0])
     assert "is_active = TRUE" in sql_text
     assert "/ekaterinburg/" in sql_text
+    assert "LIKE ANY(CAST(:oblast_patterns AS text[]))" in sql_text
     assert "detail_enriched_at IS NULL" in sql_text
     assert "(lat IS NULL) DESC" in sql_text
+    assert "CAST(:batch_size AS int)" in sql_text
+    assert "CAST(:oblast_batch_size AS int)" in sql_text
+
+    # ekb-квота передаётся batch_size БЕЗ урезания (#2576 требование "ЕКБ не
+    # деградирует") — oblast получает отдельный (не вычтенный) bind-параметр.
+    bind_params = snapshot_call.args[1]
+    assert bind_params["batch_size"] == 10
+    assert bind_params["oblast_batch_size"] == 100  # default
+    assert set(bind_params["oblast_patterns"]) == set(_OBLAST_AVITO_URL_PATTERNS)
+
+
+def test_oblast_avito_url_patterns_cover_region66_cities() -> None:
+    """#2576: _OBLAST_AVITO_URL_PATTERNS строится из CITY_LOCATIONS.avito_slug —
+    список должен покрывать реальные Avito-слаги oblast-городов (в т.ч. те, что
+    ОТЛИЧАЮТСЯ от нашего city_slug: kamensk-uralskiy через дефис, а не
+    kamensk_uralskiy).
+
+    #2578 review: '_' в слаге -- LIKE wildcard, экранируем при построении паттерна
+    ('_' -> '\\_') -- nizhniy_tagil/verhnyaya_pyshma здесь ожидаются С обратным
+    слэшем перед '_', НЕ голым подчёркиванием."""
+    assert "%/kamensk-uralskiy/%" in _OBLAST_AVITO_URL_PATTERNS
+    assert "%/nizhniy\\_tagil/%" in _OBLAST_AVITO_URL_PATTERNS
+    assert "%/pervouralsk/%" in _OBLAST_AVITO_URL_PATTERNS
+    assert "%/verhnyaya\\_pyshma/%" in _OBLAST_AVITO_URL_PATTERNS
+    assert "%/serov/%" in _OBLAST_AVITO_URL_PATTERNS
+    # ЕКБ обрабатывается отдельным жёстко закодированным паттерном (ekb CTE),
+    # НЕ через этот oblast-список — не должен в него затесаться.
+    assert not any("ekaterinburg" in p for p in _OBLAST_AVITO_URL_PATTERNS)
+
+
+def _like_pattern_to_fnmatch(pattern: str) -> str:
+    """Точный перевод семантики Postgres `LIKE` (default `ESCAPE '\\'`) в fnmatch-
+    паттерн -- посимвольно, а НЕ наивным `.replace()`.
+
+    LIKE: `%` = любая последовательность символов, `_` = РОВНО один любой символ,
+    `\\%`/`\\_`/`\\\\` = литералы (экранирование). fnmatch: `*` = любая
+    последовательность, `?` = один любой символ; голые `_`/`%` в fnmatch не
+    специальны (можно вставлять как литерал без экранирования).
+
+    #2578 review: наивный `pat.replace("%", "*")` (как было раньше) НЕ отражал бы
+    семантику `_` вообще -- fnmatch трактует `_` как литерал, LIKE -- как wildcard.
+    Из-за этого расхождения прежний тест не поймал бы латентный баг (нет
+    экранирования `_` в продовых паттернах). Посимвольный разбор здесь корректно
+    различает голый `_` (-> `?` wildcard) и экранированный `\\_` (-> литерал `_`).
+    """
+    out: list[str] = []
+    i = 0
+    n = len(pattern)
+    while i < n:
+        ch = pattern[i]
+        if ch == "\\" and i + 1 < n and pattern[i + 1] in ("%", "_", "\\"):
+            out.append(pattern[i + 1])  # экранированный символ -> литерал as-is
+            i += 2
+            continue
+        if ch == "%":
+            out.append("*")
+        elif ch == "_":
+            out.append("?")
+        else:
+            out.append(ch)
+        i += 1
+    return "".join(out)
+
+
+def _in_oblast_or_ekb_scope(source_url: str) -> bool:
+    """Локальная реплика WHERE-условия snapshot-запроса (ekb CTE OR oblast CTE)
+    через корректную LIKE-эмуляцию -- без поднятия БД."""
+    if fnmatch.fnmatchcase(source_url, _like_pattern_to_fnmatch("%/ekaterinburg/%")):
+        return True
+    return any(
+        fnmatch.fnmatchcase(source_url, _like_pattern_to_fnmatch(pat))
+        for pat in _OBLAST_AVITO_URL_PATTERNS
+    )
+
+
+def test_oblast_avito_url_patterns_include_oblast_and_ekb_exclude_foreign_region() -> None:
+    """#2576 DoD: листинг города области и екатеринбургский листинг проходят
+    scope-фильтр; листинг чужого региона (Москва/СПб) — нет.
+
+    Использует корректную LIKE-эмуляцию (_like_pattern_to_fnmatch), а не наивный
+    `%` -> `*` replace (#2578 review — тот не различал бы `_`-семантику)."""
+    # Область (Каменск-Уральский, #2576 — реальный кейс из тикета) -- проходит.
+    assert _in_oblast_or_ekb_scope("https://www.avito.ru/kamensk-uralskiy/kvartiry/prodam_123")
+    # ЕКБ — по-прежнему проходит (не деградировал).
+    assert _in_oblast_or_ekb_scope("https://www.avito.ru/ekaterinburg/kvartiry/prodam_456")
+    # Чужой регион — НЕ проходит (иначе поехали бы Москва/СПб/Тюмень legacy-строки).
+    assert not _in_oblast_or_ekb_scope("https://www.avito.ru/moskva/kvartiry/prodam_789")
+    assert not _in_oblast_or_ekb_scope("https://www.avito.ru/sankt-peterburg/kvartiry/prodam_000")
+
+
+def test_like_underscore_wildcard_regression_caught_by_escaped_patterns() -> None:
+    """#2578 deep-review latent bug: Postgres `LIKE` трактует `_` как wildcard
+    "ровно один любой символ", а НЕ литерал. Два слага из пяти (nizhniy_tagil,
+    verhnyaya_pyshma) содержат `_` -- БЕЗ экранирования 'nizhniy_tagil' молча
+    совпал бы с 'nizhniyXtagil' (X = любой символ), т.е. коллизия слагов при
+    появлении похожего города. Сегодня коллизий нет (проверено на проде: raw vs
+    escaped паттерны дают одинаковые 776 совпадений), но дыра латентная.
+
+    Этот тест ДОЛЖЕН падать на RAW (неэкранированном) варианте паттерна -- именно
+    так выглядели продовые паттерны ДО фикса #2578 (`%/nizhniy_tagil/%`, без
+    `\\`). Экранированный прод-паттерн (_OBLAST_AVITO_URL_PATTERNS, ПОСЛЕ фикса)
+    коллизию отклоняет, точный слаг по-прежнему матчит (позитивный кейс жив).
+    """
+    raw_pattern = "%/nizhniy_tagil/%"  # как было бы БЕЗ фикса #2578 (голый '_')
+    escaped_pattern = next(p for p in _OBLAST_AVITO_URL_PATTERNS if "nizhniy" in p)
+    # Сам факт экранирования: прод-паттерн ДОЛЖЕН отличаться от raw ('_' -> '\_').
+    assert escaped_pattern != raw_pattern, "фикс #2578 должен экранировать '_' в avito_slug"
+
+    collision_url = "https://www.avito.ru/nizhniyXtagil/kvartiry/prodam_1"
+    exact_url = "https://www.avito.ru/nizhniy_tagil/kvartiry/prodam_1"
+
+    # RAW: '_' -- wildcard -> ложное совпадение с ЛЮБЫМ символом на его месте.
+    assert fnmatch.fnmatchcase(collision_url, _like_pattern_to_fnmatch(raw_pattern))
+    # Экранированный прод-паттерн (после фикса): '_' -- литерал -> коллизия отклонена.
+    assert not fnmatch.fnmatchcase(collision_url, _like_pattern_to_fnmatch(escaped_pattern))
+    # Позитивный кейс не сломан: точный слаг матчит ОБА варианта паттерна.
+    assert fnmatch.fnmatchcase(exact_url, _like_pattern_to_fnmatch(raw_pattern))
+    assert fnmatch.fnmatchcase(exact_url, _like_pattern_to_fnmatch(escaped_pattern))
 
 
 @pytest.mark.asyncio