fix(tradein/yandex): «непригодных» 3535 не было — адресуем их по offerId

unenrichable_pending шесть прогонов подряд равнялся ровно 3535. Счётчик
считается живым SELECT'ом (кэша/матвьюхи нет), но множество замкнуто: после
#2235 продюсер таких строк не создаёт (0 из 6892 yandex-строк, вставленных
после самой свежей его строки id 2583989), а выйти оттуда нельзя — обогащение
недостижимо, а source_url не переписывается. Замкнутое множество и обязано
быть константой; вопрос был не «почему не растёт», а «правда ли непригодны».

Непригодны они не были. У всех 3535 в source_id лежит числовой yandex offerId
(у 3523 продублирован в yandex_offer_id), а канонический адрес из него
собирается — тот же инвариант #2235 и та же формула, которой чинила легаси
миграция 164. Живая проба прод-трактом (тот же прокси, curl_cffi chrome120,
тот же parse): 6 из 6 HTTP 200 + parse OK, включая строки, чей сохранённый
source_url — рекламный редирект na100.pro/go.php.

Откуда стухший адрес: source_url пишется только при вставке — его нет ни в
ON CONFLICT DO UPDATE, ни в reconcile-UPDATE save_listings. Поэтому #2235
вылечил только новые строки, а миграция 164 — только те легаси, чей URL делили
несколько строк (она искала дубли URL, а не непарсимость). Строки с уникальной
ссылкой на карточку застройщика не попали никуда и носят адрес, замороженный в
момент вставки, хотя свип переобходит ~511 из них в сутки.

Что сделано: очередь адресует такие строки URL, вычисленным из source_id, а
остаток очереди разделён по причине — url_from_offer_id (адрес чиним, должен
убывать) и unenrichable_pending (адресовать нечем; на проде 0). Одно число на
две разные судьбы читалось как «тут делать нечего» и держало 3535 квартир вне
обогащения неделю.

Проверено на проде (read-only): новый снапшот-SELECT отдаёт 800/800 строк с
адресом, который парсер принимает; счётчики 3535/0; разбиение точное —
queueable 13921 = pending 13921, unaddressable 0.

Не входит сюда: починка самой колонки source_url одноразовым UPDATE (тот же
164 без условия на дубли) — за миграцией; от неё зависит и
yandex_address_backfill, где 1618 из 5217 кандидатов ходят на сайты
застройщиков вместо Яндекса.

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

@@ -22,8 +22,14 @@ max_consecutive_blocks. Прогон с нулём обогащений тепе
 ведёт на сайт застройщика, а не на realty.yandex.ru/offer/<id>/. Парсер отвергает
 такие URL регуляркой ДО сети — это не капча, а предрешённый parse→None. Идут они
 пачками, поэтому «5 подряд» набиралось на первых же строках и обрывало прогон
-целиком. Теперь снапшот-SELECT берёт только то, что парсер в принципе может
+целиком. Снапшот-SELECT берёт только то, что парсер в принципе может разобрать.
-разобрать, а размер отброшенного видно в counters.unenrichable_pending.
+
+Но «не по тому URL» ≠ «нечего обогащать» (разобрано 2026-08-12, см. комментарий
+у OFFER_ID_PATTERN): у ВСЕХ таких строк в source_id лежит yandex offerId, и по
+собранному из него каноническому URL страница отдаётся и парсится. Поэтому в
+очередь они входят по адресу, ВЫЧИСЛЕННОМУ из source_id, а counters разделены:
+url_from_offer_id — сколько ждёт починки адреса, unenrichable_pending — сколько
+не адресуемо вообще (ни offer-URL, ни числового source_id).
 
 Why curl_cffi and not YandexDetailScraper.fetch_detail:
   fetch_detail uses BaseScraper._http_get (plain httpx, no proxy, no TLS
@@ -51,6 +57,8 @@ from app.services.proxy_egress import resolve_proxy_url
 logger = logging.getLogger(__name__)
 
 __all__ = [
+    "CANONICAL_URL_SQL",
+    "OFFER_ID_PATTERN",
     "OFFER_URL_PATTERN",
     "YandexDetailBackfillResult",
     "run_yandex_detail_backfill",
@@ -63,8 +71,7 @@ __all__ = [
 #
 # Замер прода 2026-08-06: из 15 511 необогащённых yandex-объявлений 3 535 имеют
 # source_url на сайт застройщика (macroserver.ru, prospect-federation.ru,
-# strana.com, …) — так карточки новостроек ведут с выдачи Яндекса. Обогащено из
+# strana.com, …) — так карточки новостроек ведут с выдачи Яндекса.
-# них за всю историю 0; все 1 210 обогащённых — вида realty.yandex.ru/offer/<id>/.
 #
 # Вред не в бесполезности, а в том, что они идут ПАЧКАМИ (один свип — один
 # застройщик) и упираются в брейкер «5 parse-None подряд», обрывающий ВЕСЬ прогон:
@@ -72,6 +79,39 @@ __all__ = [
 # Плюс каждая такая попытка — запрос на чужой сайт, который мы всё равно выбросим.
 OFFER_URL_PATTERN = "/offer/[0-9]+"
 
+# ── «Непригодных» не бывает без причины (разобрано 2026-08-12) ────────────────
+# Симптом: unenrichable_pending шесть прогонов подряд равнялся РОВНО 3535 — ни на
+# единицу, при том что очередь обогащалась по ~500/прогон. Замер на проде:
+#
+#   * счётчик считается живым SELECT'ом, кэша/матвьюхи нет — арифметика честная;
+#   * множество замкнуто: новых строк в него не приходит (0 из 6892 yandex-строк,
+#     вставленных после самой свежей его строки, id 2583989), и выйти из него
+#     нельзя (обогащение недостижимо, source_url не переписывается). Замкнутое
+#     множество и обязано быть константой — вопрос был не «почему не растёт», а
+#     «правда ли они непригодны».
+#
+# Непригодны они НЕ были. У всех 3535 в source_id лежит числовой yandex offerId
+# (у 3523 он же продублирован в yandex_offer_id), а канонический адрес оффера из
+# него собирается — это инвариант #2235 (`_canonical_source_url` в
+# scraper_kit/providers/yandex/serp.py) и та же формула, которой миграция 164
+# чинила легаси-строки. Живая проба 2026-08-12 прод-трактом (тот же прокси,
+# curl_cffi chrome120, тот же parse): 6 из 6 — HTTP 200 и parse OK, включая
+# строки, чей сохранённый source_url — рекламный редирект na100.pro/go.php.
+#
+# Откуда взялся стухший адрес: source_url пишется ТОЛЬКО при вставке — его нет ни
+# в `ON CONFLICT DO UPDATE`, ни в reconcile-UPDATE у `save_listings`. Значит #2235
+# вылечил только новые строки, миграция 164 — только те легаси, чей URL ДЕЛИЛИ
+# несколько строк (она искала дубли URL, а не непарсимость). Строки с уникальной
+# ссылкой на карточку застройщика не попали ни туда, ни туда и носят адрес,
+# замороженный в момент вставки, хотя свип переобходит ~511 из них в сутки.
+#
+# Поэтому адресуем такие строки вычисленным URL, а не сохранённым. Починка самой
+# колонки (одноразовый UPDATE, тот же 164 без условия на дубли) — за миграцией:
+# от неё зависит и yandex_address_backfill, где 1618 из 5217 кандидатов ходят
+# на сайты застройщиков вместо Яндекса.
+OFFER_ID_PATTERN = "^[0-9]+$"
+CANONICAL_URL_SQL = "'https://realty.yandex.ru/offer/' || source_id || '/'"
+
 
 @dataclass
 class YandexDetailBackfillResult:
@@ -80,6 +120,11 @@ class YandexDetailBackfillResult:
     attempted: int = 0
     enriched: int = 0
     failed: int = 0
+    # Ждут обогащения, сохранённый source_url непарсим, но адрес восстановим из
+    # source_id — идут в очередь по вычисленному URL. Должен убывать от прогона к
+    # прогону; замер на месте = очередь снова читается не тем признаком.
+    url_from_offer_id: int = 0
+    # Ждут обогащения и адресовать их НЕЧЕМ: ни offer-URL, ни числового source_id.
     unenrichable_pending: int = 0
     duration_sec: float = field(default=0.0)
 
@@ -88,6 +133,7 @@ class YandexDetailBackfillResult:
             "attempted": self.attempted,
             "enriched": self.enriched,
             "failed": self.failed,
+            "url_from_offer_id": self.url_from_offer_id,
             "unenrichable_pending": self.unenrichable_pending,
             "duration_sec": int(self.duration_sec),
         }
@@ -132,51 +178,86 @@ async def run_yandex_detail_backfill(
 
         # SNAPSHOT: single SELECT at start -- NOT re-selected in loop.
         # Priority: is_active DESC (active first), scraped_at DESC (newest first).
-        # Гейт по OFFER_URL_PATTERN — тот же признак, по которому парсер отказывает
+        # В очередь идёт то, для чего есть АДРЕС, который парсер примет: либо
-        # (см. комментарий у константы): в очередь не берём то, что заведомо
+        # сохранённый source_url подходит под OFFER_URL_PATTERN, либо адрес
-        # непарсимо, иначе пачка карточек застройщика обрывает прогон брейкером.
+        # собирается из source_id (см. комментарий у OFFER_ID_PATTERN). Что шире
+        # этого условия — гарантированный parse→None пачкой и обрыв по брейкеру.
         snapshot = (
             db.execute(
                 text(
-                    """
+                    f"""
-                    SELECT id, source_url
+                    SELECT id,
+                           CASE
+                               WHEN source_url ~ CAST(:offer_url_pattern AS text)
+                                   THEN source_url
+                               ELSE {CANONICAL_URL_SQL}
+                           END AS source_url
                     FROM listings
                     WHERE source = 'yandex'
                       AND detail_enriched_at IS NULL
-                      AND source_url IS NOT NULL
+                      AND (
-                      AND source_url ~ CAST(:offer_url_pattern AS text)
+                            (
+                                source_url IS NOT NULL
+                                AND source_url ~ CAST(:offer_url_pattern AS text)
+                            )
+                            OR source_id ~ CAST(:offer_id_pattern AS text)
+                      )
                     ORDER BY is_active DESC NULLS LAST, scraped_at DESC NULLS LAST
                     LIMIT CAST(:batch_size AS int)
                     """
+                    # f-string здесь безопасен: CANONICAL_URL_SQL — литерал модуля,
+                    # не пользовательский ввод. Всё изменяемое — bind-параметры.
                 ),
-                {"batch_size": batch_size, "offer_url_pattern": OFFER_URL_PATTERN},
+                {
+                    "batch_size": batch_size,
+                    "offer_url_pattern": OFFER_URL_PATTERN,
+                    "offer_id_pattern": OFFER_ID_PATTERN,
+                },
             )
             .mappings()
             .all()
         )
 
         # Отброшенное не должно исчезнуть из виду: без этого счётчика «обогащено
-        # 12 тыс. из 15,5 тыс.» снова стало бы необъяснимым нулём (#2674).
+        # 12 тыс. из 15,5 тыс.» снова стало бы необъяснимым нулём (#2674). И оно
-        counters.unenrichable_pending = int(
+        # разделено по ПРИЧИНЕ: одно число на две разные судьбы читалось как
-            db.execute(
+        # «тут делать нечего» и держало 3535 квартир вне обогащения неделю.
-                text(
+        pending = db.execute(
-                    """
+            text(
-                    SELECT count(*)
+                """
-                    FROM listings
+                SELECT
-                    WHERE source = 'yandex'
+                    count(*) FILTER (
-                      AND detail_enriched_at IS NULL
+                        WHERE source_id ~ CAST(:offer_id_pattern AS text)
-                      AND source_url IS NOT NULL
+                    ) AS url_from_offer_id,
-                      AND source_url !~ CAST(:offer_url_pattern AS text)
+                    count(*) FILTER (
-                    """
+                        WHERE source_id IS NULL
-                ),
+                           OR source_id !~ CAST(:offer_id_pattern AS text)
-                {"offer_url_pattern": OFFER_URL_PATTERN},
+                    ) AS unenrichable_pending
-            ).scalar_one()
+                FROM listings
-        )
+                WHERE source = 'yandex'
-        if counters.unenrichable_pending:
+                  AND detail_enriched_at IS NULL
+                  AND (
+                        source_url IS NULL
+                        OR source_url !~ CAST(:offer_url_pattern AS text)
+                  )
+                """
+            ),
+            {"offer_url_pattern": OFFER_URL_PATTERN, "offer_id_pattern": OFFER_ID_PATTERN},
+        ).one()
+        counters.url_from_offer_id = int(pending.url_from_offer_id)
+        counters.unenrichable_pending = int(pending.unenrichable_pending)
+        if counters.url_from_offer_id:
             logger.info(
-                "yandex_detail_backfill: run_id=%d — %d объявлений вне очереди: "
+                "yandex_detail_backfill: run_id=%d — у %d объявлений сохранённый "
-                "source_url ведёт не на карточку Яндекса (%s), парсер их отвергает "
+                "source_url не ведёт на карточку Яндекса; адресуем их по offerId из "
-                "до сети",
+                "source_id (колонку чинит миграция, см. OFFER_ID_PATTERN)",
+                run_id,
+                counters.url_from_offer_id,
+            )
+        if counters.unenrichable_pending:
+            logger.warning(
+                "yandex_detail_backfill: run_id=%d — %d объявлений вне очереди: нет ни "
+                "offer-URL (%s), ни числового source_id — адресовать их нечем",
                 run_id,
                 counters.unenrichable_pending,
                 OFFER_URL_PATTERN,

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

@@ -15,6 +15,7 @@ import json
 import os
 import re
 import sys
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
@@ -25,6 +26,9 @@ sys.modules.setdefault("weasyprint", _wp_mock)
 import pytest  # noqa: E402
 
 from app.tasks.yandex_detail_backfill import (  # noqa: E402
+    CANONICAL_URL_SQL,
+    OFFER_ID_PATTERN,
+    OFFER_URL_PATTERN,
     YandexDetailBackfillResult,
     run_yandex_detail_backfill,
 )
@@ -55,13 +59,21 @@ def _make_snapshot(n: int) -> list[dict]:
     ]
 
 
-def _mock_db(snapshot: list[dict], unenrichable: int = 0) -> MagicMock:
+def _mock_db(
-    """Fake Session: execute() отдаёт снапшот через .mappings().all(), а
+    snapshot: list[dict],
-    .scalar_one() — размер отброшенной (непарсимой) части очереди."""
+    unenrichable: int = 0,
+    url_from_offer_id: int = 0,
+) -> MagicMock:
+    """Fake Session: execute() отдаёт снапшот через .mappings().all(), а .one() —
+    остаток очереди вне снапшота, РАЗБИТЫЙ по причине (адрес восстановим из
+    source_id / адресовать нечем)."""
     db = MagicMock()
     sel = MagicMock()
     sel.mappings.return_value.all.return_value = snapshot
-    sel.scalar_one.return_value = unenrichable
+    sel.one.return_value = SimpleNamespace(
+        url_from_offer_id=url_from_offer_id,
+        unenrichable_pending=unenrichable,
+    )
     db.execute.return_value = sel
     return db
 
@@ -510,3 +522,72 @@ async def test_queue_gate_matches_parser_gate_and_counts_rest() -> None:
 
     assert result.unenrichable_pending == 3535
     assert runs.mark_done.call_args.args[2]["unenrichable_pending"] == 3535
+
+
+# ---------------------------------------------------------------------------
+# «Непригодно» — ярлык, а не диагноз (2026-08-12)
+# ---------------------------------------------------------------------------
+
+
+def _render_canonical_sql(offer_id: str) -> str:
+    """Считает CANONICAL_URL_SQL как строку: '||' — конкатенация, source_id — значение."""
+    parts = [p.strip() for p in CANONICAL_URL_SQL.split("||")]
+    return "".join(offer_id if p == "source_id" else p.strip("'") for p in parts)
+
+
+def test_recovered_url_equals_producer_canonical_form() -> None:
+    """Адрес, вычисленный из source_id, — тот же, что пишет продюсер, и парсер его примет.
+
+    Шесть прогонов подряд unenrichable_pending равнялся ровно 3535 не потому, что
+    счётчик застыл (SELECT живой), а потому что множество замкнуто: продюсер после
+    #2235 таких строк больше не создаёт, а выйти оттуда нельзя — source_url пишется
+    только при вставке. Ярлык «непригодны» был неверен: у всех есть offerId, и по
+    собранному из него URL страница парсится (прод-проба 2026-08-12, 6/6).
+
+    Сторож держит ровно это: формула восстановления в SQL не должна разъехаться с
+    `_canonical_source_url` продюсера, а результат — пройти гейт парсера.
+    """
+    from scraper_kit.providers.yandex.serp import _canonical_source_url
+
+    offer_id = "7416316697684470413"  # реальный source_id прод-строки с macroserver.ru
+    recovered = _render_canonical_sql(offer_id)
+
+    assert re.match(OFFER_ID_PATTERN, offer_id)
+    assert re.search(OFFER_URL_PATTERN, recovered), recovered
+    for stored_url, is_offer_url in _PROD_QUEUE_HEAD:
+        if is_offer_url:
+            continue  # у этих сохранённый адрес уже канонический, чинить нечего
+        assert _canonical_source_url(stored_url, offer_id) == recovered, stored_url
+
+
+@pytest.mark.asyncio
+async def test_pending_counter_split_by_reason() -> None:
+    """Остаток очереди делится по ПРИЧИНЕ, и восстановимое не зовётся непригодным.
+
+    Одно число на две разные судьбы («адрес чиним» и «адресовать нечем») читается
+    как «тут делать нечего» — так 3535 квартир простояли неделю вне обогащения.
+    """
+    db = _mock_db([], unenrichable=0, url_from_offer_id=3535)
+    runs = MagicMock()
+    session_cls, _session = _make_session_ctx([])
+
+    with (
+        patch(_ASYNC_SESSION, session_cls),
+        patch(_RUNS, runs),
+        patch(_RESOLVE_PROXY_URL, _mock_resolve_proxy_url()),
+    ):
+        result = await run_yandex_detail_backfill(
+            db, run_id=43, params={"batch_size": 10, "budget_sec": 60}
+        )
+
+    assert result.url_from_offer_id == 3535
+    assert result.unenrichable_pending == 0
+    counters = runs.mark_done.call_args.args[2]
+    assert counters["url_from_offer_id"] == 3535
+    assert counters["unenrichable_pending"] == 0
+
+    # Снапшот-SELECT берёт такие строки в работу по вычисленному адресу, а не
+    # выбрасывает: без этой ветки они не попадут в очередь никогда.
+    snapshot_sql = str(db.execute.call_args_list[0].args[0])
+    assert "OR source_id ~ CAST(:offer_id_pattern AS text)" in snapshot_sql
+    assert CANONICAL_URL_SQL in snapshot_sql