feat(tradein): backfill task for cian newbuilding enrichment (#972)

950-E1: app/tasks/newbuilding_enrich_backfill.py selects houses linked to
ext_source='cian_newbuilding' with a fetchable cian_zhk_url and runs the existing
fetch_newbuilding + save_newbuilding_enrichment over each, populating
houses_price_dynamics + house_reliability_checks (+ house_reviews when present).

- Idempotent + resumable: skips already-enriched houses; force= re-run dup-safe
  (price_dynamics UPSERT via dim_key, reliability dedup guard with id-DESC
  tiebreaker, reviews ON CONFLICT (source,ext_review_id)).
- Anti-bot aware: get_scraper_delay('cian') + jitter, SAVEPOINT-per-house
  (one captcha/failure logs + continues, never aborts the batch).
- limit= for bounded proof runs; sizing counters (total/fetchable/pending).

Fix cian_newbuilding._extract_transport_rate: cian transportAccessibilityRate
drifted to a nested dict and broke save_newbuilding_enrichment's CAST bind on
LIVE data (cannot adapt type 'dict') — coerce dict/float/bad -> int|None
(bool excluded before int). Fixes the live enrichment crash, not just backfill.

house_reviews stays ~0 for now: cian reviews are in a separate XHR, not the ЖК
initialState — extending fetch_newbuilding for them is a documented follow-up.

Tested: 8 new unit tests; isolated-DB proof-run landed 6 price_dynamics + 1
reliability row, idempotency proven across re-runs. Refs #972.

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

@@ -139,7 +139,7 @@ async def fetch_newbuilding(
             floors_count_max=nb.get("floorsCountMax"),
             building_class=nb.get("newbuildingClassName") or nb.get("buildingClass"),
             management_company=nb.get("managementCompany"),
-            transport_accessibility_rate=nb.get("transportAccessibilityRate"),
+            transport_accessibility_rate=_extract_transport_rate(nb),
             advantages=_extract_advantages(nb),
             builders=_safe_list(nb.get("builders")),
             banks=_safe_list(nb.get("banks")),
@@ -224,6 +224,28 @@ def _extract_deadline_year(nb: dict[str, Any]) -> int | None:
     return None
 
 
+def _extract_transport_rate(nb: dict[str, Any]) -> int | None:
+    """Extract transport accessibility rate as int|None.
+
+    Cian payload drift (#972): `transportAccessibilityRate` is now a nested dict
+    ({'transportAccessibility': <int|null>}) rather than a bare int. Older pages
+    returned a plain int. Handle both — returning a dict here would violate the
+    NewbuildingEnrichment.transport_accessibility_rate: int|None contract and break
+    the CAST(:tar AS ...) bind in save_newbuilding_enrichment (psycopg cannot adapt
+    a dict). Anything non-int collapses to None.
+    """
+    raw = nb.get("transportAccessibilityRate")
+    if isinstance(raw, dict):
+        raw = raw.get("transportAccessibility") or raw.get("rate") or raw.get("value")
+    if isinstance(raw, bool):  # bool is an int subclass — exclude explicitly
+        return None
+    if isinstance(raw, int):
+        return raw
+    if isinstance(raw, float):
+        return int(raw)
+    return None
+
+
 def _extract_advantages(nb: dict[str, Any]) -> list[dict[str, Any]]:
     """Extract typed advantages list from newbuilding.advantages."""
     raw = nb.get("advantages") or nb.get("advantagesTyped") or []

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

@@ -0,0 +1,518 @@
+"""Backfill task #972 (950-E1): enrich the 283 cian_newbuilding houses.
+
+Context
+-------
+Three newbuilding-enrichment tables are EMPTY in prod (0 rows) — the Cian
+newbuilding "Phase 4" enrichment never backfilled the existing house cache:
+
+  - houses_price_dynamics      (realtyValuation 7-month chart)
+  - 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:
+
+  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)
+
+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)
+    and reports how many of those are actually fetchable (have a zhk_url).
+  - save_newbuilding_enrichment() does NOT persist reviews and house_reliability_checks
+    has no UNIQUE constraint, so naive re-runs would duplicate it. This task adds an
+    idempotent review writer + a reliability-dedup guard WITHOUT touching the shared
+    save_newbuilding_enrichment() (keeps the SERP-sweep path untouched).
+
+Idempotency
+-----------
+  - Default: skips a house that already has rows in ALL three target tables.
+  - force=True: re-runs every selected house (price_dynamics UPSERTs via its dim_key;
+    reliability is skipped if a cian_nashdom row already exists; reviews UPSERT via
+    (source, ext_review_id)). Safe to re-run; a partial/aborted run resumes cleanly.
+
+Anti-bot / resilience
+---------------------
+  - Polite per-house delay = get_scraper_delay('cian') (default 5s) with ±20% jitter.
+  - SAVEPOINT per house: one failed fetch / captcha / save logs + continues; the batch
+    is never aborted on a single house. Counters track enriched vs failed.
+
+Execution
+---------
+  - tradein-mvp has NO Celery app (see app/tasks/refresh_search_matview.py). This is a
+    plain async callable, driven by tradein's own DB session (app.core.db.SessionLocal /
+    get_db) — NEVER gendesign's DB. Trigger via the in-app scheduler or an admin endpoint.
+  - psycopg v3 conventions: CAST(:x AS type) in SQL (never `::`); logger (never print).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import random
+import time
+from dataclasses import dataclass, field, fields
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services.scraper_settings import get_scraper_delay
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "NewbuildingEnrichBackfillResult",
+    "backfill_newbuilding_enrichment",
+    "count_cian_newbuilding_houses",
+]
+
+# Source tags written by the Cian newbuilding enrichment (kept in sync with
+# save_newbuilding_enrichment + _save_cian_reviews below).
+_PRICE_DYNAMICS_SOURCE = "cian_realty_valuation"
+_RELIABILITY_SOURCE = "cian_nashdom"
+_REVIEWS_SOURCE = "cian"
+
+
+@dataclass
+class NewbuildingEnrichBackfillResult:
+    """Per-run counters for the newbuilding-enrichment backfill."""
+
+    # 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_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
+    failed_fetch: int = 0                # fetch returned None / raised
+    failed_save: int = 0                 # save raised after a good fetch
+
+    # Row-level deltas (how much actually landed).
+    price_dynamics_rows: int = 0
+    reliability_rows: int = 0
+    review_rows: int = 0
+
+    duration_sec: float = field(default=0.0)
+
+    def to_dict(self) -> dict[str, int]:
+        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.
+#
+# "Enriched" = has price_dynamics AND reliability rows. We deliberately do NOT require
+# house_reviews here: the Cian newbuilding initialState rarely carries the reviews list
+# (verified on real ЖК pages — reviews load via a separate XHR the current scraper does
+# 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.
+_SELECT_PENDING_HOUSES = """
+    SELECT DISTINCT h.id AS house_id, h.cian_zhk_url
+    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 (
+        CAST(:force AS boolean) = TRUE
+        OR 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)
+        )
+      )
+    ORDER BY h.id
+    LIMIT :lim
+"""
+
+_COUNT_TOTAL = """
+    SELECT COUNT(DISTINCT h.id)
+    FROM houses h
+    JOIN house_sources hs ON hs.house_id = h.id
+    WHERE hs.ext_source = 'cian_newbuilding'
+"""
+
+_COUNT_FETCHABLE = """
+    SELECT COUNT(DISTINCT h.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
+"""
+
+_COUNT_PENDING = """
+    SELECT COUNT(DISTINCT h.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 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)
+      )
+"""
+
+
+def count_cian_newbuilding_houses(db: Session) -> dict[str, int]:
+    """Sizing helper: how many cian_newbuilding houses exist / are fetchable / pending.
+
+    Cheap (3 COUNTs) — safe to call before a run to gauge the population and a dry-run.
+    """
+    total = int(db.execute(text(_COUNT_TOTAL)).scalar_one())
+    fetchable = int(db.execute(text(_COUNT_FETCHABLE)).scalar_one())
+    pending = int(db.execute(text(_COUNT_PENDING)).scalar_one())
+    return {"total": total, "fetchable": fetchable, "pending": pending}
+
+
+def _save_cian_reviews(db: Session, house_id: int, reviews: list[dict]) -> int:
+    """Persist Cian ЖК reviews to house_reviews (source='cian'), idempotently.
+
+    save_newbuilding_enrichment() does not write reviews; this fills that gap so the
+    house_reviews table becomes non-zero from the cian path. The Avito review writer
+    uses source='avito'; we tag cian rows source='cian' so the two never collide on
+    the (source, ext_review_id) UNIQUE key.
+
+    Each `review` is a raw Cian review dict from NewbuildingEnrichment.reviews. We map
+    defensively (Cian field names vary by MFE version) and skip entries with no usable
+    external id — without a stable ext_review_id we cannot dedup, so we drop rather than
+    risk duplicate inserts on re-run.
+
+    Caller is responsible for the surrounding SAVEPOINT/commit.
+    """
+    if not reviews:
+        return 0
+
+    saved = 0
+    for r in reviews:
+        if not isinstance(r, dict):
+            continue
+        ext_review_id = r.get("id") or r.get("reviewId") or r.get("review_id")
+        if ext_review_id is None:
+            continue
+        try:
+            ext_review_id_int = int(ext_review_id)
+        except (TypeError, ValueError):
+            continue
+
+        author = r.get("author")
+        author_name = None
+        if isinstance(author, dict):
+            author_name = author.get("name") or author.get("title")
+        elif isinstance(author, str):
+            author_name = author
+        author_name = author_name or r.get("authorName") or r.get("userName")
+
+        score = r.get("rate") or r.get("score") or r.get("rating")
+        try:
+            score_int = int(score) if score is not None else None
+        except (TypeError, ValueError):
+            score_int = None
+        # house_reviews.score has a CHECK (1..5) — clamp out-of-range / drop bad values.
+        if score_int is not None and not (1 <= score_int <= 5):
+            score_int = None
+
+        db.execute(
+            text("""
+                INSERT INTO house_reviews (
+                    house_id, source, ext_review_id,
+                    author_name, review_title, score,
+                    model_experience, rated_date,
+                    text_main, text_pros, text_cons,
+                    raw_payload, scraped_at
+                ) VALUES (
+                    CAST(:house_id AS bigint), :source, CAST(:ext_review_id AS bigint),
+                    CAST(:author_name AS text), CAST(:review_title AS text),
+                    CAST(:score AS int),
+                    CAST(:model_experience AS text), CAST(:rated_date AS date),
+                    CAST(:text_main AS text), CAST(:text_pros AS text),
+                    CAST(:text_cons AS text),
+                    CAST(:raw_payload AS jsonb), NOW()
+                )
+                ON CONFLICT (source, ext_review_id) DO UPDATE SET
+                    author_name  = EXCLUDED.author_name,
+                    review_title = EXCLUDED.review_title,
+                    score        = EXCLUDED.score,
+                    text_main    = EXCLUDED.text_main,
+                    raw_payload  = EXCLUDED.raw_payload,
+                    scraped_at   = NOW()
+            """),
+            {
+                "house_id": house_id,
+                "source": _REVIEWS_SOURCE,
+                "ext_review_id": ext_review_id_int,
+                "author_name": author_name,
+                "review_title": r.get("title") or r.get("reviewTitle"),
+                "score": score_int,
+                "model_experience": r.get("experience") or r.get("modelExperience"),
+                "rated_date": None,  # Cian review date format varies; raw kept in payload
+                "text_main": r.get("text") or r.get("body") or r.get("comment"),
+                "text_pros": r.get("pros") or r.get("advantages"),
+                "text_cons": r.get("cons") or r.get("disadvantages"),
+                "raw_payload": json.dumps(r, ensure_ascii=False),
+            },
+        )
+        saved += 1
+    return saved
+
+
+def _house_enrichment_counts(db: Session, house_id: int) -> tuple[int, int, int]:
+    """Return (price_dynamics, reliability, reviews) row counts for a house."""
+    pd = int(
+        db.execute(
+            text("SELECT COUNT(*) FROM houses_price_dynamics WHERE house_id = CAST(:h AS bigint)"),
+            {"h": house_id},
+        ).scalar_one()
+    )
+    rc = int(
+        db.execute(
+            text(
+                "SELECT COUNT(*) FROM house_reliability_checks "
+                "WHERE house_id = CAST(:h AS bigint)"
+            ),
+            {"h": house_id},
+        ).scalar_one()
+    )
+    rv = int(
+        db.execute(
+            text("SELECT COUNT(*) FROM house_reviews WHERE house_id = CAST(:h AS bigint)"),
+            {"h": house_id},
+        ).scalar_one()
+    )
+    return pd, rc, rv
+
+
+async def backfill_newbuilding_enrichment(
+    db: Session,
+    *,
+    limit: int = 5,
+    force: bool = False,
+    request_delay_sec: float | None = None,
+    dry_run: bool = False,
+) -> NewbuildingEnrichBackfillResult:
+    """Backfill the 3 newbuilding-enrichment tables over cian_newbuilding houses.
+
+    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.
+        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.
+        dry_run: count the population + log the pending list, fetch nothing, write nothing.
+
+    Returns:
+        NewbuildingEnrichBackfillResult with population sizing, per-house outcome
+        counters, and row-level deltas across the 3 tables.
+    """
+    # Import here (not module top) so the heavy curl_cffi scraper stack is only loaded
+    # when the task actually runs — mirrors cian_history_backfill's lazy import and keeps
+    # unit tests able to patch fetch_newbuilding cheaply.
+    from app.services.scrapers.cian_newbuilding import (
+        fetch_newbuilding,
+        save_newbuilding_enrichment,
+    )
+
+    result = NewbuildingEnrichBackfillResult()
+    t0 = time.time()
+    delay = request_delay_sec if request_delay_sec is not None else get_scraper_delay("cian")
+
+    # ── Population sizing (always; cheap, helps the orchestrator decide on a full run) ──
+    sizing = count_cian_newbuilding_houses(db)
+    result.cian_houses_total = sizing["total"]
+    result.cian_houses_fetchable = sizing["fetchable"]
+    result.cian_houses_pending = sizing["pending"]
+
+    rows = (
+        db.execute(text(_SELECT_PENDING_HOUSES), {"force": force, "lim": limit})
+        .mappings()
+        .all()
+    )
+
+    logger.info(
+        "newbuilding-enrich backfill: cian_houses total=%d fetchable=%d pending=%d; "
+        "selected=%d (limit=%d force=%s dry_run=%s delay=%.1fs)",
+        result.cian_houses_total,
+        result.cian_houses_fetchable,
+        result.cian_houses_pending,
+        len(rows),
+        limit,
+        force,
+        dry_run,
+        delay,
+    )
+
+    if dry_run:
+        logger.info(
+            "dry_run: would process %d cian_newbuilding houses (ids=%s)",
+            len(rows),
+            [r["house_id"] for r in rows],
+        )
+        result.duration_sec = time.time() - t0
+        return result
+
+    for idx, row in enumerate(rows):
+        house_id: int = row["house_id"]
+        zhk_url: str = row["cian_zhk_url"]
+        result.processed += 1
+
+        # Idempotency fast-path: with force=False the SELECT already excludes enriched
+        # houses (price_dynamics + reliability present), so this branch is a belt-and-
+        # braces guard against a concurrent writer between SELECT and processing.
+        if not force:
+            pd, rc, rv = _house_enrichment_counts(db, house_id)
+            if pd > 0 and rc > 0:
+                result.skipped_already_enriched += 1
+                logger.debug(
+                    "skip house_id=%s — already enriched (pd=%d rc=%d rv=%d)",
+                    house_id,
+                    pd,
+                    rc,
+                    rv,
+                )
+                continue
+
+        # ── Fetch (network; anti-bot surface) ──────────────────────────────
+        enrichment = None
+        try:
+            enrichment = await fetch_newbuilding(zhk_url)
+        except Exception as exc:
+            logger.warning(
+                "newbuilding fetch failed house_id=%s url=%s: %s", house_id, zhk_url, exc
+            )
+            result.failed_fetch += 1
+            await _sleep_with_jitter(delay, idx, len(rows))
+            continue
+
+        if enrichment is None:
+            logger.warning(
+                "newbuilding fetch returned None house_id=%s url=%s (captcha / parse miss?)",
+                house_id,
+                zhk_url,
+            )
+            result.failed_fetch += 1
+            await _sleep_with_jitter(delay, idx, len(rows))
+            continue
+
+        # ── Save under a SAVEPOINT so one bad house can't poison the batch ──
+        # begin_nested() = SAVEPOINT; save_newbuilding_enrichment commits internally,
+        # so we snapshot the row counts BEFORE and recompute the delta AFTER its commit
+        # rather than relying on the nested transaction staying open.
+        pd_before, rc_before, rv_before = _house_enrichment_counts(db, house_id)
+        try:
+            had_reliability = rc_before > 0
+
+            # 1) price_dynamics + reliability + houses UPDATE (existing, commits inside).
+            save_newbuilding_enrichment(db, house_id, enrichment)
+
+            # 2) reviews — added here (save_newbuilding_enrichment skips them).
+            #    SAVEPOINT around the review write so a malformed review can't lose the
+            #    price/reliability rows already committed above.
+            review_written = 0
+            if enrichment.reviews:
+                sp = db.begin_nested()
+                try:
+                    review_written = _save_cian_reviews(db, house_id, enrichment.reviews)
+                    sp.commit()
+                    db.commit()
+                except Exception as rexc:
+                    sp.rollback()
+                    logger.warning(
+                        "review save failed house_id=%s (price/reliability kept): %s",
+                        house_id,
+                        rexc,
+                    )
+
+            # 3) reliability dedup guard: house_reliability_checks has no UNIQUE
+            #    constraint, so save_newbuilding_enrichment appends a fresh row every
+            #    pass. If the house already had a cian_nashdom row BEFORE this pass
+            #    (i.e. we just re-processed it), collapse to the single newest row so
+            #    repeated runs can't accumulate duplicates. Runs regardless of `force`
+            #    because a force=False resume can also re-touch a partially-saved house.
+            if had_reliability:
+                sp = db.begin_nested()
+                try:
+                    _dedup_reliability(db, house_id)
+                    sp.commit()
+                    db.commit()
+                except Exception as dexc:
+                    sp.rollback()
+                    logger.warning(
+                        "reliability dedup failed house_id=%s: %s", house_id, dexc
+                    )
+
+            pd_after, rc_after, rv_after = _house_enrichment_counts(db, house_id)
+            result.price_dynamics_rows += max(0, pd_after - pd_before)
+            result.reliability_rows += max(0, rc_after - rc_before)
+            result.review_rows += max(0, rv_after - rv_before)
+            result.succeeded += 1
+            logger.info(
+                "enriched house_id=%s: +pd=%d +reliability=%d +reviews=%d (parsed reviews=%d)",
+                house_id,
+                max(0, pd_after - pd_before),
+                max(0, rc_after - rc_before),
+                review_written,
+                len(enrichment.reviews),
+            )
+        except Exception as exc:
+            result.failed_save += 1
+            logger.warning("newbuilding save failed house_id=%s: %s", house_id, exc)
+            # save_newbuilding_enrichment may have left the session dirty — roll back so
+            # the next house starts clean (mirrors cian_history_backfill behaviour).
+            try:
+                db.rollback()
+            except Exception as rb_exc:
+                logger.warning("rollback failed house_id=%s: %s", house_id, rb_exc)
+
+        await _sleep_with_jitter(delay, idx, len(rows))
+
+    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",
+        result.processed,
+        result.succeeded,
+        result.skipped_already_enriched,
+        result.failed_fetch,
+        result.failed_save,
+        result.price_dynamics_rows,
+        result.reliability_rows,
+        result.review_rows,
+        result.duration_sec,
+    )
+    return result
+
+
+def _dedup_reliability(db: Session, house_id: int) -> None:
+    """Keep only the newest cian_nashdom reliability row for a house.
+
+    house_reliability_checks has no UNIQUE constraint; a force re-run appends a fresh
+    row each pass. Collapse to the most-recent row so re-runs don't accumulate dupes.
+    """
+    db.execute(
+        text("""
+            DELETE FROM house_reliability_checks
+            WHERE house_id = CAST(:h AS bigint)
+              AND source = :src
+              AND id NOT IN (
+                  SELECT id FROM house_reliability_checks
+                  WHERE house_id = CAST(:h AS bigint) AND source = :src
+                  ORDER BY recorded_at DESC, id DESC
+                  LIMIT 1
+              )
+        """),
+        {"h": house_id, "src": _RELIABILITY_SOURCE},
+    )
+
+
+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:
+        return
+    await asyncio.sleep(delay * random.uniform(0.8, 1.2))

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

@@ -0,0 +1,368 @@
+"""Tests for newbuilding_enrich_backfill task (#972 / 950-E1).
+
+No real network, no real Postgres. The Cian fetch is mocked; the DB session is a
+lightweight in-memory fake that models just enough of the three target tables to
+assert (a) rows land and (b) an idempotent re-run does not duplicate them.
+"""
+from __future__ import annotations
+
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+# DATABASE_URL required by config before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+# WeasyPrint stub — not installed in CI without GTK.
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+
+import pytest  # noqa: E402
+
+from app.services.scrapers.cian_newbuilding import NewbuildingEnrichment  # noqa: E402
+from app.tasks.newbuilding_enrich_backfill import (  # noqa: E402
+    NewbuildingEnrichBackfillResult,
+    _save_cian_reviews,
+    backfill_newbuilding_enrichment,
+)
+
+# ---------------------------------------------------------------------------
+# In-memory fake Session — models houses_price_dynamics / house_reliability_checks
+# / house_reviews well enough for count + idempotency assertions.
+# ---------------------------------------------------------------------------
+
+
+class _FakeResult:
+    def __init__(self, *, scalar=None, rows=None):
+        self._scalar = scalar
+        self._rows = rows or []
+
+    def scalar_one(self):
+        return self._scalar
+
+    def mappings(self):
+        return self
+
+    def all(self):
+        return self._rows
+
+
+class FakeDB:
+    """Minimal SQLAlchemy-Session stand-in keyed off SQL text fragments.
+
+    Stores price_dynamics / reliability / reviews as sets/lists keyed by their
+    natural dedup key so re-inserts are no-ops (mirrors the real ON CONFLICT /
+    dedup-guard behaviour the task relies on).
+    """
+
+    def __init__(self, pending_rows):
+        self._pending_rows = pending_rows  # list[dict house_id, cian_zhk_url]
+        # 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] = []
+
+    # -- query router ------------------------------------------------------
+    def execute(self, statement, params=None):
+        sql = str(statement)
+        params = params or {}
+
+        if "COUNT(DISTINCT h.id)" in sql and "houses_price_dynamics pd" in sql:
+            # _COUNT_PENDING
+            return _FakeResult(scalar=len(self._pending_rows))
+        if "COUNT(DISTINCT h.id)" in sql and "cian_zhk_url IS NOT NULL" in sql:
+            # _COUNT_FETCHABLE
+            return _FakeResult(scalar=len(self._pending_rows))
+        if "COUNT(DISTINCT h.id)" in sql:
+            # _COUNT_TOTAL
+            return _FakeResult(scalar=len(self._pending_rows))
+
+        if "SELECT DISTINCT h.id AS house_id" in sql:
+            lim = params.get("lim", len(self._pending_rows))
+            return _FakeResult(rows=self._pending_rows[:lim])
+
+        if "COUNT(*) FROM houses_price_dynamics" in sql:
+            h = params["h"]
+            return _FakeResult(scalar=sum(1 for k in self.price_dynamics if k[0] == h))
+        if "COUNT(*) FROM house_reliability_checks" in sql:
+            h = params["h"]
+            return _FakeResult(scalar=sum(1 for k in self.reliability if k[0] == h))
+        if "COUNT(*) FROM house_reviews" in sql:
+            h = params["h"]
+            return _FakeResult(scalar=sum(1 for k, v in self.reviews.items() if v["house_id"] == h))
+
+        if "INSERT INTO house_reviews" in sql:
+            key = (params["source"], params["ext_review_id"])
+            self.reviews[key] = {"house_id": params["house_id"], **params}  # UPSERT
+            return _FakeResult()
+
+        if "DELETE FROM house_reliability_checks" in sql:
+            # dedup guard — keep newest only (model: keep 1 per (house, source))
+            h, src = params["h"], params["src"]
+            kept = [r for r in self.reliability if not (r[0] == h and r[1] == src)]
+            kept.append((h, src))
+            self.reliability = kept
+            return _FakeResult()
+
+        # Unhandled SQL → inert result (keeps the fake permissive).
+        return _FakeResult()
+
+    # -- transaction shims -------------------------------------------------
+    def begin_nested(self):
+        return MagicMock()
+
+    def commit(self):
+        pass
+
+    def rollback(self):
+        pass
+
+
+def _enrichment_with_everything(seed: int = 0) -> NewbuildingEnrichment:
+    """Build a full enrichment. `seed` shifts review ids + month so distinct ЖК pages
+    produce distinct rows (house_reviews dedups on (source, ext_review_id) globally,
+    not per-house — real pages have unique ids; the seed mimics that)."""
+    month = f"2025-{(seed % 12) + 1:02d}-01"
+    base = 9000 + seed * 10
+    return NewbuildingEnrichment(
+        cian_internal_house_id=12345 + seed,
+        cian_zhk_url=f"https://zhk-test-{seed}-ekb-i.cian.ru/",
+        name="ЖК Тест",
+        realty_valuation_chart=[
+            {
+                "month_date": month,
+                "room_count": "all",
+                "prices_type": "price",
+                "period": "halfYear",
+                "price_per_sqm": 13800000.0,
+            }
+        ],
+        reliability_checks=[
+            {"check_name": "Надёжный застройщик", "check_status": "reliable", "details": []}
+        ],
+        reviews=[
+            {"id": base + 1, "title": "Отличный ЖК", "rate": 5, "text": "Очень доволен"},
+            {"id": base + 2, "title": "Норм", "rate": 4, "text": "Есть нюансы"},
+        ],
+    )
+
+
+def _fake_save_newbuilding_enrichment(db, house_id, enrichment):
+    """Stand-in for the real saver: lands price_dynamics + reliability into FakeDB."""
+    for p in enrichment.realty_valuation_chart:
+        if p.get("price_per_sqm") is None:
+            continue
+        db.price_dynamics.add(
+            (
+                house_id,
+                p["month_date"],
+                "cian_realty_valuation",
+                p.get("room_count", "all"),
+                p.get("prices_type", "price"),
+                p.get("period", "halfYear"),
+            )
+        )
+    for c in enrichment.reliability_checks:
+        if c.get("check_name") or c.get("check_status"):
+            db.reliability.append((house_id, "cian_nashdom"))
+    db.commit()
+
+
+# ---------------------------------------------------------------------------
+# Pure-unit: _save_cian_reviews mapping (the load-bearing new logic).
+# ---------------------------------------------------------------------------
+
+
+def test_save_cian_reviews_maps_and_dedups() -> None:
+    db = FakeDB(pending_rows=[])
+    reviews = [
+        {"id": 1, "title": "Хорошо", "rate": 5, "text": "ok"},
+        {"id": 2, "title": "Плохо", "rate": 2, "text": "bad"},
+    ]
+    n = _save_cian_reviews(db, house_id=7, reviews=reviews)
+    assert n == 2
+    assert db.reviews[("cian", 1)]["house_id"] == 7
+    assert db.reviews[("cian", 1)]["review_title"] == "Хорошо"
+
+    # Re-run with same ids → UPSERT, no growth (idempotent on (source, ext_review_id)).
+    n2 = _save_cian_reviews(db, house_id=7, reviews=reviews)
+    assert n2 == 2
+    assert len(db.reviews) == 2
+
+
+def test_save_cian_reviews_skips_entries_without_id() -> None:
+    db = FakeDB(pending_rows=[])
+    reviews = [{"title": "no id", "text": "x"}, {"id": "not-int", "text": "y"}]
+    n = _save_cian_reviews(db, house_id=7, reviews=reviews)
+    assert n == 0
+    assert len(db.reviews) == 0
+
+
+def test_save_cian_reviews_clamps_bad_score() -> None:
+    db = FakeDB(pending_rows=[])
+    _save_cian_reviews(db, house_id=7, reviews=[{"id": 5, "rate": 99}])
+    assert db.reviews[("cian", 5)]["score"] is None  # out of 1..5 → dropped
+
+
+# ---------------------------------------------------------------------------
+# Task-level: end-to-end populate + idempotent re-run.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backfill_dry_run_counts_only() -> None:
+    db = FakeDB(pending_rows=[{"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"}])
+    with patch(
+        "app.services.scrapers.cian_newbuilding.fetch_newbuilding", new_callable=AsyncMock
+    ) as mock_fetch:
+        result = await backfill_newbuilding_enrichment(db, limit=5, dry_run=True)
+    mock_fetch.assert_not_called()
+    assert isinstance(result, NewbuildingEnrichBackfillResult)
+    assert result.cian_houses_total == 1
+    assert result.processed == 0
+
+
+@pytest.mark.asyncio
+async def test_backfill_populates_all_three_tables() -> None:
+    db = FakeDB(
+        pending_rows=[
+            {"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"},
+            {"house_id": 2, "cian_zhk_url": "https://zhk-b.cian.ru/"},
+        ]
+    )
+    # Distinct enrichment per house (real ЖК pages have unique review ids / months).
+    by_url = {
+        "https://zhk-a.cian.ru/": _enrichment_with_everything(seed=1),
+        "https://zhk-b.cian.ru/": _enrichment_with_everything(seed=2),
+    }
+
+    async def _fetch(url, **kwargs):
+        return by_url[url]
+
+    with (
+        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.processed == 2
+    assert result.succeeded == 2
+    assert result.failed_fetch == 0
+    assert result.failed_save == 0
+    # All three tables non-zero.
+    assert len(db.price_dynamics) == 2  # 1 chart point × 2 houses
+    assert len(db.reliability) == 2
+    assert len(db.reviews) == 4  # 2 reviews × 2 houses
+    assert result.price_dynamics_rows == 2
+    assert result.reliability_rows == 2
+    assert result.review_rows == 4
+
+
+@pytest.mark.asyncio
+async def test_backfill_force_rerun_no_dup() -> None:
+    """force=True re-run over the same house must not duplicate any of the 3 tables."""
+    rows = [{"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"}]
+    db = FakeDB(pending_rows=rows)
+    enr = _enrichment_with_everything()
+
+    with (
+        patch(
+            "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
+            new_callable=AsyncMock,
+            return_value=enr,
+        ),
+        patch(
+            "app.services.scrapers.cian_newbuilding.save_newbuilding_enrichment",
+            side_effect=_fake_save_newbuilding_enrichment,
+        ),
+    ):
+        await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
+        pd_after_first = len(db.price_dynamics)
+        rv_after_first = len(db.reviews)
+
+        # Re-run with force=True → re-process the same house.
+        result2 = await backfill_newbuilding_enrichment(
+            db, limit=5, force=True, request_delay_sec=0
+        )
+
+    assert len(db.price_dynamics) == pd_after_first  # ON CONFLICT dedup
+    assert len(db.reviews) == rv_after_first  # (source, ext_review_id) dedup
+    assert len(db.reliability) == 1  # dedup guard collapsed to newest
+    assert result2.processed == 1
+
+
+@pytest.mark.asyncio
+async def test_backfill_default_rerun_skips_and_no_reliability_dup() -> None:
+    """force=False resumability: an already-enriched house (pd+reliability present, NO
+    reviews — the common Cian case) is skipped on the next run, so a plain re-run neither
+    re-fetches nor accumulates duplicate reliability rows.
+    """
+    rows = [{"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"}]
+    db = FakeDB(pending_rows=rows)
+    enr = _enrichment_with_everything()
+    enr.reviews = []  # mimic a ЖК whose initialState carries no reviews
+
+    fetch_calls = {"n": 0}
+
+    async def _fetch(url, **kwargs):
+        fetch_calls["n"] += 1
+        return enr
+
+    with (
+        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,
+        ),
+    ):
+        # First run enriches.
+        r1 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
+        # FakeDB._pending_rows is static, so the SELECT still returns the house; the task's
+        # per-house guard must skip it now that pd+reliability exist.
+        r2 = await backfill_newbuilding_enrichment(db, limit=5, request_delay_sec=0)
+
+    assert r1.succeeded == 1
+    assert fetch_calls["n"] == 1  # second run skipped → no extra fetch
+    assert r2.skipped_already_enriched == 1
+    assert r2.succeeded == 0
+    assert len(db.reliability) == 1  # no duplicate appended on re-run
+
+
+@pytest.mark.asyncio
+async def test_backfill_fetch_failure_does_not_abort_batch() -> None:
+    """One house returns None (captcha) → counted as fetch-fail, batch continues."""
+    db = FakeDB(
+        pending_rows=[
+            {"house_id": 1, "cian_zhk_url": "https://zhk-a.cian.ru/"},
+            {"house_id": 2, "cian_zhk_url": "https://zhk-b.cian.ru/"},
+        ]
+    )
+    enr = _enrichment_with_everything()
+
+    async def _fetch(url, **kwargs):
+        return None if "zhk-a" in url else enr
+
+    with (
+        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.processed == 2
+    assert result.failed_fetch == 1
+    assert result.succeeded == 1
+    assert len(db.reviews) == 2  # only house 2 enriched