From bcb12853413c209f61f8dad4b71cf125facfcd6a Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sun, 7 Jun 2026 19:08:48 +0500 Subject: [PATCH] feat(tradein): backfill task for cian newbuilding enrichment (#972) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../app/services/scrapers/cian_newbuilding.py | 24 +- .../app/tasks/newbuilding_enrich_backfill.py | 518 ++++++++++++++++++ .../tasks/test_newbuilding_enrich_backfill.py | 368 +++++++++++++ 3 files changed, 909 insertions(+), 1 deletion(-) create mode 100644 tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py create mode 100644 tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py diff --git a/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py b/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py index 639f9938..5d3fe442 100644 --- a/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py +++ b/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': }) 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 [] diff --git a/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py b/tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py new file mode 100644 index 00000000..b2474a60 --- /dev/null +++ b/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)) diff --git a/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py b/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py new file mode 100644 index 00000000..ac57b088 --- /dev/null +++ b/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