lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): cian newbuilding enrichment backfill task + live-data fix (#972) #1148

Merged
bot-backend merged 1 commit from feat/newbuilding-enrich-backfill-972 into main 2026-06-07 14:09:25 +00:00
Conversation 0 Commits 1 Files changed 3 +909 -1
3 changed files with 909 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py
View file

@ -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 []

518
tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py Normal file
View file

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

368
tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py Normal file
View file

@ -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