fix(tradein): working cian ЖК-url resolver via cat.php SERP (#972)

The legacy resolve_cian_zhk_url hit /zhk/<id>/ which now 404s, leaving the
318 geo-matched cian houses (house_sources.ext_id set, cian_zhk_url NULL)
unfetchable -> the newbuilding-enrichment backfill matched 0 houses.

Add resolve_cian_zhk_url_via_search(nb_id): fetch the cat.php newbuilding SERP
and extract the canonical zhk-<slug>.cian.ru url, ANCHORED on the
<h1 data-name="Title"> header anchor (NOT naive first-match — the SERP carries
promo/recommendation zhk-* links before the title that would otherwise resolve
the wrong ЖК and silently corrupt enrichment). Validated against the real
2.81MB prod SERP + an adversarial poisoned-recommendation test.

Wire into newbuilding_enrich_backfill: broaden selection to "has ext_id OR
cian_zhk_url", resolve+persist the url under a SAVEPOINT before enriching,
rate-limited + resumable + idempotent. Keep the old resolver (deprecated).

Bounded prod proof (5 houses, direct): 5/5 urls resolved+persisted, 3/5 fully
enriched (+18 price_dynamics, +3 reliability); the 2 misses were direct-mode
anti-bot on the 2nd fetch. Full 318-run gated on the cian mobile proxy
(mproxy.site) being restored. code-reviewer APPROVE (SQL/idempotency) +
resolver hardened against wrong-ЖК. 19 tests green.

Refs #972.

tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 
 import json
 import logging
+import re
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -28,6 +29,43 @@ from app.services.scrapers.cian_state_parser import extract_all_states, extract_
 
 logger = logging.getLogger(__name__)
 
+# Canonical ЖК-slug subdomain URL embedded in any Cian SERP/page that references a
+# newbuilding (e.g. https://zhk-parkovyy-kvartal-ekb-i.cian.ru). The cat.php SERP for a
+# single newobject[] id puts this slug in its title block — extracting it is the WORKING
+# nb_id → ЖК-url resolution (#972; see resolve_cian_zhk_url_via_search).
+_ZHK_SLUG_URL_RE = re.compile(r"https://zhk-[a-z0-9-]+\.cian\.ru")
+
+# Title-anchored extraction (#972 hardening). The cat.php newbuilding-SERP renders the
+# canonical ЖК-slug of the QUERIED newobject as the <a> href inside its page <h1>:
+#
+#   <h1 data-name="Title" ...>Купить квартиру в
+#       <a href="https://zhk-<slug>.cian.ru/" ...>ЖК «...»</a></h1>
+#
+# (verified on the live ~2.8 MB prod SERP for nb 48853 — the header sits inside a
+# `NewbuildingHeaderSeo` block; there is NO <link rel=canonical> / og:url on this page
+# type, so the H1 anchor is the only reliable canonical carrier). Anchoring on it avoids
+# the WRONG-ЖК hazard of a naive first-match: cat.php pages also carry banner / "похожие
+# ЖК" / recommendation blocks that can embed OTHER zhk-*.cian.ru urls, and those blocks
+# may render BEFORE the title — a plain first-match could then resolve a foreign ЖК and
+# silently enrich the house with another building's data.
+#
+# Matches the FIRST zhk-slug url that occurs inside the page's <h1 data-name="Title"> …
+# </h1>. `[^>]*` keeps the attr scan within the opening tag; `.*?` (DOTALL) spans the
+# inner markup up to the slug href. Falls back to naive first-match only when this finds
+# nothing (markup drift), preserving the previous behaviour as a safety net.
+_ZHK_TITLE_ANCHOR_RE = re.compile(
+    r'<h1[^>]*data-name="Title"[^>]*>.*?(https://zhk-[a-z0-9-]+\.cian\.ru)',
+    re.DOTALL,
+)
+
+# cat.php newbuilding-SERP template: one row per `newobject[0]=<nb_id>`. ekb. host is the
+# Свердловская-обл. mirror tradein targets; the slug URL it returns is host-agnostic
+# (the canonical zhk-*.cian.ru subdomain), so this resolves ЕКБ newbuildings correctly.
+_CATPH_NEWBUILDING_SERP = (
+    "https://ekb.cian.ru/cat.php?deal_type=sale&engine_version=2"
+    "&offer_type=flat&object_type[0]=2&newobject[0]={nb_id}"
+)
+
 
 @dataclass
 class NewbuildingEnrichment:
@@ -593,14 +631,16 @@ async def resolve_cian_zhk_url(
     *,
     session: AsyncSession | None = None,
 ) -> str | None:
-    """Resolve canonical ZHK slug URL from a Cian internal house ID.
+    """Resolve canonical ZHK slug URL from a Cian internal house ID (LEGACY — BROKEN).
 
-    Cian redirects https://cian.ru/zhk/<id>/ → canonical slug URL
-    (e.g. https://zhk-park-ekb-i.cian.ru/). We follow the redirect and
-    return the final URL.
-
-    Used by backfill scripts (scripts/backfill_cian_zhk_url.py) for houses
-    that have cian_internal_house_id set but cian_zhk_url IS NULL.
+    .. deprecated::
+        The redirect path this relies on — ``https://cian.ru/zhk/<id>/`` → canonical
+        slug — NO LONGER EXISTS. Cian now serves HTTP **404** for ``/zhk/<id>/``
+        (verified 8/8 on prod, #972), so this returns None for every real id. Use
+        :func:`resolve_cian_zhk_url_via_search` instead: it fetches the cat.php
+        newbuilding-SERP and extracts the canonical ``zhk-*.cian.ru`` slug from its
+        markup (the WORKING path). Kept only for backward-compat / reference; do not
+        wire new callers to it.
 
     Args:
         cian_internal_house_id: Cian's numeric ЖК identifier.
@@ -651,3 +691,112 @@ async def resolve_cian_zhk_url(
     finally:
         if close_session:
             await session.close()
+
+
+def _extract_zhk_url_from_serp(html: str) -> str | None:
+    """Extract the canonical ЖК-slug URL of the QUERIED newobject from a cat.php SERP.
+
+    Pure helper (no I/O) so the extraction can be unit-tested against a saved fixture.
+    The cat.php SERP for a single ``newobject[0]=<nb_id>`` renders that ЖК's canonical
+    ``https://zhk-<slug>.cian.ru`` URL as the ``<a>`` href inside its page heading
+    ``<h1 data-name="Title">`` (the ``NewbuildingHeaderSeo`` header). We anchor on that
+    heading rather than taking the first ``zhk-*.cian.ru`` match in the document, because
+    cat.php pages also carry banner / "похожие ЖК" / recommendation blocks that can embed
+    OTHER ЖК slug URLs — and some of those blocks render BEFORE the title. A naive
+    first-match could pick a foreign ЖК and enrich the house with the wrong building's
+    price_dynamics / reliability (and the backfill's idempotency would then lock the
+    corruption in). Anchoring on the title guarantees we resolve the queried ЖК.
+
+    Falls back to a naive first ``zhk-*.cian.ru`` match ONLY when the title anchor finds
+    nothing (markup drift / unexpected page shape), preserving the previous behaviour as
+    a last-resort safety net.
+
+    Returns the slug URL string, or None when the SERP carries no such anchor at all
+    (empty / no-results page, or markup drift).
+    """
+    anchored = _ZHK_TITLE_ANCHOR_RE.search(html)
+    if anchored:
+        return anchored.group(1)
+    # Fallback: no title-anchored slug found — degrade to naive first-match so a markup
+    # drift that moves/renames the header still resolves *something* rather than failing.
+    match = _ZHK_SLUG_URL_RE.search(html)
+    return match.group(0) if match else None
+
+
+async def resolve_cian_zhk_url_via_search(
+    nb_id: int,
+    *,
+    session: AsyncSession | None = None,
+) -> str | None:
+    """Resolve the canonical ЖК-slug URL for a Cian newbuilding id (the WORKING path).
+
+    Fetches the cat.php newbuilding-SERP for a single ``newobject[0]=<nb_id>`` and
+    extracts the canonical ``https://zhk-<slug>.cian.ru`` URL from its markup. This
+    replaces the legacy :func:`resolve_cian_zhk_url`, whose ``/zhk/<id>/`` redirect path
+    now 404s (verified on prod, #972). The returned slug URL is exactly what
+    :func:`fetch_newbuilding` parses, so the enrichment chain is
+    ``nb_id → cat.php SERP → zhk-slug-url → fetch_newbuilding → enrich``.
+
+    Verified live (HTTP 200, slug extracted):
+        nb 48853  → https://zhk-parkovyy-kvartal-ekb-i.cian.ru   (ЖК «Парковый квартал»)
+        nb 102791 → https://zhk-izumrudnyy-bor-ekb-i.cian.ru
+        nb 24991  → https://zhk-baltym-park-ekb-i.cian.ru
+
+    Args:
+        nb_id: Cian newbuilding id (``house_sources.ext_id`` for cian houses).
+        session: optional shared curl_cffi AsyncSession (caller owns lifecycle). When
+            None, an own session is created using the same mobile-proxy wiring as the
+            rest of the Cian scrapers (``settings.cian_proxy_url`` when set, else direct).
+
+    Returns:
+        The canonical ЖК-slug URL string, or None on non-200 / empty SERP / no match /
+        request failure (each logs a warning).
+
+    Note:
+        Makes ONE real HTTP request and does NOT sleep — the CALLER enforces the
+        anti-bot delay (matching scraper_settings 'cian'). At scale this needs the
+        mobile proxy; low-volume direct fetches work for the bounded proof.
+    """
+    close_session = False
+    if session is None:
+        # Mobile proxy wiring (#806): Cian блокирует datacenter-IP. proxy=None → прямое
+        # подключение (dev / proxy-down fallback — single fetches survive direct).
+        _proxy_url = settings.cian_proxy_url
+        _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
+        session = AsyncSession(
+            impersonate="chrome120",
+            timeout=30.0,
+            proxies=_proxies,
+            headers={
+                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+                "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
+            },
+        )
+        close_session = True
+
+    serp_url = _CATPH_NEWBUILDING_SERP.format(nb_id=nb_id)
+    try:
+        resp = await session.get(serp_url, allow_redirects=True)
+        if resp.status_code != 200:
+            logger.warning(
+                "resolve_cian_zhk_url_via_search nb_id=%s: cat.php → HTTP %d",
+                nb_id,
+                resp.status_code,
+            )
+            return None
+        zhk_url = _extract_zhk_url_from_serp(resp.text)
+        if zhk_url is None:
+            logger.warning(
+                "resolve_cian_zhk_url_via_search nb_id=%s: no zhk-slug URL in SERP "
+                "(empty results / markup drift?)",
+                nb_id,
+            )
+            return None
+        logger.info("resolve_cian_zhk_url_via_search nb_id=%s → %s", nb_id, zhk_url)
+        return zhk_url
+    except Exception as exc:
+        logger.warning("resolve_cian_zhk_url_via_search nb_id=%s failed: %s", nb_id, exc)
+        return None
+    finally:
+        if close_session:
+            await session.close()

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

@@ -9,14 +9,26 @@ newbuilding "Phase 4" enrichment never backfilled the existing house cache:
   - house_reliability_checks   (наш.дом.рф checkStatus + details[])
   - house_reviews              (ЖК resident reviews)
 
-Only the ~283 houses linked to ext_source='cian_newbuilding' (via house_sources)
-carry a fetchable Cian ЖК page (cian_zhk_url). This task selects those houses and
-runs the EXISTING enrichment over each one:
+The houses linked to ext_source='cian_newbuilding' (via house_sources) are the target
+set. Each carries `house_sources.ext_id` (= cian newbuilding id `nb_id`); most have NULL
+`cian_zhk_url`. This task resolves the url from the id when missing, then runs the
+EXISTING enrichment over each one:
 
+  resolve_cian_zhk_url_via_search(nb_id)-> cian_zhk_url   (#972 — see note below)
   fetch_newbuilding(zhk_url)            -> NewbuildingEnrichment
   save_newbuilding_enrichment(db, ...)  -> houses_price_dynamics + house_reliability_checks
   _save_cian_reviews(db, ...)           -> house_reviews   (added here — see note below)
 
+ЖК-url resolution (#972)
+------------------------
+The 318 geo-matched cian houses have `ext_id` but NULL `cian_zhk_url`. The legacy
+resolver hit https://cian.ru/zhk/<id>/ which now 404s. The working path fetches the
+cat.php newbuilding-SERP (newobject[0]=<nb_id>) and extracts the canonical
+zhk-<slug>.cian.ru URL from its markup — that slug is exactly what fetch_newbuilding
+parses. Resolved urls are persisted to houses.cian_zhk_url (CAST + SAVEPOINT) so a
+resolved-but-not-enriched house resumes without re-fetching the SERP. The full 318-house
+run is gated on the mobile proxy being up; low-volume direct fetches work for proofs.
+
 Why a dedicated task (not just the existing cian_history_backfill houses block):
   - That block keys off `houses.cian_zhk_url IS NOT NULL` only; this one anchors on
     the canonical `ext_source='cian_newbuilding'` link (the 283 set the issue names)
@@ -82,13 +94,15 @@ class NewbuildingEnrichBackfillResult:
 
     # Population sizing (independent of `limit`).
     cian_houses_total: int = 0           # houses linked to ext_source='cian_newbuilding'
-    cian_houses_fetchable: int = 0       # of those, with cian_zhk_url NOT NULL
+    cian_houses_fetchable: int = 0       # of those, with cian_zhk_url OR resolvable ext_id
     cian_houses_pending: int = 0         # fetchable AND not yet fully enriched (force=False)
 
     # Processing (bounded by `limit`).
     processed: int = 0
     skipped_already_enriched: int = 0
     succeeded: int = 0
+    resolved_zhk_url: int = 0            # ext_id → cian_zhk_url resolved + persisted (#972)
+    failed_resolve: int = 0             # had only ext_id, resolver returned None
     failed_fetch: int = 0                # fetch returned None / raised
     failed_save: int = 0                 # save raised after a good fetch
 
@@ -103,8 +117,17 @@ class NewbuildingEnrichBackfillResult:
         return {f.name: int(getattr(self, f.name)) for f in fields(self)}
 
 
-# SQL: anchor on the canonical cian_newbuilding link (the 283), require a fetchable
-# zhk_url, and (unless force) skip houses already enriched.
+# SQL: anchor on the canonical cian_newbuilding link, require a house we can FETCH, and
+# (unless force) skip houses already enriched.
+#
+# "Fetchable" (#972 fix): a house is fetchable when it has a cian_zhk_url already OR a
+# cian newbuilding id we can resolve into one. The 318 geo-matched cian houses carry
+# `house_sources.ext_id` (= cian newbuilding id `nb_id`) but NULL `cian_zhk_url` — the
+# old `cian_zhk_url IS NOT NULL` predicate matched 0 of them, so the backfill could never
+# run. We now also accept `hs.ext_id IS NOT NULL` and resolve the url per-house at
+# runtime (resolve_cian_zhk_url_via_search → cat.php SERP → zhk-slug). hs.ext_id is TEXT;
+# the resolver casts to int. We carry both columns through so the loop can decide whether
+# to resolve first.
 #
 # "Enriched" = has price_dynamics AND reliability rows. We deliberately do NOT require
 # house_reviews here: the Cian newbuilding initialState rarely carries the reviews list
@@ -112,13 +135,13 @@ class NewbuildingEnrichBackfillResult:
 # not call), so requiring reviews would leave virtually every house perpetually "pending"
 # and make every re-run re-fetch + append a duplicate reliability row. Reviews are written
 # opportunistically when present; their absence must not block the skip.
-# DISTINCT — a house can have >1 house_sources row in theory.
+# DISTINCT ON (h.id) — a house can have >1 house_sources row; pick any non-null ext_id.
 _SELECT_PENDING_HOUSES = """
-    SELECT DISTINCT h.id AS house_id, h.cian_zhk_url
+    SELECT DISTINCT ON (h.id) h.id AS house_id, h.cian_zhk_url, hs.ext_id
     FROM houses h
     JOIN house_sources hs ON hs.house_id = h.id
     WHERE hs.ext_source = 'cian_newbuilding'
-      AND h.cian_zhk_url IS NOT NULL
+      AND (h.cian_zhk_url IS NOT NULL OR hs.ext_id IS NOT NULL)
       AND (
         CAST(:force AS boolean) = TRUE
         OR NOT (
@@ -126,7 +149,7 @@ _SELECT_PENDING_HOUSES = """
             AND EXISTS (SELECT 1 FROM house_reliability_checks rc WHERE rc.house_id = h.id)
         )
       )
-    ORDER BY h.id
+    ORDER BY h.id, hs.ext_id NULLS LAST
     LIMIT :lim
 """
 
@@ -142,7 +165,7 @@ _COUNT_FETCHABLE = """
     FROM houses h
     JOIN house_sources hs ON hs.house_id = h.id
     WHERE hs.ext_source = 'cian_newbuilding'
-      AND h.cian_zhk_url IS NOT NULL
+      AND (h.cian_zhk_url IS NOT NULL OR hs.ext_id IS NOT NULL)
 """
 
 _COUNT_PENDING = """
@@ -150,13 +173,23 @@ _COUNT_PENDING = """
     FROM houses h
     JOIN house_sources hs ON hs.house_id = h.id
     WHERE hs.ext_source = 'cian_newbuilding'
-      AND h.cian_zhk_url IS NOT NULL
+      AND (h.cian_zhk_url IS NOT NULL OR hs.ext_id IS NOT NULL)
       AND NOT (
           EXISTS (SELECT 1 FROM houses_price_dynamics pd WHERE pd.house_id = h.id)
           AND EXISTS (SELECT 1 FROM house_reliability_checks rc WHERE rc.house_id = h.id)
       )
 """
 
+# UPDATE houses with the resolved ЖК url + the cian newbuilding id, idempotently. Run
+# under a SAVEPOINT before the enrich fetch so a resolved-but-not-enriched house resumes
+# on the next pass (the url is now persisted). psycopg v3: CAST(:x AS type), never `::`.
+_UPDATE_RESOLVED_ZHK_URL = """
+    UPDATE houses SET
+        cian_zhk_url           = :url,
+        cian_internal_house_id = COALESCE(CAST(:nb_id AS bigint), cian_internal_house_id)
+    WHERE id = CAST(:hid AS bigint)
+"""
+
 
 def count_cian_newbuilding_houses(db: Session) -> dict[str, int]:
     """Sizing helper: how many cian_newbuilding houses exist / are fetchable / pending.
@@ -296,15 +329,21 @@ async def backfill_newbuilding_enrichment(
 ) -> NewbuildingEnrichBackfillResult:
     """Backfill the 3 newbuilding-enrichment tables over cian_newbuilding houses.
 
+    Per house: resolve cian_zhk_url from house_sources.ext_id when missing (#972 —
+    cat.php SERP → zhk-slug, persisted under a SAVEPOINT), then fetch + save. Resumable
+    (a resolved-but-not-enriched house picks up on the next pass), idempotent, and
+    rate-limited between BOTH the resolve fetch and the enrich fetch.
+
     Args:
         db: tradein-mvp SQLAlchemy session (caller-owned; NOT gendesign's DB).
         limit: max houses to process this call. DEFAULT 5 — bounded test/proof mode;
-            raise to ~283 for the full run once the proof looks good.
+            raise to ~318 for the full run once the proof looks good AND the proxy is up.
         force: re-process houses already enriched (UPSERT/dedup-guarded). Default False
             skips houses that already have price_dynamics AND reliability rows (reviews
             are optional — see _SELECT_PENDING_HOUSES note).
         request_delay_sec: seconds between fetches. None -> get_scraper_delay('cian')
-            (default 5s). Applied with ±20% jitter; anti-bot politeness.
+            (default 5s). Applied with ±20% jitter; anti-bot politeness. A house needing
+            a resolve incurs TWO delays (resolve fetch + enrich fetch).
         dry_run: count the population + log the pending list, fetch nothing, write nothing.
 
     Returns:
@@ -316,6 +355,7 @@ async def backfill_newbuilding_enrichment(
     # unit tests able to patch fetch_newbuilding cheaply.
     from app.services.scrapers.cian_newbuilding import (
         fetch_newbuilding,
+        resolve_cian_zhk_url_via_search,
         save_newbuilding_enrichment,
     )
 
@@ -359,7 +399,8 @@ async def backfill_newbuilding_enrichment(
 
     for idx, row in enumerate(rows):
         house_id: int = row["house_id"]
-        zhk_url: str = row["cian_zhk_url"]
+        zhk_url: str | None = row["cian_zhk_url"]
+        ext_id: str | None = row["ext_id"]
         result.processed += 1
 
         # Idempotency fast-path: with force=False the SELECT already excludes enriched
@@ -378,6 +419,67 @@ async def backfill_newbuilding_enrichment(
                 )
                 continue
 
+        # ── Resolve ЖК url from ext_id when missing (#972) ──────────────────
+        # The 318 geo-matched cian houses carry house_sources.ext_id (= nb_id) but NULL
+        # cian_zhk_url; resolve it via the cat.php SERP, then PERSIST it (CAST + SAVEPOINT)
+        # so a resolved-but-not-enriched house resumes on the next pass without re-fetching
+        # the SERP. The legacy /zhk/<id>/ resolver 404s — search-based is the working path.
+        if not zhk_url:
+            nb_id = _parse_nb_id(ext_id)
+            if nb_id is None:
+                logger.warning(
+                    "house_id=%s has neither cian_zhk_url nor a numeric ext_id (%r) — skip",
+                    house_id,
+                    ext_id,
+                )
+                result.failed_resolve += 1
+                continue
+
+            try:
+                resolved = await resolve_cian_zhk_url_via_search(nb_id)
+            except Exception as exc:  # defensive — resolver already catches internally
+                logger.warning(
+                    "zhk-url resolve raised house_id=%s nb_id=%s: %s", house_id, nb_id, exc
+                )
+                resolved = None
+
+            # Rate-limit the resolve fetch itself (anti-bot): always sleep after a SERP
+            # hit, whether or not it yielded a url, before the next network call.
+            await _sleep_with_jitter(delay, idx, len(rows), force=True)
+
+            if not resolved:
+                logger.warning(
+                    "zhk-url unresolved house_id=%s nb_id=%s (404 / empty SERP / block) — skip",
+                    house_id,
+                    nb_id,
+                )
+                result.failed_resolve += 1
+                continue
+
+            # Persist under a SAVEPOINT so one bad UPDATE can't poison the batch and the
+            # resolved url survives for the next resume even if the enrich fetch later fails.
+            sp = db.begin_nested()
+            try:
+                db.execute(
+                    text(_UPDATE_RESOLVED_ZHK_URL),
+                    {"url": resolved, "nb_id": nb_id, "hid": house_id},
+                )
+                sp.commit()
+                db.commit()
+            except Exception as exc:
+                sp.rollback()
+                logger.warning(
+                    "persist resolved zhk-url failed house_id=%s nb_id=%s: %s",
+                    house_id,
+                    nb_id,
+                    exc,
+                )
+                result.failed_resolve += 1
+                continue
+
+            zhk_url = resolved
+            result.resolved_zhk_url += 1
+
         # ── Fetch (network; anti-bot surface) ──────────────────────────────
         enrichment = None
         try:
@@ -474,11 +576,14 @@ async def backfill_newbuilding_enrichment(
 
     result.duration_sec = time.time() - t0
     logger.info(
-        "newbuilding-enrich backfill done: processed=%d ok=%d skip=%d "
-        "fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d | %.1fs",
+        "newbuilding-enrich backfill done: processed=%d ok=%d skip=%d resolved=%d "
+        "resolve_fail=%d fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d "
+        "| %.1fs",
         result.processed,
         result.succeeded,
         result.skipped_already_enriched,
+        result.resolved_zhk_url,
+        result.failed_resolve,
         result.failed_fetch,
         result.failed_save,
         result.price_dynamics_rows,
@@ -511,8 +616,29 @@ def _dedup_reliability(db: Session, house_id: int) -> None:
     )
 
 
-async def _sleep_with_jitter(delay: float, idx: int, total: int) -> None:
-    """Polite anti-bot sleep with ±20% jitter; skipped after the last item."""
-    if delay <= 0 or idx >= total - 1:
+def _parse_nb_id(ext_id: str | None) -> int | None:
+    """Parse house_sources.ext_id (TEXT) into a Cian newbuilding id (int), or None.
+
+    ext_id is stored as text; cian_newbuilding links hold the numeric nb_id. Anything
+    non-numeric (or NULL) is unusable for resolution → None.
+    """
+    if ext_id is None:
+        return None
+    try:
+        return int(str(ext_id).strip())
+    except (TypeError, ValueError):
+        return None
+
+
+async def _sleep_with_jitter(delay: float, idx: int, total: int, *, force: bool = False) -> None:
+    """Polite anti-bot sleep with ±20% jitter.
+
+    Skipped after the LAST item (no further fetch follows) — UNLESS force=True, used
+    for the intra-iteration resolve→enrich gap where a second network call still follows
+    within the same house even on the last row.
+    """
+    if delay <= 0:
+        return
+    if not force and idx >= total - 1:
         return
     await asyncio.sleep(delay * random.uniform(0.8, 1.2))

tradein-mvp/backend/tests/fixtures/cian_catph_newbuilding_serp.html

@@ -0,0 +1,15 @@
+<!--
+Trimmed Cian cat.php newbuilding-SERP fixture for resolve_cian_zhk_url_via_search (#972).
+
+Captured live from prod 2026-06-08 (tradein-backend, curl_cffi impersonate=chrome120):
+  GET https://ekb.cian.ru/cat.php?deal_type=sale&engine_version=2&offer_type=flat
+      &object_type[0]=2&newobject[0]=48853
+  → HTTP 200, full body ~2.8 MB. This is a trimmed window around the canonical
+    ЖК-slug anchor (the breadcrumb/title block) — enough to exercise the regex
+    extraction without committing the megabyte SERP.
+
+The full SERP contained the slug URL ~30×; the FIRST match is the canonical one
+(nb 48853 → https://zhk-parkovyy-kvartal-ekb-i.cian.ru — ЖК «Парковый квартал»).
+The legacy resolver's path https://cian.ru/zhk/48853/ returns HTTP 404 (verified).
+-->
+<div data-name="Breadcrumbs"><span class="breadcrumbs-item"><a data-name="breadcrumbs-link" href="https://www.cian.ru/kupit/" title="Продажа"><span>Продажа</span></a></span></div><div data-name="NewbuildingHeaderPanel" class="x31de4314--_9ca1f--left"><div class="x31de4314--_9ca1f--content"><div data-name="NewbuildingHeaderSeo" class="x31de4314--c5a909--content"><h1 data-name="Title" class="x31de4314--_7735e--color_surface-neutral-pressed x31de4314--_2697e--lineHeight_4u">Купить квартиру в <a href="https://zhk-parkovyy-kvartal-ekb-i.cian.ru/" target="_blank" rel="noreferrer" class="x31de4314--_9d71c--link">ЖК «Парковый квартал»</a></h1><span data-name="Description" class="x31de4314--_7735e--color_surface-neutral-pressed">Продажа квартир в новостройке от застройщика. Актуальные цены, планировки и фотографии.</span></div></div></div>

tradein-mvp/backend/tests/services/scrapers/test_cian_resolve_zhk_url.py

@@ -0,0 +1,145 @@
+"""Tests for resolve_cian_zhk_url_via_search (#972) — the WORKING nb_id → ЖК-url resolver.
+
+The legacy resolve_cian_zhk_url hit https://cian.ru/zhk/<id>/ which now 404s; the
+search-based resolver fetches the cat.php newbuilding-SERP and extracts the canonical
+zhk-<slug>.cian.ru URL from its markup. No real network: the curl_cffi session is mocked;
+the regex is exercised against a trimmed real-prod SERP fixture.
+"""
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+# DATABASE_URL required by config before any app import (cian_newbuilding → app.core.config).
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+# WeasyPrint stub — not installed in CI without GTK; pulled in transitively by app imports.
+sys.modules.setdefault("weasyprint", MagicMock())
+
+import pytest  # noqa: E402
+
+from app.services.scrapers.cian_newbuilding import (  # noqa: E402
+    _extract_zhk_url_from_serp,
+    resolve_cian_zhk_url_via_search,
+)
+
+_FIXTURE = (
+    Path(__file__).resolve().parents[2] / "fixtures" / "cian_catph_newbuilding_serp.html"
+)
+
+
+def _serp_fixture() -> str:
+    return _FIXTURE.read_text(encoding="utf-8")
+
+
+def _mock_session(*, status_code: int, text: str) -> MagicMock:
+    """A curl_cffi AsyncSession stand-in returning one canned response."""
+    session = MagicMock()
+    response = MagicMock()
+    response.status_code = status_code
+    response.text = text
+    session.get = AsyncMock(return_value=response)
+    session.close = AsyncMock()
+    return session
+
+
+# ---- pure regex extractor -------------------------------------------------
+
+
+def test_extract_zhk_url_from_real_serp_fixture() -> None:
+    """The trimmed prod cat.php SERP yields the canonical ЖК-slug URL (nb 48853).
+
+    The fixture carries the real <h1 data-name="Title"> header block, so this exercises
+    the title-anchored extraction path against real prod markup.
+    """
+    url = _extract_zhk_url_from_serp(_serp_fixture())
+    assert url == "https://zhk-parkovyy-kvartal-ekb-i.cian.ru"
+
+
+def test_extract_zhk_url_anchors_on_title_not_recommendation() -> None:
+    """Adversarial (#972 hardening): a recommendation / "похожие ЖК" block embedding a
+    FOREIGN zhk-slug appears BEFORE the title block whose anchor is the queried ЖК.
+
+    A naive first-match would resolve the recommendation's `zhk-other-ad` and then enrich
+    the house with the WRONG building's price_dynamics/reliability (which the backfill's
+    idempotency would lock in). The title-anchored extractor must return the QUERIED ЖК
+    (`zhk-target`), proving the foreign url before the title is ignored.
+    """
+    html = (
+        '<div data-name="Breadcrumbs">'
+        '<a href="https://www.cian.ru/kupit/">Продажа</a></div>'
+        # Banner / recommendation block rendered BEFORE the title, carrying a foreign slug.
+        '<div data-name="Recommendations">'
+        '<a href="https://zhk-other-ad-ekb-i.cian.ru/">похожий ЖК (реклама)</a>'
+        'and a bare one too https://zhk-another-promo-ekb-i.cian.ru in text</div>'
+        # The canonical title block — its anchor is the queried ЖК.
+        '<div data-name="NewbuildingHeaderSeo">'
+        '<h1 data-name="Title" class="x31de4314--_7735e">Купить квартиру в '
+        '<a href="https://zhk-target-ekb-i.cian.ru/" target="_blank" '
+        'rel="noreferrer" class="x31de4314--_9d71c--link">ЖК «Целевой»</a></h1></div>'
+        # Trailing offer cards repeating the target slug (как на реальном SERP).
+        '<a data-name="LinkArea" href="https://zhk-target-ekb-i.cian.ru/kvartiry/">x</a>'
+    )
+    assert _extract_zhk_url_from_serp(html) == "https://zhk-target-ekb-i.cian.ru"
+
+
+def test_extract_zhk_url_fallback_first_match_when_no_title() -> None:
+    """Markup drift: no <h1 data-name="Title"> header → degrade to naive first-match.
+
+    The title anchor is the primary path; when it is absent (header renamed/moved) the
+    extractor falls back to the first zhk-slug occurrence so resolution still yields
+    *something* rather than failing outright.
+    """
+    html = (
+        'noise <a href="https://zhk-first-ekb-i.cian.ru/">A</a> '
+        'and later https://zhk-second-ekb-i.cian.ru more'
+    )
+    assert _extract_zhk_url_from_serp(html) == "https://zhk-first-ekb-i.cian.ru"
+
+
+def test_extract_zhk_url_no_match_returns_none() -> None:
+    """A SERP with no zhk-slug anchor (empty results / drift) → None."""
+    assert _extract_zhk_url_from_serp("<html><body>Ничего не найдено</body></html>") is None
+    # A bare cian.ru link that is NOT a zhk-* subdomain must not match.
+    assert _extract_zhk_url_from_serp('<a href="https://www.cian.ru/kupit/">x</a>') is None
+
+
+# ---- async resolver (session mocked) --------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_resolve_via_search_happy_path() -> None:
+    """HTTP 200 + a SERP containing the slug → canonical url; session reused, not closed."""
+    session = _mock_session(status_code=200, text=_serp_fixture())
+    url = await resolve_cian_zhk_url_via_search(48853, session=session)
+    assert url == "https://zhk-parkovyy-kvartal-ekb-i.cian.ru"
+    # caller owns a passed-in session — resolver must NOT close it.
+    session.close.assert_not_awaited()
+    # the requested URL targets the cat.php newbuilding SERP for this nb_id.
+    called_url = session.get.call_args.args[0]
+    assert "cat.php" in called_url
+    assert "newobject[0]=48853" in called_url
+
+
+@pytest.mark.asyncio
+async def test_resolve_via_search_no_match_returns_none() -> None:
+    """HTTP 200 but no zhk-slug in the body → None (empty SERP / markup drift)."""
+    session = _mock_session(status_code=200, text="<html>no results</html>")
+    assert await resolve_cian_zhk_url_via_search(999, session=session) is None
+
+
+@pytest.mark.asyncio
+async def test_resolve_via_search_non_200_returns_none() -> None:
+    """A non-200 (block / 404) → None without attempting extraction."""
+    session = _mock_session(status_code=429, text="too many requests")
+    assert await resolve_cian_zhk_url_via_search(48853, session=session) is None
+
+
+@pytest.mark.asyncio
+async def test_resolve_via_search_swallows_request_error() -> None:
+    """A transport error inside the fetch → None (logged), never propagates."""
+    session = MagicMock()
+    session.get = AsyncMock(side_effect=ConnectionError("proxy refused"))
+    session.close = AsyncMock()
+    assert await resolve_cian_zhk_url_via_search(48853, session=session) is None

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

@@ -56,13 +56,17 @@ class FakeDB:
     """
 
     def __init__(self, pending_rows):
-        self._pending_rows = pending_rows  # list[dict house_id, cian_zhk_url]
+        # list[dict house_id, cian_zhk_url, (optional) ext_id]; default ext_id=None so
+        # existing tests that omit it still model "url already present".
+        self._pending_rows = [{"ext_id": None, **r} for r in pending_rows]
         # price_dynamics keyed by (house_id, month_date, source, room_count, prices_type, period)
         self.price_dynamics: set[tuple] = set()
         # reviews keyed by (source, ext_review_id)
         self.reviews: dict[tuple, dict] = {}
         # reliability: list of (house_id, source)
         self.reliability: list[tuple] = []
+        # houses.cian_zhk_url written by the resolve step: {house_id: (url, nb_id)}
+        self.resolved_urls: dict[int, tuple] = {}
 
     # -- query router ------------------------------------------------------
     def execute(self, statement, params=None):
@@ -79,9 +83,21 @@ class FakeDB:
             # _COUNT_TOTAL
             return _FakeResult(scalar=len(self._pending_rows))
 
-        if "SELECT DISTINCT h.id AS house_id" in sql:
+        if "SELECT DISTINCT ON (h.id) h.id AS house_id" in sql:
             lim = params.get("lim", len(self._pending_rows))
-            return _FakeResult(rows=self._pending_rows[:lim])
+            # When a row was resolved on a prior pass, surface the persisted url (mirrors
+            # the real houses.cian_zhk_url UPDATE so a re-run sees it set).
+            out = []
+            for r in self._pending_rows[:lim]:
+                resolved = self.resolved_urls.get(r["house_id"])
+                if resolved is not None and not r.get("cian_zhk_url"):
+                    r = {**r, "cian_zhk_url": resolved[0]}
+                out.append(r)
+            return _FakeResult(rows=out)
+
+        if "UPDATE houses SET" in sql and "cian_zhk_url" in sql and "cian_internal_house_id" in sql:
+            self.resolved_urls[params["hid"]] = (params["url"], params["nb_id"])
+            return _FakeResult()
 
         if "COUNT(*) FROM houses_price_dynamics" in sql:
             h = params["h"]
@@ -366,3 +382,137 @@ async def test_backfill_fetch_failure_does_not_abort_batch() -> None:
     assert result.failed_fetch == 1
     assert result.succeeded == 1
     assert len(db.reviews) == 2  # only house 2 enriched
+
+
+# ---------------------------------------------------------------------------
+# #972 ЖК-url resolution wiring: ext_id (NULL cian_zhk_url) → resolve → persist → enrich.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backfill_resolves_zhk_url_from_ext_id_then_enriches() -> None:
+    """A house with ext_id but NULL cian_zhk_url is resolved FIRST (cat.php SERP), the
+    url is persisted, and only THEN is the enrichment fetched + saved.
+    """
+    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": None, "ext_id": "48853"}])
+    enr = _enrichment_with_everything()
+
+    async def _resolve(nb_id, **kwargs):
+        assert nb_id == 48853  # ext_id parsed to int
+        return "https://zhk-parkovyy-kvartal-ekb-i.cian.ru"
+
+    fetched_urls: list[str] = []
+
+    async def _fetch(url, **kwargs):
+        fetched_urls.append(url)
+        return enr
+
+    with (
+        patch(
+            "app.services.scrapers.cian_newbuilding.resolve_cian_zhk_url_via_search",
+            side_effect=_resolve,
+        ),
+        patch(
+            "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
+            side_effect=_fetch,
+        ),
+        patch(
+            "app.services.scrapers.cian_newbuilding.save_newbuilding_enrichment",
+            side_effect=_fake_save_newbuilding_enrichment,
+        ),
+    ):
+        result = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
+
+    # url resolved + persisted to houses (CAST/UPDATE), with the nb_id carried through.
+    assert db.resolved_urls[1] == ("https://zhk-parkovyy-kvartal-ekb-i.cian.ru", 48853)
+    # enrich fetched the RESOLVED url (not the NULL original).
+    assert fetched_urls == ["https://zhk-parkovyy-kvartal-ekb-i.cian.ru"]
+    assert result.resolved_zhk_url == 1
+    assert result.failed_resolve == 0
+    assert result.succeeded == 1
+    # enrichment landed.
+    assert len(db.price_dynamics) == 1
+    assert len(db.reliability) == 1
+
+
+@pytest.mark.asyncio
+async def test_backfill_unresolved_ext_id_counts_fail_and_skips_fetch() -> None:
+    """Resolver returns None (404 / empty SERP / block) → failed_resolve, no fetch."""
+    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": None, "ext_id": "999"}])
+
+    fetch_calls = {"n": 0}
+
+    async def _fetch(url, **kwargs):
+        fetch_calls["n"] += 1
+        return _enrichment_with_everything()
+
+    with (
+        patch(
+            "app.services.scrapers.cian_newbuilding.resolve_cian_zhk_url_via_search",
+            new_callable=AsyncMock,
+            return_value=None,
+        ),
+        patch(
+            "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
+            side_effect=_fetch,
+        ),
+        patch(
+            "app.services.scrapers.cian_newbuilding.save_newbuilding_enrichment",
+            side_effect=_fake_save_newbuilding_enrichment,
+        ),
+    ):
+        result = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
+
+    assert result.failed_resolve == 1
+    assert result.resolved_zhk_url == 0
+    assert result.succeeded == 0
+    assert fetch_calls["n"] == 0  # never fetched without a url
+    assert 1 not in db.resolved_urls
+
+
+@pytest.mark.asyncio
+async def test_backfill_resolve_then_idempotent_rerun_no_reresolve() -> None:
+    """After a resolve+enrich pass the house is enriched; a plain re-run skips it — no
+    second resolve, no second fetch, no duplicate rows.
+    """
+    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": None, "ext_id": "48853"}])
+    enr = _enrichment_with_everything()
+    enr.reviews = []  # the common Cian case (initialState carries no reviews)
+
+    resolve_calls = {"n": 0}
+    fetch_calls = {"n": 0}
+
+    async def _resolve(nb_id, **kwargs):
+        resolve_calls["n"] += 1
+        return "https://zhk-parkovyy-kvartal-ekb-i.cian.ru"
+
+    async def _fetch(url, **kwargs):
+        fetch_calls["n"] += 1
+        return enr
+
+    with (
+        patch(
+            "app.services.scrapers.cian_newbuilding.resolve_cian_zhk_url_via_search",
+            side_effect=_resolve,
+        ),
+        patch(
+            "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
+            side_effect=_fetch,
+        ),
+        patch(
+            "app.services.scrapers.cian_newbuilding.save_newbuilding_enrichment",
+            side_effect=_fake_save_newbuilding_enrichment,
+        ),
+    ):
+        r1 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
+        # Second pass: the house now has pd+reliability, so the per-house guard skips it.
+        r2 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
+
+    assert r1.resolved_zhk_url == 1
+    assert r1.succeeded == 1
+    assert resolve_calls["n"] == 1  # not re-resolved on the second pass
+    assert fetch_calls["n"] == 1  # not re-fetched on the second pass
+    assert r2.skipped_already_enriched == 1
+    assert r2.succeeded == 0
+    assert len(db.price_dynamics) == 1  # no duplicate
+    assert len(db.reliability) == 1