"""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))